Dandandan opened a new pull request, #2316:
URL: https://github.com/apache/datafusion-ballista/pull/2316
> Stacked on #2315 — until that merges, the diff here also shows its commits.
> The change only pays off when a task owns several input partitions, which
is
> what #2315 makes the default.
## Summary
A task that owns several input partitions wrote one `data.arrow` plus index
**per input partition**. Every downstream reader fetching partition `k`
therefore opened one file per input partition, and a stage left `M` files
behind, where `M` is the stage's input partition count.
The writer now emits **one file per task**.
This matters more since `ballista.scheduler.max_partitions_per_task` defaults
to `0` (#2315): a task now routinely owns several input partitions, so `M`
files per stage became `M/P`.
## Design
Each input partition still buckets and spills concurrently, and still encodes
its own buckets to IPC bytes **on its own task**, so the interleave, framing
and compression stay parallel across inputs. The coordinator — which already
awaited every input before responding — then concatenates the finished
buffers
into one file and writes one index.
The file is laid out **partition-major**: the schema header, then output
partition 0's bytes from every input in turn, then partition 1's, and so on.
```text
data.arrow: [ bucket 0 ][ bucket 1 ][ bucket 2 ]
\_ in0,in1 _/
data.arrow.index: { 0 -> off0, 1 -> off1, 2 -> off2 }
```
Keeping each output partition contiguous is what lets the index stay one
offset per partition, which is why **the reader is unchanged**:
`create_shuffle_path` already resolves a sort-shuffle summary to
`{stage_id}/{file_id}/data.arrow`, and `MultiStreamPartitionStream` already
crosses concatenated IPC streams inside a byte range.
Spill directories move from `{stage}/{file_id}/spill` to
`{stage}/{task_id}/spill-{input}`, so a task owns exactly one directory under
the stage and cleanup no longer strands an empty directory per input
partition.
## Why the encoding placement matters
An earlier revision of this change did the encoding in the coordinator. That
collapsed write parallelism from `P` to `1` and cost **+29%** on TPC-H SF10 —
the file-count reduction was real but nowhere near worth the lost
parallelism.
Encoding per input, and leaving the coordinator only byte concatenation, is
what makes the reduction free.
## Results
TPC-H SF10, 2 executors x 4 vcores, `--partitions 16`,
`max_partitions_per_task=0`. Old and new binaries alternate within each round
so drift is shared, 11 runs total:
| | median | sort-shuffle files |
|---|---|---|
| one file per input partition | 20.94 s | 1762 |
| one file per task | **19.88 s** | **442** |
-5.1% wall time and a 4x reduction in shuffle files (P = 4 vcores per
executor). The passthrough `ShuffleWriterExec` path is untouched, and its
file count is unchanged at 1710.
## Testing
`cargo test -p ballista-core --lib sort_shuffle` — 40 pass, including the
existing spill, multi-spill, empty-partition and in-memory round trips.
Those tests all drove a **single** input partition, so the consolidation path
was uncovered. Added `multiple_input_partitions_write_one_file`, which drives
3 input partitions into 4 output partitions and asserts that the stage
directory holds exactly one entry (named after the task id), that every input
row round-trips, and that no row appears in two partitions.
TPC-H SF10 end-to-end, all 22 queries verified against single-process
DataFusion with `--verify`.
--
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]
