yihua commented on code in PR #706:
URL: https://github.com/apache/hudi-rs/pull/706#discussion_r3918874882
##########
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? {
+ if instant.timestamp > earliest {
+ valid.extend(self.commits_rolled_back_by(&instant).await?);
+ }
+ }
+
+ // 4. The metadata table's own rollback and restore instants.
+ for instant in mdt.rollback_and_restore_instants().await? {
+ valid.insert(instant.timestamp.clone());
+ }
+
+ Ok(valid)
+ }
+
+ /// Source 1 — every completed data instant.
+ ///
+ /// The metadata table is written before the data instant commits, so a log
+ /// block whose instant never completed must not be read.
+ fn valid_from_completed_data_instants(&self) -> HashSet<String> {
+ self.timeline
+ .completed_commits
+ .iter()
+ .map(|instant| instant.timestamp.clone())
+ .collect()
+ }
+
+ /// Source 2 — completed metadata delta commits with no *pending* data
+ /// instant of the same name.
+ ///
+ /// An indexing delta commit has no data instant at all and is valid. One
+ /// whose data instant is still pending is the case where the data write
+ /// failed after the metadata write committed, and must not be read — that
+ /// exclusion is the whole content of this source.
+ fn valid_from_mdt_delta_commits(&self, mdt: &Table) -> HashSet<String> {
+ let pending = &self.timeline.pending_instants;
+ mdt.timeline
+ .completed_commits
+ .iter()
+ .filter(|i| i.action == Action::DeltaCommit &&
!pending.contains(&i.timestamp))
+ .map(|i| i.timestamp.clone())
+ .collect()
+ }
+
+ /// Source 5 — metadata delta commits written outside the data timeline.
+ fn valid_from_sentinel_commits(mdt: &Table) -> HashSet<String> {
+ mdt.timeline
+ .completed_commits
+ .iter()
+ .filter(|i| i.timestamp.starts_with(SOLO_COMMIT_TIMESTAMP))
+ .map(|i| i.timestamp.clone())
+ .collect()
+ }
+
+ /// Completed rollback and restore instants on this table's timeline.
+ ///
+ /// Loaded with a selector naming those actions, because they are not in
+ /// `DEFAULT_LOADING_ACTIONS` -- a rollback is not a commit, and putting it
+ /// there would change every existing read.
+ async fn rollback_and_restore_instants(&self) -> Result<Vec<Instant>> {
+ let selector = TimelineSelector::actions_in_range(
+ &[Action::Rollback, Action::Restore],
+ &[State::Completed],
+ self.hudi_configs.clone(),
+ None,
+ None,
+ )?;
+ self.timeline.load_instants(&selector, false).await
+ }
+
+ /// The commits one rollback or restore instant rolled back.
+ ///
+ /// A rollback names them in its own metadata, falling back to its
+ /// `.requested` plan when the completed file is empty -- Java does the
same
+ /// (`getRollbackedCommits`, HoodieTableMetadataUtil:2158). A restore is
+ /// several rollbacks, so its commits are the union of theirs.
+ ///
+ /// An unreadable instant yields nothing rather than failing the read: a
+ /// rollback we cannot parse should not make the whole table unreadable,
and
+ /// the cost is the same conservative exclusion that existed before.
+ async fn commits_rolled_back_by(&self, instant: &Instant) ->
Result<Vec<String>> {
Review Comment:
Java's getRollbackedCommits throws HoodieMetadataException when a
rollback/restore's metadata cannot be read, whereas this returns an empty list
at debug level (the restore arm's unwrap_or_default likewise swallows a parse
failure) — the read then silently serves listings missing the re-applied log
blocks, which is wrong query results rather than a failed read. Would you
consider propagating the error like Java, or at minimum a warn! so a corrupt
rollback file is visible?
##########
benchmark/filegroup/SKILL.md:
##########
@@ -0,0 +1,105 @@
+<!--
+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.
+-->
+
+---
+name: fg-bench-local
+description: Run the hudi-rs file-group-reader performance benchmark locally —
build the readers, generate single-FG MOR datasets, and run the containerized
resource-tier matrix comparing the Rust hudi-rs reader against the Java gold
readers (java-avro pure-JVM, java-spark columnar). Use when the user asks to
"run the fg-bench benchmark", "benchmark the hudi-rs reader locally", "compare
hudi-rs vs the Java reader", "run the FG reader perf matrix", "regenerate the
bench datasets", or similar. Encodes the moved paths, the docker/sudo shim, and
the hard-won gotchas (cp exec-bit, java-spark compaction.max.size budget key,
cgroup-peak vs true RSS, JDK17 Spark flags) that otherwise silently break a run
or make the numbers wrong.
+---
+
+# fg-bench local performance benchmark
+
+Goal: measure the **hudi-rs MOR file-group reader** (`HoodieFileGroupReader`)
in isolation under controlled CPU/memory tiers, and compare it head-to-head
against the two **Java gold readers**. Three readers, one JSON schema, one
report:
+
+| ref | what | engine | in-memory format |
+|---|---|---|---|
+| `current` | hudi-rs reader (all A1–A6 fixes), `--streaming` | none (Rust) |
Arrow columnar |
+| `java` | Java `HoodieFileGroupReader` via `HoodieAvroReaderContext` | none
(pure JVM) | Avro row (`IndexedRecord`) |
+| `java-spark` | production columnar path via embedded `local[N]` Spark |
in-process Spark driver | `InternalRow`/`ColumnarBatch` |
+
+`baseline` (pre-fix Rust binary) is a 4th optional ref if a baseline binary is
staged.
+
+## Paths (verified 2026-06-11; the original `ws3`/`operations` paths are GONE)
+- **Repo**: `/home/ubuntu/lin_root/hudi-rs-internal` — branch
`lin/mor-on-0607` already contains the A6 work (`benchmark/filegroup/` = the
`fg-bench` crate).
Review Comment:
This file (plus the ENG-XXXXX / milestone references in README.md, main.rs
and the fixture docs) points at internal machines, branches and trackers —
/home/ubuntu/lin_root/hudi-rs-internal, run_matrix.sh, ENG-42991 — none of
which exist for apache/hudi-rs readers. Could you strip these down to what is
reproducible from this repo (or drop SKILL.md entirely)?
##########
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:
non-blocking: a few editing leftovers in the docs — files_partition_reader
carries the old "single file slice" doc on top of the new one, the NOTE below
it names a partition_reader_with_valid_instants that doesn't exist,
MetadataTableV2Reader::read_files_partition has its doc block twice,
bounded_read_concurrency_for's doc has stray fragments, and
read_file_slices_bounded still links [`Self::file_slice_read_concurrency`],
which is gone. Worth one cleanup pass, since these will actively mislead the
next reader.
##########
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:
The generated log records carry an instant that never commits on the
timeline, so they're only readable because the standalone FileGroupReader's
uncommitted-block gate degrades to admit-all without timeline inputs — a
Table::read of the same generated table would drop them. Should fg-gen also
write the {instant}.deltacommit, or the README note the table is valid only
through the standalone reader path?
##########
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:
nit: this clones every routed slice only so bounded_read_concurrency can
take &[FileSlice]; computing the log_size_bytes vector directly from the
borrows would avoid the copies.
##########
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:
non-blocking: --streaming and --async-stream document two different
mechanisms (sync open() on OS threads vs open_blocking_stream()), but both
reduce to the same async read_file_slice_stream call — a sync-vs-async
comparison run with these flags measures the same path twice. Either wire the
second mode or fold the flags into one; the README's stage-timings table has
the same drift.
##########
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? {
+ if instant.timestamp > earliest {
+ valid.extend(self.commits_rolled_back_by(&instant).await?);
+ }
+ }
+
+ // 4. The metadata table's own rollback and restore instants.
+ for instant in mdt.rollback_and_restore_instants().await? {
+ valid.insert(instant.timestamp.clone());
+ }
+
+ Ok(valid)
+ }
+
+ /// Source 1 — every completed data instant.
+ ///
+ /// The metadata table is written before the data instant commits, so a log
+ /// block whose instant never completed must not be read.
+ fn valid_from_completed_data_instants(&self) -> HashSet<String> {
+ self.timeline
+ .completed_commits
+ .iter()
+ .map(|instant| instant.timestamp.clone())
+ .collect()
+ }
+
+ /// Source 2 — completed metadata delta commits with no *pending* data
+ /// instant of the same name.
+ ///
+ /// An indexing delta commit has no data instant at all and is valid. One
+ /// whose data instant is still pending is the case where the data write
+ /// failed after the metadata write committed, and must not be read — that
+ /// exclusion is the whole content of this source.
+ fn valid_from_mdt_delta_commits(&self, mdt: &Table) -> HashSet<String> {
Review Comment:
The exclusion here checks `pending_instants`, which is built from
DEFAULT_LOADING_ACTIONS (commit/deltacommit/replacecommit), while Java builds
`datasetPendingInstants` from `filterInflightsAndRequested()` over every action
— so a completed MDT delta commit for a data compaction or clean that crashed
mid-commit (a `.compaction.inflight` on the data timeline) is admitted here
where Java excludes it. Today the files partition is re-gated downstream so
this stays latent, but once column_stats/record_index gain consumers it becomes
a silently wrong read — could we widen the pending listing for this check, or
at least leave a comment naming the gap?
##########
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:
non-blocking: Java bounds this scan by the later of the earliest valid
instant and the latest MDT compaction time (rollbackFilterThreshold in
getValidInstantTimestamps), specifically to avoid sequential reads of old
rollback files during metadata reads — this loop reads every rollback newer
than the earliest valid instant, one awaited GET at a time, per metadata read.
Worth adopting the compaction bound; the comment above saying Java bounds it
"the same way" would then also be accurate.
##########
benchmark/filegroup/compare.py:
##########
@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Compare two or more fg-bench JSON reports.
+
+Prints a table of median wall, per-stage timing breakdown, and peak RSS, with a
+percent delta of every file versus the FIRST file (the baseline). Stdlib only.
+
+Usage:
+ compare.py baseline.json candidate.json [more.json ...]
+"""
+
+import json
+import sys
+
+
+# (json_key, display_label) — stage timings are medianed over measured
+# (non-warmup) iterations, same as wall.
+STAGE_FIELDS = [
+ ("base_read_ms", "base_read"),
+ ("log_block_read_ms", "log_read"),
+ ("log_block_decode_ms", "log_decode"),
+ ("merge_insert_ms", "merge_ins"),
+ ("final_merge_ms", "final_mrg"),
+ ("output_build_ms", "out_build"),
+]
+
+
+def _median(values):
+ s = sorted(values)
+ n = len(s)
+ if n == 0:
+ return 0
+ if n % 2 == 1:
+ return s[n // 2]
+ return (s[n // 2 - 1] + s[n // 2]) / 2
+
+
+def load(path):
+ with open(path) as f:
+ report = json.load(f)
+ measured = [it for it in report["iterations"] if not it.get("warmup")]
+ if not measured: # only a warmup iteration exists
+ measured = report["iterations"]
+
+ row = {
+ "file": path,
+ "wall_ms": _median([it["wall_ms"] for it in measured]),
+ "max_rss_kb": max((it["max_rss_kb"] for it in measured), default=0),
+ "rows": measured[0]["rows"] if measured else 0,
+ "contended": report.get("contended", False),
+ }
+ for key, _ in STAGE_FIELDS:
+ row[key] = _median([it["read_stats"][key] for it in measured])
Review Comment:
`it["read_stats"][key]` raises KeyError on fg-bench's own reports —
IterationReport doesn't emit a read_stats object (and core's stats fields are
_us-suffixed, not _ms), so the README's A/B workflow crashes on its first
input. Could this be aligned with the JSON the binary actually prints?
##########
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:
With the default --spill-dir /tmp, the watcher reports the directory's
absolute size, so any host with a non-empty /tmp reports spilled=true with a
multi-GB spill_peak_bytes before the read does anything (I saw 48.5 GB) — and
it re-walks all of /tmp every 100 ms. Defaulting to a fresh per-run subdir, or
reporting growth over the initial size, would make the detector mean what it
says.
##########
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) {
Review Comment:
Allocations made before ALLOC_CAP is set (tokio startup, the env string
itself) never enter ALLOC_LIVE, but once the cap is set their frees are
subtracted here, so the counter wraps and every later `live > cap` check fails
— a capped run aborts with "memory allocation of 64 bytes failed" regardless of
real usage (reproduced with a 512 MiB cap on the 36 MB fixture read). A
saturating subtract via fetch_update, or counting unconditionally from process
start, would fix it.
##########
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:
non-blocking: valid_instants=None with metadata_table=Some(..) silently
falls back to an empty set, which admits no metadata log blocks at all — the
strongest filtering, where None naturally reads as "no filtering".
fs_view_get_file_slices_with_metadata_table already exercises that degraded
path without being able to notice; bundling the metadata table and its valid
set into one parameter would make the bad state unrepresentable.
##########
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:
non-blocking: slices_in_flight treats an unrecorded log size as "cannot
estimate, admit one", but this reservation treats it as zero bytes — an
unmeasured log-heavy slice reserves only the 33 MiB floor, understating
pressure on the pool. Using the same conservative rule in both places would
keep the estimate's meaning consistent.
--
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]
