800 rows from a LIMIT 100

July 8, 2026

At SF1 we beat Trino on every suite. At SF10, SSB flipped: Trino finished the thirteen queries in 21 seconds and we took 42. Same engine, same data, twice the wall time. A scaling crossover like that is not noise. Something in our execution stopped scaling exactly where Trino’s kept going.

The suspects came in the usual order. First the filter shape: SSB queries carry key-set filters, and we had history there, so we raised the InList pushdown cap from 65,536 to a million keys and re-ran. Total went from 84.3 seconds to 84.4. rows_decoded did not move. A null result, and a useful one: the theory died in one run, and the profile counter that killed it pointed somewhere else.

The somewhere else was the same place it always is. The fact table scan fed one output partition. Every join, every aggregate, every core above it sat waiting on a single decode thread while seven cores idled. We had solved a version of this before inside the scan, but the plan above it still saw one partition, so the plan above it ran serial too.

Parallel probe, and the tax

The safe fix already existed as an opt-in rule: bump the probe-side scan of a broadcast join to N output partitions. The build side never changes, so the q72 regression we shipped once before (a dimension table collected from eleven tiny streams, 17 seconds becoming 100) cannot come back. Flag on: SSB dropped to 20.0 seconds against Trino’s 21.4. Parity, finally, and ahead by a nose.

Then we ran TPC-DS with the same flag and paid 18 percent for the privilege. q22 went from 12 to 36 seconds. That is not a tax, that is a mugging.

The profile showed a skew, not a slowdown. Our scan assigned whole files to partitions round-robin. The tables that hurt were sorted on write, and a sorted table compacts into one to four large files. Eight partitions, one file: partition zero decodes everything and the exchange above the scan disappears because the plan believes the work is spread. The fix assigns 32MB byte-range splits round-robin instead of whole files, reusing the same split math the reader already uses internally. The tax fell to 6 percent, and q22 at 10 seconds now beats its own serial baseline of 12.

The rows started lying

Speed is worthless if the answers are wrong, so every change ran under a row-diff harness against Trino. q67 failed first: 200 rows for a LIMIT 100. Two of eight partitions each produced a top-100 and the collector concatenated them. We fixed the root: if the plan still ends multi-partition, merge it back to one stream and carry the fetch onto the merge. Tests green, suite correct, done.

Except the full compare then flagged three more. q10 returned 465 rows. q14 returned 699. q51 returned exactly 800, against a LIMIT 100, and 800 is the number that gives the mechanism away: eight partitions times one hundred rows.

Here is what actually happens. DataFusion’s LimitPushdown runs while the plan is still single-partition, and it pushes the global ORDER BY ... LIMIT 100 down into a fetch on whatever mid-plan operator can absorb it: a filter, a TopK sort, a local limit. On one partition that is free and correct, “first 100 of this stream” and “LIMIT 100” are the same thing. Then our rule multiplies the partitions. The fetch is still there, still 100, but it now caps each of eight streams. The merge inserted above it carries no fetch at all, because the limit information was consumed before the plan went parallel. The root of the plan looks perfectly healthy. It is a single-partition merge in the right order. It just emits 800 rows.

The fix walks down from the root through operators that cannot change the row count, a projection, a fetchless merge, and re-applies the first stranded fetch it finds at the root. Truncating a stream that is already in final output order is exactly what the original LIMIT meant, and if the plan was already capped it is a no-op. q14 needed one more turn of the screw: its spine runs through the operator that resolves uncorrelated scalar subqueries, which passes its main input through untouched but lists the subquery plans as extra children, so the walk stopped at it and q14 kept over-returning after its siblings were fixed. The walk now knows that node passes through child zero.

The lesson generalizes. “Correct” was a property of the plan, not of the query. Every optimization that ran before ours baked single-partition assumptions into the tree, and multiplying partitions afterward silently un-proves them. If you re-optimize a plan mid-flight, you own every invariant the earlier passes established, including the ones they no longer remember establishing.

The flag that never fired

One more find from the same dig. We have a second, more general parallel-scan rule, and for weeks it measured neutral in every A/B. Neutral read as “no benefit on this workload.” The truth was less flattering: it had never bumped a single scan. Its safety check refused to touch anything beneath an operator that requires ordered input, and the merge node at the root of every ORDER BY query is such an operator. Every real plan was tainted from the root down. The rule was a no-op with a config flag.

The check was guarding against a problem the rule already solves, since it re-runs the distribution and sort passes that re-erect whatever ordering the plan needs. The only guard that earns its keep is the original one: never parallelize the build side of a broadcast join. With the taint reduced to that, the rule finally fired, and the big fact scans under partitioned joins went parallel: TPC-DS q72 from 52 seconds to 41, TPC-H q18 from 12.4 to 7.1.

A neutral A/B tells you the delta of flipping a flag. It does not tell you the flag did anything. Look at the plan.

Where it landed

Both flags now default on. On a dedicated 8-core box, both engines capped at 12GB, same Iceberg warehouse on the same S3-compatible storage, Trino 481:

Zero row differences against Trino across all 268 query runs. The SSB crossover is closed.

One gap survives, and it is honest about what it is. q72 still trails at 41 seconds against Trino’s 13, and the profile says why: we decode 1.17 billion inventory rows across eight cores and then throw 97 percent of them away, because the join’s dynamic filter runs after the decode instead of inside it. Pushing that filter below the decode is the next project. The pattern held through this whole arc: the profile counters name the suspect, the A/B convicts or acquits, and the row-diff harness keeps the speed honest.