linliu-code commented on code in PR #663: URL: https://github.com/apache/hudi-rs/pull/663#discussion_r3779917322
########## crates/core/src/file_group/reader_v2/engine.rs: ########## @@ -0,0 +1,1978 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +//! The merge-on-read file group reader. +//! +//! Ported wholesale; nothing consumes it yet, so its items are unreachable +//! from the crate's call graph until it is wired in. +#![allow(dead_code)] + +use crate::Result; +use crate::config::table::BaseFileFormatValue; +use crate::error::CoreError; +use crate::file_group::base_file::reader::{ + BaseFileReadOptions, BaseFileReader, create_base_file_reader, +}; +use crate::file_group::reader_v2::buffer::loader::{ + DefaultFileGroupRecordBufferLoader, FileGroupRecordBufferLoader, +}; +use crate::file_group::reader_v2::buffer::record_positions::ROW_INDEX_TEMPORARY_COLUMN_NAME; +use crate::file_group::reader_v2::buffered_record_converter::BufferedRecordConverter; +use crate::file_group::reader_v2::input_split::InputSplit; +use crate::file_group::reader_v2::iterator_mode::IteratorMode; +use crate::file_group::reader_v2::merge_iterator::{ + DEFAULT_BATCH_SIZE, FileGroupMergeIterator, StreamStatsHandle, new_stream_stats_handle, +}; +use crate::file_group::reader_v2::output_converter::OutputConverter; +use crate::file_group::reader_v2::profiling::profile_once; +use crate::file_group::reader_v2::read_stats::HoodieReadStats; +use crate::file_group::reader_v2::reader_context::ReaderContext; +use crate::file_group::reader_v2::reader_parameters::ReaderParameters; +use crate::file_group::reader_v2::schema_handler::FileGroupReaderSchemaHandler; +use crate::storage::{RowFilterBuilder, Storage}; +use arrow_array::RecordBatch; +use arrow_schema::SchemaRef; +use std::str::FromStr; +use std::sync::Arc; + +/// The top-level file group reader orchestrator. +/// +/// Mirrors Java's `org.apache.hudi.common.table.read.HoodieFileGroupReader<T>`. +/// +/// This is the main entry point for reading a file group. It: +/// 1. Accepts an `InputSplit` describing what to read (base file + log files) +/// 2. Creates the [`FileGroupReaderSchemaHandler`] from `data_schema` + `requested_schema` +/// 3. Creates base file iterators via storage +/// 4. Delegates log scanning + buffer creation to `FileGroupRecordBufferLoader` +/// 5. Merges base file records with log records via the buffer +/// 6. Projects output back to `requested_schema` via `OutputConverter` +/// +/// ## Construction +/// +/// Use [`HoodieFileGroupReader::builder()`] for the builder pattern, or construct +/// directly with [`HoodieFileGroupReader::new()`]. +pub struct HoodieFileGroupReader { + // ── Context (mirrors Java's HoodieReaderContext<T>) ──────────────── + /// Reader context carrying merge mode, instant range, and config maps. + reader_context: Arc<ReaderContext>, + + /// Storage for reading base files and log files. + storage: Arc<Storage>, + + // ── Input ────────────────────────────────────────────────────────── + /// Describes what to read: base file, log files, partition path. + input_split: InputSplit, + + // ── Configuration ────────────────────────────────────────────────── + /// Reader flags: use_record_position, emit_delete, sort_output, etc. + reader_parameters: ReaderParameters, + + /// The current iterator mode. + #[allow(dead_code)] + iterator_mode: IteratorMode, + + // ── Schema (mirrors Java's readerContext.getSchemaHandler()) ─────── + /// Schema handler created in the constructor from `data_schema` + + /// `requested_schema`, exactly like Java lines 119-121. + /// Owns the `required_schema` used for base file projection and the + /// `output_converter` used for final projection. + schema_handler: FileGroupReaderSchemaHandler, + + // ── Strategy ─────────────────────────────────────────────────────── + /// Buffer loader: selects buffer impl + triggers log scan. + record_buffer_loader: DefaultFileGroupRecordBufferLoader, + + // ── Mutable state (populated during read) ────────────────────────── + // NOTE: the record buffer and base-file batches are not stored on the + // reader — they are local to `init_record_iterators` and owned by the + // returned `FileGroupMergeIterator` for the rest of the read. + /// Optional converter for projecting/transforming output records. + /// Mirrors Java's `Option<UnaryOperator<T>> outputConverter`. + output_converter: Option<Box<dyn OutputConverter>>, + + /// Read statistics accumulator. + read_stats: HoodieReadStats, + + /// Stage-timing sink shared with the [`FileGroupMergeIterator`] returned by + /// [`Self::open`] (ENG-42991). The streaming iterator + /// owns the buffer once `open()` returns, so the merge-phase timings + /// (final_merge_ms, output_build_ms) and the update-processor + /// insert/update/delete counts are accumulated through this handle during + /// iteration and drained back into [`Self::read_stats`] by [`Self::read`] + /// after the stream is exhausted. Wrapped in `Arc<Mutex<…>>` because the FFI + /// path requires the iterator to be `Send` (it is boxed into an + /// `FFI_ArrowArrayStream`); the lock is taken once per emitted chunk, so the + /// cost is negligible against the per-chunk merge work. The FFI path never + /// reads these stats back — only `read()`-based callers do. + stream_stats: StreamStatsHandle, + + /// Valid block instants from log scanning. + valid_block_instants: Vec<String>, + + /// Converter for engine records to [`BufferedRecord`]. + /// Mirrors Java's `BufferedRecordConverter<T> bufferedRecordConverter`. + buffered_record_converter: Option<Box<dyn BufferedRecordConverter>>, + // NOTE: ENG-42866 — the optional parquet `RowFilter` builder used to live + // on this struct. It now lives on `reader_context` so the same builder is + // visible to (a) the base parquet read here, and (b) the parquet log + // block decoder in `file_group::log_file::content::Decoder`. The gate + // (CoW || mor_pk_safe) lives at the use sites; this file's gate is at + // `make_base_file_source` below. +} + +/// Base-file read options carrying an optional pushdown predicate. +/// +/// The three base reads below differ only in projection, so the filter is +/// attached in one place — a read that silently lost it would return extra rows +/// rather than fail, which is the hard kind of bug to notice. +fn base_read_options(row_filter: Option<RowFilterBuilder>) -> BaseFileReadOptions { + match row_filter { + Some(f) => BaseFileReadOptions::new().with_row_filter(f), + None => BaseFileReadOptions::new(), + } +} + +impl HoodieFileGroupReader { + /// Create a new file group reader. + /// + /// Mirrors Java's `HoodieFileGroupReader(readerContext, storage, tablePath, + /// latestCommitTime, dataSchema, requestedSchema, ...)` constructor. + /// + /// The constructor: + /// 1. Creates a [`FileGroupReaderSchemaHandler`] from `data_schema` + + /// `requested_schema` (Java lines 119-121) + /// 2. Calls `prepare_required_schema()` to compute the `required_schema` + /// (Java: automatic in `FileGroupReaderSchemaHandler` constructor, line 105) + /// 3. Obtains the `output_converter` from the schema handler (Java line 122) + /// + /// # Arguments + /// * `reader_context` — Engine context with merge mode, ordering fields, table config. + /// * `storage` — Storage layer for reading base files and log files. + /// * `input_split` — Describes what to read (base file path, log file paths, partition). + /// * `reader_parameters` — Reader flags (use_record_position, emit_delete, etc.). + /// * `data_schema` — Full table schema (what columns exist in the files). + /// Maps to Java's `dataSchema` / `tableSchema` parameter. + /// * `requested_schema` — Column projection requested by the caller. + /// Maps to Java's `requestedSchema` parameter. `None` means all columns. + pub fn new( + reader_context: Arc<ReaderContext>, + storage: Arc<Storage>, + input_split: InputSplit, + reader_parameters: ReaderParameters, + data_schema: Option<SchemaRef>, + requested_schema: Option<SchemaRef>, + ) -> Result<Self> { + log::debug!( + "HoodieFileGroupReader::new partition={} base_file={} log_files={} \ + ordering_fields={:?} latest_commit_time={} record_key_field={}", + input_split.partition_path, + input_split.base_file_path.as_deref().unwrap_or("<none>"), + input_split.log_file_paths.len(), + reader_context.ordering_field_names(), + reader_context.latest_commit_time, + reader_context.record_key_field(), + ); + for (i, lf) in input_split.log_file_paths.iter().enumerate() { + log::debug!(" log_file[{i}]: {lf}"); + } + + // Mirrors Java lines 119-121: + // readerContext.setSchemaHandler( + // new FileGroupReaderSchemaHandler(readerContext, dataSchema, requestedSchema, ...)); + // + // When schemas are explicitly provided (direct construction / tests), create + // a new handler. When they are not provided (FFI path via builder), use the + // handler already on reader_context — which was populated by the FFI bridge + // from the Avro JSON schemas passed through the Substrait proto. + let mut schema_handler = if data_schema.is_some() || requested_schema.is_some() { + let mut handler = FileGroupReaderSchemaHandler::new(); + if let Some(ds) = data_schema { + handler = handler.with_table_schema(ds.clone()).with_data_schema(ds); + } + if let Some(rs) = requested_schema { + handler = handler.with_requested_schema(rs); + } + handler + } else { + reader_context.schema_handler.clone() + }; + + // Mirrors Java FileGroupReaderSchemaHandler constructor line 105: + // this.requiredSchema = prepareRequiredSchema(this.deleteContext); + // + // Uses record_key_fields() (all key fields) instead of record_key_field() + // (single) to support composite record keys in virtual-key mode. + // Mirrors Java's getMandatoryFieldsForMerging() lines 250-258. + let has_instant_range = reader_context.instant_range.is_some(); + schema_handler.prepare_required_schema( + input_split.has_log_files(), + &reader_context.record_key_fields(), + reader_context.ordering_field_names(), + &reader_context.table_config, + has_instant_range, + &reader_context.merge_mode, + )?; + + // Schema-on-read (InternalSchema) evolution is not supported in hudi-rs + // (GAP-07). Gold loads an InternalSchema from the `.schema` folder and + // applies column renames / type changes through InternalSchema versioning + // when `hoodie.schema.on.read.enable=true`. hudi-rs only implements + // schema-on-write backward-compatible evolution, so silently honoring the + // flag would risk misreading evolved data. Reject it loudly at the same + // table-config chokepoint as the bootstrap gate below. + if reader_context + .table_config + .get("hoodie.schema.on.read.enable") + .map(|v| v.eq_ignore_ascii_case("true")) + .unwrap_or(false) + { + return Err(CoreError::Unsupported(format!( + "schema-on-read (InternalSchema) is not supported in hudi-rs. \ + Table at '{}' has hoodie.schema.on.read.enable=true, which requires \ + InternalSchema-based evolution (column renames / type changes) that \ + hudi-rs does not implement; only schema-on-write backward-compatible \ + evolution is supported.", + reader_context.table_path, + ))); + } + + // Bootstrap merge reordering is not yet supported in hudi-rs. + // Java's prepareRequiredSchema() (lines 280-288) partitions fields into + // meta and data columns and reorders them for bootstrap tables. Until + // that is implemented, reject bootstrap merge at construction time. + if reader_context.needs_bootstrap_merge { + // Reachable via table state (bootstrap base files), so this is a + // loud error rather than a panic. + return Err(CoreError::Unsupported(format!( + "Bootstrap merge is not yet supported in hudi-rs. \ + Table at '{}' has bootstrap base files that require \ + meta/data column reordering.", + reader_context.table_path, + ))); + } + + // Composite virtual keys ARE supported. With `hoodie.populate.meta.fields=false` + // and a multi-field recordkey, `RecordContext::record_key_array` reconstructs the + // full `field:val,field:val` merge key per row (mirroring Java + // `KeyGenerator.constructRecordKey`) on BOTH the base and log sides, so records + // sharing the first field but differing on a later one no longer collide. See + // `RecordContext::build_composite_record_key_array`. + + // Multi-field (composite) precombine/ordering keys ARE supported. + // `RecordContext::new` splits a comma-separated `hoodie.table.precombine.field` + // / `hoodie.table.ordering.fields` into `ordering_field_names`, and + // `get_ordering_values` builds one `OrderingValue::Composite` per row from + // the per-field scalars (compared lexicographically field-by-field, mirroring + // Java `OrderingValues`). A field absent from a batch, an unsupported field + // type, or a null component falls back to natural order — matching the scalar + // path — so there is no silent first-field-only degradation. (Construction + // still rejects composite *virtual keys* above: that path reconstructs the + // merge key from only the first record-key field, which would mis-collide.) + + // Mirrors Java line 122: + // this.outputConverter = readerContext.getSchemaHandler().getOutputConverter(); + let output_converter = schema_handler.get_output_converter(); + + // Propagate the prepared schema_handler back onto a new reader_context + // so downstream consumers (record buffer, log scanner) see the canonical + // schema_handler with its stored DeleteContext. Mirrors Java's + // `readerContext.setSchemaHandler(...)` — in Java the reader context is + // mutable; in Rust we create a new Arc with the updated handler. + let reader_context = { + let mut updated = (*reader_context).clone(); + updated.schema_handler = schema_handler.clone(); + Arc::new(updated) + }; + + Ok(Self { + reader_context, + storage, + input_split, + reader_parameters, + iterator_mode: IteratorMode::EngineRecord, + schema_handler, + record_buffer_loader: DefaultFileGroupRecordBufferLoader::new(), + output_converter, + read_stats: HoodieReadStats::default(), + stream_stats: new_stream_stats_handle(), + valid_block_instants: Vec::new(), + buffered_record_converter: None, + }) + } + + /// Create a builder for configuring the reader. + pub fn builder() -> HoodieFileGroupReaderBuilder { + HoodieFileGroupReaderBuilder::default() + } + + // ========================================================================= + // Main read API (mirrors Java's getClosableIterator / getBufferedRecordIterator) + // ========================================================================= + + /// Open the file group and return a streaming iterator over the merged + /// output (ENG-42991). This is the modern entry point matching Java's + /// `getClosableIterator()` semantics. + /// + /// Async work — base file decode + log file scan + buffer population — + /// runs once, here. The returned [`FileGroupMergeIterator`] is then a + /// purely synchronous [`arrow_array::RecordBatchReader`] that emits one + /// chunk per `next()` (default chunk size [`DEFAULT_BATCH_SIZE`] rows). + /// + /// This is single-use: it consumes the output_converter and (for MOR) + /// hands the buffer to the iterator. Re-opening requires constructing a + /// new `HoodieFileGroupReader`. + pub async fn open(&mut self) -> Result<FileGroupMergeIterator> { + // ENG-42992: streaming mode — the base file is held as a lazy + // `ParquetSyncReader` that does per-batch `block_on` against + // OBJECT_STORE_RUNTIME. The caller MUST consume the returned + // iterator from a synchronous context (e.g. the FFI driver), + // never from inside another tokio runtime. + self.init_record_iterators(/* streaming */ true).await + } + + /// The reader for this slice's base file format. + /// + /// Built per call rather than held: the format comes from the reader + /// context, and constructing one is cheap next to reading a file. + fn base_file_reader(&self) -> Result<std::sync::Arc<dyn BaseFileReader>> { + // An unset format means the caller did not say; parquet is the default + // base file format, and is what every non-metadata table uses here. + let format = if self.reader_context.base_file_format.is_empty() { + BaseFileFormatValue::Parquet + } else { + BaseFileFormatValue::from_str(&self.reader_context.base_file_format)? + }; + Ok(create_base_file_reader(&self.storage, &format)?) + } + + /// Stream the merged output to an async caller. + /// + /// [`Self::read`] returns the whole file group as one batch, so peak memory + /// tracks the base file. This reads the base file one row group at a time + /// instead. + /// + /// The merge loop is synchronous and has to block on the base stream, which + /// is only legal off the async worker threads. So it runs on a + /// blocking-pool thread and hands batches back over a channel; the caller + /// gets an ordinary stream and never sees the blocking. + /// + /// The channel holds one batch. That lets the producer decode row group + /// N+1 while the consumer works on N, while bounding the extra memory to a + /// single row group — a deeper channel would multiply peak memory by its + /// depth, which is what streaming is meant to avoid. + pub(crate) async fn open_blocking_stream( + &mut self, + ) -> Result<futures::stream::BoxStream<'static, Result<RecordBatch>>> { + use futures::StreamExt; + + let iter = self.init_record_iterators(/* streaming */ true).await?; + + let (tx, rx) = tokio::sync::mpsc::channel::<Result<RecordBatch>>(1); + tokio::task::spawn_blocking(move || { + for batch in iter { + let item = batch.map_err(CoreError::ArrowError); + // A send error means the consumer dropped the stream; stop + // rather than decoding row groups nobody will take. + if tx.blocking_send(item).is_err() { + break; + } + } + }); + + Ok(futures::stream::unfold(rx, |mut rx| async move { + rx.recv().await.map(|item| (item, rx)) + }) + .boxed()) + } + + /// Read the file group and return the merged output as a single + /// `RecordBatch`. Internally uses the same merge code path as + /// [`Self::open`] but with **eager** base file decode — the parquet + /// stream is drained async during this method's async body, then the + /// resulting `Vec<RecordBatch>` is wrapped in a sync + /// `RecordBatchIterator` for the merge loop. That makes the merge + /// iterator safe to consume from async callers (no nested block_on). + /// + /// New consumers that can use a sync iterator should prefer + /// [`Self::open`] for true streaming. + pub async fn read(&mut self) -> Result<RecordBatch> { + // A3 (ENG-42992): eager mode — the base parquet stream is drained + // async during `init_record_iterators` (streaming=false), so the + // returned iterator's `next()` is pure in-memory work and can be driven + // from an async caller (no nested `block_on`). `open()` (streaming=true) + // instead holds a lazy `ParquetSyncReader` for true streaming peak + // memory, but requires a sync consumer (the FFI driver). + let batch = self + .init_record_iterators(/* streaming */ false) + .await? + .collect_into_one_batch()?; + // ENG-42991 — the streaming iterator accumulated the merge-phase + // timings + insert/update/delete counts into the shared `stream_stats` + // while `collect_into_one_batch` drove it to exhaustion. Drain them back + // into `self.read_stats` so `read_stats()`-based callers (fg-bench, + // tests, reader_v1) observe the same stats the pre-streaming `read()` + // surfaced. (The FFI `open()` path does not read these back — Velox + // consumes the stream directly.) + self.drain_stream_stats(); + Ok(batch) + } + + /// Copy the accumulated streaming stage-stats into [`Self::read_stats`]. + /// Called by [`Self::read`] after the iterator is exhausted. + fn drain_stream_stats(&mut self) { + let s = self + .stream_stats + .lock() + .expect("stream_stats mutex poisoned"); + self.read_stats.final_merge_ms = s.final_merge_ms; + self.read_stats.output_build_ms = s.output_build_ms; + self.read_stats.merge_map_peak_entries = s.merge_map_peak_entries; + self.read_stats.num_inserts = s.num_inserts; + self.read_stats.num_updates = s.num_updates; + self.read_stats.num_deletes = s.num_deletes; + } + + /// Initialize record iterators: read base file + scan/merge log files, + /// hand state to a [`FileGroupMergeIterator`]. + /// + /// Mirrors Java's `HoodieFileGroupReader.initRecordIterators()`. The + /// fast path (no log files = CoW / empty) returns an `Eager` iterator + /// over the base file source; the MOR path returns a `Buffered` + /// iterator that drives `buffer.has_next() / buffer.next()` in chunks. + /// + /// The `streaming` flag controls how the base file is read: + /// - `true` — lazy `ParquetSyncReader` (one row group per + /// `next()`; does `block_on` per call → sync-context only). + /// - `false` — async drain into `Vec<RecordBatch>`, wrap in + /// `RecordBatchIterator` (no `block_on`; safe from async callers). + /// + /// `apply_instant_range_filter` requires a materialised Vec today, so + /// streaming mode falls back to eager when an instant range is active + /// (rare; see ENG-42992 follow-up to make the filter per-batch). + /// + /// ```text + /// initRecordIterators() + /// ├─ make_base_file_source(streaming) + /// └─ recordBufferLoader.getRecordBuffer(...) + /// → recordBuffer.set_base_file_source(...) + /// → FileGroupMergeIterator::new_buffered(...) + /// ``` + async fn init_record_iterators(&mut self, streaming: bool) -> Result<FileGroupMergeIterator> { + log::debug!( + "[HoodieFileGroupReader] initRecordIterators: partition={} base_file={} log_files={} streaming={streaming}", + self.input_split.partition_path, + self.input_split + .base_file_path + .as_deref() + .unwrap_or("<none>"), + self.input_split.log_file_paths.len(), + ); + + // Step 1: Open the base file source (A3 / ENG-42992). + // - streaming=true: a lazy `ParquetSyncReader` — one parquet + // row-group per `RecordBatchReader::next` call. The whole base + // file never lives in memory at once (this is the R3 fix). + // - streaming=false (or instant-range filter active, which still + // needs a materialised Vec): drain async into a `Vec<RecordBatch>`, + // optionally instant-range-filter it, wrap in a `RecordBatchIterator`. + // Stage timing (perf harness): base parquet open + (eager) read. + // In streaming mode this only covers the open; the per-row-group decode + // cost is paid lazily inside the merge loop's `next_base_row` pulls. + let base_source = profile_once!( + self.read_stats.base_read_ms, + self.make_base_file_source(streaming).await + )?; + let base_source_schema = base_source.schema(); + log::debug!( + "[HoodieFileGroupReader] makeBaseFileSource: schema_cols={} streaming={streaming}", + base_source_schema.fields().len(), + ); + + // The post-projection output schema is the same regardless of + // CoW vs MOR — used as the iterator's RecordBatchReader::schema(). + let output_converter = self.output_converter.take(); + let post_projection_schema = output_converter.as_ref().map(|c| c.target_schema()); + + // Step 2: If no records to merge (no log files), build an Eager + // iterator that yields the base file batches directly. + if self.input_split.has_no_records_to_merge() { + log::debug!("[HoodieFileGroupReader] no log files → Eager iterator"); + + // output converter runs. A3 (ENG-42992): the base source exposes + // its (post-projection) schema via `RecordBatchReader::schema()` + // without forcing a row-group decode, so we no longer peek at a + // materialised first batch. For a log-only Eager FG the source is + // an empty `RecordBatchIterator` carrying the required schema. + let merge_schema: SchemaRef = if let Some(rs) = &self.schema_handler.required_schema { + rs.clone() + } else { + base_source_schema.clone() + }; + + let output_schema = post_projection_schema.unwrap_or(merge_schema); + // Stage timing (perf harness): the Eager iterator accumulates + // per-chunk output_build_ms (concat is gone — each base batch flows + // through the converter as its own chunk) into `stream_stats`, which + // `read()` drains back into `self.read_stats`. + return Ok(FileGroupMergeIterator::new_eager( + base_source, + output_schema, + output_converter, + self.stream_stats.clone(), + )); + } + + // Step 3: MOR path — load record buffer (scan log files + create buffer). + // Mirrors Java: this.recordBuffer = recordBufferLoader.getRecordBuffer(...).getLeft(); + log::debug!( + "[HoodieFileGroupReader] scanning {} log file(s) with latest_commit_time={}", + self.input_split.log_file_paths.len(), + self.reader_context.latest_commit_time, + ); + let load_result = self + .record_buffer_loader + .get_record_buffer( + self.reader_context.clone(), + self.storage.clone(), + &self.input_split, + &self.reader_parameters, + &mut self.read_stats, + ) + .await?; + + let mut record_buffer = load_result.record_buffer; + self.valid_block_instants = load_result.valid_block_instants; + + log::debug!( + "[HoodieFileGroupReader] log scan complete: buffer_size={} valid_instants={:?} \ + stats: log_blocks={} log_records={} corrupt={} rollbacks={}", + record_buffer.size(), + self.valid_block_instants, + self.read_stats.total_log_blocks, + self.read_stats.total_log_records, + self.read_stats.total_corrupt_log_blocks, + self.read_stats.total_rollback_blocks, + ); + + // Step 4: Determine merge_schema BEFORE handing the source to the + // buffer. Schema is available via `RecordBatchReader::schema()` + // without forcing a row-group decode. + let merge_schema: SchemaRef = if let Some(rs) = &self.schema_handler.required_schema { + rs.clone() + } else if self.input_split.base_file_path.is_some() { + // The base source's schema is the parquet schema after projection. + base_source_schema.clone() + } else { + // Log-only file group: peek at any non-delete log record's batch + // (HashMap order is non-deterministic, so we must search all + // entries — the first record could be a delete). + // Find the first non-delete record's schema (`get_record()` returns + // `None` for a delete tombstone under the A2 `RecordPayload` design). + let mut schema = None; + for r in record_buffer.get_log_records().values() { + if let Some(batch) = r.get_record() { + schema = Some(batch.schema()); + break; + } + } + schema.ok_or_else(|| { + CoreError::ReadFileSliceError("No schema available for merge output".to_string()) + })? + }; + + let output_schema = post_projection_schema.unwrap_or_else(|| merge_schema.clone()); + + // Step 5: Hand the base source to the buffer + return the streaming + // iterator. The iterator owns the buffer from here on; the reader's + // role ends. + record_buffer.set_base_file_source(base_source); + log::debug!( + "[HoodieFileGroupReader] set base file source on buffer, \ + returning Buffered iterator (batch_size={DEFAULT_BATCH_SIZE})" + ); + + // Step 5: Hand the buffer to a Buffered streaming iterator. The + // iterator owns the buffer and drives `has_next/next` per chunk; it + // accumulates final_merge_ms + output_build_ms and the update-processor + // insert/update/delete counts into the shared `stream_stats`, which + // `read()` drains back into `self.read_stats` after the stream is + // exhausted (mirrors gold, where StandardUpdateProcessor increments + // HoodieReadStats during iteration). merge_map_peak_entries was already + // recorded during the log scan; the iterator reads it off the buffer up + // front (the buffer is moved into the iterator here). + self.stream_stats + .lock() + .expect("stream_stats mutex poisoned") + .merge_map_peak_entries = record_buffer.merge_map_peak_entries(); + + // Chunk size: honor `hoodie.read.stream.batch_size` from the reader + // config, falling back to DEFAULT_BATCH_SIZE when unset/unparseable. + let batch_size = self.stream_batch_size(); + + Ok(FileGroupMergeIterator::new_buffered( + record_buffer, + merge_schema, + output_schema, + output_converter, + batch_size, + self.stream_stats.clone(), + )) + } + + /// Resolve the streaming chunk size from `hoodie.read.stream.batch_size` + /// on the reader config, defaulting to [`DEFAULT_BATCH_SIZE`] when the key + /// is absent or unparseable. + /// + /// The key lives on `reader_context.hoodie_reader_config` (the same map the + /// buffer loader reads `hoodie.datasource.merge.type` from). Mirrors Java's + /// chunked `getClosableIterator` batch sizing. + fn stream_batch_size(&self) -> usize { + self.reader_context + .hoodie_reader_config + .get(crate::config::read::HudiReadConfig::StreamBatchSize.as_ref()) + .and_then(|v| v.parse::<usize>().ok()) + .filter(|&n| n > 0) + .unwrap_or(DEFAULT_BATCH_SIZE) + } + + /// [A3 / ENG-42992] Open the base file as a `RecordBatchReader` source. + /// + /// - `streaming=true`: returns a lazy [`ParquetSyncReader`] over the + /// parquet file. One row group per `RecordBatchReader::next` call. + /// Sync iteration uses `block_on(stream.next())` against + /// `OBJECT_STORE_RUNTIME` — caller MUST be in sync context. + /// - `streaming=false` (or instant range present — see note): drains + /// the parquet stream into a `Vec<RecordBatch>` async, optionally + /// applies the instant range filter, and wraps the Vec in a + /// `RecordBatchIterator` (no `block_on` at iteration time → safe + /// to consume from async callers). + /// + /// Returns an **empty** `RecordBatchIterator` (no rows) when the + /// input split has no base file (log-only file group). + /// + /// Mirrors Java's `HoodieFileGroupReader.makeBaseFileIterator()`, + /// adapted for the Rust streaming/eager split. + /// Whether this read should merge base + log records by base-file row + /// position (rather than by record key). Mirrors Java + /// `HoodieFileGroupReader`'s `setShouldMergeUseRecordPosition`: + /// `useRecordPosition && !skipMerge && hasLogFiles && parquetBaseFile`. + /// + /// When true, the base file is read with a synthetic row-index column (see + /// [`ROW_INDEX_TEMPORARY_COLUMN_NAME`]) so the position buffer can match + /// base rows to log records by position. + fn use_record_position(&self) -> bool { + if !self.reader_parameters.use_record_position { + return false; + } + if !self.input_split.has_log_files() || self.input_split.base_file_path.is_none() { + return false; + } + // Position merge needs the base file's commit time to validate log-block + // position headers. Without it the loader falls back to key-based, so the + // base read must not attach the row-index column either (keep the two + // decisions in lock-step). + if self.input_split.base_file_commit_time.is_none() { + return false; + } + let is_skip_merge = self + .reader_context + .hoodie_reader_config + .get("hoodie.datasource.merge.type") + .map(|v| v.eq_ignore_ascii_case("skip_merge")) + .unwrap_or(false); + if is_skip_merge { + return false; + } + // hudi-rs only reads parquet base files; guard defensively when the + // format is explicitly set to something else. Shared with the loader's + // buffer-selection gate so the row-index attachment and the buffer + // choice cannot diverge. + crate::file_group::reader_v2::buffer::loader::base_file_is_parquet( + &self.reader_context.base_file_format, + ) + } + + async fn make_base_file_source( + &mut self, + streaming: bool, + ) -> Result<Box<dyn arrow_array::RecordBatchReader + Send>> { + let Some(path) = self.input_split.base_file_path.clone() else { + // Log-only file group — empty base. Use the required_schema + // as the reader's reported schema when available; otherwise + // empty schema (the buffer's reader_schema fallback handles + // schema selection downstream). + let schema = self + .schema_handler + .required_schema + .clone() + .unwrap_or_else(|| Arc::new(arrow_schema::Schema::empty())); + let empty = arrow_array::RecordBatchIterator::new( + std::iter::empty::<std::result::Result<RecordBatch, arrow_schema::ArrowError>>(), + schema, + ); + return Ok(Box::new(empty)); + }; + + if self.buffered_record_converter.is_none() { + log::debug!( + "[HoodieFileGroupReader] make_base_file_source: no bufferedRecordConverter set \ + (batch-level read does not require per-record conversion)" + ); + } + + // ENG-42276 / ENG-42866 — gate parquet RowFilter pushdown. + // CoW: always safe (no merge). + // MOR: safe ONLY when every column referenced by the filter is a + // primary key (PKs are immutable across upserts, so the predicate + // outcome doesn't change post-merge — `reader_context.mor_pk_safe`, + // mirroring Java's `filterIsSafeForPrimaryKey`). + // Otherwise: drop the filter; the post-merge filter (Velox/Spark above + // the FG reader) evaluates the predicate after base+log merge. + let row_filter = if self.reader_context.can_push_row_filter() { + self.reader_context.row_filter_builder.clone() + } else { + if self.reader_context.row_filter_builder.is_some() { + log::debug!( + "[ENG-42866] MOR + non-PK predicate — skipping parquet \ + RowFilter pushdown for base file '{path}' \ + (post-merge filter still runs)" + ); + } + None + }; + + // No projection schema → fall back to the unprojected eager helper + // (rare; FFI always supplies a required_schema). Streaming variant + // of the unprojected helper is not exposed yet — eager Vec here. + // The instant-range filter (if active) must still be applied on this + // path — it gates base rows by `_hoodie_commit_time` regardless of + // projection (the filter used to live in `init_record_iterators` and + // ran on every base read; A3 moved it here, so all branches honor it). + let Some(required_schema) = self.schema_handler.required_schema.clone() else { + let batch = self + .base_file_reader()? + .read_data(&path, base_read_options(row_filter.clone())) + .await + .map_err(|e| { + CoreError::ReadFileSliceError(format!( + "Failed to read base file '{path}': {e:?}" + )) + })?; + let schema = batch.schema(); + let mut batches = vec![batch]; + if self.reader_context.instant_range.is_some() { + let pre: usize = batches.iter().map(|b| b.num_rows()).sum(); + batches = self.apply_instant_range_filter(batches)?; + let post: usize = batches.iter().map(|b| b.num_rows()).sum(); + log::debug!( + "[HoodieFileGroupReader] applyInstantRangeFilter (unprojected): {pre} → {post} rows" + ); + } + let iter = arrow_array::RecordBatchIterator::new(batches.into_iter().map(Ok), schema); + return Ok(Box::new(iter)); + }; + + // Schema-evolution intersection (gold parity, + // HoodieParquetFileFormatHelper.buildImplicitSchemaChangeInfo; A2/A1): + // 1. diff footer schema vs required by name; + // 2. ask parquet only for the INTERSECTION (in the file's own types); + // 3. project to required per batch: null-fill added columns, cast + // promotions (float→double string-mediated — C6). + // Step 3 is applied PER ROW-GROUP so it works identically on the eager + // (drain-then-project) and streaming (`ProjectingBatchReader`) paths — + // A2 risk #3: every base batch the merge interleaves must already be in + // `required_schema`. + let file_schema = self + .base_file_reader()? + .read_stream(&path, BaseFileReadOptions::new()) + .await + .map(|s| s.schema().clone()) + .map_err(|e| { + CoreError::ReadFileSliceError(format!( + "Failed to read base file footer schema '{path}': {e:?}" + )) + })?; + // Intersection by *case-insensitive* name (gold/Spark resolve field names + // case-insensitively). Project under the FILE's actual name+type so the + // parquet reader finds the column; `project_batch_to_schema` (also + // case-insensitive) then evolves each batch to `required_schema`. A + // required column absent from the footer is skipped here and null-filled + // downstream; an ambiguous footer case-collision errors loudly. + let mut present: Vec<arrow_schema::FieldRef> = + Vec::with_capacity(required_schema.fields().len()); + for rf in required_schema.fields() { + if let Some(idx) = crate::schema::batch_evolution::index_of_ci(&file_schema, rf.name())? + { + present.push(file_schema.fields()[idx].clone()); + } + } + let present_len = present.len(); + let intersection: arrow_schema::SchemaRef = Arc::new(arrow_schema::Schema::new(present)); + log::debug!( + "[base-file-evolution] path={} file_cols={} required_cols={} intersect_cols={} streaming={streaming}", + path, + file_schema.fields().len(), + required_schema.fields().len(), + present_len + ); + + // Position-based merge: append a synthetic row-index column carrying + // each row's TRUE physical base-file position (via a parquet virtual + // RowNumber column — correct even under RowFilter pushdown). It is kept + // on the base source so the position buffer can match base rows to log + // records, then stripped by the buffer before output. The column is NOT + // added to `required_schema`/`merge_schema` — only to the base source's + // physical schema (`base_read_schema` = required + row-index). + let use_position = self.use_record_position(); + // Unused: position-based merging is not wired up here. + let _row_number_col = use_position.then(|| ROW_INDEX_TEMPORARY_COLUMN_NAME.to_string()); + let base_read_schema: SchemaRef = if use_position { + let mut fields: Vec<arrow_schema::FieldRef> = + required_schema.fields().iter().cloned().collect(); + fields.push(Arc::new(arrow_schema::Field::new( Review Comment: Checked this against the code and I don't think it reproduces — the row-index column is *virtual*, not projected. `ParquetBaseFileReader` filters the row-index name out of the projection indices and emits the column anyway (`base_file/parquet.rs`, the `.filter(|name| Some(name.as_str()) != options.row_index_column.as_deref())` in the projection branch, with a comment saying exactly this). So `raw` does carry `ROW_INDEX_TEMPORARY_COLUMN_NAME` and `project_batch_to_schema` finds it rather than hitting the non-nullable-absent branch. Three pieces of evidence: 1. `test_row_index_column_is_returned_alongside_a_projection` in `base_file/parquet.rs` pins the behaviour directly. 2. `engine.rs` has a test that drives `make_base_file_source(false)` — the exact eager path here — with `use_record_position: true`, and asserts the positions come back as `[0, 1, 2]`. 3. The gold sweep runs all 62 fixtures under position merge (`every_fixture_matches_hudi_when_merging_by_record_position`), several of which were written by Hudi with `RECORD_POSITIONS` headers, so the path executes rather than erroring. No change made. Thanks for the close read — the shape of the concern was right, the projection is just not the whole story. ########## crates/core/src/avro_to_arrow/arrow_array_reader.rs: ########## @@ -849,6 +879,30 @@ fn resolve_u8(v: &Value) -> Option<u8> { } } +/// Decode an Avro decimal into the `i128` an Arrow `Decimal128Array` holds. +/// +/// Avro carries a decimal as the unscaled value in big-endian two's-complement +/// bytes, with the scale living in the schema rather than the value — so the +/// scale is applied by the array's type, not here. The bytes are narrower than +/// 16 whenever the value fits in fewer, so they are sign-extended. +/// +/// Returns `None` for a value that is not a decimal, or whose unscaled value is +/// too wide for `i128`; the row then reads as null rather than as a wrong number. +fn resolve_decimal128(v: &Value) -> Option<i128> { + let bytes: &[u8] = match v { + Value::Bytes(b) => b, + Value::Fixed(_, b) => b, + _ => return None, Review Comment: Already resolved later in this stack. `crates/core/src/avro_to_arrow/arrow_array_reader.rs` no longer exists in the squashed tree — delete-block decoding moved to arrow-avro, and `561b781 fix(core): decode Avro decimals, and carry the UTC zone on timestamps` addressed the decimal gap you're describing (the schema side produced `Decimal128` with no matching array-builder arm, so any log block with a decimal column failed the read). The case you name specifically — delete records ordered by `DecimalWrapper` — is covered by the `table_delete_ord_decimal [MorAvro]` fixture, which ships a `gold_options` manifest and matches Hudi's own output across the full option matrix under both reader versions, with no entry on the known-disagreements list. -- 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]
