🏆 Highlights
Stream results out of a distributed query with sink_batches()
lf.remote().sink_batches(fn) runs your query on the cluster and calls a Python function with each batch of results as it becomes ready, so you can consume result sets that do not fit in memory without first writing them to storage.
lf.remote(ctx).sink_batches(lambda df: my_handler(df))Return True from the callback to stop early.
chunk_size controls how many rows are buffered before each call, and maintain_order=True calls the function serially instead of in parallel across workers.
The callback may be invoked more than once for the same batch from different workers, so it should be idempotent.
This method is to provide flexibility in addition to sink_parquet, sink_csv, sink_ipc or sink_iceberg.
New experimental miso query planner
Alongside the existing planner there is now a second distributed planning algorithm, which can be selected per-query:
Stages before
Stages with miso enabled
lf.remote(ctx).distributed(planner="miso").collect()planner accepts "auto", "naive" and "miso"; "auto" still defaults to the existing naive planner.
This is experimental and not the default, but we would love for you to try it on your workloads and let us know how the resulting stage graphs compare.
Distributed pl.collect_all()
Multiple LazyFrames can now be submitted as a single distributed query:
pl.collect_all(lazyframes, lazy=True).remote(ctx).distributed().execute()HDFS support
Workers can read from hdfs:// paths, and scan_iceberg can reach Iceberg metadata over HDFS via a pyiceberg fsspec integration.
Paths use hdfs://host:port/....
The feature is off by default and enabled through the worker.extras.hdfs.enabled and worker.extras.pyiceberg.enabled configuration options.
Hive partition aware scans
The Polars distributed planner is partition aware. Meaning operations like join, group-by etc. run partitioned by key(s). If you scan hive partitioned data (single columns at the moment, multi-column is coming up), Polars can now do that without introducing a shuffle, leading to great speed ups on certain queries.
Delete anonymous query results
QueryResult.delete() removes a completed result from the anonymous storage location, and DirectQuery.delete_result() does the same directly on the query handle, without needing a QueryResult first.
Only available in direct mode, and only when the anonymous storage location is configured with allow_delete.
Try with Polars Cloud - https://cloud.pola.rs/
On-Prem Releases - https://docs.cloud.pola.rs/polars-on-premises/releases/