Dandandan opened a new pull request, #2335:
URL: https://github.com/apache/datafusion-ballista/pull/2335
Prototype, off by default behind
`ballista.shuffle.merge_ordered_passthrough`.
## What
A sorted passthrough stage is read back by an ordering-preserving reader that
merges on the same key. This merges inside the producing task instead, so the
task writes one file rather than one per partition and the consumer opens one
source per task rather than one per partition — fan-in drops from the stage's
partition count `P` to its task count `T`.
The rewrite runs on the per-task plan, not the stage plan: in the stage plan
the
writer would take its partitioning from the merge and the stage would
collapse to
a single task. Every task reports output partition 0, which `file_id` already
disambiguates — the same shape `GlobalPartitionMap::Collapsed` handles for
the
final stage of a top-N query.
No `fetch` is set. A limit here has to come from the consumer's merge, and a
consumer without one wants every row the stage produces.
## Measured
TPC-H SF10, `--partitions 64`, 2 executors × 4 vcores.
- q2, q3, q10, q18, q21 all verified against single-process DataFusion with
the
flag on and off.
- Passthrough shuffle files across those five queries: **709 → 469**. The 240
saved is exactly `5 × (64 − 16)`, i.e. `P → T` on each query's ordered
stage.
Motivation is #2321: a `SortPreservingMergeExec` over an ordered shuffle
holds
one open stream per source, so its working set scales with `P`. Reducing
sources
is the only way to shrink it without adding merge passes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
