You are viewing a plain text version of this content. The canonical link for it is here.

Posted to github@datafusion.apache.org by "avantgardnerio (via GitHub)" <gi...@apache.org> on 2026/08/24 22:12:02 UTC

[PR] feat(core): RangeShuffleWriterExec writing the seekable Arrow IPC file format [datafusion-ballista]

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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat(core): range shuffle reads only the bytes a consumer's range covers [datafusion-ballista]

Posted by "avantgardnerio (via GitHub)" <gi...@apache.org>.

avantgardnerio commented on PR #2364:
URL: https://github.com/apache/datafusion-ballista/pull/2364#issuecomment-5427810369

   CI found a real regression, I'm investigating...


-- 
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat(core): range shuffle reads only the bytes a consumer's range covers [datafusion-ballista]

Posted by "avantgardnerio (via GitHub)" <gi...@apache.org>.

avantgardnerio commented on PR #2364:
URL: https://github.com/apache/datafusion-ballista/pull/2364#issuecomment-5444395072

   @andygrove I implemented the rest of your advice, but skipped:
   
   > Might also be worth marking Action::FetchPartition #[non_exhaustive]
   
   I see that we export this and it's public and I feared breaking anyone who might consume it today.


-- 
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat(core): range shuffle reads only the bytes a consumer's range covers [datafusion-ballista]

Posted by "avantgardnerio (via GitHub)" <gi...@apache.org>.

avantgardnerio merged PR #2364:
URL: https://github.com/apache/datafusion-ballista/pull/2364


-- 
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat(core): range shuffle reads only the bytes a consumer's range covers [datafusion-ballista]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.

andygrove commented on PR #2364:
URL: https://github.com/apache/datafusion-ballista/pull/2364#issuecomment-5442474704

This looks good to me. One ask and one suggestion.

**Upgrade guide.** The `BALLISTA_PROTOCOL_VERSION` bump from 2 to 3 is the part I care most about. Anyone doing a rolling upgrade is going to see executors rejected at registration, and `docs/source/upgrading/55.0.0.md` is the only place they'd think to look for why. A short note under the behavior changes section saying schedulers and executors have to be upgraded together, and that the mismatch shows up as a registration rejection rather than a silent misread, would cover it. The strict equality check is the right design and I'd rather it be documented than discovered.

While you're in that file, the `### API changes` section around line 100 could take the public breaks in `ballista-core`:

- `Action::FetchPartition` loses `is_sort_shuffle` and gains `layout`, `file_kind`, and `byte_ranges`
- `BallistaClient::fetch_partition` and `fetch_partition_proxied` take a `ShuffleLayout` where they took a bool
- `BallistaClient::execute_do_action` gains a `header` parameter

The proto side of that is fine as it stands, reserving field 8 and taking a fresh number for `layout` is the right call. Could you also add the `api change` label.

**A thought on `execute_do_action`.** Every existing caller passes `None` for the new parameter, so this one could stay source compatible: add `execute_do_action_with_header(action, header)` and have `execute_do_action(action)` delegate with `None`. The other breaks here are unavoidable, a struct variant can't gain a field without breaking construction and the layout enum genuinely replaces the bool, but this one looks like a free save. Entirely your call if you'd rather keep the single method.

Might also be worth marking `Action::FetchPartition` `#[non_exhaustive]` while you're here, so the next field added to it isn't a break either.

--
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(core): range shuffle reads only the bytes a consumer's range covers [datafusion-ballista]

Posted by "avantgardnerio (via GitHub)" <gi...@apache.org>.

avantgardnerio commented on PR #2364:
URL: https://github.com/apache/datafusion-ballista/pull/2364#issuecomment-5419383881

   @andygrove I have an approval, but I thought I'd give you some time to look it over, since you had feedback on the last version of this PR, and its related to versioning, protocol, and shuffle.


-- 
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat(core): range shuffle reads only the bytes a consumer's range covers [datafusion-ballista]

Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.

milenkovicm commented on PR #2364:
URL: https://github.com/apache/datafusion-ballista/pull/2364#issuecomment-5415253494

   > This makes sense to me, but I do wonder if we should try to come up with a more generic shuffle writer that could handle a file format that is registered in the ballista system somehow? i.e. there isn't anything in the `RangeShuffleWriterExecNode` protobut message itself that couldn't be made generic, its just the message type itself.
   > 
   
   This would help with external shuffle services 


-- 
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org