Latest Results
fix(distributed): tolerate shuffle outputs with no partitions
`transpose_materialized_outputs` asserted that every materialized output carries
exactly `num_partitions` partitions and then indexed into each one
unconditionally. A task whose input was cut to zero rows - e.g. by a Limit
sitting below the join's shuffle, where the coordinated distributed limit lets
only the first tasks emit rows - produces a materialized output with no
partitions at all, so the assert fired in debug builds and the indexing would
run out of bounds in release builds.
Such an output carries no rows for any partition index, so it simply contributes
nothing to every partition group. Skip missing indices and relax the assert to
allow an output to be either fully partitioned or empty.
Reproducer, plain user code on main with no optimizer changes:
left = daft.range(0, 1000, partitions=8).with_column("k", col("id") % 100)
right = daft.range(0, 100, partitions=8).select(col("id").alias("k"))
left.limit(20).join(right, on="k", how="left").limit(20).to_pydict()
# DAFT_RUNNER=ray -> panicked: "Expected all outputs to have 8 partitions,
# got 0, 8, 0, 0, 0, 0, 0, 8"
Adds that query as a regression test.hello-peter-tang:fix/shuffle-transpose-empty-outputs refactor(optimizer): drop the local-limit machinery from the join limit pushdown
Reverts the `Limit.local` plumbing (logical `Limit` field, `with_local`, the
builder test helper and the distributed `LimitNode` per-partition lowering) that
was added to work around a panic on the ray runner. It did not fix that panic -
the pushed limit still tripped it - and it spread a new plan-level concept plus a
field that every `with_new_children` call has to carry by hand across two crates.
The panic is a pre-existing defect in the distributed shuffle, not something this
rule introduces: `transpose_materialized_outputs` requires every upstream output
to carry exactly `num_partitions` partitions, but a task whose input was cut to
zero rows by a limit below the shuffle emits no partitions at all. It reproduces
on main with plain user code and no optimizer change:
left = daft.range(0, 1000, partitions=8).with_column("k", col("id") % 100)
right = daft.range(0, 100, partitions=8).select(col("id").alias("k"))
left.limit(20).join(right, on="k", how="left").limit(20).to_pydict()
# DAFT_RUNNER=ray -> panicked: "Expected all outputs to have 8 partitions,
# got 0, 8, 0, 0, 0, 0, 0, 8"
That is fixed separately in the shuffle transpose helper; with that fix in place
this rule needs no special limit kind, so the change is back to a single file.
The rule itself is unchanged: push onto the left input of a Left join and the
right input of a Right join, no-op for Inner/Outer/Anti/Semi.
Partially addresses #2275 - the Outer case requested there cannot be implemented
by bounding either input, because limiting one side turns rows of the other side
that used to match into unmatched rows and the join then emits null-extended rows
the original plan never produces.hello-peter-tang:push-down-limit-outer-join Latest Branches
0%
jackylee-ch:parquet-limit-deleted-rows 0%
jackylee-ch:iceberg-partition-repr-parity 0%
zhouwenjiajia:bugfix/joinable-stream-background-error © 2026 CodSpeed Technology