linliu-code opened a new issue, #685: URL: https://github.com/apache/hudi-rs/issues/685
## Problem The merge is the second reason a read crosses the sync/async boundary. Its base input is a sync `Box<dyn arrow_array::RecordBatchReader + Send>` and its output is a sync `Iterator`, so the async parquet stream has to be converted into a sync reader on the way in, and the merged result has to be converted back into a `Stream` on the way out. Two concrete costs: - `ParquetSyncReader` iterates the async parquet stream with `block_on(stream.next())`, which is why `make_base_file_source(streaming: bool)` returns values with two different safety contracts behind one type: the streaming branch "MUST be [consumed] in sync context", the eager branch is "safe to consume from async callers". - `open_blocking_stream` runs the sync merge iterator on `spawn_blocking` behind a depth-1 channel to present it as a `Stream`. Part of #684, which removes the other reason (the log read). Neither ticket alone removes all three crossings. ## Why this should be tractable The merge is already shaped like a streaming hash join, so this is a change of interface rather than of algorithm: - **Build** — read every admitted log block and insert its records into the merge map, folding duplicates. Necessarily complete before any output, since a base row cannot be emitted until it is known whether a log record supersedes it. - **Probe** — pull base row groups one at a time; `merge_one_base_batch_kernel` walks each batch and `records.remove(key)` performs the join. - **Drain** — after the base is exhausted, emit the log-only records that never matched a base row. Build-then-probe-then-drain maps directly onto a `Stream`: await the build once, then yield one merged batch per base batch, then yield the drain in bounded chunks. The base side can stay the async parquet stream throughout, and `FileGroupMergeIterator` becomes a `Stream` impl rather than an `Iterator`. ## Depends on #684 The build phase is where log reads happen, so it can only be genuinely async once the log read is. Doing this first would mean a `Stream` whose build phase still blocks on the bridge, which trades one crossing for another rather than removing it. ## Caveats to design around **The spill tier is synchronous.** `RocksDbDiskMap::get`, `remove` and `put` are plain blocking calls, so probing a merge map that has spilled does blocking disk I/O. Inside `poll_next` that blocks the executor thread. It is local disk rather than network, and it only happens past `hoodie.memory.merge.max.size`, so it may be acceptable, but the alternative is `spawn_blocking` per batch when the map has spilled. This needs a decision rather than an accident. **Per-batch merge work is CPU-bound.** Doing it in `poll_next` is normal for DataFusion operators and is probably fine. Worth measuring before adding `spawn_blocking` for its own sake, since that would reintroduce a crossing. **Backpressure changes shape.** Today the depth-1 channel bounds the producer to one batch ahead of the consumer, which was deliberate: a deeper channel multiplies peak memory by its depth. A `Stream` is naturally demand-driven and gets that property for free, but any buffering added later would silently give it back. ## What it deletes `ParquetSyncReader` and its `block_on`, the two-contract `streaming: bool` split in `make_base_file_source`, and `open_blocking_stream` with its channel. Combined with #684, a read would cross the sync/async boundary zero times instead of three. ## Validation The gold parity sweep is the oracle: every fixture, both reader versions, output byte-identical. This is where a correctness regression would actually hide, unlike #684 which is behavior-neutral by construction, so it deserves the heavier treatment: mutation-check the probe and drain paths, and confirm the existing eager-versus-streaming agreement tests still pin the two paths to each other rather than passing vacuously once both are streams. -- 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]
