linliu-code opened a new pull request, #689:
URL: https://github.com/apache/hudi-rs/pull/689
> **Stacked on #688 → #687 → #686 → #682 → #678. Review those first.** This
targets `main`, so it shows their commits alongside its own. The delta that
belongs to this PR is the last commit: 2 files, +166/-6.
Third and last slice of #685, and it closes the one question the ticket left
open. **The answer turned out to be "don't add `spawn_blocking`"** — so this PR
is a measurement, a decision, and the bound that makes the decision hold.
## The question
#685 flagged it itself: a merge map that has spilled probes RocksDB
synchronously, and once the merge is a `Stream` (#688) that probe runs on the
task polling it rather than on a blocking thread. The ticket's own instruction
was to measure before adding `spawn_blocking` for its own sake, "since that
would reintroduce a crossing".
## The measurement
`spilled_merge_blocking_duration` (ignored; run with `--release --ignored
--nocapture`) times one `merge_base_batch` against a 50 000-key log map, with
and without the spill engaged:
| chunk rows | no-spill | spilled |
|---|---|---|
| 1024 | 0.4 – 1.1 ms | 5.7 – 6.1 ms |
| 8192 | 2.8 ms | 39 – 42 ms |
**The no-spill leg is the control, and it is what decides the question.** My
first version of this benchmark did not have it, and without it the 40 ms
figure looks like a spill problem. With it, the spill is a **~6× amplifier on
work that is already synchronous and of the same kind**. Gating
`spawn_blocking` on `merge_map_spilled()` (already available on the buffer
trait) would treat the amplifier and leave a poll that is milliseconds long
either way.
Per-chunk CPU work on the polling task is how a scan or join operator
behaves. Handing each chunk to a blocking pool would reintroduce the sync/async
crossing #686 and #688 removed, and cost a task hop plus a buffer move per
chunk, to shave single-digit milliseconds.
## What the numbers do say
The cost is linear in a chunk's rows. That makes the chunk size the thing
bounding how long one poll occupies its executor — and that bound was
**inherited, not chosen**: the base read set no batch size, so it came from
`parquet`'s own default of 1024 (`parquet-57.3.1`, `arrow_reader/mod.rs:171`,
verified rather than assumed). That is a silent bound. A larger default
upstream, or a caller passing a bigger batch size through here, multiplies
every poll's cost with nothing failing.
`base_read_options` now asks for it explicitly as `MERGE_CHUNK_ROWS`.
**Behaviour change, small but real:** a parquet base file is unaffected —
1024 is what it was already getting. A **lance** base file defaulted to 8192,
so its merge chunks become 1024: the same rows in more chunks, and the bound is
now uniform across base-file formats instead of differing eightfold by backend.
`read_data` concatenates its stream, so the single-batch path returns exactly
what it did before.
## How are the changes test-covered
- [ ] N/A
- [x] Automated tests (unit and/or integration tests)
- [ ] Manual tests
- [ ] Details are described below
`test_a_merged_chunk_is_bounded_by_the_readers_own_batch_size` writes 5000
rows in **one** row group and asserts no chunk exceeds the bound — so a chunk
that followed the file's layout fails it. It also asserts the option directly,
and that combination is deliberate: the bound agrees with parquet's default
today, so no output-level assertion can distinguish the pin from the default,
while the realistic future break is a caller passing a larger size (for
instance making `hoodie.read.stream.batch_size` effective on the merge path —
the gap recorded in #687).
Mutation-checked, each reverted: dropping the pin, and raising it eightfold.
Each fails that test.
Verified locally: format, clippy over all targets and without default
features, the full `hudi-core` suite both ways (1324 lib tests, 1303 without
defaults), the gold parity sweep green across all fixtures and both reader
versions, and `cargo check` of `hudi-cpp` and `hudi-python`. Fork PR, so none
of it ran in CI — the numbers are local.
## Limits of the measurement
Stated so nobody reads more into it than it supports:
- The spill file is freshly written, so the page cache is warm and **the
disk component is understated**. A cold spill file would be slower — though
still bounded by the same chunk size, which is the point of pinning it.
- One machine, single-threaded, release build. It is enough to separate
"milliseconds" from "hundreds of milliseconds", which is the distinction the
decision rests on, and not enough to quote as a throughput figure.
If the spilled path ever does need to come off the executor, the hook is
unchanged and cheap: `merge_map_spilled()` on the buffer trait, checked once
per merge rather than per chunk.
--
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]
