Polars 2.0 Row Order: The Same Query Returned a Different Answer on Every Run

How much is wrong

ev, cu = pl.scan_parquet(r.events_path), pl.scan_parquet(r.customers_path)
truth = {k: q.collect(engine="in-memory")
         for k, q in queries(ev, cu, explicit=True).items()}
first_wrong, total_rows, total_err, index_wrong = [], [], [], []
for _ in range(10):
    got = {k: q.collect() for k, q in queries(ev, cu, explicit=False).items()}
    a = got["first purchase per customer"]["amount"]
    first_wrong.append(int((a != truth["first purchase per customer"]["amount"]).sum()))
    d = (got["running total per customer"]["running"]
         - truth["running total per customer"]["running"]).abs()
    total_rows.append(int((d > 1e-6).sum()))
    total_err.append(float(d.max()))
    picked = set(got["row index, first 1,000"]["event_id"].to_list())
    index_wrong.append(1000 - len(picked & set(truth["row index, first 1,000"]["event_id"].to_list())))

Not much, which is the problem. Across 10 runs, the first purchase came back wrong for between 7 and 12 of the 100,000 customers. A share that small passes every eyeball check. The running totals were wrong on between 153 and 282 of the 2,000,000 rows, but a wrong one was off by as much as 1,212, in a column where the median purchase is 34. The row index was the loud failure: of the first 1,000 rows, between 961 and 1,000 were different rows from the ones 1.44.2 picked.

A small test fixture does not hide it. The same first-purchase query on synthetic logs of different sizes, 10 runs each:

import numpy as np

