avantgardnerio opened a new pull request, #2204:
URL: https://github.com/apache/datafusion-ballista/pull/2204
## Summary
Adds a value-range index (`.value.idx`) sidecar next to each
passthrough-shuffle output file, so a downstream reader with an assigned value
cut range can seek into the data file without full-file read amplification. Six
commits:
1. `ValueIndexExec` — passthrough tap resolving ORDER BY from
`input.output_ordering()` at construction. Refuses unordered input at `try_new`.
2. Writer path — accumulates `(sampled_row, ORDER BY values)` per non-empty
batch, flushes an Arrow IPC file on EOS at
`{work_dir}/{job_id}/{stage_id}/{partition}/data-{task_id}.value.idx`.
Constructor mirrors `ShuffleWriterExec::try_new` so the scheduler/executor
stamps both from the same identifying inputs.
3. Rename pass — `first_row` → `sampled_row` throughout; also generalizes
`sample_first_row` → `sample_row(batch, order_by, row_idx: usize)` so stride
sampling drops in without a rename churn.
4. `ValueIndexReader` — opens the file, validates schema envelope (version,
row-index scheme, `sampled_row` column shape), exposes `total_row_count`,
`num_leaves`, `sampled_rows`, `leaf_batch`, `leaf_row_range` (with
`total_row_count` sentinel for the last leaf).
5. Binary search — `binary_search_by_value(target, sort_options)` via
`RowConverter` (composite / arbitrary Arrow-comparable / asc/desc/nulls-first
all by construction) and `binary_search_by_sampled_row(target: u64)` (utility
for ROWS-frame halo).
6. **Format switch: passthrough shuffle now writes Arrow IPC file format
instead of stream format.** `FileReader::set_index` + footer block index is
what makes the value-index reader's `sampled_row → byte range` translation
O(1). See tradeoff section below.
## File-format agnostic by design
The sidecar itself only stores `(sampled_row, ORDER BY values)` — no byte
offsets. Byte-range resolution at read time is delegated to the data file's own
format-native chunk metadata. The tap sees only `RecordBatch`; the reader
translation step is format-specific.
**Requirement on the data file's format:** a footer (or equivalent index at
a known location) we can read once to enumerate batch/row-group boundaries,
plus per-chunk row counts. Every serious columnar format has this.
Concretely:
- **Arrow IPC file** — footer's `Block { offset, metadata_length,
body_length }` + each RecordBatch message's row count in its metadata. What
this PR implements.
- **Vortex** — chunk index in the file layout, exposed by the Vortex reader.
`ValueIndexReader` doesn't need to know the format; only whichever code sits
between it and the data file does. When Vortex-format shuffle lands (Spice has
this in their private Ballista fork per Phillip LeBlanc's DataFusion Community
Showcase Vol. 2 talk 2026-07), the tap and sidecar reader are unchanged — a
Vortex-specific `sampled_row → byte range` resolver plugs in alongside the
Arrow IPC one.
- **Parquet, ORC** — row-group index in the footer. Same shape.
So `ValueIndexExec` writing side is fully format-agnostic (it evaluates the
ORDER BY expression, records rows). The read-side is a small format-specific
translator on top of `ValueIndexReader::leaf_row_range`. This PR ships the
Arrow IPC translator via `FileReader::set_index`.
## Wins
- **Enables range-GET reads on object-store-backed shuffle.** Consumer
downloads a KB-scale index, binary-searches its assigned value cut range, then
range-GETs just the matching batch. Previously impossible on Stream format.
- Sidecar design ports to any columnar format with a chunk index (Arrow IPC
file today, Vortex when it lands upstream, Parquet/ORC if ever relevant).
- No API change downstream. Same `RecordBatch`es come out of the reader.
Sort-shuffle path is untouched.
- New surface is off by default — nothing constructs `ValueIndexExec` yet.
Planner rule to insert it will follow as a separate PR.
## Regressions (honest)
- **Full-file remote reads pay ~2 small extra range-GETs.** FileReader's
footer-open sequence is: read 10-byte tail marker → read footer flatbuffer →
then batches. StreamReader was one sequential pass. On S3 that's 2 KB-scale
GETs of overhead per shuffle file — a rounding error against typical MB-scale
batch bodies, but real. Local reads (kernel page cache) are unaffected.
- Sort-shuffle path is unchanged — its own hand-rolled `.arrow.index` offset
table + `StreamWriter` on the consolidated `data.arrow` continues to work as
before. The tradeoff above applies only to the passthrough shuffle path.
- Not strictly "purely additive" because of the format switch, but the
switch is small, motivated by the concrete downstream consumer landing in the
same PR, and touches only shuffle write + local/Flight readers.
## Test plan
- [x] `cargo test -p ballista-core --lib` (222 pass, includes 9 new
`value_index` tests)
- [x] `cargo test -p ballista-executor --lib` (53 pass — flight_service
still works with FileReader)
- [x] `cargo clippy --all-targets -p ballista-core -p ballista-executor`
clean
- [x] `cargo fmt --all`
- [x] New e2e test `range_download_via_sidecar_and_footer` runs the full
pipeline (Datasource → Sort → RuntimeStats → ORRE → ValueIndexExec →
ShuffleWriter) and demonstrates: `binary_search_by_value(target)` → leaf →
`FileReader::set_index(leaf_idx)` → batch containing target value. Also asserts
the current sampling invariant `leaf_idx == batch_idx` so a future
stride-sampling change trips this test rather than silently producing wrong
reads.
🤖 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]
