linliu-code commented on code in PR #706:
URL: https://github.com/apache/hudi-rs/pull/706#discussion_r3920272634
##########
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:
Confirmed, and widened rather than commented. `Timeline::pending_instants`
is built from `DEFAULT_LOADING_ACTIONS`, and the `Action` enum has no variant
for compaction/clean/cluster at all — `Action::from_str` rejects them, so a
selector-based listing drops those files before anything can count them.
`Timeline::all_pending_instant_times` now lists the active timeline and
parses instant times structurally (state from the suffix, instant time from
what precedes the first `.`, minus the `_completionTime` layout 2 appends).
That matches what Java actually needs from `filterInflightsAndRequested()` —
the timestamps, never the actions.
Deliberately a separate set, not a widening of `pending_instants`: that
field also feeds `completion_gate_inputs().inflight_instants`, so widening it
in place would start rejecting log blocks the MOR completion gate admits today.
Only source 2 takes the wide view.
Test is
`source_two_excludes_a_data_instant_pending_under_a_non_commit_action`: a
metadata delta commit at an instant whose only data-side file is
`{ts}.compaction.inflight`. It asserts the wide set sees it, that
`pending_instants` does *not* (pinning the gap itself), and that the instant is
excluded. Plus two loader unit tests covering both timeline layouts and the
completed-cancels-pending subtraction.
##########
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:
Propagated, matching Java. `commits_rolled_back_by` now returns
`CoreError::MetadataTable` — the counterpart to `HoodieMetadataException` —
when the instant will not load, when a restore's metadata will not parse, and
when the rollback *plan* will not load or parse.
The one fallback Java keeps is kept: an unreadable or empty completed
rollback file falls back to its `.requested` plan, and that now logs at `warn!`
(Java's `LOG.warn`) instead of passing in silence. A plan naming no instant
stays a non-error — that is a rollback of nothing, not a failure.
Your framing is the reason: the swallow dropped exactly the commits rollback
re-applied, so the listing came back missing their log blocks — a wrong answer
presented as a right one, which is worse than a read that stops.
Regression test
`an_unreadable_rollback_fails_the_read_rather_than_shrinking_the_set` writes a
corrupt completed rollback with no plan beside it and asserts the error names
the instant.
##########
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:
Dropped SKILL.md entirely — it was wholly about one internal host. Also
stripped the `ENG-` ids, the `A1`/`A3`/`A6e` milestone labels, and the `perf
M2` / `02-bench-harness-design.md` / `03-milestones.md` references from
README.md and main.rs.
The README needed more than a trim: its "Stage timings" table and the
`read_stats` field documented an interface the harness lost when it moved to
the public reader surface. Rewritten to describe what the binary actually emits
and what is reproducible from this repo alone.
##########
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:
Good catch, and reproduced. Minimal repro of the mechanism:
```
cap installed; LIVE=0
after freeing pre-cap alloc; LIVE=18446744073701163072
now allocating 1 KiB under a 512 MiB cap...
memory allocation of 1024 bytes failed
memory allocation of 64 bytes failed
```
`#[tokio::main]` builds the runtime before the body runs, so those
allocations — and the env string itself, dropped at the end of the `if let`
*after* the store — are never added but are subtracted once the cap is set.
Went with counting unconditionally from the process's first allocation
rather than a saturating subtract: with balanced accounting the counter cannot
underflow, so no CAS loop on the dealloc path. The cap is still only *checked*
when set.
Worth noting the severity is data-dependent, which is probably why it
survived some runs — `fetch_add` wraps, so a small underflow is healed by the
next allocation larger than it. A multi-MB one is not, and that is your 64-byte
abort.
##########
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:
Right on both counts — `IterationReport` has no `read_stats` field, and the
stage timings it named went away when the harness moved to the public reader
surface (`HoodieReadStats` is not on it). README.md:84 was advertising the same
phantom object.
`compare.py` now compares what the binary emits: `wall_ms`, `user_ms`,
`sys_ms` and `spill_peak_bytes` by median, `max_rss_kb` by peak (it is a
high-water mark, so the max across iterations is the number a bound has to hold
against), plus rows/spilled/contended on the footer. Reads use `.get(key, 0)`
so a report from an older build still compares instead of raising.
Verified end to end against reports the binary actually wrote.
--
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]
