Polars DataFrames skill for Codex

Done well means: the pipeline is lazy from scan to sink, every transformation is an expression (no Python lambdas in the hot path), the query plan shows predicate and projection pushdown, and results match the old implementation row for row.

Before writing code

1. Read AGENTS.md and the lockfile; confirm python -c "import polars; print(polars.__version__)". Everything here targets Polars 1.x. Pre-1.0 code (groupby, melt, dtypes= in read_csv, pivot(columns=...), how="outer", pl.count()) needs renaming.
2. If porting from pandas, keep the pandas version runnable so outputs can be compared.
3. Inspect schema without loading data: pl.scan_parquet(path).collect_schema().

Core model

Concept	Rule
LazyFrame	Default. Build the whole query, then collect() once
DataFrame	Only for small data, interactive work, or eager-only APIs (pivot)
Expression pl.col(...)	Describes a computation; runs only inside a context
Contexts	select (new frame), with_columns (add/replace), filter, group_by().agg
Nulls	null is missing; NaN is a float value. fill_null and fill_nan are different

There is no index. Row order is preserved by most operations but group_by output order is not guaranteed unless maintain_order=True.

Workflow checklist

- [ ] 1. Scan sources lazily with explicit schema overrides where inference is risky
- [ ] 2. Filter and select columns as early as possible
- [ ] 3. Express logic with expressions; no map_elements in hot paths
- [ ] 4. Check lf.explain() for pushdown
- [ ] 5. collect() (streaming engine for big inputs) or sink_parquet()
- [ ] 6. Compare against the reference output with polars.testing

A typical lazy pipeline

import polars as pl
import polars.selectors as cs

orders = (
    pl.scan_parquet("data/orders/*.parquet")
    .filter(pl.col("created_at") >= pl.datetime(2025, 1, 1))
    .select("order_id", "customer_id", "status", "amount", "created_at")
)
customers = pl.scan_csv(
    "data/customers.csv",
    schema_overrides={"customer_id": pl.String, "signup_date": pl.Date},
)

result = (
    orders
    .join(customers, on="customer_id", how="left", validate="m:1")
    .with_columns(
        month=pl.col("created_at").dt.truncate("1mo"),
        tier=pl.when(pl.col("amount") >= 1_000).then(pl.lit("high"))
               .when(pl.col("amount") >= 100).then(pl.lit("mid"))
               .otherwise(pl.lit("low")),
    )
    .group_by("month", "country", maintain_order=True)
    .agg(
        orders=pl.len(),
        revenue=pl.col("amount").sum(),
        p95_amount=pl.col("amount").quantile(0.95),
    )
    .sort("month", "country")
)

print(result.explain())               # check: filters pushed into the Parquet scan
df = result.collect(engine="streaming")  # or result.sink_parquet("out/monthly.parquet")

Expressions you will need

# window functions: per-group values without a join
pl.col("amount").sum().over("customer_id").alias("customer_total")
(pl.col("amount") / pl.col("amount").sum().over("customer_id")).alias("share")
pl.col("created_at").rank("ordinal").over("customer_id").alias("nth_order")

# strings and dates
pl.col("email").str.to_lowercase().str.strip_chars()
pl.col("raw_date").str.to_date("%d/%m/%Y", strict=False)   # bad values -> null
pl.col("ts").dt.convert_time_zone("Europe/Berlin")

# many columns at once
df.with_columns(cs.numeric().fill_null(0), cs.string().str.strip_chars())

# conditional dedupe: latest row per key
lf.sort("updated_at").unique(subset="id", keep="last", maintain_order=True)

Joins and reshaping

Need	Polars 1.x
Left join with cardinality check	join(other, on="k", how="left", validate="m:1")
Full outer with single key column	join(other, on="k", how="full", coalesce=True)
Rows without a match	how="anti"; rows with a match: how="semi"
Nearest earlier timestamp	join_asof(other, on="ts", by="sku", strategy="backward") (both sides sorted by ts)
Wide -> long	unpivot(index="store", on=["jan", "feb"], variable_name="month", value_name="sales")
Long -> wide	df.pivot(on="month", index="store", values="sales", aggregate_function="sum") (eager only)
List column -> rows	explode("tags")

pandas to Polars translation

pandas	Polars
df["c"] = df["a"] * 2	df.with_columns((pl.col("a") * 2).alias("c"))
df[df.a > 1]	df.filter(pl.col("a") > 1)
df.groupby("k").agg(total=("x", "sum"))	df.group_by("k").agg(total=pl.col("x").sum())
groupby().transform("sum")	pl.col("x").sum().over("k")
np.where(cond, a, b)	pl.when(cond).then(a).otherwise(b)
df.apply(f, axis=1)	Rewrite as expressions; last resort map_elements with return_dtype=
df.merge(..., validate="many_to_one")	join(..., validate="m:1")
pd.read_csv(dtype=...)	pl.read_csv(schema_overrides=...)
df.reset_index()	Not needed; use with_row_index() if you need a row number

Interop: pl.from_pandas(pdf) and df.to_pandas(use_pyarrow_extension_array=True) (needs pyarrow). Convert at the boundary only; round-tripping in a loop wastes the speedup.

Performance rules

• Scan, do not read: scan_* lets the optimizer push filters and column selection into the file reader. read_* then .lazy() loses that.
• map_elements and map_batches with Python functions run single-threaded per element and block optimisation. Polars warns when it can translate one into an expression; take the hint.
• Large inputs: collect(engine="streaming") processes in batches. Some operations fall back to in-memory; sink_parquet fails loudly when a plan cannot stream, which is useful.
• Hive-partitioned folders: scan_parquet("root/**/*.parquet", hive_partitioning=True) and filter on the partition columns.
• Set CSV inference deliberately: infer_schema_length=10_000 or infer_schema=False (all strings) plus casts. Default inference reads only the first 100 rows.

Gotchas

• pl.col("a") == None does not test for null; use .is_null().
• Integer division: // floors, / always returns float.
• str.to_date(..., strict=True) (default) raises on the first bad value; with strict=False bad values become null. Count them before moving on.
• Joining on columns with different dtypes raises; cast both sides explicitly.
• unique() without maintain_order=True returns rows in arbitrary order; tests that compare ordered output will flake.
• Polars uses all cores by default. In containers set POLARS_MAX_THREADS before import if CPU quota is lower than the visible core count.
• Installing polars needs network; in a sandbox, check whether it is already in the environment before asking for approval.

Verification

from polars.testing import assert_frame_equal

new = run_polars_pipeline().sort("order_id")
old = pl.from_pandas(run_pandas_pipeline()).sort("order_id")
assert_frame_equal(new, old, check_dtypes=False, abs_tol=1e-9)

• Compare row counts and key column totals between implementations before switching.
• Time both on the same input with time.perf_counter(), and record peak memory (/usr/bin/time -v on Linux, /usr/bin/time -l on macOS).
• If the plan in explain() shows a filter above the scan rather than inside it, find the expression that blocks pushdown (often a Python UDF or a cast of the filtered column).

Report template

**Change:** ported monthly_revenue.py from pandas to Polars LazyFrame
**Parity:** assert_frame_equal passes on 2025-09 data (1.8M rows); revenue totals identical
**Performance:** 38.2 s / 6.1 GB peak -> 3.4 s / 0.9 GB peak (streaming engine)
**Behaviour notes:** group order now explicit (sorted by month, country); null countries kept as null
**Follow-ups:** customers.csv dates use two formats; 14 rows parsed to null, listed in rejects.parquet