feat(core): ValueIndexExec + ValueIndexReader + shuffle FileWriter switch

[avantgardnerio]

Draft status. This PR ships the file format, reader, and search primitives as an additive lego so they can be reviewed in isolation. The ultimate shape is likely a bundled RangeShuffleWriter / RangeShuffleReader pair (parallel to SortShuffleWriter) that owns both the data file and the .value.idx file end-to-end. Bundling removes a silent alignment risk in the current tap-based shape: today's leaf_idx == batch_idx invariant would break if any future op between ValueIndexExec and ShuffleWriter re-chunked batches (coalesce, byte-budget splitter, etc.) — no panic, just wrong reads. Landing the pieces additively first lets the format and reader primitives get review; the bundled writer/reader consumes them unchanged.

Downside of the bundled shape: one bundled writer variant per data file format (RangeArrowIpcShuffleWriter, RangeVortexShuffleWriter, ...). The tap-based design is compositional — a single format-agnostic ValueIndexExec pairs with whichever data writer already exists per format. Under bundling, the value-index-writing logic can factor into a shared trait/generic so it's not literally duplicated code, but the writer struct itself is per-format. Whether that matters depends on how many formats we care about long-term.

Summary

Adds a value-range index file (.value.idx) 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.

Switches passthrough shuffle from Arrow IPC stream format to Arrow IPC file format so FileReader::set_index + footer block index can translate sampled_row → byte range in O(1) rather than the O(N) message walk stream format would force on S3.

File-format agnostic by design

The index file 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) that can be 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 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.
• 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 RecordBatches 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.

[avantgardnerio]

@phillipleblanc I'd be curious to hear how this works with your Vortex shuffles. I cloned a few of your repos, but couldn't find them. I also considered implementing this functionality in the flight server (binary search on read), but it seems like both Corlogix and SpiceAI flush shuffles to s3 in case an executor fails? This PR solves that problem with a (round-trip-heavy) pure S3 range-request answer.

[andygrove]

I understand the motivation, and at a high level this makes sense to me.

This is obviously a breaking change - how does this impact rolling upgrades? do we need to start adding ballista version to the protocol so that nodes can reject requests from older versions? I think we discussed this in the past, but I don't remember if that got implemented already.

[avantgardnerio]

I don't remember if that got implemented already.

The field is hooked up, including the scheduler side rejection, and the health check. But that only covers the normal protocol path - it's just for schedulers & executors speaking to each other. For the hard part:

This is obviously a breaking change - how does this impact rolling upgrades?

1. Currently, we don't currently ship shared scheduler state. I'm going to assume folks run in k8s. The scheduler now has the readiness endpoint. So new scheduler fires up, new protocol version. New executors register with it. When it has enough to do work, the readiness endpoint turns green, k8s starts routing queries to it and kills the old one. During the transition, in-flight queries probably die. Solution: k8s operator, bring back KV stores.

2. There is no persistent shuffles. If there were, this would be a breaking change. But since shuffles die with their host executor, this PR effectively falls under # 1 above.

3. Ballista is a library. @phillipleblanc likely has better answers for how he uses it. I'm pretty sure he persists shuffles to s3 I saw in his video? In which case this PR could be a deeper breaking change if new executors try to read old shuffles. I think we could create a fallback path for that.

[avantgardnerio]

I'm also increasingly leaning towards it being its own shuffle writer, only invoked for ordered range reparations, which would both: 1. Save a round trip 2. Not break backwards compatibility

But it would require another impl for vortex shuffles

[phillipleblanc]

I think having the per-format implementations of RangeShuffleWriter and RangeShuffleReader that own both the data + index makes sense to me. Its unlikely we'll want to add a bunch of formats to ballista, and this would open the door for us to have our Vortex shuffle implementation out of our fork and fully contained in our own project code.

As long as the traits RangeShuffleWriter and RangeShuffleReader are public and there is some way for us to register our implementation of it then this makes sense to me!

Also RE: the breaking change - I checked and this won't be a problem for us.

[andygrove]

