yihua commented on code in PR #660: URL: https://github.com/apache/hudi-rs/pull/660#discussion_r3788042170
########## crates/core/src/file_group/reader_v2/log_record_reader.rs: ########## @@ -0,0 +1,1805 @@ +/* + * 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 `org.apache.hudi.common.table.log.BaseHoodieLogRecordReader`. +//! +//! The 2-pass log scanning engine that reads log blocks, filters them through +//! 5 gates, resolves compaction, and dispatches to the record buffer. +//! +//! ## 3-pass algorithm (matching Java's `scanInternal`): +//! +//! **Pass 1 — Forward scan with 5 gates:** +//! - Gate 1: Corrupt → skip +//! - Gate 2: Future (instant > latestInstantTime) → skip +//! - Gate 3: Completed/inflight → skip uncommitted/inflight data/delete blocks (table version < 8; +//! applied only when the completion sets are supplied) +//! - Gate 4: Instant range filter → skip +//! - Rollback command → remove target instant +//! - Data/Delete → add to `instant_to_blocks_map` +//! +//! **Pass 2 — Reverse iteration with compaction resolution:** +//! Iterate instants newest→oldest, `Collections.reverse(logBlocks)` then `addLast`. +//! Result: deque has latest-first at head, earliest-last at tail. +//! +//! **Pass 3 — `processQueuedBlocksForInstant`:** +//! Drain deque via `pollLast` (tail-first) = oldest instant processed first. + +use crate::Result; +use crate::file_group::log_file::log_block::{BlockMetadataKey, BlockType, LogBlock}; +use crate::file_group::log_file::reader::LogFileReader; +use crate::file_group::reader_v2::buffer::HoodieFileGroupRecordBuffer; +use crate::file_group::reader_v2::profiling::profile_once; +use crate::file_group::reader_v2::reader_context::{CompletionGateInputs, ReaderContext}; +use crate::storage::Storage; +use crate::timeline::selector::InstantRange; +use std::collections::{HashMap, HashSet, VecDeque}; +use std::sync::Arc; + +// ========================================================================= +// Pass 1 result +// ========================================================================= + +/// Output of Pass 1 (forward scan with 5 gates). +#[derive(Debug)] +pub struct Pass1Result { + /// Instants in first-seen order (no duplicates). + pub ordered_instants_list: Vec<String>, + /// instant time → blocks for that instant (in file-read order). + pub instant_to_blocks_map: HashMap<String, Vec<LogBlock>>, + /// Stats + pub total_log_blocks: u64, + pub total_corrupt_blocks: u64, + pub total_rollbacks: u64, +} + +/// Borrowed view of [`CompletionGateInputs`] used inside a single scan pass. +pub struct CompletionGate<'a> { + completed_instants: &'a HashSet<String>, + inflight_instants: &'a HashSet<String>, + archived_boundary: Option<&'a str>, +} + +impl<'a> CompletionGate<'a> { + /// Build a borrowing gate from owned inputs. + pub fn new(inputs: &'a CompletionGateInputs) -> Self { + Self { + completed_instants: &inputs.completed_instants, + inflight_instants: &inputs.inflight_instants, + archived_boundary: inputs.archived_boundary.as_deref(), + } + } + + /// Whether a data/delete block at `instant_time` is admitted (committed and not inflight). + /// + /// Mirrors `TimelineView::is_committed` (completed-set membership OR older than the archived + /// boundary) plus the explicit inflight exclusion, matching Java's scanInternalV1 filter. + /// + /// # No-positive-information fallback + /// + /// When the gate carries NO positive completion information — an empty + /// `completed_instants` set AND no `archived_boundary` — it cannot establish + /// committedness for any instant, so it admits everything (a no-op) rather + /// than excluding all blocks. Java always builds the gate from a non-empty + /// active timeline (`filterCompletedInstants()`), so an empty set here means + /// the inputs were not populated by the caller (e.g. the FFI bridge forwarded + /// an empty list). Excluding on that basis would silently drop EVERY log delta + /// — including committed ones — and return base-file-only data (the + /// C-INFLIGHT silent-wrong: a committed later delta wrongly dropped). Deferring + /// to the other gates preserves the pre-gate behavior for a mis-wired gate; a + /// correctly-populated gate is unaffected because its completed set is non-empty. + fn admits(&self, instant_time: &str) -> bool { + if self.is_unpopulated() { + return true; + } + let is_committed = self.completed_instants.contains(instant_time) + || self + .archived_boundary + .is_some_and(|boundary| instant_time < boundary); + is_committed && !self.inflight_instants.contains(instant_time) + } + + /// Whether the gate carries NO positive completion information — an empty + /// `completed_instants` set AND no `archived_boundary`. Such a gate cannot establish + /// committedness for any instant, so [`admits`](Self::admits) degrades to admit-all. + /// + /// When the gate was supplied deliberately (the caller passes `Some` only for a table + /// version < 8 snapshot), being unpopulated is a mis-wire signal, not a genuine empty + /// timeline: Java always builds the gate from a non-empty active timeline + /// (`filterCompletedInstants()`), and a v1 snapshot always has >= 1 completed instant. + /// `forward_scan_pass1` warns once when this holds so a gate whose inputs were dropped + /// across the FFI boundary is diagnosable rather than a silent no-op. + fn is_unpopulated(&self) -> bool { + self.completed_instants.is_empty() && self.archived_boundary.is_none() + } +} + +/// Run Pass 1: forward scan with 5 gates. +/// +/// Processes a flat list of blocks (from all log files, in file-read order) +/// and classifies them into `instant_to_blocks_map` + `ordered_instants_list`. +/// +/// ## 5 Gates (per-block): +/// 1. Corrupt → skip +/// 2. Future (instant > latest_instant_time) → skip +/// 3. Completed/inflight → skip data/delete blocks whose instant is uncommitted or inflight +/// (C-INFLIGHT-DELTA). Applied only when `completion_gate` is `Some` — callers pass it for +/// table version < 8 and leave it `None` for v8+ (and when no timeline is available), which +/// preserves the prior behavior of admitting such blocks. +/// 4. Instant range → skip if not in range +/// 5. Rollback command → remove target instant from maps +pub fn forward_scan_pass1( + all_blocks: Vec<LogBlock>, + latest_instant_time: &str, + instant_range: &Option<InstantRange>, + timezone: &str, + completion_gate: Option<&CompletionGate>, +) -> Result<Pass1Result> { + let mut target_rollback_instants: HashSet<String> = HashSet::new(); + let mut instant_to_blocks_map: HashMap<String, Vec<LogBlock>> = HashMap::new(); + let mut ordered_instants_list: Vec<String> = Vec::new(); + let mut total_log_blocks: u64 = 0; + let mut total_corrupt_blocks: u64 = 0; + let mut total_rollbacks: u64 = 0; + + log::debug!( + "[Pass1] forward_scan: {} total blocks, latest_instant_time={}, has_instant_range={}", + all_blocks.len(), + latest_instant_time, + instant_range.is_some(), + ); + + // The completion gate was supplied (table version < 8 snapshot) but carries no positive + // completion info — its inputs were not populated across the FFI boundary. Rather than + // silently degrade to admit-all (Gate 3 becomes a no-op, so a straddling uncommitted delta + // could be merged), warn once so the mis-wire is diagnosable. Behavior is unchanged (still + // fail-open); this only makes the condition visible. + if completion_gate.is_some_and(CompletionGate::is_unpopulated) { + log::warn!( + "[Pass1] completion gate supplied but unpopulated (empty completed set, no archived \ + boundary): Gate 3 (C-INFLIGHT) admits all delta blocks. A table version < 8 snapshot \ + always has >= 1 completed instant, so this indicates the gate inputs were dropped \ + across the FFI boundary. latest_instant_time={latest_instant_time}" + ); + } + + for block in all_blocks { + total_log_blocks += 1; + + // Gate 1: Corrupt blocks → skip + if block.block_type == BlockType::Corrupted { + log::debug!("[Pass1] Gate1: corrupt block #{total_log_blocks} skipped"); + total_corrupt_blocks += 1; + continue; + } + + let instant_time = match block.instant_time() { + Ok(t) => t.to_string(), + Err(_) => { + log::debug!("[Pass1] block #{total_log_blocks} has no instant time, skipping"); + continue; + } + }; + + // Gate 2: Future blocks → skip (instant > latestInstantTime) + if block.block_type != BlockType::Command && instant_time.as_str() > latest_instant_time { + log::debug!( + "[Pass1] Gate2: future block #{total_log_blocks} instant={instant_time} > {latest_instant_time}, skipped" + ); + continue; + } + + // Gate 3: Completed/inflight check (Java scanInternalV1, tableVersion < 8). + // Applied only when the caller supplies the timeline sets (None for v8+ or + // when no timeline is available). A data/delete block whose instant is not + // committed, or is inflight, is skipped -- without this an uncommitted + // "straddling" instant (time < latest completed, no in-log rollback block) + // is wrongly merged (C-INFLIGHT-DELTA). Command (rollback) blocks are never + // gated here so they can still remove their target instant below. + if block.block_type != BlockType::Command + && let Some(gate) = completion_gate + && !gate.admits(&instant_time) + { + log::debug!( + "[Pass1] Gate3: block #{total_log_blocks} instant={instant_time} uncommitted/inflight, skipped" + ); + continue; + } + + // Gate 4: Instant range filter (not for command blocks) + if block.block_type != BlockType::Command + && let Some(range) = instant_range + && range.not_in_range(&instant_time, timezone)? + { + log::debug!( + "[Pass1] Gate4: block #{total_log_blocks} instant={instant_time} out of range, skipped" + ); + continue; + } + + log::debug!( + "[Pass1] block #{total_log_blocks} passed all gates: type={:?} instant={instant_time}", + block.block_type, + ); + + // Classify the block + match block.block_type { + BlockType::AvroData | BlockType::ParquetData | BlockType::Delete => { + let blocks_list = instant_to_blocks_map + .entry(instant_time.clone()) + .or_default(); + if blocks_list.is_empty() { + ordered_instants_list.push(instant_time.clone()); + } + blocks_list.push(block); + } + BlockType::Command if block.is_rollback_block() => { + total_rollbacks += 1; + if let Ok(target) = block.target_instant_time() { + let target = target.to_string(); + log::debug!("[Pass1] ROLLBACK: removing instant={target}"); + target_rollback_instants.insert(target.clone()); + ordered_instants_list.retain(|t| t != &target); + instant_to_blocks_map.remove(&target); + } + } + _ => {} + } + } + + log::debug!( + "[Pass1] complete: ordered_instants={ordered_instants_list:?} \ + total_blocks={total_log_blocks} corrupt={total_corrupt_blocks} \ + rollbacks={total_rollbacks}", + ); + for (instant, blocks) in &instant_to_blocks_map { + log::debug!( + "[Pass1] instant={instant}: {} block(s) [{:?}]", + blocks.len(), + blocks + .iter() + .map(|b| format!("{:?}", b.block_type)) + .collect::<Vec<_>>(), + ); + } + + Ok(Pass1Result { + ordered_instants_list, + instant_to_blocks_map, + total_log_blocks, + total_corrupt_blocks, + total_rollbacks, + }) +} + +// ========================================================================= +// Pass 2 result +// ========================================================================= + +/// Output of Pass 2 (reverse iteration with compaction resolution). +#[derive(Debug)] +pub struct Pass2Result { + /// The final ordered deque of blocks to be processed. + /// Latest instant at head (front), earliest at tail (back). + /// Drain via `pop_back` for oldest-first processing. + pub current_instant_log_blocks: VecDeque<LogBlock>, + /// Dedup guard — instants whose blocks are enqueued. + /// Recorded for parity with Java's scan result; nothing reads it back here. + #[allow(dead_code)] + pub instant_times_included: HashSet<String>, + /// Which instant times actually contributed blocks (ordered output). + pub valid_block_instants: Vec<String>, +} + +/// Run Pass 2: reverse iteration with compaction resolution. +/// +/// Iterates `ordered_instants_list` from newest→oldest, applying: +/// - `Collections.reverse(logBlocks)` then `addLast` (Rust: `reverse()` + `push_back`) +/// - Compaction: if block has `COMPACTED_BLOCK_TIMES` header, register mapping +/// - Deduplication: `instant_times_included` prevents double-enqueue +/// +/// ## Invariants (from FS_logFileReadInputs_fileOrderRequestTime.md): +/// +/// **Invariant 1**: `instant_times_included` and `valid_block_instants` contain +/// exactly the same instant strings (Set vs List). +/// +/// **Invariant 2**: Every instant in `instant_times_included`/`valid_block_instants` +/// has its blocks in `current_instant_log_blocks`, and vice versa. +/// +/// **Invariant 3**: Each instant appears in `instant_times_included` at most once (dedup). +/// +/// **Invariant 4**: `current_instant_log_blocks` is ordered latest-first (reverse chronological). +/// Drain via `pop_back` produces oldest-first processing order. +pub fn reverse_scan_pass2(pass1: &mut Pass1Result) -> Pass2Result { + log::debug!( + "[Pass2] reverse_scan: {} instants to process (newest→oldest)", + pass1.ordered_instants_list.len(), + ); + + let mut current_instant_log_blocks: VecDeque<LogBlock> = VecDeque::new(); + let mut instant_times_included: HashSet<String> = HashSet::new(); + let mut valid_block_instants: Vec<String> = Vec::new(); + let mut block_time_to_compaction_block_time_map: HashMap<String, String> = HashMap::new(); + + for i in (0..pass1.ordered_instants_list.len()).rev() { + let instant_time = &pass1.ordered_instants_list[i]; + let instants_blocks = match pass1.instant_to_blocks_map.get(instant_time) { + Some(blocks) if !blocks.is_empty() => blocks, + _ => continue, + }; + + let first_block = &instants_blocks[0]; + + // Check for compacted blocks (has COMPACTED_BLOCK_TIMES header) + if let Some(compacted_times) = first_block + .header + .get(&BlockMetadataKey::CompactedBlockTimes) + { + for original_instant in compacted_times.split(',') { + let original_instant = original_instant.trim().to_string(); + let final_instant = block_time_to_compaction_block_time_map + .get(instant_time) + .cloned() + .unwrap_or_else(|| instant_time.clone()); + block_time_to_compaction_block_time_map.insert(original_instant, final_instant); + } + } else { + // Normal data block — check if it's been compacted + let compacted_final = block_time_to_compaction_block_time_map.get(instant_time); + + if let Some(final_instant) = compacted_final { + let final_instant = final_instant.clone(); + if instant_times_included.contains(&final_instant) { + continue; + } + if let Some(blocks) = pass1.instant_to_blocks_map.get(&final_instant) { + let mut reversed = blocks.clone(); + reversed.reverse(); + for block in reversed { + current_instant_log_blocks.push_back(block); + } + } + instant_times_included.insert(final_instant.clone()); + valid_block_instants.push(final_instant); + } else { + // Not compacted — add blocks directly + // Java: Collections.reverse(logBlocks) then forEach(addLast) + let blocks = pass1 + .instant_to_blocks_map + .get(instant_time) + .cloned() + .unwrap_or_default(); + let mut reversed = blocks; + reversed.reverse(); + for block in reversed { + current_instant_log_blocks.push_back(block); + } + instant_times_included.insert(instant_time.clone()); + valid_block_instants.push(instant_time.clone()); + } + } + } + + log::debug!( + "[Pass2] complete: {} blocks in deque, valid_instants={:?}", + current_instant_log_blocks.len(), + valid_block_instants, + ); + + Pass2Result { + current_instant_log_blocks, + instant_times_included, + valid_block_instants, + } +} + +// ========================================================================= +// BaseHoodieLogRecordReader +// ========================================================================= + +/// The base log record reader implementing the 3-pass scanning algorithm. +/// +/// Mirrors Java's `BaseHoodieLogRecordReader<T>` (abstract base class). +/// +/// ## Java hierarchy: +/// ```text +/// BaseHoodieLogRecordReader<T> (abstract — this struct) +/// └─ HoodieMergedLogRecordReader<T> (concrete — wraps this via composition) +/// ``` +/// +/// ## Fields matching Java: +/// - `reader_context` ↔ `readerContext` +/// - `storage` ↔ `storage` +/// - `log_file_paths` ↔ `logFiles` +/// - `latest_instant_time` ↔ `latestInstantTime` +/// - `instant_range` ↔ `instantRange` +/// - `force_full_scan` ↔ `forceFullScan` +/// - `record_buffer` ↔ `recordBuffer` +/// - `allow_inflight_instants` ↔ `allowInflightInstants` +pub struct BaseHoodieLogRecordReader { + pub reader_context: Arc<ReaderContext>, + pub storage: Arc<Storage>, + pub log_file_paths: Vec<String>, + pub latest_instant_time: String, + /// Mirrors Java's `private final Option<InstantRange> instantRange`. + pub instant_range: Option<InstantRange>, + /// Mirrors Java's `protected final boolean forceFullScan`. + /// When true, scanning happens eagerly in the constructor. + pub force_full_scan: bool, + pub record_buffer: Box<dyn HoodieFileGroupRecordBuffer>, + /// Carried for parity. Nothing reads it, so it is inert — the completion gate + /// below is what actually decides whether an inflight instant is admitted, + /// and `ReaderParameters` never sets this true. + #[allow(dead_code)] + pub allow_inflight_instants: bool, + /// Inputs for the Gate-3 completed/inflight check (C-INFLIGHT-DELTA). `Some` only for + /// table version < 8 (v1 timeline layout); `None` for v8+ and when no timeline is available, + /// in which case Gate 3 is a no-op. Populated by the builder / FFI wiring from the active + /// timeline (completed + inflight instant sets + the first active instant). + pub completion_gate_inputs: Option<CompletionGateInputs>, + + // ── Stats / state (mirrors Java's AtomicLong counters + progress) ── + pub valid_block_instants: Vec<String>, + pub total_log_files: u64, + pub total_log_blocks: u64, + pub total_log_records: u64, + pub total_corrupt_blocks: u64, + pub total_rollbacks: u64, + /// Mirrors Java's `private float progress` (0.0 → 1.0). + pub progress: f32, + + // ── Stage timings (perf harness) ────────────────────── + /// Wall ms spent reading log-block metadata + bytes off storage (Pass 1). + pub log_block_read_ms: u64, + /// Wall ms spent in Pass-3 block dispatch (inflate/decode + merge insert). + pub merge_insert_ms: u64, +} + +impl BaseHoodieLogRecordReader { + /// Mirrors Java's `scanInternal(Option<KeySpec> keySpecOpt, boolean skipProcessingBlocks)`. + /// + /// # Arguments + /// - `skip_processing_blocks` — when `true`, Pass 3 (block processing) is skipped. + /// Java's `HoodieMergedLogRecordReader.scan(true)` uses this to collect block + /// metadata without actually merging records. + pub async fn scan_internal(&mut self, skip_processing_blocks: bool) -> Result<()> { + // Reset state (mirrors Java: currentInstantLogBlocks = new ArrayDeque<>(), progress = 0.0f, ...) + self.valid_block_instants.clear(); + self.progress = 0.0; + self.total_log_files = 0; + self.total_log_blocks = 0; + self.total_log_records = 0; + self.total_corrupt_blocks = 0; + self.total_rollbacks = 0; + + let timezone = self.reader_context.timezone(); + + // Read all blocks as metadata-only (no content decoding). Review Comment: This comment describes the lazy sweep, but the loop below uses the eager whole-file reader — the lazy tier (`new_streaming`, `read_all_blocks_metadata_only`, `LogBlockFetcher`/`DeferredContent`, and the `hoodie.memory.dfs.buffer.max.size` window knob) has no production caller, only the `#[cfg(test)]` memory_bench; the one production `load_content` call in Pass 3 no-ops because content is always already read. Since the range here is deliberately unbounded (rollback command blocks must always process), eager decode means out-of-window and rolled-back blocks pay a full Arrow decode before the gates discard them — v1 skips those at the header. Is wiring the sweep in a follow-up, or should the tier and the knob wait until it lands? _⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality._ ########## crates/core/src/file_group/log_file/content.rs: ########## @@ -17,35 +17,128 @@ * under the License. */ use crate::Result; -use crate::avro_to_arrow::arrow_array_reader::AvroArrowArrayReader; use crate::config::HudiConfigs; use crate::error::CoreError; -use crate::file_group::log_file::avro::AvroDataBlockContentReader; +use crate::file_group::log_file::avro::AvroBlockDecoder; use crate::file_group::log_file::log_block::{ BlockMetadataKey, BlockType, LogBlockContent, LogBlockVersion, }; use crate::file_group::log_file::log_format::LogFormatVersion; use crate::file_group::record_batches::RecordBatches; use crate::hfile::{HFileReader, HFileRecord}; -use crate::schema::delete::{avro_schema_for_delete_record, avro_schema_for_delete_record_list}; -use apache_avro::{Schema as AvroSchema, from_avro_datum}; +use crate::schema::delete::delete_record_list_schema_json; +use crate::schema::extended_promotion::record_needs_rewrite_for_extended_promotion; +use crate::schema::parquet_list_norm::normalize_parquet_metadata; +use crate::schema::resolver::avro_json_to_arrow_schema; +use crate::storage::RowFilterBuilder; +use arrow_array::{Array, ArrayRef, ListArray, RecordBatch, StructArray, UnionArray}; +use arrow_schema::{DataType, Field, Schema}; use bytes::Bytes; -use parquet::arrow::arrow_reader::ParquetRecordBatchReader; +use parquet::arrow::arrow_reader::{ + ArrowReaderMetadata, ArrowReaderOptions, ParquetRecordBatchReaderBuilder, +}; +use parquet::file::metadata::ParquetMetaDataReader; use std::collections::HashMap; use std::io::{Read, Seek}; use std::sync::Arc; +/// Turn the wrapped ordering values into a plain column. +/// +/// Hudi writes `orderingVal` as a union of per-type wrapper records, so a decode +/// against that schema yields a union of one-field structs. The merge wants the +/// value itself. +/// +/// A block writes one ordering type, so exactly one branch is populated; that +/// branch's `value` child is the column. A block mixing branches is rejected +/// rather than silently reduced to one of them. +fn unwrap_ordering_values(ordering: &ArrayRef) -> Result<ArrayRef> { + let union = ordering + .as_any() + .downcast_ref::<UnionArray>() + .ok_or_else(|| { + CoreError::LogBlockError(format!( + "Expected orderingVal to be a union, got {}", + ordering.data_type() + )) + })?; + + let mut active: Option<i8> = None; + for i in 0..union.len() { + let type_id = union.type_id(i); + match active { + None => active = Some(type_id), + Some(seen) if seen == type_id => {} + Some(seen) => { + return Err(CoreError::LogBlockError(format!( + "Delete block mixes ordering types (union branches {seen} and {type_id})" Review Comment: Null is a union branch too, so a delete block over a nullable ordering column where some deleted rows carry null mixes branch 0 with a wrapper branch and fails here. Java resolves the branch per record and treats a null as the default ordering, so the delete wins naturally. Not a regression — the old per-record narrowing couldn't read this shape either — but the message reads as if two value types mixed, when null+value is the realistic case. Could the null branch be exempted from the mixing check and those slots treated as default-ordering deletes? _⚠️ 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]
