linliu-code opened a new pull request, #688:
URL: https://github.com/apache/hudi-rs/pull/688
> **Stacked on #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: 7 files, +711/-707.
Second slice of #685, and the one that removes the crossings. With #686 (the
log read) and this, a file-slice read crosses the sync/async boundary **zero**
times instead of three.
## Description
The merge was a synchronous `Iterator`, so the async base file had to be
turned back into a sync reader to feed it, and the sync result turned back into
a `Stream` to return it. Both conversions are gone.
`FileGroupMergeStream` — renamed from `FileGroupMergeIterator`, because it
is not one any more — produces a chunk with `next_chunk().await`, or becomes a
`Stream` with `into_stream()`. Its base side is the parquet stream itself. The
only await is the base pull; the merge kernel, the log-only drain and the
output projection are synchronous work over data already in memory.
### What this deletes
| Deleted | Why it existed |
|---|---|
| `BlockingBatchReader` | Pulled the async base stream with `block_on`
against a runtime handle; could only be driven off the worker threads. **Note
for anyone searching:** #685 calls this `ParquetSyncReader`, a name that
existed nowhere but in ten comments — also corrected here. |
| `open_blocking_stream` | Ran the whole merge on `spawn_blocking` behind a
depth-1 channel, to bound the producer to one batch ahead of the consumer. A
stream is demand-driven, so it has that bound by construction. |
| The `streaming: bool` split | `make_base_file_source` returned the same
type from both branches with two different safety contracts — one "MUST be
consumed in sync context", the other safe from async callers. There is one base
source shape now, and it is always async. |
| `HoodieFileGroupReader::open()` | Nothing. It had no caller in core, hudi,
datafusion, python or cpp, and `reader_v2` is `pub(crate)`, so nothing
downstream could call it either. The cxx bridge reads through
`read_file_slice*` with its own runtime and wraps a materialised
`Vec<RecordBatch>`. |
| `apply_instant_range_filter` | It never inspected a batch — it kept all of
them or none, by the base file's own commit instant. The decision moved ahead
of the read as `base_file_in_range`, so an excluded base file is no longer read
and discarded, and the eager fallback that existed only to materialise rows for
that filter is gone with it. |
### `read()` keeps merging the base as one batch, on purpose
`collapse_base` collects the base stream into a single batch before merging,
and this is **not** a safety concern — both paths are async now. It is about
output order: merging per row group emits kept base rows and log-side
replacements interleaved per group, while merging one batch emits all of the
former then all of the latter. Same rows, different sequence.
Nothing in the suite would have caught a change here — the eager/stream
agreement test compared only row counts, and the gold parity sweep sorts before
comparing — so changing a public entry point's output order would have been an
unverifiable change. The two paths now differ in chunking alone, stated in one
place.
### One stat corrected
`final_merge_ms` now times only the merge. Timing the loop, which is what
the code shape invites, would fold the base file's read latency into a stat
that means merging.
## How are the changes test-covered
- [ ] N/A
- [x] Automated tests (unit and/or integration tests)
- [ ] Manual tests
- [ ] Details are described below
Two coverage gaps surfaced, both the same shape — **fewer rows, no error**:
- **`test_every_base_row_group_reaches_the_output`** reads a base file
written in four row groups, through both entry points. Making `collapse_base`
keep only the first row group **passed all 1321 tests** before this test
existed: every base file in the suite fits in one row group, so keeping one and
keeping all look identical. The stream side asserts more than one chunk, which
is what proves the fixture really spans several groups.
- **`streaming_and_eager_reads_agree`** compared row counts only, which
passes for a stream returning the right number of wrong rows. It now compares
row content, sorted — sorted because the two entry points chunk the base
differently and Hudi promises no row order.
The schema-evolution test no longer has an eager and a streaming source to
compare, so it compares the base source against its collapsed form instead —
still two shapes, still required to agree.
Mutation-checked, each reverted afterwards:
| Mutation | Result |
|---|---|
| `collapse_base` drops all but the first row group | fails only
`test_every_base_row_group_reaches_the_output` |
| No-merge path forwards only the first base batch | fails 4 tests |
| Invert the hoisted instant-range decision | fails 8 tests |
Verified locally: format, clippy on `hudi-core` with and without default
features and over all targets, the full `hudi-core` suite both ways (1322 lib
tests, 1302 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.
## Known, deliberate, and next
A merge map that has **spilled** probes RocksDB synchronously, and that
probe now runs on an async worker rather than a blocking-pool thread. It is
local disk, and only past `hoodie.memory.merge.max.size`. This is the caveat
#685 raises itself, and the honest position is that it needs a measurement
before a fix: gating it on `merge_map_spilled()` (already on the buffer trait)
is cheap, but adding `spawn_blocking` per batch on principle would reintroduce
a crossing this PR exists to remove.
The bindings still keep runtimes of their own — Python's process-wide one,
the cxx per-reader `current_thread` one. Those are #685's stated non-goals and
do not go away because the core became async. What is gone is the runtime the
*core* kept for its own blocking I/O (#686) and the blocking thread it kept for
its own merge (here).
Also worth recording, not fixed: `hoodie.read.stream.batch_size` is honoured
for a base-file-only slice and has no effect on a slice with log files, where
chunking follows the base file's row-group cadence. Flagged in #687; unchanged
here.
--
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]
