yihua commented on code in PR #660: URL: https://github.com/apache/hudi-rs/pull/660#discussion_r3787843003
########## crates/core/src/file_group/reader_v2/resolver.rs: ########## @@ -0,0 +1,908 @@ +/* + * 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. + */ + +//! Derives a [`ReaderContext`] from a table's resolved configs. + +use crate::Result; +use crate::config::HudiConfigs; +use crate::config::error::ConfigError; +use crate::config::read::HudiReadConfig; +use crate::config::table::{BaseFileFormatValue, HudiTableConfig}; +use crate::error::CoreError; +use crate::file_group::reader_v2::buffer::spillable_map; +use crate::file_group::reader_v2::reader_context::{CONFIG_MERGE_TYPE, MergeMode, ReaderContext}; +use crate::file_group::reader_v2::record_context::RecordContext; +use crate::file_group::reader_v2::schema_handler::FileGroupReaderSchemaHandler; +use crate::timeline::selector::InstantRange; +use std::collections::HashMap; + +/// Resolve the MOR reader context from `hudi_configs`, which the caller has +/// already merged read options into. +#[allow(dead_code)] +pub(crate) fn resolve_reader_context( + hudi_configs: &HudiConfigs, + has_log_files: bool, +) -> Result<ReaderContext> { + let table_path: String = hudi_configs.get(HudiTableConfig::BasePath)?.into(); + + // Resolved by `Table::prepare_reader_options` for table-level reads. A + // `FileGroupReader` built straight from a base URI never loads the timeline, + // so the caller must supply it; defaulting here would silently widen the read. + let latest_commit_time: String = hudi_configs + .try_get(HudiReadConfig::EndTimestamp)? + .ok_or_else(|| ConfigError::NotFound(HudiReadConfig::EndTimestamp.as_ref().to_string()))? + .into(); + + let merge_mode = resolve_merge_mode(hudi_configs)?; + let base_file_format = BaseFileFormatValue::resolve_from_configs(hudi_configs, None)?; + let instant_range = resolve_instant_range(hudi_configs)?; + let (table_config, hoodie_reader_config) = partition_configs(hudi_configs); + + Ok(ReaderContext { + table_path, + latest_commit_time, + merge_mode: merge_mode.as_ref().to_string(), + base_file_format: base_file_format.as_ref().to_string(), + has_log_files, + instant_range: Some(instant_range), + table_config, + hoodie_reader_config, + should_merge_use_record_position: resolve_use_record_position(hudi_configs)?, + // Only one iterator mode is implemented. + iterator_mode: "ENGINE_RECORD".to_string(), + // Dispatch is on `merge_mode`; the strategy id is carried, not consulted. + merge_strategy_id: String::new(), + // No bootstrap support in this crate. + has_bootstrap_base_file: false, + needs_bootstrap_merge: false, + enable_logical_timestamp_field_repair: false, + // Predicate pushdown into the merge path has no caller here, so no + // filter is installed and the primary-key-safety gate is irrelevant. + row_filter_builder: None, + mor_pk_safe: false, + // The table-version < 8 completion gate needs a timeline the caller + // has not loaded; leaving it unset keeps the gate a no-op. + completion_gate_inputs: None, Review Comment: This is the only production producer of `completion_gate_inputs`, so Gate 3 in `forward_scan_pass1` is inert for every read through this crate — the straddling case it exists for (writer A inflight at T1, writer B commits T2 > T1, so A's uncommitted blocks pass Gate 2) still merges. Not a regression — version 1 has no committed-set check either — but `Table` already holds the timeline this needs. Is wiring it through a follow-up, or deliberately off? _⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality._ ########## crates/core/src/schema/delete.rs: ########## @@ -37,9 +37,54 @@ static DELETE_RECORD_AVRO_SCHEMA_IN_JSON: Lazy<Result<JsonValue>> = Lazy::new(|| .map_err(|e| CoreError::Schema(format!("Failed to parse schema to JSON: {e}"))) }); +/// Union position of `ArrayWrapper`, which carries a list rather than a scalar. +/// Nothing orders records by a list, so it is rejected rather than mapped. +const ARRAY_WRAPPER_POSITION: u32 = 12; + +/// Replace a wrapped ordering value with the primitive inside it. +/// +/// Hudi writes `orderingVal` as a union of per-type wrapper records — `LongWrapper` +/// is a record whose single `value` field is a `long`. The Arrow side wants the +/// primitive, and [`avro_schema_for_delete_record`] narrows the schema to match, +/// so the value has to be unwrapped to agree with it. +/// +/// The narrowed schema has two branches, `[null, <primitive>]`, so the surviving +/// branch is position 1 regardless of where the wrapper sat in the full union. +/// +/// Records that are already primitives pass through: this runs over decoded +/// values, and only the wrapper shape is rewritten. +pub fn unwrap_ordering_value(delete_record: AvroValue) -> Result<AvroValue> { + let AvroValue::Record(mut fields) = delete_record else { + return Err(CoreError::Schema( + "Expected a record for delete record".to_string(), + )); + }; + let Some((_, ordering_val)) = fields.get_mut(2) else { + return Err(CoreError::Schema( + "Delete record has no orderingVal field".to_string(), + )); + }; + if let AvroValue::Union(pos, inner) = ordering_val { + if *pos == ARRAY_WRAPPER_POSITION { Review Comment: `ArrayWrapper` is what multiple ordering fields serialize to (HUDI-9569), and composite ordering is otherwise supported here — so a table with `hoodie.table.ordering.fields=ts,seq` reads fine until its first delete block, then the whole read errors. Could this degrade to natural order, as other unusable ordering values do? All six `table_delete_ord_*` fixtures are single-field, so the sweep can't see it. _⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality._ ########## crates/core/src/file_group/reader_v2/engine.rs: ########## @@ -0,0 +1,2077 @@ +/* + * 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. + +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::BufferType; +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`]. 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: 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, and the +/// row-position column when the merge is by position. +/// +/// The three base reads below differ only in projection, so both are attached in +/// one place — a read that silently lost the filter would return extra rows +/// rather than fail, which is the hard kind of bug to notice, and one that lost +/// the row-position column would fail in the buffer with the column named but +/// not the read that dropped it. +fn base_read_options( + row_filter: Option<RowFilterBuilder>, + use_record_position: bool, +) -> BaseFileReadOptions { + let mut options = BaseFileReadOptions::new(); + if let Some(row_filter) = row_filter { + options = options.with_row_filter(row_filter); + } + if use_record_position { + options = options.with_row_index_column(ROW_INDEX_TEMPORARY_COLUMN_NAME); + } + options +} + +/// `schema` without the internal row-position column. +/// +/// The column belongs to the base read and the position buffer; it is not the +/// table's, so it must not reach a caller. Every schema derived from a base +/// source's own schema goes through here. +fn without_row_index(schema: SchemaRef) -> SchemaRef { + if schema + .column_with_name(ROW_INDEX_TEMPORARY_COLUMN_NAME) + .is_none() + { + return schema; + } + Arc::new(arrow_schema::Schema::new( + schema + .fields() + .iter() + .filter(|f| f.name() != ROW_INDEX_TEMPORARY_COLUMN_NAME) + .cloned() + .collect::<Vec<_>>(), + )) +} + +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 Review Comment: This closing sentence — "Construction still rejects composite *virtual keys* above" — contradicts both the start of this same comment and `test_composite_virtual_keys_accepted_at_construction` below; the rejection it refers to no longer exists. _⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality._ ########## crates/core/src/file_group/reader_v2/buffer/position_based.rs: ########## @@ -0,0 +1,1491 @@ +/* + * 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. Some items exist only for parity +//! with the Java reader and have no caller here. + +//! Position-based file-group record buffer. +//! +//! Mirrors Java `org.apache.hudi.common.table.read.buffer.PositionBasedFileGroupRecordBuffer`, +//! which `extends KeyBasedFileGroupRecordBuffer`. Instead of keying the merge +//! map by record key, it keys it by each log record's **position in the base +//! file** — the row index recorded in the log block's `RECORD_POSITIONS` header. +//! Base rows are matched to log records by their physical position (read from +//! the synthetic row-index column attached to the base read), which is cheaper +//! than key extraction + string hashing and is the layout Spark writes when +//! `hoodie.merge.use.record.positions` is enabled. +//! +//! ## Composition +//! +//! Java extends the key-based buffer; here we **compose** it +//! ([`inner`](PositionBasedFileGroupRecordBuffer::inner)) and reuse its +//! vectorized/scalar merge kernels via the [`BaseMatch`] parameter, so the merge +//! logic (winner selection, keep-mask, reconcile, update stats, spill map) is +//! shared rather than duplicated. Positions are encoded as decimal-string map +//! keys in the existing `String`-keyed spillable map; the map holds a single key +//! domain at any instant — positions before fallback, record keys after. +//! +//! ## Fallback to key-based +//! +//! When a log block has no valid positions (missing / mismatched base-file +//! instant, or an empty bitmap), the buffer permanently switches to key-based +//! merge for the rest of the scan: it re-keys every already-buffered record from +//! its position to its record key and delegates subsequent processing to the +//! key-based super methods. This mirrors Java `fallbackToKeyBasedBuffer`. + +use std::collections::HashMap; +use std::sync::Arc; + +use arrow_array::RecordBatch; +use arrow_schema::SchemaRef; + +use crate::Result; +use crate::error::CoreError; +use crate::file_group::log_file::log_block::{LogBlock, LogBlockContent}; +use crate::file_group::reader_v2::buffer::key_based::{BaseMatch, KeyBasedFileGroupRecordBuffer}; +use crate::file_group::reader_v2::buffer::record_positions::{ + base_row_position_array, decode_record_positions, +}; +use crate::file_group::reader_v2::buffer::row_extraction::records_to_batch; +use crate::file_group::reader_v2::buffer::spillable_map::{SpillConfig, SpillableRecordMap}; +use crate::file_group::reader_v2::buffer::{BufferType, HoodieFileGroupRecordBuffer}; +use crate::file_group::reader_v2::buffered_record::{BufferedRecord, DeleteRecord}; +use crate::file_group::reader_v2::merge_iterator::DEFAULT_BATCH_SIZE; +use crate::file_group::reader_v2::reader_context::ReaderContext; +use crate::file_group::reader_v2::update_processor::UpdateStats; + +/// A merge buffer that matches base rows to log records by base-file row +/// position. See the module docs. Wraps a [`KeyBasedFileGroupRecordBuffer`] and +/// falls back to it when positions are unavailable. +#[derive(Debug)] +pub struct PositionBasedFileGroupRecordBuffer { + /// The key-based buffer whose merge kernels/state this reuses. After a + /// fallback its records map holds record-key entries (position entries + /// otherwise). + inner: KeyBasedFileGroupRecordBuffer, + /// Commit/instant time of the base file being merged. Positions in a log + /// block are only usable when the block's + /// `BASE_FILE_INSTANT_TIME_OF_RECORD_POSITIONS` header equals this. + base_file_instant_time: String, + /// Whether to still merge by position. Set true at construction; flipped + /// false permanently on the first fallback (mirrors Java + /// `readerContext.setShouldMergeUseRecordPosition(false)`). + should_use_position: bool, + /// Set during fallback when a position-only delete (a delete record without + /// a usable record key) could not be re-keyed and must keep being matched by + /// position. Effectively unreachable in hudi-rs (delete blocks always carry + /// record keys), but tracked for parity with Java's hybrid strategy. + needs_hybrid_strategy: bool, + /// Position-only deletes retained across a fallback (see + /// [`needs_hybrid_strategy`](Self::needs_hybrid_strategy)). + hybrid_deletes: HashMap<u64, BufferedRecord>, +} + +impl PositionBasedFileGroupRecordBuffer { + /// Construct a position-based buffer. `base_file_instant_time` is the commit + /// time of the base file in this file slice (Java passes + /// `baseFile.getCommitTime()`), used to validate that a block's positions + /// were computed against this base file. + pub fn new( + reader_context: Arc<ReaderContext>, + merge_mode: String, + emit_delete: bool, + base_file_instant_time: String, + ) -> Result<Self> { + let inner = KeyBasedFileGroupRecordBuffer::new(reader_context, merge_mode, emit_delete)?; + Ok(Self { + inner, + base_file_instant_time, + should_use_position: true, + needs_hybrid_strategy: false, + hybrid_deletes: HashMap::new(), + }) + } + + /// The map key for a base-file position: its decimal string. Positions and + /// record keys never share the map at the same time (positions before + /// fallback, record keys after), so the string domains do not collide. + fn position_key(position: u64) -> String { + position.to_string() + } + + /// Extract the block's record positions, validating the base-file instant. + /// + /// Mirrors Java `extractRecordPositions`: returns `Ok(None)` (→ fallback) + /// when the block's `BASE_FILE_INSTANT_TIME_OF_RECORD_POSITIONS` header is + /// missing / empty / different from this file group's base file instant, or + /// when the `RECORD_POSITIONS` bitmap is missing / empty. + fn extract_positions(&self, block: &LogBlock) -> Result<Option<Vec<u64>>> { + match block.base_file_instant_time_of_positions() { + Some(t) if !t.is_empty() && t == self.base_file_instant_time => {} + _ => { + log::debug!( + "[PositionBasedBuffer] falling back: block base-file instant of positions \ + absent or != {}", + self.base_file_instant_time, + ); + return Ok(None); + } + } + match block.record_positions_header() { + Some(encoded) if !encoded.is_empty() => { + // Writer-side invariant: `decode_record_positions` yields the + // Roaring64 bitmap's positions in ASCENDING order, and the + // caller zips them index-by-index with the block's records in + // file order. This matches Java's + // `PositionBasedFileGroupRecordBuffer` + // (`recordPositions.get(recordIndex++)`) and is only correct + // because the Hudi writer emits records within a block in + // ascending base-position order. + let positions = decode_record_positions(encoded)?; + if positions.is_empty() { + Ok(None) + } else { + Ok(Some(positions)) + } + } + _ => Ok(None), + } + } + + /// Permanently switch to key-based merge, re-keying every already-buffered + /// record from its position to its record key. Mirrors Java + /// `fallbackToKeyBasedBuffer`. + fn fallback_to_key_based(&mut self) -> Result<()> { + if !self.should_use_position { + return Ok(()); + } + self.should_use_position = false; + + // Rebuild the (possibly-spilled) position-keyed map as a record-key-keyed + // map. `drain_iter` yields values; the position keys are dropped. + // + // A record's `record_key` field is NOT trustworthy here: the spill tier + // reconstructs a drained record's `record_key` from its MAP key (see + // `RocksDbDiskMap::record_from_entry`), which in this buffer is the + // POSITION string, not the record key. So for data records the true key + // is re-extracted from the payload row itself. A delete tombstone has no + // payload to re-extract from — its `record_key` is only correct if the + // tombstone never spilled (hudi-rs delete blocks always carry record + // keys, and `process_next_deleted_record` preserves them in the + // in-memory tier). A delete without a usable record key cannot be + // re-keyed — Java keeps it position-keyed under the hybrid strategy, but + // that requires the position, which `drain_iter` does not surface; fail + // loudly rather than silently drop a delete (which would resurrect a + // deleted row). A SPILLED position-keyed delete comes back with its + // position string as `record_key` and is indistinguishable from a real + // numeric key — persisting the record key through the spill tier is the + // real fix (tracked as a follow-up outside this buffer). Review Comment: This looks already fixed: `record_from_entry` in `spillable_map.rs` persists and restores a delete's real key via `DiskLoc::Delete { record_key }`, with a comment naming this exact hazard — and I couldn't reproduce the mis-key described here. Could the comment be updated so the next reader doesn't chase a fixed bug? _⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality._ ########## crates/core/src/file_group/reader_v2/reader_context.rs: ########## @@ -0,0 +1,445 @@ +/* + * 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. + +//! Mirrors Java's `HoodieReaderContext<T>` — the engine-agnostic reader context +//! that flows through the entire file group reader call stack. +//! +//! In Java, `HoodieReaderContext<T>` is an engine-specific object (Spark, Flink) +//! that carries merge mode, instant range, schema handler, etc. In Rust, we work +//! directly with Arrow `RecordBatch`es, so this is a plain data struct carrying +//! the structured configuration that the reader stack needs. + +use super::record_context::RecordContext; +use super::schema_handler::FileGroupReaderSchemaHandler; +use crate::config::table::HudiTableConfig; +use crate::storage::RowFilterBuilder; +use crate::timeline::selector::InstantRange; +use std::collections::{HashMap, HashSet}; + +/// Owned inputs for the Gate-3 completed/inflight check (C-INFLIGHT-DELTA). +/// +/// Mirrors the SET check Java's `BaseHoodieLogRecordReader.scanInternalV1` performs for +/// `tableVersion < 8`: a data/delete block is skipped when its instant is in +/// `filterInflights()` OR not in `filterCompletedInstants()` (with archived instants treated as +/// committed). Held by the reader so it can be built once from the active timeline and borrowed for +/// each scan. +/// +/// The gate is only meaningful for tables with a v1 timeline layout (table version < 8). For v8+ +/// tables the per-delta-commit log files are already excluded at the completion-time file-slice +/// level, so callers leave this `None` (see `forward_scan_pass1`), preserving prior behavior. +/// +/// Lives here rather than with the scan that consumes it: [`ReaderContext`] holds it as a field, so +/// defining it alongside the scanner would make the context and the scanner import each other. +#[derive(Debug, Clone, Default)] +pub struct CompletionGateInputs { + /// Active completed-commit instant times (Java `filterCompletedInstants()`). + pub completed_instants: HashSet<String>, + /// Active inflight/requested instant times (Java `filterInflights()`). + pub inflight_instants: HashSet<String>, + /// First active-timeline instant. An instant strictly before it is archived; archival only + /// removes completed instants, so an archived instant is committed. `None` disables the + /// archived fallback (every candidate must then be in `completed_instants`). + pub archived_boundary: Option<String>, +} + +/// Config key selecting how a slice's base and log records are combined. +/// `skip_merge` asks for them unmerged, which this reader does not implement and +/// refuses in the buffer loader. Hudi's own spelling, so it is looked up by name +/// rather than modelled as a [`HudiReadConfig`](crate::config::read::HudiReadConfig). +pub const CONFIG_MERGE_TYPE: &str = "hoodie.datasource.merge.type"; + +/// Reader context that flows through the file group reader call stack. +/// +/// Mirrors Java's `HoodieReaderContext<T>`, carrying structured reader +/// configuration instead of raw config key-value maps. +/// +/// ## Java counterpart +/// +/// | Java field / method | Rust field | +/// |--------------------------------------------|-----------------------------------------| +/// | `readerContext.getTablePath()` | `table_path` | +/// | `readerContext.getLatestCommitTime()` | `latest_commit_time` | +/// | `readerContext.getMergeMode()` | `merge_mode` | +/// | `readerContext.getInstantRange()` | `instant_range` | +/// | `readerContext.getRecordContext().format()` | `base_file_format` | +/// | `readerContext.getHasLogFiles()` | `has_log_files` | +/// | `readerContext.getRecordContext()` | `record_context` | +/// | `readerContext.getSchemaHandler()` | `schema_handler` | +/// | `metaClient.getTableConfig()` (config map) | `table_config` | +/// | `props` (hoodie reader config overrides) | `hoodie_reader_config` | +#[derive(Clone)] +pub struct ReaderContext { + pub table_path: String, + pub latest_commit_time: String, + pub base_file_format: String, + pub has_log_files: bool, + pub has_bootstrap_base_file: bool, + pub needs_bootstrap_merge: bool, + pub should_merge_use_record_position: bool, + pub enable_logical_timestamp_field_repair: bool, + pub iterator_mode: String, + pub merge_mode: String, + pub merge_strategy_id: String, + pub instant_range: Option<InstantRange>, + /// The engine-specific record context for record-level operations. + /// + /// Mirrors Java's `HoodieReaderContext.recordContext` field. + /// In Java this is a persistent mutable field set at construction and + /// shared across all consumers. In Rust it is set once and shared via + /// `Arc<ReaderContext>`. + /// + /// Constructed from `table_config` + `partition_path`, mirroring Java's + /// `RecordContext(tableConfig, typeConverter)`. + pub record_context: RecordContext, + /// Schema management for the read pipeline. + /// + /// Mirrors Java's `HoodieReaderContext.schemaHandler` field + /// (`FileGroupReaderSchemaHandler<T>`). + pub schema_handler: FileGroupReaderSchemaHandler, + pub table_config: HashMap<String, String>, + /// Per-read overrides. Populated by + /// [`resolve_reader_context`](crate::file_group::reader_v2::resolver::resolve_reader_context) + /// from the `hoodie.read.*` keys plus the individually-named ones its + /// `READER_CONFIG_KEYS` lists — anything a consumer looks up here must + /// appear in one of those two, or it reads as unset. + pub hoodie_reader_config: HashMap<String, String>, + /// Optional parquet `RowFilter` builder for predicate pushdown into base + /// parquet files and parquet-format log blocks. Set by the FFI bridge from Review Comment: There are ~95 comments across `reader_v2/` referencing infrastructure that isn't in this repo — "the FFI bridge", "decoded substrait predicate", "Velox consumes the stream directly", `cpp/src/predicate.rs` (cpp/src has only lib.rs and util.rs), "#76/#95 review". For `row_filter_builder` and `completion_gate_inputs` it's more than tidiness: their only documented producer doesn't exist here, so inert-by-design and inert-by-accident read the same. Could these be rewritten against this crate's actual call graph? _⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality._ -- 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]
