Tracking issue for the new streaming engine

[coastalwhite]

From 1.31.1, Polars has a new streaming engine. In time, it will become the default engine, as it is usually faster and uses less memory. All queries that run on the in-memory engine should run on the streaming engine (please file a bug otherwise), but certain operations might not have a native streaming implementation yet (in which case they will transparently fall back to the in-memory engine).

This issue tracks the progress of adding functionality natively to the streaming engine.

The streaming engine can be used by using LazyFrame.collect(engine='streaming') or LazyFrame.sink_*(...). The physical plan can be visualized as a dot graph by using .show_graph(engine='streaming', plan_stage='physical').

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
• IO plugins (feat: Add support for io-plugins in new-streaming #21870)
• PyArrow dataset (feat(python): Streaming pyarrow datasets sources #27230)
• AnonymousScan
• scan_delta
• scan_iceberg

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)
• Partitioned Writers
  • ByKey / Hive (feat: Add PartitionByKey for new streaming sinks #21689)
  • MaxSize
  • Parted
• Anonymous Sink
• Python Sink (feat: Add LazyFrame.{sink,collect}_batches #23980)
  • Push (sink_batches)
  • Pull (collect_batches)
• Sink delta (feat(python): ArrowStreamExportable and sink_delta #25994)
• Sink iceberg (feat: Add unstable LazyFrame.sink_iceberg #26799)

Out-of-core

• Multiplexers (feat: Lock-free memory manager with spill-to-disk and fully OOC multiplexer #26774)
• Group-by
• Equi-join
• Sort

Streaming Nodes

• Projection
• Group-by
  • single key optimization
• Equi-join
  • single key optimization
• Cross-join
• AsOf join (feat: Add streaming AsOf join node #26398)
  • With by parameter (Streaming implementation for grouped AsOf join #26500)
• Inequality join (feat: Add a streaming range-join #26790)
• 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 (perf: Add streaming GatherNode #27465)
• Merge-Sorted (feat: Implement a streaming merge sorted node #20960)
• Sorted unique (perf: Add sorted unique node to streaming engine #26990)
• {Expr,LazyFrame}.rolling (feat: Streaming {Expr,LazyFrame}.rolling #25058)
  • Sorted group_by column
• LazyFrame.group_by_dynamic
  • group_by=None (perf: Add streaming native LazyFrame.group_by_dynamic #25342)
  • Sorted group_by column
• Merge / Sorted Equality Join (perf(rust): Add streaming merge-join #25964)
• Sorted Group By (perf: Add streaming sorted Group-By #25013)
• Rolling functions like rolling_{sum,std,var,...}
• strptime without format (to_time, to_datetime, to_date) (perf: Streaming strptime with format=None #27056)

Aggregates

• Sum
• Mean
• Min/max
• Last/first
• Var/std
• Count
• NUnique (normal)
• NUnique (grouped) (perf: Lower streaming group_by n_unique to unique().len() #26109)
• Implode
• Median/quantile
• Mode (perf: Native streaming .mode() expression #24459)
• ApproxNUnique
• arg_min, arg_max (perf: Lower arg_{min,max} to streaming engine #26845)
• any / all
• corr (perf: Streaming cov and corr #27008)
• cov (perf: Streaming cov and corr #27008)
• kurtosis / skew (perf: Lower skew / kurtosis to streaming aggregations #27176)
• entropy (perf: Lower entropy to streaming reductions #27174)
• str.join

Plan translation to streaming

• Literal series in selections
• Aggregates in selections
• Sorts in selections
• Filters in selections
• shift (perf: Lower Expr.shift to streaming engine #24106)
• .over() to group-by + join
  • .over(mapping_strategy="group_to_rows") perf: Lower basic over() to streaming primitives #27303
• .over(keys) where keys are sorted to sorted-group-by + explode
• .replace() to map (small replacement dictionary) or join (large replacement dictionary)
• is_first_distinct (perf: Streaming is_first_distinct and unique(maintain_order=True) #27052)
• is_last_distinct
• is_unique
• is_duplicated
• arg_where (perf: Lower arg_where natively to streaming engine #24088)
• index_of (perf: Lower index_of to streaming engine #26923)
• drop_nans / drop_nulls
• reshape
• arg_unique (perf: Lower arg_unique natively to the streaming engine #24279)
• top_k / bottom_k (perf: Lower top-k to streaming engine #23979)
• cum_count, cum_sum, cum_prod, cum_min, cum_max (feat: Add cum_* as native streaming nodes #23977)
• diff (feat: Implement diff() in streaming engine #24189)
• unique
• pl.concat
• peak_min / peak_max (feat: Add native streaming for peaks_{min,max} #24039)
• cut (perf: Make cut output Enum and mark as elementwise #27173)
• qcut
• rle (feat: Lower rle to a native streaming engine node #23929)
• rle_id (feat: Lower rle_id to a native streaming node #23894)
• sample
• repeat (perf: Lower pl.repeat to streaming engine #23804)
• extend_constant (perf: Lower extend_constant to the streaming engine #23824)
• unique_counts / value_counts (feat: Lower unique_counts and value_counts to streaming engine #23890)
• mode (perf: Native streaming .mode() expression #24459)
• pct_change
• interpolate (perf: Add native streaming interpolate #27185)
  • interpolate_by
• ewm_{mean,std,var} (feat: Support ewm_mean() in streaming engine #25003, feat: Support ewm_var/std in streaming engine #25109)
• ewm_{mean,std,var}_by
• gather_every (perf: Native streaming gather_every #24700)
• fill_null(strategy="min"/"max"/"mean")
• forward_fill (perf: Native streaming forward_fill #26922)
• backward_fill (perf: Streaming native backward_fill #26967)
• search_sorted
• random
• rank
• arg_sort
• hist

Datatypes

• Everything

Other

• Show plan with show_graph (feat: Show physical stage graph #22491)
• Add support for profile (Remove LazyFrame.profile() #28274)
• Add morsel size configuration to set_streaming_chunk_size

[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.

[jhirsch-mhp]

Categoricals are currently the only thing that are simply broken on the new streaming engine (to my knowledge)

Guess there is another bug for the new streaming engine when it comes to split -> process -> concat as described in this comment here .

[deanm0000]

I saw that var/std are supported in new streaming so I did

import os
os.environ["POLARS_FORCE_NEW_STREAMING"]="1"
import numpy as np
import polars as pl
df=pl.DataFrame({"a":np.random.random(100)})
print(df.lazy().select(pl.col("a").std()).explain(streaming=True))

 SELECT [col("a").std()] FROM
  STREAMING:
    DF ["a"]; PROJECT 1/1 COLUMNS

The std operation is outside of the streaming. I'm not sure how to tell if that's because explain doesn't give a new_streaming query plan or if something is awry with the new_streaming var/cov.

[coastalwhite]

Explain does not yet reflect the new streaming engine. To see what it is actually doing, use POLARS_VISUALIZE_PHYSICAL_PLAN=/path/to/graph.dot. I added explain to the list of things that needs to be done.

[ion-elgreco]

Sinks should also be possible with deltalake since v0.25 if you pass a RecordBatchStream to the writer

[jayceslesar]

Any insight as to when this will be available with sink_* methods?

[ritchie46]

Any insight as to when this will be available with sink_* methods?

Next release the sinks will point to the new streaming engine.

[MPJansen]

Any insight as to when this will be available with sink_* methods?

Next release the sinks will point to the new streaming engine.

Hi!

Just checking, does that imply that writing partitioned lazyframes will be supported by the sink_ methods?
Or is that out of scope? If so, is that something Polars is interested in supporting?

KR