Lazy queries, query plans and streaming in polars

For a 100,000-order report, a lazy query planlazy query plan: A description of a query whose result is computed by collection or a sink; resolving its schema can still inspect source metadata. in polars describes a whole computation first and runs it only when you ask for the result. You start one with pl.scan_csv or pl.scan_parquet instead of read_csv, or with .lazy() on a DataFrame you already have, and chain the same methods as in eager code, but each method adds a step to a query plan instead of running. collect() hands the plan to an optimiser before the result is materialized; schema resolution may already inspect source metadata: it applies filters while the file is read (predicate pushdownpredicate pushdown: Moving a filter closer to a query's data source when the transformation preserves the intended result.) and reads only the columns the query uses (projection pushdownprojection pushdown: Retaining only the input columns a query needs so that later stages handle fewer values.), then runs the result. explain() prints the optimised plan, so you can see what will be read before you read it. Streamingstreaming engine: Query execution that can process data in batches; operator state, fallback operations and collected output can still require growing memory. in polars runs the same plan in batches, which lets a query work through data that does not fit in memory.

A lazy plan defers result computation

Lazy queries pay off most when the data is large or the query is long, and they cost almost nothing to adopt: the methods are the ones you already know from the previous lesson.

Here is one report written both ways: revenue per category for the city of Pune, from 100,000 generated orders in a CSV file of 10 columns. make_sales.py makes the same rows every time it runs.

import polars as pl

from make_sales import sales

sales(100_000).write_csv("sales.csv")  # 100,000 orders, 10 columns

# Eager: every line runs as soon as it is reached and returns a DataFrame.
df = pl.read_csv("sales.csv")
df = df.with_columns(amount=pl.col("qty") * pl.col("price"))
df = df.filter(pl.col("city") == "Pune")
eager = df.group_by("category").agg(pl.col("amount").sum()).sort("category")

# Lazy: the steps build a plan; schema resolution may inspect the CSV before result collection.
query = (
    pl.scan_csv("sales.csv")
    .with_columns(amount=pl.col("qty") * pl.col("price"))
    .filter(pl.col("city") == "Pune")
    .group_by("category")
    .agg(pl.col("amount").sum())
    .sort("category")
)
print(type(query).__name__)
print(query.collect_schema())  # resolve columns and types; this may read source metadata

lazy = query.collect()  # optimise the plan, then run it
print(lazy)
print("same result as eager:", lazy.equals(eager))

LazyFrame
Schema({'category': String, 'amount': Int64})
shape: (5, 2)
┌───────────┬─────────┐
│ category  ┆ amount  │
│ ---       ┆ ---     │
│ str       ┆ i64     │
╞═══════════╪═════════╡
│ bakery    ┆ 2547784 │
│ beverages ┆ 2561638 │
│ dairy     ┆ 2574859 │
│ grocery   ┆ 2618408 │
│ snacks    ┆ 2528771 │
└───────────┴─────────┘
same result as eager: True

"""A synthetic sales table for the lazy-query examples: n orders in 10 columns, always the same rows for the same n."""
import numpy as np
import polars as pl

CITIES = pl.Series(["Pune", "Nagpur", "Nashik", "Thane", "Solapur", "Kolhapur"])
CATEGORIES = pl.Series(["grocery", "snacks", "dairy", "bakery", "beverages"])
CHANNELS = pl.Series(["app", "web", "store"])
CUSTOMERS = pl.Series([f"Customer {i:05d}" for i in range(5000)])
ADDRESSES = pl.Series([f"Flat {i % 90 + 1}, Lane {i % 37 + 1}, Sector {i % 29 + 1}" for i in range(20000)])


def sales(n: int, seed: int = 2026) -> pl.DataFrame:
    rng = np.random.default_rng(seed)
    return pl.DataFrame({
        "order_id": np.arange(1, n + 1),
        "city": CITIES.gather(rng.integers(0, len(CITIES), n)),
        "category": CATEGORIES.gather(rng.integers(0, len(CATEGORIES), n)),
        "qty": rng.integers(1, 6, n),
        "price": rng.integers(10, 500, n),
        "discount": rng.integers(0, 30, n),
        "channel": CHANNELS.gather(rng.integers(0, len(CHANNELS), n)),
        "pin": rng.integers(411001, 411062, n),
        "customer": CUSTOMERS.gather(rng.integers(0, len(CUSTOMERS), n)),
        "address": ADDRESSES.gather(rng.integers(0, len(ADDRESSES), n)),
    })

The lazy code builds a LazyFrame rather than computing the report immediately. collect_schema() resolves the result’s String and Int64 columns, potentially reading file metadata or CSV values for inference. It does not materialize the complete result, but it is not a promise of zero I/O or negligible cost. Use it to check the schema before a long run. Both versions give the same five totals.

Follow the input and compare the result with the recorded example.

Pushdown keeps four columns and Pune rows

polars keeps two query plans for every lazy query: the one you wrote and the one the optimiser made from it.

import polars as pl

from make_sales import sales

sales(1_000).write_csv("sales.csv")

query = (
    pl.scan_csv("sales.csv")
    .with_columns(amount=pl.col("qty") * pl.col("price"))
    .filter(pl.col("city") == "Pune")
    .group_by("category")
    .agg(pl.col("amount").sum())
    .sort("category")
)

# Plans print from the last step at the top down to the scan at the bottom: read them bottom up.
print("As written:")
print(query.explain(optimized=False))
print()
print("Optimised:")
print(query.explain())

As written:
SORT BY [col("category")]
  AGGREGATE[maintain_order: false]
    [col("amount").sum()] BY [col("category")]
    FROM
    FILTER col("city") == "Pune"
    FROM
       WITH_COLUMNS:
       [(col("qty") * col("price")).alias("amount")] 
        Csv SCAN [sales.csv]
        PROJECT */10 COLUMNS
        ESTIMATED ROWS: 1000

Optimised:
SORT BY [col("category")]
  AGGREGATE[maintain_order: false]
    [col("amount").sum()] BY [col("category")]
    FROM
    simple π 2/2 ["category", "amount"]
       WITH_COLUMNS:
       [(col("qty") * col("price")).alias("amount")] 
        simple π 3/3 ["category", "qty", "price"]
          Csv SCAN [sales.csv]
          PROJECT 4/10 COLUMNS
          SELECTION: col("city") == "Pune"
          ESTIMATED ROWS: 1000

Query plans print with the last step at the top and the scan at the bottom, so read them from the bottom up. As written, the scan reads all 10 columns (PROJECT */10 COLUMNS) of every row, and the filter runs only after amount has been computed for all of them. In the optimised plan the condition has moved into the scan as SELECTION, so orders from other cities are dropped while the file is read, and the scan keeps 4 of the 10 columns (PROJECT 4/10 COLUMNS), the ones the query uses. The lines that start with π are projections: after the scan, only the columns still needed move on.

The optimiser has more rewrites than these two. Slice pushdown reads only the rows a head() needs. Common subplan elimination reads a file once when the plan uses it twice. Expression simplification replaces costly expressions with cheaper ones that give the same values. Each rewrite is allowed only when the result stays the same, and the plan shows which ones happened.

Follow the input and compare the result with the recorded example.

Streaming reduces buffers while order needs a rule

Streaming in polars means running a lazy query in batches: collect(engine="streaming"). Batched execution can reduce input buffering and handle files larger than memory. Aggregation, joins, sorts, fallback operations and collected output can still retain growing state. sink_parquet or sink_csv writes output in batches without returning the whole result as one DataFrame; it does not remove every operator’s memory needs. In polars 1.x a plain collect() uses the in-memory engine, which processes each step’s whole input at once. In polars 2.0, lazy queries run on the streaming engine by default.

That default has a cost. The streaming engine does not keep row order for group_by, joins or unpivot, and the order of groups is not promised by either engine, so a pipeline that depends on the order rows happen to arrive in can change its output after an upgrade. An explicit sort, or maintain_order where an operation offers it, makes the order part of the query. LazyFrame.profile(), which timed each step, is deprecated in 1.43 and removed in 2.0: time a whole query with time.perf_counter() instead.

Follow the input and compare the result with the recorded example.

Eager, lazy and streaming, measured on two million orders Pro: opens with a Pro, Premium or Ultimate pass

Pipelines that return the same rows on every engine and version Pro: opens with a Pro, Premium or Ultimate pass

What the optimiser can and cannot move, read in six plans Pro: opens with a Pro, Premium or Ultimate pass

Practise the stated transformations

Interview questions

Key takeaways

• scan_* or .lazy() builds a plan; collect() optimises and runs it; explain() prints the optimised plan, which reads from the bottom up.
• In a good plan, the filter appears as SELECTION inside the scan and the scan’s PROJECT line names only the columns the query uses.
• The streaming engine runs a plan in batches and is the default for lazy queries in polars 2.0; sink_parquet writes a result without collecting it.
• Neither engine promises group order, and the streaming engine does not keep row order for joins or unpivot: sort explicitly or pass maintain_order.