hudi-agent commented on code in PR #663: URL: https://github.com/apache/hudi-rs/pull/663#discussion_r3770823951
########## crates/core/src/file_group/reader_v2/memory_limit_tests.rs: ########## @@ -0,0 +1,291 @@ +/* + * 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. + */ + +//! Spill-budget config propagation (ENG-45062 / I-33 / G-16). +//! +//! `SpillConfig::from_config` derives the in-memory spill threshold from the +//! reader config map only (`reader_context.hoodie_reader_config`, read at +//! `crates/core/src/file_group/reader/buffer/key_based.rs`). The gluten adapter +//! forwards `hoodie.memory.merge.max.size` (the computed `maxMemoryPerCompaction`) +//! into that map, so the operator/computed budget reaches spill sizing. +//! +//! Regression guard for I-33: gluten previously put the budget only in the props +//! map, which hudi-rs never sees, so `SpillConfig` fell back to the 1 GiB default +//! on every read (in-memory threshold pinned at ~779 MiB regardless of the +//! operator's setting -> OOM risk). This asserts `from_config` honors the budget +//! when it is present (as gluten now forwards it) and only defaults when it is +//! genuinely absent. +//! +//! ## Peak-memory hard cap (ENG-44436 / 44437) +//! +//! The second group of tests here covers the hudi-rs-side foundation for the +//! velox memory-reservation work: a queryable current-footprint getter +//! ([`SpillableRecordMap::current_in_memory_bytes`]) and a configurable HARD +//! peak cap ([`CONFIG_MAX_PEAK_MEMORY`]) that fails loudly with +//! [`CoreError::MemoryLimitExceeded`] instead of letting the executor OOM. These +//! are cargo-only (no gluten/velox bundle); the FFI + velox reservation wiring +//! is a later increment. +//! +//! Run: `cargo test -p hudi-core --test nonfunctional_gaps_repro -- --nocapture` + +use std::collections::HashMap; +use std::sync::Arc; + +use crate::error::CoreError; +use crate::file_group::reader_v2::buffer::spillable_map::{ + CONFIG_MAX_PEAK_MEMORY, CONFIG_MERGE_MAX_SIZE, DEFAULT_MERGE_MAX_SIZE_BYTES, DiskMapType, + ENTRY_OVERHEAD_BYTES, SpillConfig, SpillableRecordMap, +}; +use crate::file_group::reader_v2::buffered_record::{BufferedRecord, OrderingValue}; +use arrow_array::{Int32Array, RecordBatch, StringArray}; +use arrow_schema::{DataType, Field, Schema}; + +/// The merge-type key gluten forwards in `hoodieReaderConfig` +/// (`HoodieReaderConfig.MERGE_TYPE.key()`). +const MERGE_TYPE_KEY: &str = "hoodie.datasource.merge.type"; +/// Default merge type gluten resolves when unset (`REALTIME_PAYLOAD_COMBINE`). +const MERGE_TYPE_PAYLOAD_COMBINE: &str = "payload_combine"; + +const MIB: u64 = 1024 * 1024; + +/// Threshold `SpillConfig` derives from a given `hoodie_reader_config` map. +fn threshold_bytes(config: &HashMap<String, String>) -> u64 { + SpillConfig::from_config(config) + .expect("SpillConfig::from_config") + .max_in_memory_size +} Review Comment: π€ nit: could you add the `test_` prefix here (and to the other `oom_*` functions below)? The project convention is `test_<function>_<scenario>_<expected>`, and names like `i33_merge_budget_honored_when_forwarded` / `oom_peak_cap_rejects_oversized_insertion_loudly` don't show up clearly when filtering test output with `--test-filter test_`. <sub><i>β οΈ AI-generated; verify before applying. React π/π to flag quality.</i></sub> ########## crates/core/src/file_group/reader_v2/update_processor.rs: ########## @@ -0,0 +1,190 @@ +/* + * 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. + */ + +//! Ported from the merge-on-read reader. Nothing consumes it yet, so its +//! items are unreachable from the crate's call graph until the reader wires in. +#![allow(dead_code)] + +//! Mirrors `org.apache.hudi.common.table.read.UpdateProcessor`. Review Comment: π€ nit: having two inner `//!` doc blocks in a row is a bit odd β the second one (starting at line 25) overwrites the first in rustdoc. Could you merge them into a single `//!` block at the top, keeping both the porting note and the Java mirror description? <sub><i>β οΈ AI-generated; verify before applying. React π/π to flag quality.</i></sub> ########## crates/core/src/file_group/reader_v2/input_split.rs: ########## @@ -0,0 +1,258 @@ +/* + * 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. + */ + +//! Ported from the merge-on-read reader. Nothing consumes it yet, so its +//! items are unreachable from the crate's call graph until the reader wires in. +#![allow(dead_code)] + +//! Mirrors `org.apache.hudi.common.table.read.InputSplit`. +//! +//! Describes the data to be read from a file group: an optional base file, +//! a list of log files, and the partition path context. + +use crate::file_group::log_file::LogFile; +use std::str::FromStr; + +/// Describes the input data for a file group read. +/// +/// Carries the base file (if any), the list of log files to scan, +/// the partition path, and the byte range to read from the base file. +#[derive(Debug, Clone)] +pub struct InputSplit { + /// Path to the base file (relative to table root), if present. + pub base_file_path: Option<String>, + + /// Commit time of the base file, if present. + pub base_file_commit_time: Option<String>, + + /// Relative paths to log files to scan. + pub log_file_paths: Vec<String>, + + /// Partition path for this file group (e.g. "year=2024/month=01"). + pub partition_path: String, + + /// Byte offset to start reading from in the base file. + pub start: i64, + + /// Number of bytes to read from the base file. + pub length: i64, +} + +/// CDC log-file suffix. Mirrors Java's `HoodieCDCUtils.CDC_LOGFILE_SUFFIX` (".cdc"). +/// CDC log files carry change-data-capture blocks and must be excluded from a +/// normal snapshot read β gold drops them in `InputSplit`'s constructor +/// (`InputSplit.java:57`). +const CDC_LOGFILE_SUFFIX: &str = ".cdc"; + +impl InputSplit { + pub fn new( + base_file_path: Option<String>, + base_file_commit_time: Option<String>, + log_file_paths: Vec<String>, + partition_path: String, + ) -> Self { + // Filter out CDC log files, then sort ascending by + // deltaCommitTime β logVersion β writeToken. This mirrors Java's + // InputSplit constructor (InputSplit.java:56-58), which sorts via + // HoodieLogFile.getLogFileComparator() and filters file names ending in + // HoodieCDCUtils.CDC_LOGFILE_SUFFIX. The C++ side may send log files in + // descending order (from FileSlice's reverse TreeSet), so we re-sort. + let log_file_paths = Self::filter_cdc_log_files(log_file_paths); + let log_file_paths = Self::sort_log_file_paths(log_file_paths); + Self { + base_file_path, + base_file_commit_time, + log_file_paths, + partition_path, + start: 0, + length: -1, + } + } + + /// Drop CDC log files from the scan list. + /// + /// Mirrors Java's `InputSplit` constructor filter + /// (`InputSplit.java:57`): `!logFile.getFileName().endsWith(CDC_LOGFILE_SUFFIX)`. + /// The match is on the file-name portion (after the last `/`), matching gold's + /// `getFileName()` semantics. + fn filter_cdc_log_files(paths: Vec<String>) -> Vec<String> { + paths + .into_iter() + .filter(|p| { + let name = p.rsplit('/').next().unwrap_or(p); + !name.ends_with(CDC_LOGFILE_SUFFIX) + }) + .collect() + } + + /// Sort log file paths ascending by deltaCommitTime β logVersion β writeToken. + /// + /// Mirrors Java's `InputSplit` constructor which sorts via + /// `HoodieLogFile.getLogFileComparator()`. + fn sort_log_file_paths(mut paths: Vec<String>) -> Vec<String> { + if paths.len() <= 1 { + return paths; + } + paths.sort_by(|a, b| { + let name_a = a.rsplit('/').next().unwrap_or(a); + let name_b = b.rsplit('/').next().unwrap_or(b); + match (LogFile::from_str(name_a), LogFile::from_str(name_b)) { + (Ok(lf_a), Ok(lf_b)) => lf_a.cmp(&lf_b), + _ => a.cmp(b), // fallback to lexicographic if parsing fails + } + }); + paths + } Review Comment: π€ nit: the doc comment says "no base file and no log files" but the implementation only checks `!self.has_log_files()` β a base-only split returns `true` here too. Could you correct the doc (e.g. "Returns true when there are no log files to merge") and maybe rename to `is_base_only` or `needs_no_merge` to match? <sub><i>β οΈ AI-generated; verify before applying. React π/π to flag quality.</i></sub> ########## 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: π€ `resolve_decimal128` handles `Value::Bytes`/`Value::Fixed`, but `from_avro_datum` decodes a decimal-logical-type field to `Value::Decimal` β and since the `Decimal128` arm is only reached for `Schema::Decimal` fields, every real decimal value falls to `_ => return None` and reads as null. Could you add a `Value::Decimal(d)` arm (its big-endian bytes come from `Vec::<u8>::try_from(d)`)? The same nulling hits delete records ordered by `DecimalWrapper` (bytes + logicalType decimal). <sub><i>β οΈ AI-generated; verify before applying. React π/π to flag quality.</i></sub> ########## 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: π€ When `use_record_position` is true, `base_read_schema` gains a non-nullable `ROW_INDEX_TEMPORARY_COLUMN_NAME`, but the base read only projects the fileβ©required intersection, so that column is never present in `raw`. `project_batch_to_schema` (line 894, and the streaming map at ~930) then hits its non-nullable-absent branch and errors the whole read. Since the comment below says the row number is "dropped rather than faked," did you mean to leave the column out of `base_read_schema` too (or make it nullable) until position merge is wired up? <sub><i>β οΈ AI-generated; verify before applying. React π/π to flag quality.</i></sub> ########## crates/core/src/file_group/reader_v2/update_processor.rs: ########## @@ -0,0 +1,190 @@ +/* + * 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. + */ + +//! Ported from the merge-on-read reader. Nothing consumes it yet, so its +//! items are unreachable from the crate's call graph until the reader wires in. +#![allow(dead_code)] + +//! Mirrors `org.apache.hudi.common.table.read.UpdateProcessor`. +//! +//! Strategy interface for processing record updates during the +//! base-file-vs-log merge phase. The default implementation passes through the +//! merged record and increments the read-stats counters (inserts / updates / +//! deletes), mirroring gold's `StandardUpdateProcessor`. + +use super::buffered_record::BufferedRecord; +use crate::Result; +use std::sync::atomic::{AtomicU64, Ordering}; + +/// Per-merge counts of insert / update / delete operations. +/// +/// Mirrors the `numInserts` / `numUpdates` / `numDeletes` that gold's +/// `StandardUpdateProcessor` increments on `HoodieReadStats`. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct UpdateStats { + pub num_inserts: u64, + pub num_updates: u64, + pub num_deletes: u64, +} + +/// Strategy for processing record updates during merge iteration. +/// +/// Mirrors Java's `UpdateProcessor<T>` interface. +/// +/// Created by factory method based on merge mode and configuration. +/// Wrapped optionally in a `CallbackProcessor` for update callbacks. +pub trait UpdateProcessor: Send + Sync + std::fmt::Debug { + /// Process an update (base record merged with log record). + /// + /// `previous_record` is the pre-merge base record when this record came + /// from a base+log merge (the update path), and `None` when it came from a + /// log-only record (the insert path) β matching gold's + /// `processUpdate(recordKey, previousRecord, mergedRecord, isDelete)`. + /// + /// Returns the record to emit, or `None` to skip. + fn process_update( + &self, + record_key: &str, + previous_record: Option<&BufferedRecord>, + merged_record: &BufferedRecord, + is_delete: bool, + ) -> Result<Option<BufferedRecord>>; + + /// Snapshot the accumulated insert / update / delete counts. + /// + /// Mirrors the counters gold's `StandardUpdateProcessor` writes onto + /// `HoodieReadStats` (`incrementNumInserts/Updates/Deletes`). The reader + /// drains these into its `HoodieReadStats` after the merge completes. + fn read_stats_counts(&self) -> UpdateStats { + UpdateStats::default() + } +} + +/// Default update processor: passes through the merged record and counts +/// inserts / updates / deletes. +/// +/// Corresponds to Java's `StandardUpdateProcessor<T>` +/// (`UpdateProcessor.java:75-120`). Counting uses interior mutability +/// (`AtomicU64`) because the processor is shared behind a `&self` trait +/// method, matching the way the buffer iterator calls `processUpdate` while +/// holding the processor by reference. +/// +/// NOTE: `emit_delete` row-emission is intentionally NOT implemented here. +/// Gold's `emitDeletes` path emits a synthesized delete row +/// (`recordContext.getDeleteRow`) and tags `HoodieOperation`; that is a larger +/// change (delete-row synthesis + operation tagging) gated loudly at the reader +/// construction boundary (`HoodieFileGroupReader::new`). See the gaps registry. +#[derive(Debug, Default)] +pub struct StandardUpdateProcessor { + num_inserts: AtomicU64, + num_updates: AtomicU64, + num_deletes: AtomicU64, +} + +impl StandardUpdateProcessor { + pub fn new() -> Self { + Self::default() + } +} + +impl UpdateProcessor for StandardUpdateProcessor { + fn process_update( + &self, + _record_key: &str, + previous_record: Option<&BufferedRecord>, + merged_record: &BufferedRecord, + is_delete: bool, + ) -> Result<Option<BufferedRecord>> { + // Mirrors gold StandardUpdateProcessor.processUpdate (UpdateProcessor.java:88-119). + if is_delete { + // readStats.incrementNumDeletes(); emitDeletes is gated off upstream, + // so a delete is always dropped from the output here. + self.num_deletes.fetch_add(1, Ordering::Relaxed); + return Ok(None); + } + // handleNonDeletes: prev present β update; prev absent β insert. + if previous_record.is_some() { + self.num_updates.fetch_add(1, Ordering::Relaxed); + } else { + self.num_inserts.fetch_add(1, Ordering::Relaxed); + } + Ok(Some(merged_record.clone())) Review Comment: π€ nit: the `_emit_deletes` parameter is always `false` in practice (as the doc note explains), so it currently has no effect on the returned type. Could you rename it to `emit_deletes` and add a `todo!()` or `debug_assert!(!emit_deletes)` to make the unimplemented branch explicitly visible, rather than silently ignoring it? <sub><i>β οΈ AI-generated; verify before applying. React π/π to flag quality.</i></sub> -- 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]
