NoahKusaba opened a new pull request, #2406:
URL: https://github.com/apache/datafusion-ballista/pull/2406
# Which issue does this PR close?
Closes #2403
# Rationale for this change
Look at the linked issue to understand the problem being solved.
Performance improvements + AI Benchmarking:
**Fetch ordering is cheaper than what it replaces.** The old path sorted all
M
blocks and then ran an O(M) shuffle over them. `interleave_by_executor` sorts
only the N producers. Per `execute()` call, on the synchronous setup path
before any fetch is issued:
| M | N | old (sort + shuffle) | new (round-robin) |
|---|---|---|---|
| 1000 | 20 | 837 us | 285 us |
| 5000 | 50 | 2.59 ms | 1.45 ms |
| 20000 | 200 | 16.96 ms | 6.11 ms |
**Draining off the driver.** 24 blocks of ZSTD-compressed Arrow IPC, 337 KiB
per block on the wire, decoded in `poll_next` as `BlockDataStream` does,
driven
from a single task through `buffer_unordered`:
| drain | wall time (4 runs) |
|---|---|
| inline, in the driver | 58.8, 61.0, 61.5, 63.2 ms |
| `collect_off_task` | 11.5, 12.5, 14.4, 19.1 ms |
Inline, all 24 decodes serialize onto one worker while the rest of the
runtime
idles. This is why the single driver issues requests but does not drain them.
**Help wanted: cluster numbers.** The ordering benefit itself is unmeasured.
I
do not have the hardware, and it cannot be reproduced on one machine: a
single-executor standalone cluster reads every block locally so the remote
path
never runs, and loopback has no contention to avoid. Everything above is CPU
cost, not the producer spread this PR exists to improve.
If you have a multi-executor cluster, the numbers I would find most useful,
against `main` on a shuffle-heavy query such as TPC-H q4 or q18:
- `ShuffleReaderExec`'s `permit_wait_time` and `fetch_time` metrics, which is
where converging on one producer shows up
- reduce-stage wall time, especially its spread across tasks: the change is
aimed at tail latency more than the mean
- whether more executors widens or narrows the gap, since N is what the
round-robin has to spread across
A confirmation that nothing regresses would be just as valuable as a win.
# What changes are included in this PR?
**Ordering.** `interleave_by_executor` replaces the sort-then-shuffle. Blocks
are round-robined across the executors that produced them, starting at the
producer the consumer's own index selects. A skewed producer keeps its turn
each round, then drains alone.
**Peer identity.** New `ShuffleReaderExecNode.output_partition_indices`
(proto field 8), stamped by `restrict_plan_to_partitions` alongside the slice
it selects. Absent on the wire decodes to the identity mapping, so a plan
from
an older encoder is unaffected; a stamp that is not parallel to `partition`
is
a decode error rather than a silent fallback.
**One fetch driver instead of a task per block.** Ordering only matters if it
survives to the wire. With a task per block, M tasks race to acquire
`req_sem`
and the admitted subset is up to the runtime, so the round-robin is computed
and then discarded. A single driver over `buffer_unordered` initiates in list
order; completion stays unordered.
**Decode off the driver.** IPC decode and decompression are CPU and hold the
polling thread, so `collect_off_task` drains each stream on its own task. The
request is still issued from the driver, so fetch order is unaffected.
**Separate local and remote lanes.** A queued local result pins an open fd
and
a 256 KB buffer, so locals get a depth-1 lane; remotes mirror `req_sem`. The
two are merged with `select`, so locals arrive interleaved rather than
serialized behind remotes.
`permit_wait_time` now times from when a block becomes eligible rather than
from when the buffer constructs its future, keeping the metric comparable
with
the previous task-per-block version.
`itertools` is no longer used by `ballista-core` and can be dropped.
# Tests
Ordering: `interleave_visits_every_executor_once_per_round`,
`interleave_drains_a_skewed_producer_last`,
`interleave_preserves_the_fetch_set`,
`producer_order_is_independent_of_input_order`,
`interleave_handles_no_locations`,
`stamped_peers_round_robin_across_producers`,
`rotation_comes_from_the_stamp_not_the_local_index`.
Reaching the wire: `remote_fetches_reach_the_wire_in_list_order` records
fetch
order through a stub client pool, on both sides of the governor window.
Concurrency: `decode_does_not_run_on_the_callers_task` fails unless the
drains
run on threads of their own; `local_reads_do_not_run_ahead_of_the_consumer`
counts open descriptors under the work dir.
Stamping: `restriction_stamps_the_global_output_partition_indices`,
`stamp_stays_parallel_when_an_index_is_out_of_range`,
`broadcast_reader_under_a_collected_join_is_still_stamped`, plus codec
roundtrips for the stamped, absent, mismatched, and broadcast cases.
Every test above was checked by mutation: reverting the behavior it covers
fails that test and no other.
# Are there any user-facing changes?
No
--
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]
