Dandandan opened a new pull request, #2317:
URL: https://github.com/apache/datafusion-ballista/pull/2317
> Draft. Correct and tested, but **neutral at its default budget** — see
> Results. Opening it for the measurement it produced as much as the code.
> Stacked on #2315 and #2316.
## The measurement that motivates this
Pointing `--work-dir` at a RAM disk, identical binaries and nothing else
changed, on TPC-H SF10:
| work_dir | median of 5 alternating rounds |
|---|---|
| normal volume | 23.27 s |
| RAM disk | **13.81 s** |
-40.6%, with non-overlapping distributions. Single-process DataFusion on the
same data is 13.30 s.
So shuffle disk I/O is essentially the whole remaining gap to single-node. It
is not the scheduler (planning is 1-2 ms per query, dead time 10-20 ms), not
join strategy, and not the IPC format — the RAM disk still pays full IPC
encode/decode and lz4, and only file I/O is removed. It also explains why
disabling lz4 makes things 2.2x *worse*: bytes written is the whole game.
## What this adds
An executor-wide store holding sort-shuffle output in memory, skipping the
file write. It composes with #2316: after that change the writer already
holds
each partition's finished IPC bytes, and those bytes are byte-identical to
the
range the on-disk reader addresses through the index, so storing them is a
matter of *not* writing.
**Admission, not eviction.** A task that skipped its write has nowhere else
to
serve from, so evicting an entry would lose data a downstream stage still
needs. The budget is checked when the entry is offered; a task that does not
fit writes to disk exactly as before. The store is a fast path, never a new
failure mode.
**All three read paths** consult the store before building a path: the
co-located local read, the Flight `do_get`, and the `IO_BLOCK_TRANSPORT`
`do_action` that serves remote reads by default. The block path prepends the
schema-header stream the way the on-disk path does, so the receiver cannot
tell them apart — which is why the store carries the header bytes.
**Release** happens when a job's data is reclaimed. Intermediate stages
already get that immediately on job completion via
`clean_up_intermediate_job_data`, rather than waiting for the delayed
whole-job cleanup.
`ballista.shuffle.memory_store_limit_bytes` bounds the executor, 1 GiB by
default; `0` disables it.
## Results
TPC-H SF10, 2 executors x 4 vcores, `max_partitions_per_task=0`, store on vs
off within alternating rounds:
| budget | median | sort-shuffle files left on disk |
|---|---|---|
| off | 22.12 s | 443 |
| 1 GiB (default) | 22.54 s | ~270 |
| 12 GiB | 20.98 s | 16 |
Neutral at the default, ~5% when sized to hold everything. That is well short
of the 40% the RAM disk shows, and the reason is **coverage**: only
sort-shuffle goes through this path. The passthrough `ShuffleWriterExec`
still
writes ~855 files per run — including every query's final result, which makes
a full disk round trip even for a one-row answer — and those bytes were in
RAM
in the RAM-disk experiment.
Extending the store to that writer is the follow-up that would realise the
rest, and is the reason this is a draft rather than a merge candidate on its
own numbers.
## Testing
- `cargo test -p ballista-core --lib` — 287 pass, including 7 new store tests
(admission, budget exhaustion, per-job and per-stage release, disabled via
`0`, shared global).
- `small_shuffle_is_kept_in_memory_and_readable` drives a two-input task and
asserts the job directory is **empty** — the write was skipped, not moved —
then reads every row back out of the store.
- Existing on-disk layout tests now pin `memory_store_limit_bytes = 0`, since
they assert on files the memory path deliberately does not create.
- TPC-H SF10 end-to-end: all 22 queries verified against single-process
DataFusion with `--verify`, with the store on.
--
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]
