avantgardnerio opened a new pull request, #2364:
URL: https://github.com/apache/datafusion-ballista/pull/2364
> **Draft.** Substrate only — this changes the shuffle file's framing and
nothing else. Read volume is unchanged, so please don't review it expecting a
speedup. The saving it exists to enable is described under *What this is for*,
and lands in the follow-up.
**Purely additive.** Nothing constructs the new writer unless
`ballista.shuffle.range.enabled=true`, which defaults to `false`. The existing
passthrough and sort-based shuffles keep their formats byte-for-byte, and every
existing reader is untouched on its own path.
## What this is for
A stage that routes rows by value range hands each consumer a set of
producer files, of which the consumer wants only the rows inside its own cut
range. It reads them whole and `RangeFilterExec` drops the rest.
Measured on h2o Q8 (`sum(v2) OVER (ORDER BY v2 RANGE BETWEEN 3 PRECEDING AND
CURRENT ROW)`), 1e8 rows, 2 executors × 4 vcores. Read amplification needs no
new instrumentation — `RangeShuffleReaderExec` does no filtering, so the ratio
between what it emits and what the `RangeFilterExec` above it keeps is exactly
the over-fetch:
| K | reader emits | halo-trim keeps | read amp |
|---|---|---|---|
| 8 | 275.0 M / 8.2 GB | 121.2 M / 3.6 GB | **2.27×** |
| 32 | 266.5 M / 7.9 GB | 193.9 M / 5.8 GB | 1.37× |
| 128 | 567.4 M / 16.9 GB | 480.4 M / 14.3 GB | 1.18× |
At K=8 — also the fastest configuration, 85 s against 214 s at K=128 — 56%
of the shuffle read is thrown away after being fetched and decoded.
Skipping it means seeking, and seeking means the data file needs a chunk
index. The Arrow IPC **stream** format the passthrough shuffle writes has none:
finding the bytes holding a given value means walking every message from the
head. The IPC **file** format ends with a footer listing one `Block { offset,
metadata_length, body_length }` per record batch.
So this PR writes the file format for those stages, and captures each
batch's byte offset at write time. Reads stay whole-file. What lands is a
footer to seek against and the offsets to seek with — the content of the value
index that follows.
## Why a separate writer rather than a flag on `ShuffleWriterExec`
The two IPC framings are not interchangeable on the read side: a stream
decoder rejects a file's leading magic and vice versa. A mode flag on
`ShuffleWriterExec` would put every existing reader — local, Flight `do_get`,
and the raw block transport — one config change away from failing to decode
what it is handed.
As a distinct operator, range shuffle is a new file type that only the
readers taught to recognise it will open, and the existing shuffle's format
never moves. This also answers @phillipleblanc's point on #2204 — a writer that
owns its own file type, which a per-format implementation can sit alongside —
and it avoids the concern @andygrove raised there about a wire-format switch
with no compatibility path, because nothing that exists today changes format.
## How the pair is chosen
Both sides are planted in the AQE adapter, the only place that sees the
writer and the reader together, under the same condition: the stage's child
declares an output ordering. That is already what selects
`RangeShuffleReaderExec`, so writer and reader cannot disagree.
The static planner never plants either — it plants the arrival-order reader,
which would be handed a format it cannot decode.
## Format detection
Readers pick a decoder by sniffing the file's `ARROW1` magic rather than by
a flag on `PartitionLocation`, so the format stays out of the wire protocol and
a consumer, the Flight server, and a human with `xxd` all reach the same answer
from the file alone. This follows the existing precedent in
`is_sort_shuffle_output`, where a stage's on-disk layout is authoritative.
## Honest regressions
- **Remote range reads route over `do_get` while the flag is on.** The
`IO_BLOCK_TRANSPORT` action ships raw file bytes for the client to decode with
a `StreamDecoder`, which only speaks the stream format, so range sources come
back decoded instead. Local reads are unaffected. A byte-range action is what
makes block transport viable here again, and it is the same mechanism the value
index needs — so this resolves in the follow-up rather than lingering.
- **Crash semantics change for range stages.** A stream-format file is
readable up to its last complete message; a file-format file with no footer
does not open at all. A task killed mid-write therefore leaves a file that
hard-fails on open rather than one that partially reads. Task-level retry
covers it, but it is a real behaviour change and I would rather name it than
have it found later.
- **Nothing consumes the captured offsets yet.** They are logged at debug.
The tests pin that they tile the file contiguously and that their count matches
the footer, but nothing yet proves they correctly *address* a batch — that
proof arrives with the reader-side seek.
## Testing
- Round trip driving `RangeShuffleWriterExec` through to
`fetch_partition_local`, rather than fabricating files and asserting the reader
agrees with itself.
- The writer's output is asserted to actually be the file format, since a
silent fallback to stream would go unnoticed until a consumer tried to seek.
- Block offsets tile the file with no gaps, start past the magic, and count
the same as the footer's batches.
- Format detection separates the two IPC framings and treats a missing file
as neither.
- h2o Q8 at 1e7 verified against single-process DataFusion with the flag
both on and off. On disk the split is exactly as intended: the range-routed
stage is `ARROW1`, the result stage stays a stream, and with the flag off every
file is a stream.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
