linliu-code commented on code in PR #706: URL: https://github.com/apache/hudi-rs/pull/706#discussion_r3920273635
########## benchmark/filegroup/src/main.rs: ########## @@ -0,0 +1,734 @@ +/* + * 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. + */ + +//! `fg-bench` — minimal file-group reader benchmark harness (perf effort M2). +//! +//! Opens a Hudi table, discovers its latest file slice(s), drives a +//! `HoodieFileGroupReader` per slice to completion, and records per-iteration +//! wall/CPU/RSS plus the full `HoodieReadStats` (including the stage +//! timings). Output is a single JSON document; see `benchmark/filegroup/README.md`. + +mod host; +mod rusage; + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Instant; + +use arrow_schema::{Schema, SchemaRef}; +use clap::Parser; +use std::alloc::{GlobalAlloc, Layout, System}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + +use futures::{StreamExt, TryStreamExt}; +use hudi_core::error::Result; +use hudi_core::file_group::file_slice::FileSlice; +use hudi_core::file_group::reader::FileGroupReader; +use hudi_core::table::Table; +use hudi_core::table::builder::OptionResolver; +use serde::Serialize; + +use host::HostSnapshot; +use rusage::Rusage; + +#[derive(Parser, Debug)] +#[command( + name = "fg-bench", + about = "Minimal HoodieFileGroupReader benchmark harness (perf M2)" +)] +struct Args { + /// Path to the Hudi table (local filesystem path or file:// URI). + #[arg(long)] + table: String, + + /// Number of measured iterations. The first is always a warmup and is + /// excluded from the summary statistics. + #[arg(long, default_value_t = 3)] + iterations: usize, + + /// Write the JSON report to this file instead of stdout. + #[arg(long)] + output_json: Option<String>, + + /// Comma-separated projection columns (requested schema). When omitted the + /// reader reads all columns. + #[arg(long, value_delimiter = ',')] + columns: Option<Vec<String>>, + + /// Streaming chunk size (`hoodie.read.stream.batch_size`, ENG-42991). When + /// omitted the reader uses its built-in default (DEFAULT_BATCH_SIZE = 4096). + /// Set this to measure the effect of chunk granularity on wall time. (The + /// peak-RSS effect shows only on the streaming `open()` path; this bench + /// drives `read()`, which retains the full output, so `max_rss` is + /// dominated by output retention here, not chunk size.) + #[arg(long)] + batch_size: Option<usize>, + + /// Merge memory budget in bytes (`hoodie.memory.merge.max.size`, A1/ENG-42993). + /// When omitted the reader uses its 1 GiB default (no spill on smoke + /// datasets). Set it low (e.g. a few MiB) to force the size-tracked merge map + /// to spill to RocksDB and measure bounded RSS without OOM. + #[arg(long)] + merge_max_size: Option<u64>, + + /// Drive the true-streaming `open()` path (A3/ENG-42992) instead of the + /// eager `read()` path. In streaming mode the base parquet file is decoded + /// one row-group at a time (`ParquetSyncReader`) and never fully + /// materialised — this is the path the R3 base-file-memory fix optimizes. + /// The streaming iterator does `block_on` per row-group, so each slice is + /// driven on a dedicated OS thread (off the harness's tokio runtime), + /// mirroring the FFI driver's sync-consumer contract. + #[arg(long, default_value_t = false)] + streaming: bool, + + /// Drive the async streaming path (`open_blocking_stream`) from this + /// process's tokio runtime, rather than the sync `open()` path on a plain + /// OS thread. This is how a Rust async caller consumes a streaming read. + #[arg(long, default_value_t = false)] + async_stream: bool, + + /// Merge base + log records by base-file row POSITION instead of record key + /// (`hoodie.merge.use.record.positions`). Selects the + /// PositionBasedFileGroupRecordBuffer, which matches base rows to log + /// records by their physical position (read via a parquet virtual + /// row-number column) and falls back to key-based merge when a log block has + /// no valid positions. Use with a table whose log blocks carry + /// RECORD_POSITIONS headers (written by Spark with record positions enabled) + /// to compare position-based vs key-based merge on the same table. + #[arg(long, default_value_t = false)] + use_record_position: bool, + /// Read this many file slices concurrently, mirroring the reader's own + /// `buffer_unordered` fan-out. 1 is sequential with no coordination cost and + /// is the baseline any bounded-memory claim must not regress. + #[arg(long, default_value_t = 1)] + slice_concurrency: usize, + /// Fail the run when peak RSS exceeds this many bytes. Turns the harness + /// from a measurement into a gate: a read may be slower under pressure, but + /// it may not grow without bound. + #[arg(long)] + max_rss_bytes: Option<u64>, + /// Total memory the scan may use for concurrent slice reads + /// (`hoodie.read.scan.max.memory.size`). When set, slices-in-flight is + /// derived from it by `hudi_core::file_group::admission::slices_in_flight` + /// — the same function the DataFusion plan uses — and `--slice-concurrency` + /// becomes the ceiling it may not exceed rather than the value used. + #[arg(long)] + scan_memory_budget: Option<u64>, + + /// Directory the merge map spills into (`hoodie.memory.spillable.map.path`). + /// Watched to report whether a run actually exercised the disk tier. + #[arg(long, default_value = "/tmp")] Review Comment: Fixed both halves. `--spill-dir` now defaults to a fresh per-run subdirectory of the system temp dir (`fg-bench-spill-<pid>`), which also keeps the 100 ms sampler off large unrelated trees. And `SpillWatcher` samples the directory's size when it starts and reports `peak - baseline`, saturating — so growth, not absolute size, even when `--spill-dir` is pointed somewhere shared. Checked against a directory pre-filled with 200 MB: `spilled=false, peak=0` both by default and when explicitly pointed at it. Confirmed it still detects a real spill (`--merge-max-size 65536` → `spilled=true, peak=27718`), so the fix did not trade the false positive for a false negative. ########## crates/core/src/metadata/table/mod.rs: ########## @@ -294,9 +416,181 @@ impl Table { // where per-block fixed cost cannot amortise; on larger blocks it overtakes // the reader it replaces. So the ratio above is not the production ratio, and // nothing measured here establishes what that is. - v2_reader::MetadataTableV2Reader::new(configs, storage) - .read_files_partition(&file_slice, keys) - .await + Ok(Some(( + v2_reader::MetadataTableV2Reader::new(configs, storage), + file_slices, + ))) + // NOTE: the valid-instant set is attached by `partition_reader_with_valid_instants`, + // which has the data table in hand; this constructor does not. + } +} + +/// Hudi's sentinel prefix for a metadata delta commit written outside the data +/// timeline (`HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP`). +const SOLO_COMMIT_TIMESTAMP: &str = "00000000000000"; + +impl Table { + /// The instants whose metadata log blocks may be read. + /// + /// Mirrors Java's `HoodieTableMetadataUtil.getValidInstantTimestamps` (:2081). + /// This is a **set, not a window**: it has holes -- a pending data instant is + /// excluded while instants either side of it are included -- and members from + /// outside the data timeline entirely. See + /// [`InstantRange::exact_match`](crate::timeline::selector::InstantRange::exact_match) + /// for why a bounded range cannot stand in for it. + /// + /// `self` is the data table; `mdt` its metadata table. Both timelines are + /// needed, which is why this lives here rather than on either one alone. + pub(crate) async fn valid_instant_timestamps(&self, mdt: &Table) -> Result<HashSet<String>> { + let mut valid: HashSet<String> = self.valid_from_completed_data_instants(); + valid.extend(self.valid_from_mdt_delta_commits(mdt)); + valid.extend(Self::valid_from_sentinel_commits(mdt)); + + // 3. Commits rolled back by the data table's rollbacks and restores. + // Their log blocks were written, rolled back, and re-applied, so + // excluding them drops records that are genuinely present. + // + // Only rollbacks newer than the earliest valid instant can have + // rolled back anything we hold a log block for; Java bounds the scan + // the same way, falling back to the sentinel when the set is empty. + let earliest = valid + .iter() + .min() + .cloned() + .unwrap_or_else(|| SOLO_COMMIT_TIMESTAMP.to_string()); + for instant in self.rollback_and_restore_instants().await? { Review Comment: Checked this against the source and I think the premise is off — `rollbackFilterThreshold` is not in apache/hudi. apache/hudi `HoodieTableMetadataUtil.getValidInstantTimestamps`:2101-2104 bounds the scan on `earliestInstantTime` alone: ```java final String earliestInstantTime = validInstantTimestamps.isEmpty() ? SOLO_COMMIT_TIMESTAMP : Collections.min(validInstantTimestamps); datasetTimeline.getRollbackAndRestoreTimeline().filterCompletedInstants().getInstantsAsStream() .filter(instant -> compareTimestamps(instant.requestedTime(), GREATER_THAN, earliestInstantTime)) ``` `grep -rn rollbackFilterThreshold` over apache/hudi master returns nothing; the only hit is in the Onehouse fork (`hudi-internal` master, the 0.14.1 line), where it was added as a latency optimization. So the existing "Java bounds the scan the same way" comment is accurate for the Java this project targets, and adopting the compaction bound here would be importing fork-only behavior — and narrowing the set on an argument about MDT compaction semantics I would not want to land unverified in an ASF repo. Happy to do it as its own change if you would rather hudi-rs lead apache/hudi here. The cost you identified is real regardless of the bound, so I fixed that part: the reads are no longer a sequential awaited-GET walk. They now run through `bounded_in_order` at a concurrency of 8. Same set, off the critical path. I also added the `:2101` line reference to the comment so the claim is checkable. ########## crates/core/src/table/fs_view.rs: ########## @@ -278,17 +278,33 @@ impl FileSystemView { /// * `table_schema` - Table schema for statistics extraction /// * `timeline_view` - The timeline view containing query context /// * `metadata_table` - Optional metadata table instance for file listing + // Eight parameters, one over clippy's limit. The eighth is the valid-instant + // set, which cannot be derived here: this view holds the metadata table's + // timeline and the set needs the data table's too. Bundling the existing + // seven to make room would be a wider refactor than this change. + #[allow(clippy::too_many_arguments)] pub(crate) async fn get_file_slices( &self, partition_pruner: &PartitionPruner, file_pruner: &FilePruner, table_schema: &Schema, timeline_view: &TimelineView, metadata_table: Option<&Table>, + // The instants whose metadata log blocks may be read. Built by the data + // table, which holds both timelines, and passed in because this view + // holds only the metadata one. + valid_instants: Option<&std::collections::HashSet<String>>, Review Comment: Agreed, and made unrepresentable. The two parameters are now one: ```rust pub(crate) struct MetadataListing<'a> { pub table: &'a Table, pub valid_instants: &'a HashSet<String>, } ``` `get_file_slices` takes `Option<MetadataListing<'_>>`, so there is no way to pass a metadata table without its set and no `unwrap_or(&empty)` left. That also drops the parameter count back under clippy's limit, so the `#[allow(too_many_arguments)]` and the comment explaining it are gone. `fs_view_get_file_slices_with_metadata_table` — the test you noted was riding the degraded path — now builds the real set and asserts it is non-empty first, so it cannot pass while silently reading a metadata table it had filtered down to nothing. ########## benchmark/filegroup/src/main.rs: ########## @@ -0,0 +1,734 @@ +/* + * 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. + */ + +//! `fg-bench` — minimal file-group reader benchmark harness (perf effort M2). +//! +//! Opens a Hudi table, discovers its latest file slice(s), drives a +//! `HoodieFileGroupReader` per slice to completion, and records per-iteration +//! wall/CPU/RSS plus the full `HoodieReadStats` (including the stage +//! timings). Output is a single JSON document; see `benchmark/filegroup/README.md`. + +mod host; +mod rusage; + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Instant; + +use arrow_schema::{Schema, SchemaRef}; +use clap::Parser; +use std::alloc::{GlobalAlloc, Layout, System}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + +use futures::{StreamExt, TryStreamExt}; +use hudi_core::error::Result; +use hudi_core::file_group::file_slice::FileSlice; +use hudi_core::file_group::reader::FileGroupReader; +use hudi_core::table::Table; +use hudi_core::table::builder::OptionResolver; +use serde::Serialize; + +use host::HostSnapshot; +use rusage::Rusage; + +#[derive(Parser, Debug)] +#[command( + name = "fg-bench", + about = "Minimal HoodieFileGroupReader benchmark harness (perf M2)" +)] +struct Args { + /// Path to the Hudi table (local filesystem path or file:// URI). + #[arg(long)] + table: String, + + /// Number of measured iterations. The first is always a warmup and is + /// excluded from the summary statistics. + #[arg(long, default_value_t = 3)] + iterations: usize, + + /// Write the JSON report to this file instead of stdout. + #[arg(long)] + output_json: Option<String>, + + /// Comma-separated projection columns (requested schema). When omitted the + /// reader reads all columns. + #[arg(long, value_delimiter = ',')] + columns: Option<Vec<String>>, + + /// Streaming chunk size (`hoodie.read.stream.batch_size`, ENG-42991). When + /// omitted the reader uses its built-in default (DEFAULT_BATCH_SIZE = 4096). + /// Set this to measure the effect of chunk granularity on wall time. (The + /// peak-RSS effect shows only on the streaming `open()` path; this bench + /// drives `read()`, which retains the full output, so `max_rss` is + /// dominated by output retention here, not chunk size.) + #[arg(long)] + batch_size: Option<usize>, + + /// Merge memory budget in bytes (`hoodie.memory.merge.max.size`, A1/ENG-42993). + /// When omitted the reader uses its 1 GiB default (no spill on smoke + /// datasets). Set it low (e.g. a few MiB) to force the size-tracked merge map + /// to spill to RocksDB and measure bounded RSS without OOM. + #[arg(long)] + merge_max_size: Option<u64>, + + /// Drive the true-streaming `open()` path (A3/ENG-42992) instead of the + /// eager `read()` path. In streaming mode the base parquet file is decoded + /// one row-group at a time (`ParquetSyncReader`) and never fully + /// materialised — this is the path the R3 base-file-memory fix optimizes. + /// The streaming iterator does `block_on` per row-group, so each slice is + /// driven on a dedicated OS thread (off the harness's tokio runtime), + /// mirroring the FFI driver's sync-consumer contract. + #[arg(long, default_value_t = false)] + streaming: bool, + + /// Drive the async streaming path (`open_blocking_stream`) from this + /// process's tokio runtime, rather than the sync `open()` path on a plain + /// OS thread. This is how a Rust async caller consumes a streaming read. + #[arg(long, default_value_t = false)] + async_stream: bool, + + /// Merge base + log records by base-file row POSITION instead of record key + /// (`hoodie.merge.use.record.positions`). Selects the + /// PositionBasedFileGroupRecordBuffer, which matches base rows to log + /// records by their physical position (read via a parquet virtual + /// row-number column) and falls back to key-based merge when a log block has + /// no valid positions. Use with a table whose log blocks carry + /// RECORD_POSITIONS headers (written by Spark with record positions enabled) + /// to compare position-based vs key-based merge on the same table. + #[arg(long, default_value_t = false)] + use_record_position: bool, + /// Read this many file slices concurrently, mirroring the reader's own + /// `buffer_unordered` fan-out. 1 is sequential with no coordination cost and + /// is the baseline any bounded-memory claim must not regress. + #[arg(long, default_value_t = 1)] + slice_concurrency: usize, + /// Fail the run when peak RSS exceeds this many bytes. Turns the harness + /// from a measurement into a gate: a read may be slower under pressure, but + /// it may not grow without bound. + #[arg(long)] + max_rss_bytes: Option<u64>, + /// Total memory the scan may use for concurrent slice reads + /// (`hoodie.read.scan.max.memory.size`). When set, slices-in-flight is + /// derived from it by `hudi_core::file_group::admission::slices_in_flight` + /// — the same function the DataFusion plan uses — and `--slice-concurrency` + /// becomes the ceiling it may not exceed rather than the value used. + #[arg(long)] + scan_memory_budget: Option<u64>, + + /// Directory the merge map spills into (`hoodie.memory.spillable.map.path`). + /// Watched to report whether a run actually exercised the disk tier. + #[arg(long, default_value = "/tmp")] + spill_dir: String, +} + +/// Per-slice read configuration, bundled to keep the read helpers under the +/// clippy argument-count limit and to make the eager/streaming split explicit. +#[derive(Clone)] +struct ReadConfig { + requested_schema: Option<SchemaRef>, + batch_size: Option<usize>, + merge_max_size: Option<u64>, + /// True → drive the streaming `open()` path (A3); false → eager `read()`. + streaming: bool, + /// True → drive `open_blocking_stream()` from the tokio runtime. + async_stream: bool, + /// True → position-based merge (`use_record_position`); false → key-based. + use_record_position: bool, + /// How many slices to read concurrently — a ceiling when a budget is set. + slice_concurrency: usize, + /// Scan-wide memory budget, in bytes. + scan_memory_budget: Option<u64>, + /// Where the merge map spills (`hoodie.memory.spillable.map.path`). Passed + /// to the reader, not only watched: the reader's own default is `/tmp`, and + /// on a host where anything else lives under `/tmp` a watcher pointed there + /// measures that instead of spill. + spill_dir: String, +} + +/// A global allocator that refuses to exceed a byte ceiling. +/// +/// `--max-rss-bytes` asserts *after* the fact: it reads peak RSS once the +/// allocation has already happened, so it answers "how much did this read want" +/// on a machine with memory to spare. It cannot answer "does this read survive +/// on a small machine", because nothing ever told the process no. +/// +/// This does. Past the ceiling, `alloc` returns null. Rust's runtime turns that +/// into an abort, which is deliberately not graceful: an abort proves the +/// ceiling is real. A read that wants to degrade instead has to stay under it by +/// spilling, which is why this pairs with a low `--merge-max-size` rather than +/// standing alone. +/// +/// Off unless `FG_BENCH_ALLOC_CAP_BYTES` is set, so ordinary runs are unaffected +/// and pay only a relaxed atomic add per allocation. +struct CappedAllocator; + +static ALLOC_CAP: AtomicU64 = AtomicU64::new(0); +static ALLOC_LIVE: AtomicU64 = AtomicU64::new(0); + +unsafe impl GlobalAlloc for CappedAllocator { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + let cap = ALLOC_CAP.load(Ordering::Relaxed); + if cap > 0 { + let live = ALLOC_LIVE.fetch_add(layout.size() as u64, Ordering::Relaxed) + + layout.size() as u64; + if live > cap { + ALLOC_LIVE.fetch_sub(layout.size() as u64, Ordering::Relaxed); + // Null here, rather than a panic: a panic would unwind through + // an allocation path that has just been told there is no memory. + return std::ptr::null_mut(); + } + } + unsafe { System.alloc(layout) } + } + + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + if ALLOC_CAP.load(Ordering::Relaxed) > 0 { + ALLOC_LIVE.fetch_sub(layout.size() as u64, Ordering::Relaxed); + } + unsafe { System.dealloc(ptr, layout) } + } +} + +#[global_allocator] +static GLOBAL: CappedAllocator = CappedAllocator; + +#[tokio::main(flavor = "multi_thread")] +async fn main() { + env_logger::init(); + // Read before anything large is allocated, so the cap covers the run rather + // than only its tail. + if let Ok(v) = std::env::var("FG_BENCH_ALLOC_CAP_BYTES") + && let Ok(n) = v.parse::<u64>() + { + ALLOC_CAP.store(n, Ordering::Relaxed); + eprintln!( + "[fg-bench] allocation ceiling: {} MiB (hard)", + n / (1024 * 1024) + ); + } + let args = Args::parse(); + if let Err(e) = run(&args).await { + eprintln!("fg-bench failed: {e:?}"); + std::process::exit(1); + } +} + +async fn run(args: &Args) -> Result<()> { + let nproc = host::nproc(); + + // 1. Open the table + resolve the full hoodie option map (includes raw + // hoodie.properties: record key, ordering, merge mode, table type). + let table = Table::new(&args.table).await?; + let hoodie_options = resolve_hoodie_options(&args.table).await?; + + // 2. Discover the latest file slice(s) (snapshot, no partition filters). + // OSS `get_file_slices` takes `&ReadOptions` (the internal fork takes filter + // tuples). Default options = snapshot, no partition filters. + let no_filters = hudi_core::config::ReadOptions::default(); + let file_slices = table.get_file_slices(&no_filters).await?; + if file_slices.is_empty() { + return Err(hudi_core::error::CoreError::ReadFileSliceError(format!( + "no file slices found at '{}'", + args.table + ))); + } + eprintln!( + "[fg-bench] table={} slices={} nproc={} iterations={} (1 warmup)", + args.table, + file_slices.len(), + nproc, + args.iterations + ); + + // Requested schema. When --columns is given we project to those columns; + // otherwise we request the FULL table schema (with meta fields). Always + // supplying a requested schema mirrors the production FFI path and ensures + // the reader's required/reader schema is set even for log-only slices + // (otherwise merge_and_collect has no schema source). Field types come from + // the table schema so they match exactly; the per-slice base footer + // supplies the data schema (mirrors the FG harness Projection path). + let table_data_schema: SchemaRef = table.get_schema_with_meta_fields().await.map(Arc::new)?; + let requested_schema: Option<SchemaRef> = match &args.columns { + Some(cols) => Some(build_requested_schema(&table, cols).await?), + None => Some(table_data_schema.clone()), + }; + + let mut iterations = Vec::with_capacity(args.iterations); + let mut over_budget_any = false; + let spill_dir = std::path::PathBuf::from(&args.spill_dir); + let mut any_contended = false; + + for iter_idx in 0..args.iterations { + let warmup = iter_idx == 0; + + let host_pre = HostSnapshot::capture(); + let contended = host_pre.is_contended(nproc); + if contended { + any_contended = true; + eprintln!( + "[fg-bench] !!! HOST CONTENDED: load1={:.2} over {} cores (ratio {:.2} > {:.2}); \ + iteration {} numbers are unreliable", + host_pre.load1, + nproc, + host_pre.load1 / nproc as f64, + host::LOAD_THRESHOLD, + iter_idx + ); + } + + let ru_before = Rusage::capture(); + let wall_start = Instant::now(); + + let read_config = ReadConfig { + requested_schema: requested_schema.clone(), + batch_size: args.batch_size, + merge_max_size: args.merge_max_size, + streaming: args.streaming, + async_stream: args.async_stream, + use_record_position: args.use_record_position, + slice_concurrency: args.slice_concurrency, + scan_memory_budget: args.scan_memory_budget, + spill_dir: args.spill_dir.clone(), + }; + let spill_watch = SpillWatcher::start(spill_dir.clone()); + let rows = + read_all_slices(&file_slices, &hoodie_options, &args.table, &read_config).await?; + + let wall_ms = wall_start.elapsed().as_millis() as u64; + let ru_delta = Rusage::capture().delta(&ru_before); + + eprintln!( + "[fg-bench] iter {}{}: wall={}ms user={}ms sys={}ms rss={}MB rows={}", + iter_idx, + if warmup { " (warmup)" } else { "" }, + wall_ms, + ru_delta.user_ms, + ru_delta.sys_ms, + ru_delta.max_rss_kb / 1024, + rows + ); + + // The memory gate. `HoodieReadStats` carried the merge map's accounted + // peak, which the old drift detector compared against RSS; the public + // reader surface does not expose it, and widening that surface for a + // benchmark is the wrong trade. What the gate actually needs is simpler + // and stronger: measured RSS against the budget the caller declared. + let max_rss_bytes = ru_delta.max_rss_kb.saturating_mul(1024); + let over_budget = args.max_rss_bytes.is_some_and(|cap| max_rss_bytes > cap); + if over_budget { + eprintln!( + "[fg-bench] !!! OVER BUDGET: max_rss={}MB exceeds --max-rss-bytes={}MB on \ + iteration {}. A read must degrade in throughput rather than grow without \ + bound.", + max_rss_bytes / (1024 * 1024), + args.max_rss_bytes.unwrap_or(0) / (1024 * 1024), + iter_idx, + ); + } + over_budget_any |= over_budget; + + // Spill is observed rather than reported: the disk tier writes under + // `hoodie.memory.spillable.map.path`, so a directory that grew during + // the iteration means the merge map spilled. Without this a passing run + // cannot distinguish "stayed under budget because it spilled correctly" + // from "stayed under because the data never got large". + let spill_peak_bytes = spill_watch.finish(); + let spilled = spill_peak_bytes > 0; + + iterations.push(IterationReport { + warmup, + wall_ms, + user_ms: ru_delta.user_ms, + sys_ms: ru_delta.sys_ms, + max_rss_kb: ru_delta.max_rss_kb, + rows, + contended, + accounting_drift: over_budget, + spilled, + spill_peak_bytes, + host: HostReport { + load1: host_pre.load1, + mem_available_kb: host_pre.mem_available_kb, + }, + }); + } + + let report = Report { + env: EnvReport::capture(nproc), + table: args.table.clone(), + columns: args.columns.clone(), + num_slices: file_slices.len(), + merge_strategy: if args.use_record_position { + "position" + } else { + "key" + }, + contended: any_contended, + summary: Summary::from_iterations(&iterations), + iterations, + }; + + let json = serde_json::to_string_pretty(&report) + .map_err(|e| hudi_core::error::CoreError::ReadFileSliceError(e.to_string()))?; + match &args.output_json { + Some(path) => { + std::fs::write(path, &json) + .map_err(|e| hudi_core::error::CoreError::ReadFileSliceError(e.to_string()))?; + eprintln!("[fg-bench] wrote JSON report to {path}"); + } + None => println!("{json}"), + } + if over_budget_any { + // A non-zero exit is what makes this a gate rather than a report. The + // budget was declared by the caller; exceeding it is a failure even + // though every read returned correct rows. + return Err(hudi_core::error::CoreError::Unsupported(format!( + "peak RSS exceeded --max-rss-bytes={} on at least one iteration", + args.max_rss_bytes.unwrap_or(0) + ))); + } + + Ok(()) +} + +/// Watches a spill directory for its high-water mark while a read runs. +/// +/// A before/after comparison cannot work here. RocksDB creates its directory, +/// writes, compacts, and removes it when the reader closes -- which happens +/// before the read call returns. Sampling at iteration boundaries therefore sees +/// an empty directory both times and reports "never spilled" for a read that +/// spilled a gigabyte. Only sampling *during* the read observes it. +struct SpillWatcher { + stop: Arc<AtomicBool>, + peak: Arc<AtomicU64>, + handle: Option<std::thread::JoinHandle<()>>, +} + +impl SpillWatcher { + fn start(dir: std::path::PathBuf) -> Self { + let stop = Arc::new(AtomicBool::new(false)); + let peak = Arc::new(AtomicU64::new(0)); + let (s, pk) = (stop.clone(), peak.clone()); + let handle = std::thread::spawn(move || { + while !s.load(Ordering::Relaxed) { + let now = spill_dir_bytes(&dir); + pk.fetch_max(now, Ordering::Relaxed); + std::thread::sleep(std::time::Duration::from_millis(100)); + } + }); + Self { + stop, + peak, + handle: Some(handle), + } + } + + /// Stops sampling and returns the largest total seen, in bytes. + fn finish(mut self) -> u64 { + self.stop.store(true, Ordering::Relaxed); + if let Some(h) = self.handle.take() { + let _ = h.join(); + } + self.peak.load(Ordering::Relaxed) + } +} + +/// Total bytes of files under `dir`, walked recursively, or 0 when unreadable. +/// +/// Recursive because the disk tier is RocksDB, which writes into a subdirectory +/// rather than into `dir` itself. A shallow scan reports zero while a gigabyte +/// of spill sits one level down, which reads as "never spilled" -- the opposite +/// of the truth. +/// +/// Even so this only detects spill that is still on disk when it runs: RocksDB +/// compacts as it goes, so a sample after the read can miss a spill the read +/// certainly did. Treat a `true` as proof and a `false` as unproven, never as +/// proof of absence. +/// +/// Failure-tolerant by intent: an unreadable directory is an observation +/// problem, not a benchmark failure. +fn spill_dir_bytes(dir: &std::path::Path) -> u64 { + let Ok(entries) = std::fs::read_dir(dir) else { + return 0; + }; + entries + .filter_map(|e| e.ok()) + .map(|e| match e.file_type() { + Ok(t) if t.is_dir() => spill_dir_bytes(&e.path()), + Ok(t) if t.is_file() => e.metadata().map(|m| m.len()).unwrap_or(0), + _ => 0, + }) + .sum() +} + +/// Read every discovered file slice to completion, returning total rows. +/// +/// Built on the public `FileGroupReader` surface rather than assembling a +/// reader from `reader_v2` internals, which are `pub(crate)`. Every knob this +/// harness offers is a Hudi config key, so they survive the move: batch size, +/// merge budget and record-position merge all travel in the options bag. What +/// does not survive is `HoodieReadStats` -- stage timings and the spill flag are +/// not on the public surface. Spill is instead observed from outside, by +/// watching the spill directory, so the harness reports it without the reader +/// having to expose it. +async fn read_all_slices( + file_slices: &[FileSlice], + hoodie_options: &HashMap<String, String>, + table_path: &str, + cfg: &ReadConfig, +) -> Result<usize> { + let mut options: Vec<(String, String)> = hoodie_options + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(); + if let Some(n) = cfg.batch_size { + options.push(("hoodie.read.stream.batch_size".to_string(), n.to_string())); + } + if let Some(n) = cfg.merge_max_size { + options.push(("hoodie.memory.merge.max.size".to_string(), n.to_string())); + } + // Point the reader at the directory the harness watches. Without this the + // reader spills to its `/tmp` default while the watcher looks elsewhere, and + // `spilled` reports the watched directory rather than the read. + options.push(( + "hoodie.memory.spillable.map.path".to_string(), + cfg.spill_dir.clone(), + )); + if cfg.use_record_position { + options.push(( + "hoodie.merge.use.record.positions".to_string(), + "true".to_string(), + )); + } + let reader = FileGroupReader::new_with_options(table_path, options).await?; + + let read_options = hudi_core::config::read_options::ReadOptions { + projection: cfg.requested_schema.as_ref().map(|schema| { + schema + .fields() + .iter() + .map(|f| f.name().clone()) + .collect::<Vec<String>>() + }), + ..Default::default() + }; + + // The fan-out under test. `buffer_unordered(1)` is sequential with no + // coordination cost, which is what the single-slice baseline must stay. + // Derived by the library, not by the harness: the gate has to exercise the + // shipped decision, or it measures the benchmark instead of the product. + // One benchmark process is one engine partition, hence `partitions = 1`. + let slice_log_bytes: Vec<Option<u64>> = file_slices + .iter() + .map(hudi_core::file_group::file_slice::FileSlice::log_size_bytes) + .collect(); + let concurrency = hudi_core::file_group::admission::slices_in_flight( + cfg.scan_memory_budget, + 1, + &slice_log_bytes, + cfg.slice_concurrency, + ); + if cfg.scan_memory_budget.is_some() { + eprintln!( + "[fg-bench] budget {} MiB over {} slices -> concurrency {concurrency} (ceiling {})", + cfg.scan_memory_budget.unwrap_or(0) / (1024 * 1024), + file_slices.len(), + cfg.slice_concurrency, + ); + } + let total_rows = futures::stream::iter(file_slices.iter()) + .map(|slice| { + let reader = &reader; + let read_options = &read_options; + let streaming = cfg.streaming || cfg.async_stream; Review Comment: Correct — `let streaming = cfg.streaming || cfg.async_stream;` was the whole of it, and the docs described `open()` and `open_blocking_stream()`, neither of which exists on the public `FileGroupReader` (it has one streaming entry point, `read_file_slice_stream`). Folded into a single `--streaming`, with the doc saying what it actually contrasts: streamed batches dropped as counted, versus the eager read retaining the whole output. The README's stage-timings table went too — it documented per-stage fields no report carries. ########## benchmark/filegroup/src/gen.rs: ########## @@ -0,0 +1,368 @@ +// 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. + +//! `fg-gen` — generate a copy-on-write Hudi table of a target size. +//! +//! Exists because the memory benchmark needs roughly a gigabyte spread over +//! several file groups, and a gigabyte cannot be committed to the repository — +//! the checked-in fixtures are tens of kilobytes. So the data is produced on +//! demand and thrown away. +//! +//! The table is **table version 6, timeline layout 1**, whose commit metadata is +//! JSON. Version 8 encodes it as Avro, which a generator would have to +//! reimplement; the read path under test does not depend on which, so the +//! simpler one is the honest choice. +//! +//! ```text +//! cargo run --release -p fg-bench --bin fg-gen -- \ +//! --out /tmp/bench_table --files 10 --total-bytes 1073741824 +//! ``` + +use std::collections::HashMap; +use std::fs; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use arrow_array::{ArrayRef, Float64Array, Int64Array, RecordBatch, StringArray}; +use arrow_schema::{DataType, Field, Schema}; +use clap::Parser; +use parquet::arrow::ArrowWriter; +use parquet::basic::Compression; +use parquet::file::properties::WriterProperties; + +#[derive(Parser, Debug)] +#[command(name = "fg-gen", about = "Generate a Hudi table of a target size")] +struct Args { + /// Directory to create the table in. Must not already exist. + #[arg(long)] + out: String, + /// Number of file groups, one base file each. + #[arg(long, default_value_t = 10)] + files: usize, + /// Approximate total size of the base files, in bytes. + #[arg(long, default_value_t = 1024 * 1024 * 1024)] + total_bytes: u64, + /// Log files to write per file group. 0 keeps the table copy-on-write; any + /// positive number makes it merge-on-read, so the read builds a merge map -- + /// the structure `hoodie.memory.merge.max.size` bounds and which a streaming + /// read does not avoid. + #[arg(long, default_value_t = 0)] + log_files: usize, + /// Shift the log records' key range by this many keys, making them inserts + /// rather than updates. + /// + /// Zero (the default) reuses the base file's keys, so every log record + /// updates a base row and the merged row count equals the base row count. + /// A table generated that way cannot show whether excluding the base file + /// worked: excluding it leaves updates with nothing to update, so the read + /// returns nothing either way. Offsetting past the base's key range gives + /// log records that survive on their own. + #[arg(long, default_value_t = 0)] + log_key_offset: u64, + + /// Records per log block. Each log file gets one block. + #[arg(long, default_value_t = 200_000)] + log_records: usize, + /// Rows per parquet row group. Smaller groups let a reader release memory + /// sooner, so this is the knob that makes a bounded read possible at all. + #[arg(long, default_value_t = 100_000)] + row_group_rows: usize, +} + +const INSTANT: &str = "20250101000000000"; +/// The log files are written at a later instant than the base file, so their +/// records win under commit-time ordering and the merge actually has work to do. +const LOG_INSTANT: &str = "20250102000000000"; Review Comment: Confirmed, with numbers, and fixed in the generator. On a table from `fg-gen --files 2 --log-files 2 --log-records 5000 --log-key-offset 50000000`: - `Table::read` with the delta commit: **90,000** rows - same table, delta commit removed: **80,000** rows So the 10,000 log records were being silently dropped on the `Table` path while the standalone reader returned all of them — measurements taken on such a table would have been comparing two different row sets. `fg-gen` now writes `{LOG_INSTANT}.deltacommit` whenever it writes log files. Its write stats carry `baseFile` alongside `path`, because schema resolution reads the first write stat's `path`, gets nothing from a `.log.` extension, and falls through to the base file — without it the generated table loses its resolvable schema once the delta commit becomes the latest. Regression test `benchmark/filegroup/tests/fg_gen_timeline.rs` runs the generator and asserts the delta commit exists for a MOR table and does not for `--log-files 0`. ########## crates/core/src/metadata/table/mod.rs: ########## @@ -233,18 +271,104 @@ impl Table { async fn read_files_partition( &self, keys: &[&str], + valid_instants: &HashSet<String>, ) -> Result<HashMap<String, FilesPartitionRecord>> { - let Some(timestamp) = self.timeline.get_latest_commit_timestamp_as_option() else { + let Some((reader, file_slices)) = self.files_partition_reader().await? else { return Ok(HashMap::new()); }; + let reader = reader.with_valid_instants(valid_instants.clone()); + // Each slice holds a disjoint set of keys -- a metadata partition shards + // by hashing the record key -- so the per-slice maps combine by insertion + // with no merge rule needed. Bounded fan-out, the same ceiling the data + // read uses, so a sharded partition cannot open more readers at once than + // a table scan would. + // Only the shards the keys route to. On a partition with ten file + // groups a single-key lookup opens one slice, not ten. + let targets = routing::slices_for_keys(&file_slices, keys); + let per_slice = crate::util::concurrency::bounded_in_order( + &targets, + self.bounded_read_concurrency_for(&targets), + |file_slice| reader.read_files_partition(file_slice, keys), + ) + .await?; + Ok(per_slice.into_iter().flatten().collect()) + } + + /// The `files` partition's merged batch, for a caller that wants Arrow. + /// + /// An empty batch when the metadata table has no commits, matching the empty + /// map the decoded form returns in that case. + /// + /// # Note + /// Must be called on a METADATA table instance. + pub(crate) async fn read_files_partition_batch( + &self, + keys: &[&str], + valid_instants: &HashSet<String>, + ) -> Result<arrow_array::RecordBatch> { + let Some((reader, file_slices)) = self.files_partition_reader().await? else { + return Ok(arrow_array::RecordBatch::new_empty(std::sync::Arc::new( + Schema::empty(), + ))); + }; + let reader = reader.with_valid_instants(valid_instants.clone()); + let targets = routing::slices_for_keys(&file_slices, keys); + let batches = crate::util::concurrency::bounded_in_order( + &targets, + self.bounded_read_concurrency_for(&targets), + |file_slice| reader.read_files_partition_batch(file_slice, keys), + ) + .await?; + concat_metadata_batches(batches) + } + + /// Resolve the `files` partition's single file slice and build a reader for it. Review Comment: All five fixed: - `files_partition_reader` — the superseded single-slice block above the current one is gone; one doc, and it says "every file slice" rather than "single file slice". - the NOTE — rewritten to name `MetadataTableV2Reader::with_valid_instants`, which is what actually attaches the set, and to say why this constructor cannot (it holds only the metadata table's timeline). - `read_files_partition` — the duplicated paragraph merged. - `bounded_read_concurrency_for` — its doc had been spliced into the middle of `bounded_read_concurrency`'s; the two are separated again. - `read_file_slices_bounded` — now links `Self::bounded_read_concurrency`. Checked `cargo doc` before and after: 13 unresolved links both ways, all pre-existing in files this PR does not touch. ########## crates/core/src/table/mod.rs: ########## @@ -842,25 +851,67 @@ impl Table { FileGroupReader::new_with_overrides(self.hudi_configs.clone(), hudi_opts, storage_opts) } + /// The scan's memory budget, when the table set one. + /// + /// Absent means `hoodie.read.file.slice.read.concurrency` stands on its own, + /// which is the behaviour every existing table keeps. + fn scan_max_memory_size(&self) -> Option<u64> { + self.hudi_configs + .try_get(HudiReadConfig::ScanMaxMemorySize) + .ok() + .flatten() + .map(|v| -> usize { v.into() }) + .map(|v| v as u64) + } + /// How many file slices to read at once. /// /// `try_join_all` over every slice was unbounded, so a table with a thousand /// file groups issued a thousand concurrent reads and held a thousand merged - /// batches at the same time — and under file group reader version 2 each of those - /// also carries its own merge map and, on spill, its own RocksDB instance. - /// `hoodie.read.file.slice.read.concurrency` already bounds the DataFusion - /// scan; honoring it here puts the direct and Python read paths on the same - /// knob instead of leaving them unbounded. - fn file_slice_read_concurrency(&self) -> usize { - let configured: usize = self + /// batches at once — and under file group reader version 2 each of those also + /// carries its own merge map and, on spill, its own RocksDB instance. + /// + /// The same derivation the DataFusion plan uses, so the two paths cannot + /// bound a scan differently. This is the **only** way to obtain a + /// concurrency for a read: the raw `hoodie.read.file.slice.read.concurrency` + /// lookup lives inside it rather than in a method of its own, so a read + /// cannot reach the ceiling while bypassing the budget. A test can miss that + /// bypass — one did — but a compile error cannot. One `Table` read is one partition's worth of + /// work, so the budget is not divided further here — dividing twice would + /// bound this path below what the caller asked for. + /// + /// Separated from the read so it can be asserted directly: it is a single + /// call whose absence compiles perfectly, and the arithmetic behind it is + /// tested elsewhere in isolation, which says nothing about whether anything + /// calls it. + /// [`Self::bounded_read_concurrency`] for a borrowed selection. + /// + /// The metadata read routes keys to a subset of a partition's slices, so it + /// holds `&FileSlice` rather than owning them. The budget must see that + /// subset, not the whole partition: routing a single key to one shard should + /// admit on that shard's cost. + pub(crate) fn bounded_read_concurrency_for(&self, file_slices: &[&FileSlice]) -> usize { Review Comment: Done. Both forms now funnel through `bounded_read_concurrency_from_log_bytes`, which takes an `Iterator<Item = Option<u64>>`, so each derives the log-size vector from its own borrows and the owned round-trip is gone. Admission needs nothing from a slice except its log size, so neither caller has to own one. ########## crates/datafusion/src/hudi_exec.rs: ########## @@ -293,11 +317,29 @@ impl ExecutionPlan for HudiScanExec { let reader = self.file_group_reader.clone(); let options = self.read_options.clone(); - let concurrency = self + let planned = self .file_slice_read_concurrency .min(file_slices.len()) .max(1); + // The plan-time budget cannot see what else is running. Registering with + // the pool lets this scan account for what it is about to hold and, when + // the pool is already under pressure, read fewer slices at once instead + // of allocating anyway. Dropping the reservation with the stream returns + // the bytes. + let mut reservation = MemoryConsumer::new(format!("HudiScanExec[{partition}]")) + .register(context.memory_pool()); + let per_slice = file_slices + .iter() + .map(|s| { + hudi_core::file_group::admission::estimated_slice_bytes( + s.log_size_bytes().unwrap_or(0), Review Comment: Agreed — the two halves of the same budget disagreed about what an unmeasured slice costs. Extracted the rule into `HudiScanExec::planned_slices`, which applies `slices_in_flight`'s: any slice with an unrecorded log size drops the plan to one, instead of letting `unwrap_or(0)` reserve just the 33 MiB floor for the slices least is known about. Test `an_unmeasured_log_file_plans_one_slice` pins them together — 2 slices with known log sizes plan 2 under a ceiling of 4; adding one unmeasured slice drops it to 1. -- 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]
