Tracking issue for the new streaming engine

[coastalwhite]

Note

TLDR
Starting 1.23.0, the old streaming engine will no longer be maintained and might start becoming less usable with time. People currently using it should pin their versions to 1.22.0 or before.

The old streaming engine is being deprecated, and the new streaming engine is coming to replace it soon™. The new streaming engine is far along, but does not yet have 100% feature parity with the old streaming engine. At the same time, the maintenance burden of maintaining both the old and new streaming engine is becoming too large. Therefore, we are deprecating the old streaming engine starting on Polars version 1.23.0 and telling people to pin their Polars versions to any version before 1.23.0.

This issue tracks the progress on the new streaming engine. All non-supported (unless mentioned otherwise) features fall back to the in-memory engine.

The streaming engine can be used already by using .collect(new_streaming=True) or by setting the POLARS_FORCE_NEW_STREAMING=1. The physical plan can be visualized as a dot graph by using POLARS_VISUALIZE_PHYSICAL_PLAN=filename.dot.

Sources

• Parquet (refactor(rust): Add new streaming CSV source #19694)
• CSV (refactor(rust): Add parquet source node to new streaming engine #18152)
• IPC (feat: Add IPC source node for new streaming engine #19454)
• NDJson
• Multifile / Hive (currently supported, but in-memory)
• IO plugins

Sinks

• Parquet (feat: Add Parquet Sink to new streaming engine #20690)
• CSV (feat: Add CSV sink for the new streaming engine #20804)
• IPC (feat: Add IPC sink in new streaming engine #19431)
• NDJson (feat: Add NDJson sink for the new streaming engine #20805)
• Multifile / Hive

Out-of-core

• Group-by
• Equi-join
• Sort

Streaming Nodes

• Projection
• Group-by
• Equi-join
• Nested loop join
• Anti/semi join
• Filter
• Positive slice / head
• Horizontal concat
• Vertical concat
• Negative slice / tail (feat: Add negative slice support to new-streaming engine #21001)
• Gather
• Merge-Sorted (feat: Implement a streaming merge sorted node #20960)
• Window (rolling) functions

Aggregates

• Sum
• Mean
• Min/max
• Last/first
• Var/std
• NUnique/count
• Implode
• Median/quantile

Plan translation to streaming

• Literal series in selections
• Aggregates in selections
• Sorts in selections
• Filters in selections
• .over() to group-by + join

Datatypes

• Categorical (currently no fallback - simply broken)
• Everything else

[deanm0000]

what's the difference between out-of-core group_by/equi-join and streaming nodes group_by/equi-join?

[orlp]

@deanm0000 Out-of-core operators will automatically spill to disk when the data you're processing is larger than available memory.

[daviskirk]

non-supported (unless mentioned otherwise) features fall back to the in-memory engine

What happens to the other non-supported features like the broken categoricals? Is there any way to opt out of streaming as soon as they are encountered?

[orlp]

@daviskirk Categoricals are currently the only thing that are simply broken on the new streaming engine (to my knowledge). Almost all unsupported things automatically fall back to the eager engine (just for the parts of the computation that aren't supported), some give a hard error.

My plan is to fix categoricals on the new streaming engine soon.