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 fix(optimizer): only push limit onto the Left/Right join outer side
Pushing a Limit onto an input of a full Outer join is unsound: dropping rows
from one input turns rows on the other input that used to match into
unmatched rows, so the join emits null-extended rows the original plan never
produces. With l = [1, 2], r = [2] and LIMIT 1, pushing onto the left input
can return (a=2, b=null, c=r2), a row that is absent from the unoptimized
result. This holds for a one-sided push as well, so Outer joins are now a
no-op and only Left (push onto left) and Right (push onto right) are pushed
into.
Mark the pushed-down limit as a local (per-partition) limit and lower it in
the distributed executor to a plain per-partition limit instead of the
coordinated DistributedLimit, which hangs when a shuffle is scheduled
directly above it - the join repartitions the very side the limit was pushed
onto. The retained outer global limit still performs the exact final cut, so
a per-partition limit below the join is sufficient.
Make the idempotency guard offset-agnostic: a child Limit emits at most
`limit` rows regardless of its offset, so an input that is already
`Limit{limit, offset}` is no longer re-wrapped. Previously such an input was
re-wrapped on every pass and folded back by the Limit-Limit rule, making the
rule oscillate instead of reaching a fixed point.
Tests: Rust unit tests for the Left/Right push, the offset case, both
idempotency guards and the Inner/Outer/Anti/Semi no-ops, plus Python
regression tests that run a multi-partition Left/Right join with an outer
limit.
Closes #2275hello-peter-tang:push-down-limit-outer-join Latest Branches
0%
hello-peter-tang:push-down-limit-outer-join 0%
hello-peter-tang:fix/shuffle-transpose-empty-outputs 0%
zhouwenjiajia:bugfix/joinable-stream-background-error © 2026 CodSpeed Technology