Latest Results
feat(optimizer): push limit through row-preserving join sides (#7439)
## Changes Made
One optimizer feature plus the two distributed-execution defects it
exposes. Folds in #7442, which is now closed.
### 1. Push a `Limit` onto the row-preserving side of a join
Extends the `PushDownLimit` rule:
- **Left** join -> push the limit onto the **left** input
- **Right** join -> push the limit onto the **right** input
- **Inner / Outer / Anti / Semi** -> no-op
For a Left join every output row is attributable to exactly one left
row, and the rows a given left row contributes depend only on that row
and the (untouched) right input. Limiting the left input therefore only
drops whole groups of output rows, and the retained outer `Limit`
performs the final trim, which keeps results correct under Daft's
unordered `LIMIT` semantics. The Right join case is symmetric. The
pushed limit is sized `limit + offset` and carries no offset itself.
**Outer joins are deliberately excluded.** Limiting either input of a
full Outer join turns rows of the other input that used to match into
unmatched rows, so the join emits null-extended rows that 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, not only when both inputs are limited, which is why there is no
sound way to bound either input of an Outer join. Spark's
`LimitPushDown` restricts itself to LeftOuter/RightOuter for the same
reason.
The idempotency guard ignores the child's offset: a `Limit` emits at
most `limit` rows regardless of its offset, so an input that is already
`Limit{limit, offset}` is not re-wrapped. Without that, such an input
was re-wrapped on every pass and folded back by the `Limit`-`Limit`
rule, so the rule oscillated instead of reaching a fixed point.
### 2. 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 a 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
`debug_assert!` fires 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
contributes nothing to any partition group. This skips missing indices
and relaxes the assert to allow an output to be either fully partitioned
or empty. For well-formed outputs the code path is unchanged.
Reproducible on `main` with plain user code, no optimizer change
involved:
```python
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()
```
```
DaftCoreException: task 4 panicked with message
"Expected all outputs to have 8 partitions, got 0, 8, 0, 0, 0, 0, 0, 8"
```
### 3. Preserve notify tokens when fusing task builders
`SwordfishTaskBuilder::combine_with`, the one-to-one fusion used by
`HashJoinNode`, built the merged builder with `notify_tokens: vec![]`
and `cancel_token: None`, silently dropping both inputs' tokens. A
dropped oneshot resolves to a cancellation, so an upstream node waiting
on the fused task saw it as finished before it ever ran.
`LimitNode` is exactly such a node: it waits for every forwarded task to
report before tearing down its Ray `_LimitCounterImpl` actor. With the
token dropped it took the natural-drain exit straight away, `ray.kill`ed
the actor, and the fused task then failed inside
`DistributedLimitSink::start_task` with `ActorDiedError`.
`flotilla.py::_get_result` catches that - Ray builds `RayTaskError` as a
dynamic subclass of its cause, so `except ActorDiedError` also matches
an actor death raised *inside* a task - and reports `worker_died()`, so
the scheduler retried on a fresh worker forever, the actor being
permanently dead. `tests/dataframe/test_joins.py::test_limit_after_join`
then burned the job's full 90 minute timeout.
The shape is only reachable when both join inputs report exactly one
partition, because `gen_hash_join_nodes` skips the `RepartitionNode`
wrap in that case and the `LimitNode` builder goes straight into
`combine_with`. With a shuffle in between, `materialize()` submits the
task and the token survives. Change 1 is what puts a `LimitNode`
directly under a join in the first place, so this is the first PR to hit
it.
Both inputs are now consumed and their tokens merged. `CrossJoinNode`
fans one builder out against many partners and a oneshot cannot report
several completions, so it keeps the borrowed form, now named
`combine_fanout`.
## Tests
- Rust unit tests for the rule: Left push, Right push, the `limit +
offset` case, both idempotency guards (with and without a child offset),
and the Inner / Outer / Anti / Semi no-ops.
- Rust unit test `join_preserves_notify_tokens_from_both_sides` for
change 3.
- Python regression test for change 2 in
`tests/dataframe/test_limit_offset.py`.
- Change 3 is covered by the existing `test_limit_after_join`
parametrization, which is the test that was timing out.
## Known follow-ups (not addressed here)
- A `Limit` feeding a `CrossJoinNode` still hangs, before and after this
change, for the same underlying reason: the fan-out needs a completion
signal that can fire more than once, i.e. materialising the cross join's
inputs. Change 1 never produces that shape (`PushDownLimit` only pushes
onto Left/Right sides, and `JoinStrategy::Cross` requires `Inner`), but
plain user code does - `df.limit(1).join(other, on=[], how="inner")`.
- `flotilla.py::_get_result` should test for `RayTaskError` before
`ActorDiedError` / `ActorUnschedulableError`, so an actor death raised
inside a task surfaces as a real error instead of an endless
`worker_died()` retry.
## Related Issues
Partially addresses #2275 - the Outer join case requested there is not
implementable by bounding either input, see above.
---------
Co-authored-by: caobiao.cao <caobiao.cao@alibaba-inc.com>
Co-authored-by: peter <132056702+hello-tang-rui@users.noreply.github.com> Latest Branches
0%
hello-peter-tang:fix/7471-regexp-replace-backslash 0%
hello-peter-tang:fix/7468-var-stddev-precision 0%
hello-peter-tang:fix/7469-todatetime-date-only © 2026 CodSpeed Technology