Heads up that CI is red here on three jobs (macos test, test linux ballista, test linux crates), and I think it's real rather than flaky.

should_force_local_read::case_1_standalone and case_2_remote fail with IpcError("Premature end of the stream while decoding schema"). That traces back to the block transfer path, which I don't think the format switch covers. flight_service.rs around line 340 opens the shuffle file with tokio::fs::File and ships the raw bytes over ReaderStream for the do_action block transfer from #1315, and the client decodes those bytes with StreamDecoder in client.rs. StreamDecoder only speaks stream format, so once write_stream_to_disk emits file format the bytes lead with the ARROW1 magic and trail a footer, and BlockDataStream::try_new never finds a schema. That one isn't a type swap unfortunately, since raw block transfer is inherently stream shaped. The client would need to skip the magic and footer, or use a different decoder.

Related, and probably worth fixing in the same pass: I think the writer side is only half migrated. write_stream_to_disk covers the output_partitioning: None branch of ShuffleWriterExec, but the hash repartition branch builds its own writer inline around shuffle_writer.rs:325 and still uses StreamWriter, while both readers moved to FileReader. That path would write stream and read file. It isn't failing in CI today because sort shuffle is the default and has its own untouched writer and reader, which is also why TPC-H SF10 stayed green.

One thing that made both of these easy to miss: the three updated tests in shuffle_reader.rs fabricate their files with FileWriter directly, so they only assert the reader agrees with itself. A round trip that drives ShuffleWriterExec through both branches into the reader, plus one exercising forced remote block transfer, would cover the gap.

I have more feedback on the rest of the PR but will hold it until CI is green.

[andygrove]

Here's the rest of my read, separate from the CI issues above.

On the format switch and the personas. This trips a red flag in user-personas.md under Persona 3, specifically "wire-format or serialization changes that break externally-produced plans with no compatibility path". The on disk shuffle format moves from Arrow IPC stream to Arrow IPC file with no version marker and no fallback for reading what a previous version wrote. That doesn't make it forbidden, it just means it needs a deliberate call rather than landing quietly, and an entry in the upgrade guide. I appreciate that the description already says this isn't purely additive, I'd only push back on the framing of the cost as roughly two extra range GETs. The bigger cost is that the switch has to land atomically across every writer and reader of that file, and as the CI failures show there are more of those than it first looks.

Personas 1 and 2 look fine to me. No stage or task model changes, no AQE changes, and results are unaffected once the format is consistent everywhere.

Crash semantics, one for the "Regressions (honest)" list. I liked that you wrote that section, so here's one more for it. A stream format file is readable up to the last complete message, and a file format file with no footer doesn't open at all. So a task killed partway through a write now leaves behind a file that hard fails on open rather than one that partially reads. Task level retry probably makes that a non issue in practice, but it's a real behavior change and I'd rather see it named than discovered later.

write_stream_to_disk is public. Its signature is unchanged while its output format changes, so anyone using ballista-core as a library gets a silent behavior change with nothing at the type level to warn them. Worth an upgrade guide entry alongside the format switch itself.

The value index code itself. No concerns. ValueIndexExec and ValueIndexReader are cleanly additive, nothing constructs the exec yet, and the format agnostic split between "index stores sampled rows and ORDER BY values" and "a per format resolver turns a row into a byte range" reads well to me. The Vortex and Parquet cases in the description make a convincing argument that the split lands in the right place.

On the bundled RangeShuffleWriter / RangeShuffleReader question you raise at the top: I lean toward the tap based shape you have, and I'd rather fix the leaf_idx == batch_idx alignment risk directly than give it up. A debug assert comparing the recorded row counts against what the writer actually emitted would turn that silent wrong read into a loud failure, and it keeps the compositional property of one format agnostic ValueIndexExec per data writer. Happy to be argued out of it though, since you've been closer to this than I have.

[avantgardnerio]

Thanks @andygrove ! There's a lot to think about here. I hope to land e2e parallel windows then come back to this shortly thereafter. I appreciate the discussion!