linliu-code commented on code in PR #663: URL: https://github.com/apache/hudi-rs/pull/663#discussion_r3779919799
########## 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: Fixed. Renamed to `emit_deletes` and added a `debug_assert!(!emit_deletes, "emitting deletes is not implemented; buffer/loader.rs rejects it upstream")`, with the doc saying the `true` branch is not yet implemented. `todo!()` would have been wrong here — the parameter exists for signature parity with gold and the value is always `false`, so panicking on a value nothing produces would be dead code that only fires if the upstream rejection regresses. The `debug_assert` catches exactly that case without changing release behaviour. ########## 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: Fixed. All five now carry the `test_` prefix: `test_i33_merge_budget_honored_when_forwarded`, `test_oom_current_footprint_getter_reflects_in_memory_bytes`, `test_oom_peak_cap_rejects_oversized_insertion_loudly`, `test_oom_no_cap_preserves_unchanged_behavior`, `test_oom_peak_cap_config_parses_and_defaults_to_none`. -- 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]
