Support streaming MOR merge with projection pushdown on base file
- Dominant language
- Rust
- Stars
- 279
- Forks
- 67
- Avg merge
- 3d 1h
- Merged PRs (30d)
- 40
Description
## Summary
`FileGroupReader::read_file_slice_stream` currently falls back to collect-and-merge for MOR slices with log files (yields the merged result as a single batch on the stream). The eager merge path also reads the entire base parquet file before applying any user projection. Both gaps have known fixes that map directly to Apache Hudi's Java implementation.
## Current state
- `RecordMerger::merge_record_batches` is an "all-batches → one-batch" primitive (`crates/core/src/merge/record_merger.rs:89`), so the MOR merge path cannot yield batches incrementally.
- `read_base_file_eager` materializes the entire parquet file into a single `RecordBatch`; `apply_eager_options` then applies projection post-merge.
- The streaming-without-log-files path already widens the parquet projection to include filter columns and narrows after filtering. The merge path does not push projection down at all.
## Reference: Apache Hudi Java
- Base file is **streamed** record-by-record via `ClosableIterator` (`hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodieFileGroupReader.java:128-170`).
- Log records are loaded into a hash map keyed by record key, populated by `KeyBasedFileGroupRecordBuffer` from `HoodieMergedLogRecordReader`. Spillable to disk via `ExternalSpillableMap`.
- Merge: for each base record, look up the map → merge if hit, emit base as-is if miss. After the base iterator is exhausted, emit log-only insert records.
- Projection **is** pushed down to the base file, with widening to include the merge keys (record key + precombine field + delete-marker fields) via `FileGroupReaderSchemaHandler.generateRequiredSchema()` (lines 167-208). The merged output is narrowed back to the user-requested schema by an output converter (lines 127-131).
- Predicates are **not** pushed down to the base file — applied after merge so log updates that change a row's predicate-relevant column are handled correctly.
## Proposed work
1. **Per-batch merge primitive on `RecordMerger`:**
- `build_log_index(log_batches) -> LogRecordIndex` (hash-keyed by record key, deduped/precombined)
- `merge_base_batch(base, &mut index) -> RecordBatch` (consumes matched keys from the index)
- `drain_remaining(index) -> RecordBatch` (insert-only log records as a final stream item)
2. Replace the `read_file_slice_from_paths_stream` MOR fallback with the streaming merge path built on the new primitive.
3. In the merge path, widen the base-file parquet projection to include record key + precombine field; reuse `util::arrow::project_batch_by_names` to narrow the merged result back to the user-requested columns.
4. Apply the same widening to the eager merge path so projection pushdown benefits both eager and streaming reads.
## Out of scope
- Predicate pushdown to the base file (Java doesn't do this either; correctness requires post-merge filter evaluation).
- Spillable log-record index (parity with Java's `ExternalSpillableMap`) — reasonable follow-up for tables where the log set doesn't fit in memory, but not required for the initial streaming-merge implementation.
Contributor guide
Research direction
Start with crates/core/src/merge/record_merger.rs:89 and the read_file_slice_from_paths_stream MOR fallback, then trace read_base_file_eager, apply_eager_options, and util::arrow::project_batch_by_names. Verify the merge can yield batches incrementally, widens the base projection for merge keys, narrows output to requested columns, and preserves post-merge predicate handling.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- data-engineering
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 55/100