This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new b0f89adf feat(reader): merge-aware pushdown gate, row-group pruning,
and read-volume counters (#708)
b0f89adf is described below
commit b0f89adf55a299845362b765eea641f7b4bf6f42
Author: Lin Liu <[email protected]>
AuthorDate: Wed Sep 2 17:52:39 2026 -0700
feat(reader): merge-aware pushdown gate, row-group pruning, and read-volume
counters (#708)
Co-authored-by: Y Ethan Guo <[email protected]>
---
crates/core/src/file_group/base_file/parquet.rs | 198 +++++++++-
crates/core/src/file_group/base_file/reader.rs | 41 ++-
crates/core/src/file_group/reader_v2/engine.rs | 403 +++++++++++++++++++--
crates/core/src/file_group/reader_v2/harness.rs | 31 +-
.../core/src/file_group/reader_v2/harness_tests.rs | 57 +--
.../src/file_group/reader_v2/log_record_reader.rs | 15 +-
.../src/file_group/reader_v2/reader_context.rs | 89 +----
crates/core/src/file_group/reader_v2/resolver.rs | 4 +-
crates/core/src/storage/mod.rs | 131 +++++++
9 files changed, 820 insertions(+), 149 deletions(-)
diff --git a/crates/core/src/file_group/base_file/parquet.rs
b/crates/core/src/file_group/base_file/parquet.rs
index 56c9be2f..c2b5e494 100644
--- a/crates/core/src/file_group/base_file/parquet.rs
+++ b/crates/core/src/file_group/base_file/parquet.rs
@@ -31,11 +31,65 @@ use parquet::file::metadata::ParquetMetaData;
use super::reader::{BaseFileReadOptions, BaseFileReader, BaseFileStream};
use crate::schema::parquet_list_norm::normalize_parquet_metadata;
use crate::statistics::StatisticsContainer;
+use crate::storage::ReadVolume;
use crate::storage::Storage;
use crate::storage::error::{Result, StorageError};
use crate::storage::file_metadata::FileMetadata;
use crate::storage::util::join_url_segments;
+/// `AsyncFileReader` wrapper that accumulates read volume into a shared
+/// [`ReadVolume`].
+///
+/// Delegates everything and changes no behaviour. Counting happens here rather
+/// than inside the object store because the store is shared between readers
and
+/// therefore cannot carry per-read counters.
+struct CountingReader<R: AsyncFileReader> {
+ inner: R,
+ volume: Arc<ReadVolume>,
+}
+
+impl<R: AsyncFileReader> AsyncFileReader for CountingReader<R> {
+ fn get_bytes(
+ &mut self,
+ range: std::ops::Range<u64>,
+ ) -> BoxFuture<'_, parquet::errors::Result<bytes::Bytes>> {
+ let volume = self.volume.clone();
+ let fut = self.inner.get_bytes(range);
+ Box::pin(async move {
+ let bytes = fut.await?;
+ volume.add_bytes(bytes.len() as u64);
+ Ok(bytes)
+ })
+ }
+
+ fn get_byte_ranges(
+ &mut self,
+ ranges: Vec<std::ops::Range<u64>>,
+ ) -> BoxFuture<'_, parquet::errors::Result<Vec<bytes::Bytes>>> {
+ let volume = self.volume.clone();
+ let fut = self.inner.get_byte_ranges(ranges);
+ Box::pin(async move {
+ let chunks = fut.await?;
+ // One call, many ranges: count the call once and every byte it
+ // returned, so `io_calls` stays a count of round trips.
+ let total: u64 = chunks.iter().map(|b| b.len() as u64).sum();
+ volume.add_bytes(total);
+ Ok(chunks)
+ })
+ }
+
+ fn get_metadata<'a>(
+ &'a mut self,
+ options: Option<&'a ArrowReaderOptions>,
+ ) -> BoxFuture<'a, parquet::errors::Result<Arc<ParquetMetaData>>> {
+ self.inner.get_metadata(options)
+ }
+}
+
+/// The builder every read here is driven from: a parquet object reader with
the
+/// read-volume counters wrapped around it.
+type CountedBuilder =
ParquetRecordBatchStreamBuilder<CountingReader<ParquetObjectReader>>;
+
/// Parquet implementation of [`BaseFileReader`].
///
/// Reads Parquet files directly via `object_store` and the `parquet` crate.
@@ -78,7 +132,7 @@ impl ParquetBaseFileReader {
obj_path: ObjPath,
file_size: u64,
row_index_column: Option<&str>,
- ) -> Result<ParquetRecordBatchStreamBuilder<ParquetObjectReader>> {
+ ) -> Result<CountedBuilder> {
let mut reader =
ParquetObjectReader::new(self.storage.object_store.clone(), obj_path)
.with_file_size(file_size);
@@ -94,6 +148,11 @@ impl ParquetBaseFileReader {
normalized,
Self::arrow_reader_options(row_index_column)?,
)?;
+
+ let reader = CountingReader {
+ inner: reader,
+ volume: self.storage.read_volume.clone(),
+ };
Ok(ParquetRecordBatchStreamBuilder::new_with_metadata(
reader,
arrow_metadata,
@@ -104,16 +163,28 @@ impl ParquetBaseFileReader {
&self,
relative_path: &str,
row_index_column: Option<&str>,
- ) -> Result<ParquetRecordBatchStreamBuilder<ParquetObjectReader>> {
+ ) -> Result<CountedBuilder> {
let (obj_path, file_size) =
self.object_path_and_size(relative_path).await?;
self.open_builder_with_size(obj_path, file_size, row_index_column)
.await
}
fn apply_options(
- mut builder: ParquetRecordBatchStreamBuilder<ParquetObjectReader>,
+ &self,
+ mut builder: CountedBuilder,
options: &BaseFileReadOptions,
- ) -> Result<ParquetRecordBatchStreamBuilder<ParquetObjectReader>> {
+ ) -> Result<CountedBuilder> {
+ // What the file holds, taken from the footer that has already been
+ // fetched, so it costs no extra IO. Recorded here rather than at every
+ // builder open so it counts once per read OF THE DATA: opening the
same
+ // file again for its schema alone would otherwise inflate the
+ // denominator that `row_groups_read` and `rows_out` are read against.
+ let metadata = builder.metadata();
+ self.storage.read_volume.record_file_shape(
+ metadata.num_row_groups() as u64,
+ metadata.file_metadata().num_rows().max(0) as u64,
+ );
+
if let Some(batch_size) = options.batch_size {
builder = builder.with_batch_size(batch_size);
}
@@ -150,6 +221,34 @@ impl ParquetBaseFileReader {
builder = builder.with_projection(projection_mask);
}
+ // Prune row groups from footer statistics, BEFORE the row filter is
+ // installed: a group excluded here is never fetched, so the filter
only
+ // ever sees groups that survived.
+ let volume = &self.storage.read_volume;
+ let total_row_groups = builder.metadata().num_row_groups();
+ let mut row_groups_read = total_row_groups;
+ if let Some(keep) =
options.row_group_selector.as_ref().and_then(|select| {
+ // Count the CALL, not just a successful prune, so "ran and found
+ // nothing" stays distinguishable from "never installed".
+ volume.record_selector_call();
+ select(builder.metadata())
+ }) {
+ // A selector that names a row group the file does not have is a
bug
+ // in the selector. parquet-rs does not validate the indices; the
+ // read panics when the stream reaches the bad index, release
builds
+ // included. The `debug_assert!` surfaces the bug at this choke
+ // point with a clear message in tests; no release check is added
+ // because the failure is already loud, a panic rather than lost
+ // rows.
+ debug_assert!(
+ keep.iter().all(|&i| i < total_row_groups),
+ "selector returned an out-of-range row-group index"
+ );
+ row_groups_read = keep.len();
+ builder = builder.with_row_groups(keep);
+ }
+ volume.add_row_groups_read(row_groups_read as u64);
+
// Built here rather than by the caller because the predicate has to be
// resolved against the file's own schema, which only exists once the
// footer is open. A builder that returns `None` — typically because
the
@@ -229,7 +328,7 @@ impl BaseFileReader for ParquetBaseFileReader {
let builder = self
.open_builder(relative_path,
options.row_index_column.as_deref())
.await?;
- let builder = Self::apply_options(builder, &options)?;
+ let builder = self.apply_options(builder, &options)?;
let full_schema = builder.schema().clone();
let stream = builder.build()?;
let schema = Self::schema_with_row_index(
@@ -237,14 +336,31 @@ impl BaseFileReader for ParquetBaseFileReader {
&full_schema,
options.row_index_column.as_deref(),
)?;
+ // Rows the stream actually yields — after any row filter. Against
+ // `file_rows` this is the read's selectivity; against
`bytes_read`,
+ // what that selectivity cost.
+ let volume = self.storage.read_volume.clone();
let mapped_stream = stream
- .map(|result| result.map_err(StorageError::from))
+ .map(move |result| {
+ let batch = result.map_err(StorageError::from)?;
+ volume.add_rows_out(batch.num_rows() as u64);
+ Ok(batch)
+ })
.boxed();
Ok(BaseFileStream::new(schema, mapped_stream))
})
}
+ /// Answered from the footer alone: no stream is built, and no read-volume
+ /// counter moves for a call that reads no data.
+ fn read_schema<'a>(
+ &'a self,
+ relative_path: &'a str,
+ ) -> BoxFuture<'a, Result<arrow_schema::SchemaRef>> {
+ Box::pin(async move {
Ok(Arc::new(self.get_schema(relative_path).await?)) })
+ }
+
fn get_metadata_and_stats<'a>(
&'a self,
relative_path: &'a str,
@@ -379,6 +495,76 @@ mod tests {
assert_eq!(batch.num_rows(), 0, "the predicate rejected every row");
}
+ /// The volume counters describe a real read: what the file held, what was
+ /// fetched to read it, and what came back.
+ #[tokio::test]
+ async fn read_volume_counts_what_the_read_actually_moved() {
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let storage = test_storage();
+ let volume = storage.read_volume();
+ let reader = ParquetBaseFileReader::new(storage);
+
+ let batch = reader
+ .read_data("a.parquet", BaseFileReadOptions::default())
+ .await
+ .unwrap();
+
+ assert_eq!(batch.num_rows(), 5);
+ assert_eq!(volume.file_rows.load(Relaxed), 5, "footer row count");
+ assert_eq!(volume.rows_out.load(Relaxed), 5, "every row was yielded");
+ let file_row_groups = volume.file_row_groups.load(Relaxed);
+ assert!(file_row_groups >= 1, "the file has at least one row group");
+ assert_eq!(
+ volume.row_groups_read.load(Relaxed),
+ file_row_groups,
+ "nothing prunes, so every row group is scanned"
+ );
+ assert!(volume.bytes_read.load(Relaxed) > 0, "bytes were fetched");
+ assert!(volume.io_calls.load(Relaxed) > 0, "round trips were made");
+ }
+
+ /// Why the counters exist. A `RowFilter` decides per row *after* the
+ /// predicate columns are decoded, so a filter that rejects everything
still
+ /// reads the file: `rows_out` collapses to zero while `bytes_read` does
not,
+ /// and `row_groups_read` still covers the whole file. A disposition flag
+ /// ("was a predicate pushed?") reports the same thing here as it would
for a
+ /// read that skipped the file entirely.
+ #[tokio::test]
+ async fn a_row_filter_that_rejects_everything_still_reads_the_file() {
+ use arrow_array::BooleanArray;
+ use parquet::arrow::ProjectionMask;
+ use parquet::arrow::arrow_reader::{ArrowPredicateFn, RowFilter};
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let storage = test_storage();
+ let volume = storage.read_volume();
+ let reader = ParquetBaseFileReader::new(storage);
+
+ let opts =
BaseFileReadOptions::default().with_row_filter(Arc::new(|descr, _| {
+ let mask = ProjectionMask::roots(descr, [0]);
+ Some(RowFilter::new(vec![Box::new(ArrowPredicateFn::new(
+ mask,
+ |batch| Ok(BooleanArray::from(vec![false; batch.num_rows()])),
+ ))]))
+ }));
+
+ let batch = reader.read_data("a.parquet", opts).await.unwrap();
+
+ assert_eq!(batch.num_rows(), 0);
+ assert_eq!(volume.rows_out.load(Relaxed), 0, "no row survived");
+ assert_eq!(volume.file_rows.load(Relaxed), 5, "the file still held 5");
+ assert_eq!(
+ volume.row_groups_read.load(Relaxed),
+ volume.file_row_groups.load(Relaxed),
+ "a row filter skips no row group"
+ );
+ assert!(
+ volume.bytes_read.load(Relaxed) > 0,
+ "rejecting every row still cost IO"
+ );
+ }
+
/// A builder that declines — typically because the file has none of the
/// columns the predicate names — reads every row. Returning no rows would
/// silently drop data on a file the predicate cannot speak about.
diff --git a/crates/core/src/file_group/base_file/reader.rs
b/crates/core/src/file_group/base_file/reader.rs
index d9a9c519..c062ed2a 100644
--- a/crates/core/src/file_group/base_file/reader.rs
+++ b/crates/core/src/file_group/base_file/reader.rs
@@ -31,7 +31,7 @@ use crate::config::table::BaseFileFormatValue;
use crate::statistics::StatisticsContainer;
use crate::storage::error::Result;
use crate::storage::file_metadata::FileMetadata;
-use crate::storage::{RowFilterBuilder, Storage};
+use crate::storage::{RowFilterBuilder, RowGroupSelector, Storage};
/// Which record keys a read is interested in.
///
@@ -130,6 +130,18 @@ pub struct BaseFileReadOptions {
/// excluded from [`projection`](Self::projection) matching, since the
column
/// is not one of the file's own.
pub row_index_column: Option<String>,
+ /// Picks which row groups to read, from the file's footer statistics.
+ ///
+ /// Only the Parquet reader honors this; other formats ignore it. Unlike
+ /// [`row_filter`](Self::row_filter), which discards rows once they have
been
+ /// decoded, a row group this excludes is never fetched — so the two are
+ /// complementary, and this is the one that saves IO. It runs against the
+ /// footer the reader has already read, so consulting it costs nothing.
+ ///
+ /// It carries the same safety obligation as `row_filter`, and a stronger
+ /// one: pruning removes base rows before a log merge could have updated
them
+ /// into a match. The caller installs it only when that cannot happen.
+ pub row_group_selector: Option<RowGroupSelector>,
}
// `row_filter` holds a closure, which has no `Debug`. Report whether one is
set
@@ -143,6 +155,7 @@ impl std::fmt::Debug for BaseFileReadOptions {
.field("key_predicate", &self.key_predicate)
.field("row_filter", &self.row_filter.is_some())
.field("row_index_column", &self.row_index_column)
+ .field("row_group_selector", &self.row_group_selector.is_some())
.finish()
}
}
@@ -158,6 +171,13 @@ impl BaseFileReadOptions {
self
}
+ /// Read only the row groups a selector keeps. See
+ /// [`Self::row_group_selector`].
+ pub fn with_row_group_selector(mut self, selector: RowGroupSelector) ->
Self {
+ self.row_group_selector = Some(selector);
+ self
+ }
+
/// Read only the records a key predicate admits, where the format can seek
/// by key. See [`Self::key_predicate`].
pub fn with_key_predicate(mut self, predicate: KeyPredicate) -> Self {
@@ -253,6 +273,25 @@ pub trait BaseFileReader: Send + Sync {
})
}
+ /// The schema of a base file, without reading its data.
+ ///
+ /// The default opens a stream and takes the schema it reports, which is
what
+ /// a caller would otherwise write by hand. A format that can answer from
+ /// metadata alone should override it: the caller wants the schema in
order to
+ /// decide what to read next, so the read that follows is a second open,
and
+ /// only the override keeps this one from setting up a decode nobody polls.
+ fn read_schema<'a>(
+ &'a self,
+ relative_path: &'a str,
+ ) -> BoxFuture<'a, Result<arrow_schema::SchemaRef>> {
+ Box::pin(async move {
+ let stream = self
+ .read_stream(relative_path, BaseFileReadOptions::new())
+ .await?;
+ Ok(stream.schema().clone())
+ })
+ }
+
/// Read data from a base file as a stream of RecordBatches.
fn read_stream<'a>(
&'a self,
diff --git a/crates/core/src/file_group/reader_v2/engine.rs
b/crates/core/src/file_group/reader_v2/engine.rs
index 3a4d8e28..f5f089a3 100644
--- a/crates/core/src/file_group/reader_v2/engine.rs
+++ b/crates/core/src/file_group/reader_v2/engine.rs
@@ -46,7 +46,7 @@ use crate::file_group::reader_v2::read_stats::HoodieReadStats;
use crate::file_group::reader_v2::reader_context::ReaderContext;
use crate::file_group::reader_v2::reader_parameters::ReaderParameters;
use crate::file_group::reader_v2::schema_handler::FileGroupReaderSchemaHandler;
-use crate::storage::{RowFilterBuilder, Storage};
+use crate::storage::{RowFilterBuilder, RowGroupSelector, Storage};
use arrow_array::RecordBatch;
use arrow_schema::SchemaRef;
use futures::StreamExt;
@@ -164,6 +164,7 @@ const MERGE_CHUNK_ROWS: usize = 1024;
/// not the read that dropped it.
fn base_read_options(
row_filter: Option<RowFilterBuilder>,
+ row_group_selector: Option<RowGroupSelector>,
key_predicate: Option<crate::file_group::base_file::reader::KeyPredicate>,
use_record_position: bool,
) -> BaseFileReadOptions {
@@ -172,6 +173,9 @@ fn base_read_options(
if let Some(row_filter) = row_filter {
options = options.with_row_filter(row_filter);
}
+ if let Some(row_group_selector) = row_group_selector {
+ options = options.with_row_group_selector(row_group_selector);
+ }
if let Some(key_predicate) = key_predicate {
options = options.with_key_predicate(key_predicate);
}
@@ -659,6 +663,64 @@ impl HoodieFileGroupReader {
))
}
+ /// Is it safe to push the predicate into the BASE read of this split?
+ ///
+ /// Safe exactly when no log merge can change the predicate's outcome. Two
+ /// ways that holds:
+ /// - the split carries no log files, so nothing merges and the base rows
+ /// are final; or
+ /// - the predicate references only record-key columns, which are
immutable
+ /// across upserts, so the outcome survives the merge (`mor_pk_safe`).
+ ///
+ /// Mirrors Java's `SparkFileFormatInternalRowReaderContext
+ /// .getSchemaAndFiltersForRead`, which branches on `getHasLogFiles()` and
+ /// never on the table type: `allFilters` when there are no log files,
+ /// `morFilters` when there are. The table type does not appear here
either —
+ /// a CoW slice has no log files, so it takes the first branch on its own.
+ ///
+ /// # Why the split and not `ReaderContext`
+ ///
+ /// [`ReaderContext::has_log_files`] is a different fact with a different
+ /// source: it is set by whoever built the context, whereas
+ /// [`InputSplit::log_file_paths`] is the split's own file list. Nothing
+ /// derives one from the other, so gating on the context flag would rest a
+ /// safety decision on a caller-supplied boolean. If it were ever false
for a
+ /// slice that does have logs, a non-PK predicate would reach the base read
+ /// and drop rows the merge would have updated — silently, because a filter
+ /// above the reader can only remove rows, never restore them.
+ ///
+ /// Using the split also keeps this in lock-step with
+ /// [`Self::use_record_position`], which reads the same
+ /// `input_split.has_log_files()`. The gate and the merge therefore cannot
+ /// disagree about a split.
+ ///
+ /// # Why there is no bootstrap term, unlike Java
+ ///
+ /// Java has a third branch — `!getHasLogFiles && hasRowIndexField` selects
+ /// `bootstrapSafeFilters` — because a bootstrap read pairs skeleton and
data
+ /// files by row position, and filters that physically drop records
misalign
+ /// that pairing. `has_bootstrap_base_file` reaches `ReaderContext` here
and
+ /// is consulted nowhere, so a bootstrap slice with
+ /// `needs_bootstrap_merge == false` does arrive at this gate. It is still
+ /// safe, for two independent reasons:
+ ///
+ /// 1. The positional mechanism here is the virtual `RowNumber` column,
which
+ /// carries each row's TRUE physical position and stays correct under
+ /// row-group selection and `RowFilter` pushdown. Dropping rows cannot
+ /// shift it, which is exactly the failure Java's tier avoids by not
+ /// pushing.
+ /// 2. The row-index column is requested only from
+ /// [`Self::use_record_position`], which returns false when the split
has
+ /// no log files — so on the branch this gate widens, there is no
+ /// positional pairing to misalign at all.
+ ///
+ /// Anyone adding a bootstrap term should re-check both: lifting the
+ /// `needs_bootstrap_merge` rejection in `new()` without revisiting this
gate
+ /// is how the Java hazard would arrive here.
+ fn base_read_pushdown_is_safe(&self) -> bool {
+ !self.input_split.has_log_files() || self.reader_context.mor_pk_safe
+ }
+
/// Whether this read should merge base + log records by base-file row
/// position (rather than by record key). Mirrors Java
/// `HoodieFileGroupReader`'s `setShouldMergeUseRecordPosition`:
@@ -734,26 +796,52 @@ impl HoodieFileGroupReader {
);
}
- // gate parquet RowFilter pushdown.
- // CoW: always safe (no merge).
- // MOR: safe ONLY when every column referenced by the filter is a
- // primary key (PKs are immutable across upserts, so the
predicate
- // outcome doesn't change post-merge —
`reader_context.mor_pk_safe`,
- // mirroring Java's `filterIsSafeForPrimaryKey`).
+ // gate parquet RowFilter pushdown on whether this read MERGES.
+ // No log files on this split: always safe (nothing merges). A CoW
+ // slice reaches the gate this way.
+ // Log files present: safe ONLY when every column referenced by the
+ // filter is a primary key (PKs are immutable across upserts, so
+ // the predicate outcome doesn't change post-merge —
+ // `reader_context.mor_pk_safe`, mirroring Java's
+ // `filterIsSafeForPrimaryKey`).
// Otherwise: drop the filter; the post-merge filter (Velox/Spark
above
// the FG reader) evaluates the predicate after base+log merge.
- let row_filter = if self.reader_context.can_push_row_filter() {
+ // ONE gate, bound once and shared by both mechanisms. Bound to a local
+ // rather than called twice so the sharing is structural: an edit that
+ // changes the condition for one can no longer leave the other behind,
+ // and pruning is the one that must not be left behind — it drops rows
+ // before the merge can see them.
+ let pushdown_is_safe = self.base_read_pushdown_is_safe();
+ let row_filter = if pushdown_is_safe {
self.reader_context.row_filter_builder.clone()
} else {
if self.reader_context.row_filter_builder.is_some() {
log::debug!(
- "MOR + non-PK predicate — skipping parquet \
+ "merging read with a non-PK predicate — skipping parquet \
RowFilter pushdown for base file '{path}' \
(post-merge filter still runs)"
);
}
None
};
+ let row_group_selector = if pushdown_is_safe {
+ self.reader_context.row_group_selector.clone()
+ } else {
+ // Record the suppression. The gate and the selector are each
correct
+ // alone; what does not compose is the observability.
+ // `row_group_selector_calls` exists to separate "ran and found
+ // nothing" from "never installed", and a selector the gate refuses
+ // is a third state that also reads zero calls. Counting it here
+ // keeps that counter answerable.
+ if self.reader_context.row_group_selector.is_some() {
+ self.storage.read_volume().record_selector_suppressed();
+ log::debug!(
+ "merging read with a non-PK predicate — skipping row-group
\
+ pruning for base file '{path}' (post-merge filter still
runs)"
+ );
+ }
+ None
+ };
// The key predicate needs no such gate. It narrows *which blocks are
read*
// and the reader filters the records it brings back, so it cannot
change the
@@ -780,7 +868,12 @@ impl HoodieFileGroupReader {
.base_file_reader()?
.read_data(
&path,
- base_read_options(row_filter.clone(),
key_predicate.clone(), use_position),
+ base_read_options(
+ row_filter.clone(),
+ row_group_selector.clone(),
+ key_predicate.clone(),
+ use_position,
+ ),
)
.await
.map_err(|e| {
@@ -808,9 +901,8 @@ impl HoodieFileGroupReader {
// interleaves is already in `required_schema`.
let file_schema = self
.base_file_reader()?
- .read_stream(&path, BaseFileReadOptions::new())
+ .read_schema(&path)
.await
- .map(|s| s.schema().clone())
.map_err(|e| {
CoreError::ReadFileSliceError(format!(
"Failed to read base file footer schema '{path}': {e:?}"
@@ -861,16 +953,22 @@ impl HoodieFileGroupReader {
}
// Open the base file as a stream. The whole file never lives in
memory;
- // one batch does. The (CoW-gated) RowFilter is threaded through the
- // intersection read so row groups can be pruned via column-index stats
- // (the builder resolves predicate columns by name and returns None
when
- // any referenced column is absent — safe even for evolved/added cols).
+ // one batch does. The gated RowFilter and row-group selector are both
+ // threaded through the intersection read. Only the SELECTOR skips IO:
a
+ // RowFilter decides per row once the predicate columns are decoded.
The
+ // filter builder resolves predicate columns by name and returns None
when
+ // any referenced column is absent — safe even for evolved/added cols.
let base_stream = self
.base_file_reader()?
.read_stream(
&path,
- base_read_options(row_filter.clone(), key_predicate.clone(),
use_position)
- .with_projection(intersection.fields().iter().map(|f|
f.name())),
+ base_read_options(
+ row_filter.clone(),
+ row_group_selector.clone(),
+ key_predicate.clone(),
+ use_position,
+ )
+ .with_projection(intersection.fields().iter().map(|f|
f.name())),
)
.await
.map_err(|e| {
@@ -1084,6 +1182,9 @@ pub struct HoodieFileGroupReaderBuilder {
/// at build time so the same builder is visible to base parquet reads
/// (this file) and parquet log block decodes (`log_file::content`).
row_filter_builder: Option<RowFilterBuilder>,
+ /// Set by `with_row_group_selector`; copied onto a cloned reader_context
in
+ /// `build()`, exactly like `row_filter_builder`.
+ row_group_selector: Option<RowGroupSelector>,
/// Set by `with_mor_pk_safe`; copied onto the cloned reader_context.
mor_pk_safe: Option<bool>,
}
@@ -1128,7 +1229,7 @@ impl HoodieFileGroupReaderBuilder {
/// install a parquet `RowFilter` builder.
///
/// Whether the builder is actually used at scan time is gated by
- /// `reader_context.can_push_row_filter()`:
+ /// `base_read_pushdown_is_safe()`:
/// - CoW table → always pushed
/// - MOR table → pushed only if `mor_pk_safe` is true (see
/// [`Self::with_mor_pk_safe`])
@@ -1140,6 +1241,18 @@ impl HoodieFileGroupReaderBuilder {
self
}
+ /// Install a row-group selector, pruning base reads from footer
statistics.
+ ///
+ /// Routed onto `reader_context` exactly like
+ /// [`Self::with_row_filter_builder`], and gated at scan time by the same
+ /// `base_read_pushdown_is_safe()`. Setting one without the other is
+ /// supported: they are independent mechanisms over the same predicate, and
+ /// only this one avoids IO.
+ pub fn with_row_group_selector(mut self, selector: RowGroupSelector) ->
Self {
+ self.row_group_selector = Some(selector);
+ self
+ }
+
/// mark the pushed predicate as safe for MOR (i.e. it
/// references only primary-key columns). When true, the row filter
/// pushes into both base parquet files and parquet log blocks on MOR
@@ -1167,11 +1280,17 @@ impl HoodieFileGroupReaderBuilder {
// builder API, copy them onto the reader_context. Clone-and-replace
// mirrors the same pattern HoodieFileGroupReader::new() uses to
// update the schema_handler on its reader_context.
- let reader_context = if self.row_filter_builder.is_some() ||
self.mor_pk_safe.is_some() {
+ let reader_context = if self.row_filter_builder.is_some()
+ || self.row_group_selector.is_some()
+ || self.mor_pk_safe.is_some()
+ {
let mut updated = (*reader_context).clone();
if let Some(b) = self.row_filter_builder {
updated.row_filter_builder = Some(b);
}
+ if let Some(selector) = self.row_group_selector {
+ updated.row_group_selector = Some(selector);
+ }
if let Some(s) = self.mor_pk_safe {
updated.mor_pk_safe = s;
}
@@ -1583,6 +1702,19 @@ mod tests {
InputSplit::new(None, None, vec![], "p1".to_string())
}
+ /// A split that merges: one base file and one log file. The gate reduces
to
+ /// `mor_pk_safe` only on a split like this — with no log files it is open
+ /// whatever `mor_pk_safe` says, so a PK-safety assertion made on a bare
+ /// split would pass without testing anything.
+ fn merging_input_split() -> InputSplit {
+ InputSplit::new(
+ Some("base.parquet".to_string()),
+ None,
+ vec![".log.1".to_string()],
+ "p1".to_string(),
+ )
+ }
+
fn make_row_filter_builder() -> RowFilterBuilder {
// Closure that always returns None — we only care that the builder
// was installed, not what it produces.
@@ -1611,14 +1743,14 @@ mod tests {
let reader = HoodieFileGroupReader::builder()
.with_reader_context(dummy_reader_context("MERGE_ON_READ"))
.with_storage(storage)
- .with_input_split(dummy_input_split())
+ .with_input_split(merging_input_split())
.with_row_filter_builder(make_row_filter_builder())
.with_mor_pk_safe(true)
.build()
.unwrap();
assert!(reader.reader_context.mor_pk_safe);
assert!(
- reader.reader_context.can_push_row_filter(),
+ reader.base_read_pushdown_is_safe(),
"MOR + mor_pk_safe=true must push"
);
}
@@ -1629,18 +1761,235 @@ mod tests {
let reader = HoodieFileGroupReader::builder()
.with_reader_context(dummy_reader_context("MERGE_ON_READ"))
.with_storage(storage)
- .with_input_split(dummy_input_split())
+ .with_input_split(merging_input_split())
.with_row_filter_builder(make_row_filter_builder())
// mor_pk_safe defaults to false
.build()
.unwrap();
assert!(!reader.reader_context.mor_pk_safe);
assert!(
- !reader.reader_context.can_push_row_filter(),
+ !reader.base_read_pushdown_is_safe(),
"MOR without PK-safety must NOT push (mirrors Java's morFilters
gate)"
);
}
+ /// A MOR slice with no log files does not merge, so the predicate is safe
to
+ /// push whatever `mor_pk_safe` says. Parameterized over both values so the
+ /// "does it merge" rule is shown to be independent of PK safety.
+ #[test]
+ fn base_only_mor_slice_allows_pushdown_regardless_of_pk_safety() {
+ for mor_pk_safe in [false, true] {
+ let storage =
Storage::new_with_base_url(parse_uri("file:///tmp").unwrap()).unwrap();
+ let reader = HoodieFileGroupReader::builder()
+ .with_reader_context(dummy_reader_context("MERGE_ON_READ"))
+ .with_storage(storage)
+ .with_input_split(InputSplit::new(
+ Some("base.parquet".to_string()),
+ None,
+ // No log files => no merge => nothing can flip the
predicate.
+ vec![],
+ "p1".to_string(),
+ ))
+ .with_row_filter_builder(make_row_filter_builder())
+ .with_mor_pk_safe(mor_pk_safe)
+ .build()
+ .unwrap();
+ assert!(
+ reader.base_read_pushdown_is_safe(),
+ "base-only slice must push regardless of mor_pk_safe
({mor_pk_safe})"
+ );
+ }
+ }
+
+ /// The split rule must not weaken the real MOR case: with log files
present
+ /// the merge can supersede or delete a base row, so a non-PK-safe
predicate
+ /// still may not be pushed.
+ #[test]
+ fn mor_slice_with_log_files_still_blocks_pushdown_when_not_pk_safe() {
+ let storage =
Storage::new_with_base_url(parse_uri("file:///tmp").unwrap()).unwrap();
+ let reader = HoodieFileGroupReader::builder()
+ .with_reader_context(dummy_reader_context("MERGE_ON_READ"))
+ .with_storage(storage)
+ .with_input_split(InputSplit::new(
+ Some("base.parquet".to_string()),
+ None,
+ vec![".log.1".to_string()],
+ "p1".to_string(),
+ ))
+ .with_row_filter_builder(make_row_filter_builder())
+ // mor_pk_safe defaults to false
+ .build()
+ .unwrap();
+ assert!(reader.input_split.has_log_files());
+ assert!(
+ !reader.base_read_pushdown_is_safe(),
+ "MOR with log files and no PK safety must NOT push"
+ );
+ }
+
+ #[test]
+ fn builder_routes_row_group_selector_into_reader_context() {
+ let storage =
Storage::new_with_base_url(parse_uri("file:///tmp").unwrap()).unwrap();
+ let reader = HoodieFileGroupReader::builder()
+ .with_reader_context(dummy_reader_context("MERGE_ON_READ"))
+ .with_storage(storage)
+ .with_input_split(dummy_input_split())
+ .with_row_group_selector(std::sync::Arc::new(|_| None))
+ .build()
+ .unwrap();
+ assert!(
+ reader.reader_context.row_group_selector.is_some(),
+ "with_row_group_selector should land on reader_context"
+ );
+ assert!(
+ reader.reader_context.row_filter_builder.is_none(),
+ "the two mechanisms are independent: one may be set without the
other"
+ );
+ }
+
+ /// Three rows, one per row group. A selector keeping only the first must
+ /// leave the read with that row group's row and no other -- and the volume
+ /// counters must show that the other two were never scanned, which is the
+ /// difference between pruning and filtering.
+ #[tokio::test]
+ async fn a_selector_prunes_row_groups_when_the_read_does_not_merge() {
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let (tmp, base_name, schema) = three_row_groups();
+ let mut reader = test_file_group_reader_for_base_file(tmp.path(),
&base_name, schema).await;
+ let volume = reader.storage.read_volume();
+ install_selector(&mut reader, |_| Some(vec![0]), false);
+
+ let out =
drain_base_source(reader.base_file_source().await.unwrap()).await;
+
+ assert_eq!(out.num_rows(), 1, "only the kept row group was read");
+ assert_eq!(volume.row_group_selector_calls.load(Relaxed), 1);
+ assert_eq!(volume.row_group_selector_suppressed.load(Relaxed), 0);
+ assert_eq!(volume.file_row_groups.load(Relaxed), 3);
+ assert_eq!(
+ volume.row_groups_read.load(Relaxed),
+ 1,
+ "the other two row groups were never fetched"
+ );
+ }
+
+ /// The same selector on a slice that merges, with a predicate that is not
+ /// primary-key-safe. Pruning would drop base rows before the merge could
+ /// update them into a match, so the gate refuses it -- and counts the
+ /// refusal, because a suppressed selector otherwise reads as "no caller
ever
+ /// installed one": both are zero calls.
+ #[tokio::test]
+ async fn a_selector_the_gate_refuses_is_counted_not_silently_dropped() {
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let (tmp, base_name, schema) = three_row_groups();
+ let mut reader = test_file_group_reader_for_base_file(tmp.path(),
&base_name, schema).await;
+ let volume = reader.storage.read_volume();
+ reader.input_split = InputSplit::new(
+ Some(base_name.clone()),
+ Some("20240101120000000".to_string()),
+ vec![".f1-0_20240101130000000.log.1_0-1-1".to_string()],
+ String::new(),
+ );
+ install_selector(&mut reader, |_| Some(vec![0]), false);
+
+ let out =
drain_base_source(reader.base_file_source().await.unwrap()).await;
+
+ assert_eq!(out.num_rows(), 3, "every base row still reaches the
merge");
+ assert_eq!(
+ volume.row_group_selector_calls.load(Relaxed),
+ 0,
+ "the selector never ran"
+ );
+ assert_eq!(
+ volume.row_group_selector_suppressed.load(Relaxed),
+ 1,
+ "and the reason it never ran is on the record"
+ );
+ assert_eq!(volume.row_groups_read.load(Relaxed), 3);
+ }
+
+ /// The same merging slice with a primary-key-safe predicate: the gate
opens,
+ /// so the selector runs. Pairs with the case above -- same file, same
+ /// selector, opposite outcome from `mor_pk_safe` alone.
+ #[tokio::test]
+ async fn a_pk_safe_predicate_lets_the_selector_run_on_a_merging_slice() {
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let (tmp, base_name, schema) = three_row_groups();
+ let mut reader = test_file_group_reader_for_base_file(tmp.path(),
&base_name, schema).await;
+ let volume = reader.storage.read_volume();
+ reader.input_split = InputSplit::new(
+ Some(base_name.clone()),
+ Some("20240101120000000".to_string()),
+ vec![".f1-0_20240101130000000.log.1_0-1-1".to_string()],
+ String::new(),
+ );
+ install_selector(&mut reader, |_| Some(vec![0]), true);
+
+ let out =
drain_base_source(reader.base_file_source().await.unwrap()).await;
+
+ assert_eq!(out.num_rows(), 1);
+ assert_eq!(volume.row_group_selector_calls.load(Relaxed), 1);
+ assert_eq!(volume.row_group_selector_suppressed.load(Relaxed), 0);
+ }
+
+ /// A selector with no opinion reads the whole file -- but the call is
still
+ /// counted, which is what separates "ran and found nothing" from "never
+ /// installed".
+ #[tokio::test]
+ async fn a_selector_that_declines_reads_every_row_group_and_still_counts()
{
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let (tmp, base_name, schema) = three_row_groups();
+ let mut reader = test_file_group_reader_for_base_file(tmp.path(),
&base_name, schema).await;
+ let volume = reader.storage.read_volume();
+ install_selector(&mut reader, |_| None, false);
+
+ let out =
drain_base_source(reader.base_file_source().await.unwrap()).await;
+
+ assert_eq!(out.num_rows(), 3);
+ assert_eq!(volume.row_group_selector_calls.load(Relaxed), 1);
+ assert_eq!(
+ volume.row_groups_read.load(Relaxed),
+ volume.file_row_groups.load(Relaxed),
+ "declining prunes nothing"
+ );
+ }
+
+ /// A three-row base file written one row per row group, so a selector has
+ /// something to choose between.
+ fn three_row_groups() -> (tempfile::TempDir, String, SchemaRef) {
+ use arrow_array::Int32Array;
+
+ let tmp = tempfile::tempdir().unwrap();
+ let schema =
Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
+ "id",
+ arrow_schema::DataType::Int32,
+ true,
+ )]));
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![Arc::new(Int32Array::from(vec![7, 8, 9]))],
+ )
+ .unwrap();
+ let base_name = "f1-0_0-1-1_20240101120000000.parquet".to_string();
+ write_parquet_file_in_row_groups(tmp.path(), &base_name, &batch, 1);
+ (tmp, base_name, schema)
+ }
+
+ /// Put a selector and a PK-safety verdict on a reader that was already
built.
+ fn install_selector(
+ reader: &mut HoodieFileGroupReader,
+ select: fn(&parquet::file::metadata::ParquetMetaData) ->
Option<Vec<usize>>,
+ mor_pk_safe: bool,
+ ) {
+ let mut context = (*reader.reader_context).clone();
+ context.row_group_selector = Some(std::sync::Arc::new(select));
+ context.mor_pk_safe = mor_pk_safe;
+ reader.reader_context = Arc::new(context);
+ }
+
// Bootstrap base files are rejected loudly at reader construction.
// `needs_bootstrap_merge = true` (set when the table has bootstrap base
files
// requiring meta/data column reordering) must surface as
CoreError::Unsupported
@@ -1862,6 +2211,8 @@ mod tests {
);
}
+ /// A CoW slice never carries log files, so it reaches the gate through the
+ /// "nothing merges" branch rather than through a table-type test.
#[test]
fn builder_cow_always_pushes_regardless_of_mor_pk_safe() {
let storage =
Storage::new_with_base_url(parse_uri("file:///tmp").unwrap()).unwrap();
@@ -1874,7 +2225,7 @@ mod tests {
.build()
.unwrap();
assert!(
- reader.reader_context.can_push_row_filter(),
+ reader.base_read_pushdown_is_safe(),
"CoW path always pushes regardless of mor_pk_safe"
);
}
@@ -1903,7 +2254,7 @@ mod tests {
for use_position in [false, true] {
for filter in [None, Some(make_row_filter_builder())] {
assert_eq!(
- base_read_options(filter, None, use_position).batch_size,
+ base_read_options(filter, None, None,
use_position).batch_size,
Some(MERGE_CHUNK_ROWS),
"the base read must ask for the merge's chunk bound rather
than \
inherit one (use_position={use_position})"
diff --git a/crates/core/src/file_group/reader_v2/harness.rs
b/crates/core/src/file_group/reader_v2/harness.rs
index ad048efb..05ba3bb1 100644
--- a/crates/core/src/file_group/reader_v2/harness.rs
+++ b/crates/core/src/file_group/reader_v2/harness.rs
@@ -133,10 +133,11 @@ pub struct RowFilterSpec {
/// Marks the filter PK-safe. Mirrors Java's
/// `SparkFileFormatInternalRowReaderContext.filterIsSafeForPrimaryKey`:
/// only record-key filters are safe to push under merge, because PKs are
- /// immutable across upserts. Sets `builder.with_mor_pk_safe`; the
- /// `can_push_row_filter` gate (`is_cow() || mor_pk_safe`) decides whether
- /// the filter is actually installed on the base parquet + parquet log
- /// blocks.
+ /// immutable across upserts. Sets `builder.with_mor_pk_safe`; the reader's
+ /// per-split gate (`base_read_pushdown_is_safe`) decides whether the
+ /// filter is actually installed on the base parquet + parquet log blocks —
+ /// a slice with no log files pushes whatever this says, because nothing
+ /// merges.
pub mor_pk_safe: bool,
}
@@ -766,21 +767,17 @@ async fn read_case_with_filter(
let mut reader_context = base_reader_context(case, has_log_files);
reader_context.schema_handler = schema_handler;
- // The `can_push_row_filter` gate is `is_cow() || mor_pk_safe`, and
- // `is_cow()` reads `hoodie.table.type` from the table_config. The
FFI/Spark
- // path populates this from `hoodie.properties`; the harness's
- // `ReaderContext::empty()` does not, so set it here to match gold's
- // pushdown gate. A base-only slice (no log files) is read as COPY_ON_WRITE
- // (no merge can flip the predicate outcome — the CoW gate branch); a slice
- // with log files is MERGE_ON_READ (only PK-safe filters may push).
- let table_type = if has_log_files {
- "MERGE_ON_READ"
- } else {
- "COPY_ON_WRITE"
- };
+ // `hoodie.table.type` reaches the reader from `hoodie.properties` on the
+ // engine path; the harness's `ReaderContext::empty()` has none, so set it
+ // here. Every fixture these cases read is a MERGE_ON_READ table, and that
is
+ // what goes in — a base-only slice of one is no longer declared
+ // COPY_ON_WRITE to unlock pushdown. The reader decides that per split now
+ // (`base_read_pushdown_is_safe`): no log files, no merge, so nothing can
+ // flip a predicate's outcome. Faking the table type would hide whether
that
+ // decision works.
reader_context.table_config.insert(
HudiTableConfig::TableType.as_ref().to_string(),
- table_type.to_string(),
+ "MERGE_ON_READ".to_string(),
);
let mut reader = HoodieFileGroupReader::builder()
diff --git a/crates/core/src/file_group/reader_v2/harness_tests.rs
b/crates/core/src/file_group/reader_v2/harness_tests.rs
index f4ddd360..94c4d86f 100644
--- a/crates/core/src/file_group/reader_v2/harness_tests.rs
+++ b/crates/core/src/file_group/reader_v2/harness_tests.rs
@@ -1307,9 +1307,10 @@ fg_case_test!(
// Java contract
(SparkFileFormatInternalRowReaderContext.filterIsSafeForPrimaryKey):
// - PK-safe filters (record-key columns) may be pushed under MOR merge.
// - Data-column filters may be pushed only on the CoW path (no log files).
-// - The gate `can_push_row_filter() = is_cow() || mor_pk_safe` blocks unsafe
-// pushes; when blocked, ALL rows return and the post-merge filter (above
-// the FG reader, e.g. Velox/Spark) evaluates the predicate.
+// - The gate `base_read_pushdown_is_safe() = no log files on the split ||
+// mor_pk_safe` blocks unsafe pushes; when blocked, ALL rows return and the
+// post-merge filter (above the FG reader, e.g. Velox/Spark) evaluates the
+// predicate.
//
// Record-key format for V9Mor8I4UCommitTime: plain id value ("1", "2", ...)
// (verified by reading the sf base parquet's _hoodie_record_key column).
@@ -1379,17 +1380,18 @@ fg_case_test!(
}
);
-// (3) Data-column filter on the CoW path (base-only, no logs -> is_cow gate):
-// `age` Gt "27", mor_pk_safe=false. The gate is open via is_cow(), so the
-// filter is pushed: Alice(30) survives, Bob(25) is filtered out.
+// (3) Data-column filter on a base-only slice of a MOR table: `age` Gt "27",
+// mor_pk_safe=false. No log files, so nothing merges and the gate is open on
+// the split alone: the filter is pushed, Alice(30) survives, Bob(25) is
+// filtered out. Poisons if the split-level gate stops opening.
fg_case_test!(
- harness_filter_data_col_cow,
+ harness_filter_data_col_base_only,
FgReaderCase {
- name: "filter_data_col_cow",
+ name: "filter_data_col_base_only",
fixture: QuickstartTripsTable::V9Mor8I4UCommitTime,
partition: "city=sf",
base_file: I8I4U_SF_BASE,
- log_files: &[], // base-only => COW path => can_push_row_filter via
is_cow()
+ log_files: &[], // base-only => nothing merges => the split gate opens
expect_output_columns: Some(&["id", "name", "age"]),
row_filter: Some(RowFilterSpec {
column: "age",
@@ -1405,18 +1407,18 @@ fg_case_test!(
}
);
-// (4) Logical/typed-column filter on the CoW path: MorLayoutAllDataTypes
+// (4) Logical/typed-column filter on a base-only slice: MorLayoutAllDataTypes
// base-only read, `long_field` (Int64) Gt "250". Base values long_field =
// 100,200,300,400,500 for keys k1..k5 (read from the base parquet, pre-merge).
// => k3,k4,k5 (300,400,500) survive.
fg_case_test!(
- harness_filter_logical_type_cow,
+ harness_filter_logical_type_base_only,
FgReaderCase {
- name: "filter_logical_type_cow",
+ name: "filter_logical_type_base_only",
fixture: QuickstartTripsTable::MorLayoutAllDataTypes,
partition: "",
base_file:
"c887c1e8-5fb9-475e-8171-769c5cf10c61-0_0-240-395_20260409030537482.parquet",
- log_files: &[], // base-only => COW path
+ log_files: &[], // base-only => nothing merges => the split gate opens
expect_output_columns: Some(&["key", "long_field", "severity"]),
row_filter: Some(RowFilterSpec {
column: "long_field",
@@ -1439,8 +1441,7 @@ fg_case_test!(
// (5) NEGATIVE gate case: data-column filter under MOR merge with
// mor_pk_safe=false. `age` Gt "27" — pushing this under merge could DROP a
// base row (Bob, age 25) whose log update might later have made it match, so
-// the gate (can_push_row_filter = is_cow() || mor_pk_safe = false) BLOCKS the
-// push. The post-merge filter (above the FG reader) is responsible instead, so
+// the gate (no log files = false, mor_pk_safe = false) BLOCKS the push. The
post-merge filter (above the FG reader) is responsible instead, so
// the FG reader returns ALL merged rows: Alice-V2(31) AND Bob(25).
// (Unsafe because the predicate evaluated on BASE values can disagree with
post-merge values for
// the same key: a log update may change the column so a pruned base row would
have matched after
@@ -1452,7 +1453,7 @@ fg_case_test!(
fixture: QuickstartTripsTable::V9Mor8I4UCommitTime,
partition: "city=sf",
base_file: I8I4U_SF_BASE,
- log_files: &[I8I4U_SF_LOG], // MOR merge => is_cow() false
+ log_files: &[I8I4U_SF_LOG], // MOR merge => the split gate stays shut
expect_output_columns: Some(&["id", "name", "age"]),
row_filter: Some(RowFilterSpec {
column: "age",
@@ -1821,9 +1822,9 @@ fg_case_test!(
// 0-6; IN ("1","4") keeps exactly those two original rows. Unfiltered would
// be 7 rows.
fg_case_test!(
- harness_filter_pk_in_cow,
+ harness_filter_pk_in_base_only,
FgReaderCase {
- name: "filter_pk_in_cow",
+ name: "filter_pk_in_base_only",
fixture: QuickstartTripsTable::V9MorNonpart3Commits,
partition: "",
base_file: NONPART_BASE_FILE,
@@ -1850,9 +1851,9 @@ fg_case_test!(
// Boolean Eq: true rows are k2, k4.
fg_case_test!(
- harness_filter_boolean_cow,
+ harness_filter_boolean_base_only,
FgReaderCase {
- name: "filter_boolean_cow",
+ name: "filter_boolean_base_only",
fixture: QuickstartTripsTable::MorLayoutAllDataTypes,
partition: "",
base_file:
"c887c1e8-5fb9-475e-8171-769c5cf10c61-0_0-240-395_20260409030537482.parquet",
@@ -1874,9 +1875,9 @@ fg_case_test!(
// Date32 Gt: days > 19754 (2024-02-01) are k3, k4, k5.
fg_case_test!(
- harness_filter_date_cow,
+ harness_filter_date_base_only,
FgReaderCase {
- name: "filter_date_cow",
+ name: "filter_date_base_only",
fixture: QuickstartTripsTable::MorLayoutAllDataTypes,
partition: "",
base_file:
"c887c1e8-5fb9-475e-8171-769c5cf10c61-0_0-240-395_20260409030537482.parquet",
@@ -1898,9 +1899,9 @@ fg_case_test!(
// Timestamp(us, UTC) Lt: strictly before k3's 2024-03-01T00:00:03Z are k1, k2.
fg_case_test!(
- harness_filter_timestamp_cow,
+ harness_filter_timestamp_base_only,
FgReaderCase {
- name: "filter_timestamp_cow",
+ name: "filter_timestamp_base_only",
fixture: QuickstartTripsTable::MorLayoutAllDataTypes,
partition: "",
base_file:
"c887c1e8-5fb9-475e-8171-769c5cf10c61-0_0-240-395_20260409030537482.parquet",
@@ -1922,9 +1923,9 @@ fg_case_test!(
// Decimal128(20,2) Gt: values > 300.30 are k4 (400.40), k5 (500.50).
fg_case_test!(
- harness_filter_decimal_cow,
+ harness_filter_decimal_base_only,
FgReaderCase {
- name: "filter_decimal_cow",
+ name: "filter_decimal_base_only",
fixture: QuickstartTripsTable::MorLayoutAllDataTypes,
partition: "",
base_file:
"c887c1e8-5fb9-475e-8171-769c5cf10c61-0_0-240-395_20260409030537482.parquet",
@@ -1947,9 +1948,9 @@ fg_case_test!(
// Float32 Gt: 3.4f32 equals k3's stored value exactly (same literal), so the
// strictly-greater rows are k4 (4.5), k5 (5.6).
fg_case_test!(
- harness_filter_float32_cow,
+ harness_filter_float32_base_only,
FgReaderCase {
- name: "filter_float32_cow",
+ name: "filter_float32_base_only",
fixture: QuickstartTripsTable::MorLayoutAllDataTypes,
partition: "",
base_file:
"c887c1e8-5fb9-475e-8171-769c5cf10c61-0_0-240-395_20260409030537482.parquet",
diff --git a/crates/core/src/file_group/reader_v2/log_record_reader.rs
b/crates/core/src/file_group/reader_v2/log_record_reader.rs
index 3338b111..e398041d 100644
--- a/crates/core/src/file_group/reader_v2/log_record_reader.rs
+++ b/crates/core/src/file_group/reader_v2/log_record_reader.rs
@@ -867,7 +867,20 @@ impl BaseHoodieLogRecordReader {
// for cannot change what the requested keys merge to, and dropping it
before
// the Avro decode is the whole saving.
.with_key_predicate(self.reader_context.key_predicate.clone())
- .with_row_filter(if self.reader_context.can_push_row_filter() {
+ // The log gate is `mor_pk_safe`. A log block only exists when the
slice
+ // HAS log files, so the base read's "does this merge?" disjunct is
+ // false here by construction and its condition reduces to exactly
this;
+ // naming it directly says what the log path requires.
+ //
+ // One case is not yet excluded: under position-based merge, Java's
+ // `SparkFileFormatInternalRowReaderContext` pushes NO filters into log
+ // files, because the RECORD_POSITIONS bitmap pairs positions to block
+ // records by index and a decode that drops records misaligns the
+ // pairing. Here that combination fails loudly instead (the position
+ // buffer's count-mismatch check): a PK-safe filter that removes
+ // records from a parquet log block errors a position-based read
+ // rather than corrupting it.
+ .with_row_filter(if self.reader_context.mor_pk_safe {
self.reader_context.row_filter_builder.clone()
} else {
None
diff --git a/crates/core/src/file_group/reader_v2/reader_context.rs
b/crates/core/src/file_group/reader_v2/reader_context.rs
index fe995bbd..883d5ccd 100644
--- a/crates/core/src/file_group/reader_v2/reader_context.rs
+++ b/crates/core/src/file_group/reader_v2/reader_context.rs
@@ -28,7 +28,7 @@
use super::record_context::RecordContext;
use super::schema_handler::FileGroupReaderSchemaHandler;
use crate::config::table::HudiTableConfig;
-use crate::storage::RowFilterBuilder;
+use crate::storage::{RowFilterBuilder, RowGroupSelector};
use crate::timeline::selector::InstantRange;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
@@ -130,9 +130,11 @@ pub struct ReaderContext {
/// parquet files and parquet-format log blocks. Set by the FFI bridge from
/// the decoded substrait predicate; `None` when no predicate was pushed.
///
- /// Whether it actually gets installed is gated by [`Self::mor_pk_safe`] +
- /// table type (see callers in `file_group::reader::HoodieFileGroupReader`
- /// and `file_group::log_file::content::Decoder`).
+ /// Whether it actually gets installed is gated by whether the read merges:
+ /// `HoodieFileGroupReader::base_read_pushdown_is_safe` for the base file
+ /// (no log files on the split, or [`Self::mor_pk_safe`]), and
+ /// [`Self::mor_pk_safe`] alone for a parquet log block, which by
definition
+ /// only exists on a slice that does merge.
///
/// **Not active for reads through this crate.**
///
[`resolve_reader_context`](crate::file_group::reader_v2::resolver::resolve_reader_context)
@@ -142,6 +144,15 @@ pub struct ReaderContext {
/// Groundwork for a caller that supplies one; no performance claim about
this
/// crate's reads rests on it.
pub row_filter_builder: Option<RowFilterBuilder>,
+ /// Optional row-group selector for pruning base parquet reads from footer
+ /// statistics; `None` when no caller supplied one.
+ ///
+ /// Gated by the same "does this read merge?" condition as
+ /// [`Self::row_filter_builder`], and for a stronger reason: pruning drops
+ /// base rows before the log merge runs, so on a merging read it could
remove
+ /// a row the merge would have updated into a match. The row filter at
least
+ /// sees every row.
+ pub row_group_selector: Option<RowGroupSelector>,
/// Record keys or key prefixes the read is interested in, pushed into the
base
/// file reader so a key-ordered format seeks instead of scanning.
///
@@ -214,6 +225,10 @@ impl std::fmt::Debug for ReaderContext {
"row_filter_builder",
&self.row_filter_builder.as_ref().map(|_| "<closure>"),
)
+ .field(
+ "row_group_selector",
+ &self.row_group_selector.as_ref().map(|_| "<closure>"),
+ )
.field("mor_pk_safe", &self.mor_pk_safe)
.field("completion_gate_inputs", &self.completion_gate_inputs)
.finish()
@@ -320,82 +335,18 @@ impl ReaderContext {
table_config: HashMap::new(),
hoodie_reader_config: HashMap::new(),
row_filter_builder: None,
+ row_group_selector: None,
key_predicate: None,
mor_pk_safe: false,
completion_gate_inputs: None,
}
}
-
- /// Returns true iff the table is COPY_ON_WRITE per `hoodie.table.type`.
- /// Defaults to `false` (treat as MOR) on missing/unparseable values so
- /// callers err on the side of NOT pushing predicates down.
- pub fn is_cow(&self) -> bool {
- use crate::config::table::TableTypeValue;
- use std::str::FromStr;
- self.table_config
- .get("hoodie.table.type")
- .and_then(|v| TableTypeValue::from_str(v).ok())
- .map(|t| matches!(t, TableTypeValue::CopyOnWrite))
- .unwrap_or(false)
- }
-
- /// Returns true iff this context allows installing the parquet `RowFilter`
- /// for the current scan. Either the table is CoW (the merge can't flip
- /// predicate outcomes) or the filter is PK-safe (PKs are immutable across
- /// upserts, so the predicate's outcome is stable across the base+log
- /// merge). Mirrors the gate Java applies via the `morFilters`/`allFilters`
- /// selection in `SparkFileFormatInternalRowReaderContext`.
- pub fn can_push_row_filter(&self) -> bool {
- self.is_cow() || self.mor_pk_safe
- }
}
#[cfg(test)]
mod tests {
use super::*;
- fn ctx_with_table_type(t: &str) -> ReaderContext {
- let mut ctx = ReaderContext::empty();
- ctx.table_config
- .insert("hoodie.table.type".to_string(), t.to_string());
- ctx
- }
-
- #[test]
- fn is_cow_true_for_copy_on_write() {
- assert!(ctx_with_table_type("COPY_ON_WRITE").is_cow());
- }
-
- #[test]
- fn is_cow_false_for_merge_on_read() {
- assert!(!ctx_with_table_type("MERGE_ON_READ").is_cow());
- }
-
- #[test]
- fn is_cow_false_when_table_type_missing() {
- // Defaults to MOR (conservative) so non-PK predicates don't sneak in.
- assert!(!ReaderContext::empty().is_cow());
- }
-
- #[test]
- fn can_push_row_filter_cow_unconditionally() {
- // CoW: pushdown is always safe regardless of mor_pk_safe.
- let mut ctx = ctx_with_table_type("COPY_ON_WRITE");
- ctx.mor_pk_safe = false;
- assert!(ctx.can_push_row_filter());
- ctx.mor_pk_safe = true;
- assert!(ctx.can_push_row_filter());
- }
-
- #[test]
- fn can_push_row_filter_mor_only_when_pk_safe() {
- let mut ctx = ctx_with_table_type("MERGE_ON_READ");
- ctx.mor_pk_safe = false;
- assert!(!ctx.can_push_row_filter());
- ctx.mor_pk_safe = true;
- assert!(ctx.can_push_row_filter());
- }
-
#[test]
fn record_key_fields_from_default_meta_field_mode() {
// Default (populate.meta.fields absent → true): record_key_fields()
diff --git a/crates/core/src/file_group/reader_v2/resolver.rs
b/crates/core/src/file_group/reader_v2/resolver.rs
index 3930869a..89076f3a 100644
--- a/crates/core/src/file_group/reader_v2/resolver.rs
+++ b/crates/core/src/file_group/reader_v2/resolver.rs
@@ -97,8 +97,10 @@ pub(crate) fn resolve_reader_context(
needs_bootstrap_merge: false,
enable_logical_timestamp_field_repair: false,
// Predicate pushdown into the merge path has no caller here, so no
- // filter is installed and the primary-key-safety gate is irrelevant.
+ // filter is installed, nothing prunes row groups, and the
+ // primary-key-safety gate is irrelevant.
row_filter_builder: None,
+ row_group_selector: None,
key_predicate: None,
mor_pk_safe: false,
// The table-version < 8 completion gate needs a timeline the caller
diff --git a/crates/core/src/storage/mod.rs b/crates/core/src/storage/mod.rs
index 2f7c80ff..69e7474a 100644
--- a/crates/core/src/storage/mod.rs
+++ b/crates/core/src/storage/mod.rs
@@ -21,6 +21,7 @@
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
use async_recursion::async_recursion;
use bytes::Bytes;
@@ -59,12 +60,134 @@ pub type RowFilterBuilder = Arc<
+ Sync,
>;
+/// Chooses which row groups a read fetches, from the file's parsed footer.
+/// Returning `None` means "no opinion, read them all".
+///
+/// This is the only mechanism on the read path that can avoid reading bytes: a
+/// [`RowFilterBuilder`] decides per row after the predicate columns are
decoded,
+/// so it saves decode, never IO, while a row group excluded here is never
+/// fetched.
+///
+/// The selector must be CONSERVATIVE. Keeping a row group that cannot match
+/// costs only time; dropping one that can match silently loses rows, and
nothing
+/// downstream can restore them.
+///
+/// `Arc` for the same reason as [`RowFilterBuilder`]: options holding it stay
+/// `Clone`, and it may run on any worker thread.
+pub type RowGroupSelector =
+ Arc<dyn Fn(&parquet::file::metadata::ParquetMetaData) ->
Option<Vec<usize>> + Send + Sync>;
+
#[derive(Clone, Debug)]
pub struct Storage {
pub(crate) base_url: Arc<Url>,
pub(crate) object_store: Arc<dyn ObjectStore>,
pub(crate) options: Arc<HashMap<String, String>>,
pub(crate) hudi_configs: Arc<HudiConfigs>,
+ /// Read-volume counters for the base-file reads made through this
+ /// `Storage`.
+ ///
+ /// Here rather than on a read parameter so no read signature changes. The
+ /// scope is this `Storage`'s lifetime, which is as narrow as the caller
+ /// makes it: a per-read `Storage` yields per-read counters, while a
+ /// `FileGroupReader` builds one `Storage` in its constructor and reuses
+ /// it, so every slice read through that reader accumulates into the same
+ /// counters. The `object_store` inside could not carry them: it is shared
+ /// even wider, so its counts would be everyone's.
+ pub(crate) read_volume: Arc<ReadVolume>,
+}
+
+/// Read-volume counters for one [`Storage`]'s lifetime.
+///
+/// Whether a predicate was pushed is not the same question as what a push
+/// bought: a parquet `RowFilter` can be installed on every file and still read
+/// every byte, because it decides per row after the predicate columns are
+/// decoded. Only pruning avoids IO. These counters separate the two.
+///
+/// `bytes_read` and `io_calls` are counted at the `AsyncFileReader` boundary,
+/// which makes them exact and independent of the OS page cache: a warm re-read
+/// reports the same bytes as a cold one. Wall-clock does not have that
property,
+/// which is what makes these the transferable numbers when comparing read
paths.
+/// The boundary also bounds the scope: footer fetches happen before the
+/// counting wrapper is installed and log-file IO never crosses it, so these
+/// two count base-file column-chunk reads only.
+///
+/// All fields are `AtomicU64` under an `Arc` because the parquet stream is
+/// polled on whichever worker thread drives it, while a consumer may read the
+/// counters from another. `Relaxed` throughout: these are advisory counters,
and
+/// the happens-before that makes them visible is the consumer draining the
+/// stream.
+#[derive(Debug, Default)]
+pub struct ReadVolume {
+ /// Bytes actually fetched from the object store, summed over every range
read.
+ pub bytes_read: AtomicU64,
+ /// Number of `get_bytes` / `get_byte_ranges` calls — round trips, not
ranges.
+ /// A two-pass read (predicate columns, then the selected rows) shows up
here
+ /// as roughly double the calls of a single-pass read over the same file.
+ pub io_calls: AtomicU64,
+ /// Row groups the reader was configured to scan. Equal to
`file_row_groups`
+ /// until something prunes; the gap between the two is what pruning bought.
+ pub row_groups_read: AtomicU64,
+ /// Row groups the file contains. Denominator for the line above.
+ pub file_row_groups: AtomicU64,
+ /// Times the row-group selector closure actually RAN.
+ ///
+ /// Separate from its outcome on purpose. A selector returns `None` when it
+ /// cannot prune anything, so `row_groups_read == file_row_groups` reads
the
+ /// same whether the selector ran and found nothing or was never installed.
+ /// Only this counter separates them.
+ pub row_group_selector_calls: AtomicU64,
+ /// Times a selector WAS installed by the caller but the merge-safety gate
+ /// refused to pass it down.
+ ///
+ /// Without this the gate silently defeats the counter above: a suppressed
+ /// selector is a third state that also reads zero calls. Read the two
+ /// together:
+ /// calls > 0 the selector ran
+ /// calls == 0, suppressed > 0 the gate refused it (the read merges, and
+ /// the predicate is not primary-key-safe)
+ /// calls == 0, suppressed == 0 no caller ever installed one
+ pub row_group_selector_suppressed: AtomicU64,
+ /// Rows the file contains, from parquet metadata.
+ pub file_rows: AtomicU64,
+ /// Rows the stream actually yielded, after any row filter. `file_rows -
+ /// rows_out` is what filtering removed; `bytes_read` says what it cost to
+ /// remove it.
+ pub rows_out: AtomicU64,
+}
+
+impl ReadVolume {
+ /// One completed fetch: its bytes, and the round trip that carried them.
+ pub(crate) fn add_bytes(&self, n: u64) {
+ self.bytes_read.fetch_add(n, Ordering::Relaxed);
+ self.io_calls.fetch_add(1, Ordering::Relaxed);
+ }
+
+ /// What the file holds, read off the footer the reader has already
fetched.
+ pub(crate) fn record_file_shape(&self, row_groups: u64, rows: u64) {
+ self.file_row_groups
+ .fetch_add(row_groups, Ordering::Relaxed);
+ self.file_rows.fetch_add(rows, Ordering::Relaxed);
+ }
+
+ pub(crate) fn add_row_groups_read(&self, n: u64) {
+ self.row_groups_read.fetch_add(n, Ordering::Relaxed);
+ }
+
+ /// The selector ran. Counted whether or not it managed to prune.
+ pub(crate) fn record_selector_call(&self) {
+ self.row_group_selector_calls
+ .fetch_add(1, Ordering::Relaxed);
+ }
+
+ /// A caller installed a selector and the safety gate declined to pass it
on.
+ pub(crate) fn record_selector_suppressed(&self) {
+ self.row_group_selector_suppressed
+ .fetch_add(1, Ordering::Relaxed);
+ }
+
+ pub(crate) fn add_rows_out(&self, n: u64) {
+ self.rows_out.fetch_add(n, Ordering::Relaxed);
+ }
}
impl Storage {
@@ -93,11 +216,18 @@ impl Storage {
object_store: Arc::new(object_store),
options,
hudi_configs,
+ read_volume: Arc::new(ReadVolume::default()),
})),
Err(e) => Err(Creation(format!("Failed to create storage: {e}"))),
}
}
+ /// Clone of this `Storage`'s read-volume counters, for a consumer that
+ /// outlives the read and reports them once the stream has drained.
+ pub fn read_volume(&self) -> Arc<ReadVolume> {
+ self.read_volume.clone()
+ }
+
/// Build storage over a caller-supplied object store.
///
/// Test-only, so a test can wrap the real store and observe the requests a
@@ -116,6 +246,7 @@ impl Storage {
object_store,
options: Arc::new(HashMap::new()),
hudi_configs,
+ read_volume: Arc::new(ReadVolume::default()),
})
}