def first_purchase_errors(n_events, runs=10, seed=1):
    rng = np.random.default_rng(seed)
    n_c = max(n_events // 20, 10)
    c = pl.LazyFrame({"customer_id": np.arange(n_c)})
    e = pl.LazyFrame({"event_id": np.arange(n_events),
                      "customer_id": np.sort(rng.integers(0, n_c, n_events)),
                      "amount": rng.gamma(2.0, 20.0, n_events).round(2)})
    q = queries(e, c, explicit=False)["first purchase per customer"]
    ref = queries(e, c, explicit=True)["first purchase per customer"].collect(engine="in-memory")
    return n_c, [int((q.collect()["amount"] != ref["amount"]).sum()) for _ in range(runs)]

scale = {}
for n in [1_000, 10_000, 100_000, 1_000_000, 10_000_000]:
    n_c, bad = first_purchase_errors(n)
    scale[n] = [n_c, min(bad), max(bad), sum(b > 0 for b in bad)]
    print(f"{n:>10,} events  {n_c:>7,} customers  wrong in {scale[n][3]:>2}/10 runs, "
          f"{min(bad)} to {max(bad)} customers")
##      1,000 events       50 customers  wrong in 10/10 runs, 3 to 6 customers
##     10,000 events      500 customers  wrong in 10/10 runs, 2 to 7 customers
##    100,000 events    5,000 customers  wrong in 10/10 runs, 3 to 8 customers
##  1,000,000 events   50,000 customers  wrong in 10/10 runs, 5 to 8 customers
## 10,000,000 events  500,000 customers  wrong in 10/10 runs, 28 to 44 customers

On a thousand rows the error reaches up to 12% of the customers, so a test that compares output with a stored answer fails on every run. On ten million rows it is 28 to 44 customers out of 500,000.

Why it happens

The streaming engine splits the data into chunks and works on them in parallel. Operations that do not need order, such as a join, may hand chunks on in whatever order they finish. first(), cum_sum(), unique(keep="first"), and with_row_index() all read the order they are given, so they inherit the shuffle. head(20) came through intact here, but nothing promises that it will.

None of this is a bug. A Polars issue opened in November 2024 set out that left-join order should not be guaranteed, since it had only ever been an implementation detail, and the upgrade guide says the same thing about the new default. Code written against 1.x could rely on an order Polars never promised, and 2.0 is the first release whose default engine stops providing it.

The fix is one argument

import time

def first_q(join_order=None, expr=pl.col("amount").first()):
    if join_order:
        j = ev.join(cu, on="customer_id", how="left", maintain_order=join_order)
    else:
        j = ev.join(cu, on="customer_id", how="left")
    return j.group_by("customer_id").agg(expr).sort("customer_id")

ref = truth["first purchase per customer"]
variants = {
    "default": lambda: first_q().collect(),
    "join(maintain_order='left')": lambda: first_q("left").collect(),
    "sort_by inside the expression":
        lambda: first_q(expr=pl.col("amount").sort_by("event_id").first()).collect(),
    "collect(engine='in-memory')": lambda: first_q().collect(engine="in-memory"),
}
plan = [k for k in variants for _ in range(20)]
np.random.default_rng(7).shuffle(plan)           # interleave, so timing is fair
secs = {k: [] for k in variants}
wrong = {k: 0 for k in variants}
most = {k: 0 for k in variants}
for k in plan:
    t0 = time.perf_counter()
    out = variants[k]()
    secs[k].append(time.perf_counter() - t0)
    wrong[k] += int(not out.equals(ref))
    most[k] = max(most[k], int((out["amount"] != ref["amount"]).sum()))
for k in variants:
    print(f"{k:31s} wrong {wrong[k]:>2}/20   worst run {most[k]:>2} customers   "
          f"median {np.median(secs[k]):.3f} s")
## default                         wrong 20/20   worst run 14 customers   median 0.028 s
## join(maintain_order='left')     wrong  0/20   worst run  0 customers   median 0.028 s
## sort_by inside the expression   wrong  0/20   worst run  0 customers   median 0.035 s
## collect(engine='in-memory')     wrong 17/20   worst run  1 customers   median 0.044 s

Asking for order worked. Keeping the left table’s order in the join and putting sort_by() inside the expression both returned the right answer on all 20 runs. The join fix ran about as fast as the default here (median 0.028 against 0.028 seconds), so on this data it costs nothing. sort_by() is the more robust of the two, because the answer no longer depends on any upstream step keeping order.

Going back to the old engine was not enough. The upgrade guide offers collect(engine="in-memory") to restore the previous default, and it does restore the engine. It did not restore the order: on a Parquet source, 17 of the 20 runs still came back wrong, by at most 1 customer. Neither engine promises order, so the fixes that hold are the ones that ask for it. One more caution from the tracker: an optimizer bug that dropped maintain_order=True when a head() sat between a group-by and a sort (issue 30079) is fixed in 2.1.0, so upgrade before relying on maintain_order=True in that pattern.

Will your tests catch it?

from polars.testing import assert_frame_equal

bad = first_q().collect()
print("same shape:", bad.shape == ref.shape, "  same schema:", bad.schema == ref.schema)
## same shape: True   same schema: True
try:
    assert_frame_equal(bad, ref, check_row_order=False)
except AssertionError as err:
    print(str(err).splitlines()[0])
## DataFrames are different (value mismatch for column "amount")

A test that compares output with a stored answer will catch it, even one that ignores row order, because the values themselves differ. A test that checks shape, schema, or column names will pass.

R users are in the same position

data.table works through rows in the order it holds them, so amount[1L] by customer is the first purchase. DuckDB, which runs in parallel like the streaming engine, documents that first() can be non-deterministic without an ORDER BY. Same file, same question:

library(data.table)
library(DBI)
ev <- as.data.table(events)
truth_r <- ev[order(event_id), .(amount = amount[1L]), by = customer_id][order(customer_id)]

con <- dbConnect(duckdb::duckdb())
duck <- function(agg) dbGetQuery(con, sprintf(
  "SELECT customer_id, %s AS amount
   FROM read_parquet('%s') e LEFT JOIN read_parquet('%s') c USING (customer_id)
   GROUP BY customer_id ORDER BY customer_id",
  agg, normalizePath(events_path, "/"), normalizePath(customers_path, "/")))$amount

engines <- list(
  "data.table, amount[1L]" = function()
    as.data.table(customers)[ev, on = "customer_id"][
      , .(amount = amount[1L]), by = customer_id][order(customer_id)]$amount,
  "DuckDB, first(amount)" = function() duck("first(amount)"),
  "DuckDB, first(amount ORDER BY event_id)" = function() duck("first(amount ORDER BY event_id)"))
r_res <- t(sapply(engines, function(f) {
  bad <- replicate(20, sum(f() != truth_r$amount))
  c(runs_wrong = sum(bad > 0), most_customers_wrong = max(bad))
}))
dbDisconnect(con, shutdown = TRUE)
r_res
##                                         runs_wrong most_customers_wrong
## data.table, amount[1L]                           0                    0
## DuckDB, first(amount)                           17                    1
## DuckDB, first(amount ORDER BY event_id)          0                    0

DuckDB without an ORDER BY was wrong on 17 of 20 runs, and with one it was right on all of them. The rule is the same in every parallel engine: if the answer depends on order, say so in the query.

What this does not show

• The data are a synthetic event log on one machine with 16 threads. How many rows go wrong depends on the thread count and on how the engine splits the work.
• Polars 1.44.2 being right on every run shows what it did, not what it promised.
• head(20) was right here. Nothing guarantees that either.

What to do on Monday

1. Search lazy pipelines for first(), last(), cum_sum(), unique(keep=...), with_row_index(), and head() downstream of a join or a group-by.
2. Add maintain_order="left" to the join that feeds them, or put sort_by() inside the expression.
3. Do not count on engine="in-memory" to bring the old order back.
4. Keep one test that compares output with a stored answer. check_row_order=False is fine.
5. In SQL, write first(x ORDER BY t) or arg_min(x, t).

Whether someone else can rerun your pipeline and get your numbers is the replication standard, covered in A Guide on Data Analysis. The two scripts used here, audit.py and run_audit.py, sit next to this post.