linliu-code opened a new pull request, #687:
URL: https://github.com/apache/hudi-rs/pull/687
> **Stacked on #686, which is stacked on #682, which is stacked on #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: 5 files,
+373/-107.
First slice of #685. Does not close it: the merge is still a synchronous
`Iterator` and still runs on a blocking thread. This moves the seam that lets
the next slice make it a `Stream`.
## Description
The base file is the only part of the merge that has to be read rather than
computed, and the record buffer owned it: `set_base_file_source` handed the
buffer a `RecordBatchReader`, and `next_merged_base_batch` pulled from it
before merging. So every byte of the base file came out of a synchronous reader
buried under the merge, which is what forces the whole loop onto
`spawn_blocking` today.
The pull and the merge were already separable -
`pull_and_merge_next_base_batch` is a loop around `merge_one_base_batch_kernel`
- so this splits them at that seam. The buffer gains `merge_base_batch`, which
merges a batch it is handed and holds no source. `FileGroupMergeIterator` owns
the source and does the pulling. Nothing about when or where the bytes are read
changes yet.
The rejected alternative was to leave the source with the buffer and make
the buffer itself async (`set_base_file_stream`, async trait methods). On a
`dyn` trait that needs boxed futures, it infects both buffer implementations
and roughly 7k lines of buffer tests, and it puts an await point in the middle
of merge state.
Two contracts get sharper in the move:
- **Only the source says the base is exhausted.** The old code returned the
merge kernel's answer straight through, so a merge that produced no batch would
have been read as end-of-base and would have skipped every later base row. The
kernel only answers that way for a zero-row input, which the pull already
filters, so it was unreachable rather than wrong - but the new split states it
in the trait and enforces it in the state machine.
- **A base source that fails is not a base source that ended.** The read is
dropped and the error surfaced, because rows already emitted make a truncated
read look like a short successful one.
The position buffer's `merge_base_batch` refuses the hybrid strategy rather
than quietly merging the ordinary way: that strategy resolves position-only
deletes row by row against a source this entry point does not have, so
answering without it would drop deletes and return rows that should not exist.
Nothing sets `needs_hybrid_strategy` today - it is initialised `false` and
never assigned anywhere - so this only matters if it is ever switched on, which
is exactly when a silent wrong answer would be worst.
## A config that does nothing on one path
Not changed here, but worth recording. `stream_batch_size` read
`hoodie.read.stream.batch_size` and handed it to a parameter its own
documentation described as ignored; this removes it as dead code. The config is
honoured for a **base-file-only** slice, which takes a separate path in
`FileGroupReader::read_base_file_stream` and passes it to the parquet reader.
It has **no effect** on a slice **with log files**, where chunking follows the
base file's row-group cadence. The integration tests that cover it all use CoW
tables, so nothing currently fails. Whether the MOR path should honour it is a
behaviour question that belongs in its own change.
## How are the changes test-covered
- [ ] N/A
- [x] Automated tests (unit and/or integration tests)
- [ ] Manual tests
- [ ] Details are described below
`merge_base_batch_matches_the_source_owning_pull` drives the same base
batches through both routes and compares rows, including the log-only drain
that follows.
Its first version was **vacuous**, and the mutation check is what caught it:
every base key in the fixture had a winning log entry, so a buffer that merged
nothing and drained everything produced exactly the same rows, and making
`merge_base_batch` return `Ok(None)` still passed. The fixture now carries a
base row with no log entry at all, which is what makes merging distinguishable
from draining, and the test asserts that row reaches the output so the property
cannot rot back.
Two more cover the pull itself, which is new code: a merge that yields no
batch must not end the base scan, and a base source error must surface rather
than truncate.
Mutation-checked, each reverted afterwards:
| Mutation | Result |
|---|---|
| Read an empty merge as end-of-base | fails
`buffered_merge_yielding_nothing_is_not_the_end_of_the_base` |
| Swallow a base source error as end-of-base | fails
`buffered_base_source_error_surfaces_rather_than_truncating` |
| Skip the log-only drain entirely | fails 8 tests |
| Make `merge_base_batch` merge nothing | fails
`merge_base_batch_matches_the_source_owning_pull` |
Verified locally: format, clippy on `hudi-core` with and without default
features, the full `hudi-core` suite both ways (1321 lib tests, 1301 without
defaults), the gold parity sweep green across all fixtures and both reader
versions, and `cargo check` of `hudi-cpp` and `hudi-python`. This is a fork PR,
so none of it ran in CI - the numbers above are local.
## What is next
Slice 2 makes `FileGroupMergeIterator` a `Stream` whose base side is the
async parquet stream, and deletes `BlockingBatchReader`, the `streaming: bool`
split in `make_base_file_source`, `open_blocking_stream`, and
`HoodieFileGroupReader::open()` - which, checked across core, hudi, datafusion,
python and cpp, has no caller at all.
That slice moves any RocksDB spill probe from a blocking thread onto an
async worker, since the merge map's probe is synchronous. That is a real
regression in one dimension and is tracked as slice 3, gated on measurement
rather than added blind; `merge_map_spilled()` already exists on the buffer
trait to gate it.
--
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]
