Latest Results
refactor(shuffle): migrate repartition onto shared backend and unify pipeline dispatch (#7275)
## Summary
Follows #7221 (merged). Completes the local shuffle backend unification
(#7212) by migrating `RepartitionSink` onto the shared
`LocalShuffleBackend` and collapsing the three per-operator Ray/Flight
dispatch blocks in `pipeline.rs` into a single resolution point. Removes
the last `// TODO: unify shuffle backends in all local operations`
markers.
## Why
After #7221, gather and into_partitions share `LocalShuffleBackend`, but
`repartition.rs` still carried its own `RepartitionBackend` enum, and
`physical_plan_to_pipeline` repeated the same Ray/Flight match
(including an identical `ctx.shuffle_server().expect(...)`) three times.
This is the remaining phase 2 cleanup from #6472.
## Changes Made
- `sinks/shuffle_backend.rs`: make `FlightShuffleContext` transport-only
(`shuffle_id`, `shuffle_dirs`, `compression`, `local_server`,
`shuffle_address`); sink-specific inputs (schema, per-partition spill
target) are owned by each sink. Add `LocalShuffleBackend::from_plan` as
the single place that resolves a plan-level `ShuffleBackend` and the
worker'''s Flight server
- `sinks/repartition.rs`: replace `RepartitionBackend` with
`LocalShuffleBackend`; parse compression lazily at `finalize` instead of
eagerly at construction; collapse `new_ray` / `try_new_flight` into one
`new(..., backend)`
- `sinks/gather.rs`, `sinks/into_partitions.rs`: each cache sink derives
its own per-partition spill target locally (a module const for gather, a
`num_partitions`-derived value for into_partitions) and sources its
schema from its own field instead of the shared context; collapse
constructors to take the resolved backend
- `pipeline.rs`: the three shuffle-sink arms now call
`LocalShuffleBackend::from_plan(...)` and the unified constructors; the
duplicated shuffle-server `expect` lives once in `from_plan`
## Behavior
Functionally equivalent for all inputs. Spill thresholds, state
machines, output types, and node display names are unchanged.
Repartition no longer parses the compression string eagerly at pipeline
construction; it parses at `finalize` like the write step of the other
sinks. This is unobservable in practice: `flight_shuffle_compression` is
validated at config-set time (`set_execution_config` accepts only `lz4`,
`zstd`, or `none`), so no invalid value reaches any sink.
## Test Plan
- `cargo check` / `cargo fmt` / `cargo clippy -p daft-local-execution
--lib --no-deps` clean for the touched files
- `DAFT_RUNNER=ray pytest tests/dataframe/test_shuffles.py`: 39 passed
(gather / into_partitions / repartition under Ray and Flight backends)
- `DAFT_RUNNER=ray pytest tests/dataframe/test_sort.py`: 188 passed, 8
skipped (repartition `Range` spec via sort)
- `DAFT_RUNNER=native pytest tests/dataframe/test_sort.py
tests/dataframe/test_into_partitions.py
tests/dataframe/test_sort_shuffle_elision.py`: 316 passed, 58 skipped
## Related Issues
Second of two PRs for #7212. Part of #6472. Latest Branches
0%
FANNG1:fix/count-rows-ignores-row-groups 0%
kyo-tom:support_iceberg_rest_catalog_on_gravitino 0%
jiangxt2:fix/gravitino-error-handling © 2026 CodSpeed Technology