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 3a031458 fix(core): decline base-read pushdown when a predicate reads
a repaired column (#748)
3a031458 is described below
commit 3a03145816f23906151cd61976b44822be17e7ff
Author: Lin Liu <[email protected]>
AuthorDate: Fri Sep 4 18:17:37 2026 -0700
fix(core): decline base-read pushdown when a predicate reads a repaired
column (#748)
---
crates/core/src/file_group/reader_v2/engine.rs | 603 ++++++++++++++++++++-
.../src/file_group/reader_v2/reader_context.rs | 30 +
crates/core/src/file_group/reader_v2/resolver.rs | 3 +
crates/core/src/schema/batch_evolution.rs | 480 ++++++++++++++++
crates/core/src/storage/mod.rs | 41 +-
5 files changed, 1148 insertions(+), 9 deletions(-)
diff --git a/crates/core/src/file_group/reader_v2/engine.rs
b/crates/core/src/file_group/reader_v2/engine.rs
index 4483abe2..afbbee94 100644
--- a/crates/core/src/file_group/reader_v2/engine.rs
+++ b/crates/core/src/file_group/reader_v2/engine.rs
@@ -812,7 +812,7 @@ impl HoodieFileGroupReader {
// 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 {
+ let mut row_filter = if pushdown_is_safe {
self.reader_context.row_filter_builder.clone()
} else {
if self.reader_context.row_filter_builder.is_some() {
@@ -824,7 +824,7 @@ impl HoodieFileGroupReader {
}
None
};
- let row_group_selector = if pushdown_is_safe {
+ let mut 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
@@ -932,6 +932,63 @@ impl HoodieFileGroupReader {
present_len
);
+ // Parquet evaluates a pushed predicate against the file's PHYSICAL
values,
+ // before `project_batch_to_schema` runs. Sound only while a physical
value
+ // means what its physical type says, which the apache/hudi#18132
repair
+ // breaks: the file labels a tz-aware column micros while the stored
i64 is
+ // MILLIS, so a millis-semantics literal reads those rows as 1970 and
the
+ // filter drops rows that match. The post-scan filter cannot restore
them.
+ //
+ // Two gates, cheapest first. `repair_risk_columns` was decided ONCE
per scan
+ // from the table schema and the predicate's own referenced columns,
and is
+ // empty unless the predicate touches a tz-aware millis column — so the
+ // common scan never reaches the footer comparison below and never
loses
+ // pushdown. The footer schema itself is already fetched
unconditionally
+ // above, so gate 1 buys predicate scoping and the per-file name walk,
not
+ // avoided IO.
+ //
+ // The table side is `table_schema`, NOT `required_schema`: a filter
column
+ // absent from the projection is still decoded and still misread,
because a
+ // `RowFilter` builder derives its own `ProjectionMask` from the
parquet
+ // schema rather than from `intersection`.
+ let repair_conflict =
+ if pushdown_is_safe &&
!self.reader_context.repair_risk_columns.is_empty() {
+ let table_side = self
+ .schema_handler
+ .table_schema
+ .as_ref()
+ .unwrap_or(&required_schema);
+ crate::schema::batch_evolution::reinterpreted_columns(
+ &file_schema,
+ table_side,
+ &self.reader_context.repair_risk_columns,
+ )?
+ } else {
+ Vec::new()
+ };
+
+ // ONE verdict, both consumers, withdrawn in one block. That is the
same
+ // property the merge-safety gate is bound once for, one layer in: an
edit
+ // to the condition cannot leave the row-group selector behind, and
pruning
+ // is the one that must not be left behind — it drops rows before
anything
+ // downstream can see them.
+ if !repair_conflict.is_empty() {
+ let volume = self.storage.read_volume();
+ // Counted for every withdrawal; `row_group_selector_suppressed`
can
+ // only speak for a selector the caller actually installed.
+ volume.record_pushdown_suppressed_by_repair();
+ if row_group_selector.is_some() {
+ volume.record_selector_suppressed();
+ }
+ log::debug!(
+ "base file '{path}' needs a value-reinterpreting logical-type
repair \
+ on {repair_conflict:?} — skipping parquet RowFilter pushdown
and \
+ row-group pruning (post-merge filter still runs)"
+ );
+ row_filter = None;
+ row_group_selector = None;
+ }
+
let base_read_schema: SchemaRef = if use_position {
let mut fields: Vec<arrow_schema::FieldRef> =
required_schema.fields().iter().cloned().collect();
@@ -1187,6 +1244,9 @@ pub struct HoodieFileGroupReaderBuilder {
row_group_selector: Option<RowGroupSelector>,
/// Set by `with_mor_pk_safe`; copied onto the cloned reader_context.
mor_pk_safe: Option<bool>,
+ /// Set by `with_repair_risk_columns`; copied onto the cloned
reader_context.
+ /// Absent leaves the repair guard OFF.
+ repair_risk_columns: Option<Vec<String>>,
}
/// Reached only from the test harness — see the builder's own note.
@@ -1236,6 +1296,9 @@ impl HoodieFileGroupReaderBuilder {
///
/// The builder is also visible to the parquet log block decoder via the
/// same `reader_context` channel.
+ ///
+ /// Pair this with [`Self::with_repair_risk_columns`], or the repair guard
is
+ /// off and a base file mislabelling a predicate column over-drops rows.
pub fn with_row_filter_builder(mut self, b: RowFilterBuilder) -> Self {
self.row_filter_builder = Some(b);
self
@@ -1265,6 +1328,20 @@ impl HoodieFileGroupReaderBuilder {
self
}
+ /// Arm the value-reinterpreting repair guard with the predicate columns
the
+ /// apache/hudi#18132 logical-type repair could make a pushed filter
misread.
+ ///
+ /// Required alongside [`Self::with_row_filter_builder`] whenever the
table may
+ /// hold legacy base files labelling a tz-aware column micros while the
stored
+ /// i64 is millis. Left unset the guard is OFF and such a file over-drops
rows —
+ /// this is not a perf knob. Compute via
+ /// [`crate::schema::batch_evolution::repair_risk_columns`]; the empty vec
is the
+ /// explicit "no column is at risk".
+ pub fn with_repair_risk_columns(mut self, columns: Vec<String>) -> Self {
+ self.repair_risk_columns = Some(columns);
+ self
+ }
+
pub fn build(self) -> Result<HoodieFileGroupReader> {
let reader_context = self
.reader_context
@@ -1283,6 +1360,7 @@ impl HoodieFileGroupReaderBuilder {
let reader_context = if self.row_filter_builder.is_some()
|| self.row_group_selector.is_some()
|| self.mor_pk_safe.is_some()
+ || self.repair_risk_columns.is_some()
{
let mut updated = (*reader_context).clone();
if let Some(b) = self.row_filter_builder {
@@ -1294,6 +1372,9 @@ impl HoodieFileGroupReaderBuilder {
if let Some(s) = self.mor_pk_safe {
updated.mor_pk_safe = s;
}
+ if let Some(cols) = self.repair_risk_columns {
+ updated.repair_risk_columns = cols;
+ }
Arc::new(updated)
} else {
reader_context
@@ -2589,4 +2670,522 @@ mod tests {
"the streamed read must return the same rows as the single-batch
read"
);
}
+
+ // ── pushdown vs. the apache/hudi#18132 logical-type repair
────────────────
+
+ /// A `ts > threshold` row filter that normalises the column to NANOSECONDS
+ /// from its own DECLARED unit — the shape an engine's timestamp comparison
+ /// takes when it reconciles a literal against the column type parquet
reports.
+ /// Counts its own invocations, so a test can assert the filter was never
even
+ /// built rather than inferring it from rows.
+ fn nanos_gt_filter_builder(
+ column: &'static str,
+ threshold_nanos: i64,
+ invocations: Arc<std::sync::atomic::AtomicUsize>,
+ ) -> RowFilterBuilder {
+ use parquet::arrow::ProjectionMask;
+ use parquet::arrow::arrow_reader::{ArrowPredicateFn, RowFilter};
+ use std::sync::atomic::Ordering::Relaxed;
+ Arc::new(move |parquet_schema, _projected_schema| {
+ invocations.fetch_add(1, Relaxed);
+ let root = parquet_schema.root_schema();
+ let idx = root.get_fields().iter().position(|f| f.name() ==
column)?;
+ let mask = ProjectionMask::roots(parquet_schema, [idx]);
+ let predicate = ArrowPredicateFn::new(mask, move |batch:
RecordBatch| {
+ use arrow_array::cast::AsArray;
+ use arrow_array::types::{TimestampMicrosecondType,
TimestampMillisecondType};
+ let col = batch.column_by_name(column).ok_or_else(|| {
+ arrow_schema::ArrowError::ComputeError(format!(
+ "predicate column '{column}' missing from the
predicate batch"
+ ))
+ })?;
+ // Scale the raw i64 to nanos using the unit the column
DECLARES.
+ // That declaration is precisely what a mislabelled file gets
wrong,
+ // so the scaling inherits the lie.
+ let (values, per_unit): (Vec<i64>, i64) = match
col.data_type() {
+
arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Microsecond, _) => {
+ let a = col.as_primitive::<TimestampMicrosecondType>();
+ ((0..a.len()).map(|i| a.value(i)).collect(), 1_000)
+ }
+
arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Millisecond, _) => {
+ let a = col.as_primitive::<TimestampMillisecondType>();
+ ((0..a.len()).map(|i| a.value(i)).collect(), 1_000_000)
+ }
+ other => {
+ return
Err(arrow_schema::ArrowError::ComputeError(format!(
+ "unsupported predicate column type {other}"
+ )));
+ }
+ };
+ Ok(arrow_array::BooleanArray::from_iter(
+ values.iter().map(|v| Some(v * per_unit >
threshold_nanos)),
+ ))
+ });
+ Some(RowFilter::new(vec![Box::new(predicate)]))
+ })
+ }
+
+ /// Like [`test_file_group_reader_for_base_file`], but also installs a row
+ /// filter builder so the base read exercises the pushdown path.
+ async fn test_file_group_reader_with_row_filter(
+ dir: &std::path::Path,
+ base_name: &str,
+ required: SchemaRef,
+ row_filter_builder: RowFilterBuilder,
+ row_group_selector: Option<RowGroupSelector>,
+ repair_risk_columns: &[&str],
+ ) -> HoodieFileGroupReader {
+ let mut reader = test_file_group_reader_for_base_file(dir, base_name,
required).await;
+ let mut context = (*reader.reader_context).clone();
+ context.row_filter_builder = Some(row_filter_builder);
+ context.row_group_selector = row_group_selector;
+ // What `batch_evolution::repair_risk_columns` would have produced for
this
+ // predicate against this table schema — the gate that arms the
per-file check.
+ context.repair_risk_columns = repair_risk_columns.iter().map(|c|
c.to_string()).collect();
+ reader.reader_context = Arc::new(context);
+ reader
+ }
+
+ /// 2020-01-01T00:00:00Z — the threshold the failing fixtures straddle.
+ const THRESHOLD_NANOS: i64 = 1_577_836_800_000_000_000;
+ /// 2020-01-01T00:00:00.001Z as MILLIS — above the threshold.
+ const ABOVE_MS: i64 = 1_577_836_800_001;
+ /// 2019-12-31T23:59:59.999Z as MILLIS — below it.
+ const BELOW_MS: i64 = 1_577_836_799_999;
+
+ fn ts_field(name: &str, unit: arrow_schema::TimeUnit) ->
arrow_schema::Field {
+ arrow_schema::Field::new(
+ name,
+ arrow_schema::DataType::Timestamp(unit, Some("UTC".into())),
+ true,
+ )
+ }
+
+ /// The table's view of the straddling file: `ts` is tz-aware MILLIS,
which is
+ /// what the stored i64s have always been.
+ fn straddling_table_schema() -> SchemaRef {
+ Arc::new(arrow_schema::Schema::new(vec![
+ arrow_schema::Field::new("_hoodie_record_key",
arrow_schema::DataType::Utf8, true),
+ ts_field("ts", arrow_schema::TimeUnit::Millisecond),
+ ]))
+ }
+
+ /// Write a two-row base file whose `ts` column is DECLARED with
`declared_unit`
+ /// while its values are always the millisecond counts above. When
+ /// `declared_unit` is micros this is the apache/hudi#18132 shape: the
label is
+ /// a lie and the repair has to reinterpret it on read.
+ fn write_straddling_base_file(
+ dir: &std::path::Path,
+ name: &str,
+ declared_unit: arrow_schema::TimeUnit,
+ ) {
+ let file_schema = Arc::new(arrow_schema::Schema::new(vec![
+ arrow_schema::Field::new("_hoodie_record_key",
arrow_schema::DataType::Utf8, true),
+ ts_field("ts", declared_unit),
+ ]));
+ let ts: arrow_array::ArrayRef = match declared_unit {
+ arrow_schema::TimeUnit::Microsecond => Arc::new(
+ arrow_array::TimestampMicrosecondArray::from(vec![ABOVE_MS,
BELOW_MS])
+ .with_timezone("UTC"),
+ ),
+ _ => Arc::new(
+ arrow_array::TimestampMillisecondArray::from(vec![ABOVE_MS,
BELOW_MS])
+ .with_timezone("UTC"),
+ ),
+ };
+ let batch = RecordBatch::try_new(
+ file_schema,
+ vec![
+ Arc::new(arrow_array::StringArray::from(vec!["k1", "k2"])),
+ ts,
+ ],
+ )
+ .unwrap();
+ write_parquet_file(dir, name, &batch);
+ }
+
+ /// THE REGRESSION. The file declares `ts` as tz-aware micros while the
stored
+ /// i64s are MILLIS, so a nanos-normalised predicate reads them as 1970 and
+ /// `ts > 2020-01-01` matches nothing. The post-scan filter cannot restore
the
+ /// rows the scan already dropped.
+ #[tokio::test]
+ async fn
base_read_declines_pushdown_when_the_file_needs_a_reinterpreting_repair() {
+ use std::sync::atomic::AtomicUsize;
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let tmp = tempfile::tempdir().unwrap();
+ let base_name = "f1-0_0-1-1_001.parquet";
+ write_straddling_base_file(
+ tmp.path(),
+ base_name,
+ arrow_schema::TimeUnit::Microsecond, // the LIE
+ );
+
+ let required = straddling_table_schema();
+ let invocations = Arc::new(AtomicUsize::new(0));
+ let builder = nanos_gt_filter_builder("ts", THRESHOLD_NANOS,
invocations.clone());
+
+ let mut reader = test_file_group_reader_with_row_filter(
+ tmp.path(),
+ base_name,
+ required.clone(),
+ builder,
+ None,
+ &["ts"],
+ )
+ .await;
+ // The existing merge gate is satisfied: no log files, so nothing
merges.
+ assert!(
+ reader.base_read_pushdown_is_safe(),
+ "a slice with no log files clears the merge gate; the repair check
\
+ is what must decline this read"
+ );
+ let out =
drain_base_source(reader.base_file_source().await.unwrap()).await;
+
+ assert_eq!(
+ invocations.load(Relaxed),
+ 0,
+ "the row filter must never even be BUILT for a file whose physical
\
+ timestamp labelling is repaired on read"
+ );
+ assert_eq!(out.schema(), required);
+ assert_eq!(
+ out.num_rows(),
+ 2,
+ "both rows must reach the post-scan filter; dropping one inside
the \
+ scan is unrecoverable"
+ );
+ let ts = out
+ .column(1)
+ .as_any()
+ .downcast_ref::<arrow_array::TimestampMillisecondArray>()
+ .expect("the repair must relabel the column to millis");
+ assert_eq!(
+ (ts.value(0), ts.value(1)),
+ (ABOVE_MS, BELOW_MS),
+ "and it must relabel the i64, not rescale it"
+ );
+ }
+
+ /// The other half of the rule. Same values and predicate, but the file
declares
+ /// the unit it actually uses, so no repair applies and pushdown must
survive —
+ /// otherwise the guard is a blanket regression on every well-formed table.
+ #[tokio::test]
+ async fn base_read_keeps_pushdown_when_the_file_is_honestly_labelled() {
+ use std::sync::atomic::AtomicUsize;
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let tmp = tempfile::tempdir().unwrap();
+ let base_name = "f1-0_0-1-1_001.parquet";
+ write_straddling_base_file(tmp.path(), base_name,
arrow_schema::TimeUnit::Millisecond);
+
+ let required = straddling_table_schema();
+ let invocations = Arc::new(AtomicUsize::new(0));
+ let builder = nanos_gt_filter_builder("ts", THRESHOLD_NANOS,
invocations.clone());
+
+ let mut reader = test_file_group_reader_with_row_filter(
+ tmp.path(),
+ base_name,
+ required.clone(),
+ builder,
+ None,
+ &["ts"],
+ )
+ .await;
+ let out =
drain_base_source(reader.base_file_source().await.unwrap()).await;
+
+ assert_eq!(
+ invocations.load(Relaxed),
+ 1,
+ "an honestly labelled file must keep its pushdown — the guard keys
on \
+ the FILE's own schema, not on the table's"
+ );
+ assert_eq!(
+ out.num_rows(),
+ 1,
+ "the pushed predicate keeps only the row above the threshold"
+ );
+ }
+
+ /// The narrowing. A file mislabels `ts`, but the predicate reads `other`,
so
+ /// nothing the predicate touches is misread and pushdown must be kept.
Without
+ /// the per-column scoping this file would lose pushdown for a predicate
the
+ /// repair cannot affect.
+ #[tokio::test]
+ async fn
base_read_keeps_pushdown_for_a_predicate_on_an_unaffected_column() {
+ use std::sync::atomic::AtomicUsize;
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let tmp = tempfile::tempdir().unwrap();
+ let base_name = "f1-0_0-1-1_001.parquet";
+ // `ts` mislabelled micros; `other` is honestly labelled millis.
+ let file_schema = Arc::new(arrow_schema::Schema::new(vec![
+ ts_field("ts", arrow_schema::TimeUnit::Microsecond),
+ ts_field("other", arrow_schema::TimeUnit::Millisecond),
+ ]));
+ let batch = RecordBatch::try_new(
+ file_schema,
+ vec![
+ Arc::new(
+
arrow_array::TimestampMicrosecondArray::from(vec![ABOVE_MS, BELOW_MS])
+ .with_timezone("UTC"),
+ ),
+ Arc::new(
+
arrow_array::TimestampMillisecondArray::from(vec![ABOVE_MS, BELOW_MS])
+ .with_timezone("UTC"),
+ ),
+ ],
+ )
+ .unwrap();
+ write_parquet_file(tmp.path(), base_name, &batch);
+
+ let required: SchemaRef = Arc::new(arrow_schema::Schema::new(vec![
+ ts_field("ts", arrow_schema::TimeUnit::Millisecond),
+ ts_field("other", arrow_schema::TimeUnit::Millisecond),
+ ]));
+ let invocations = Arc::new(AtomicUsize::new(0));
+ let builder = nanos_gt_filter_builder("other", THRESHOLD_NANOS,
invocations.clone());
+
+ let mut reader = test_file_group_reader_with_row_filter(
+ tmp.path(),
+ base_name,
+ required,
+ builder,
+ None,
+ // Gate 1 saw only `other`: it is the sole column the predicate
reads.
+ &["other"],
+ )
+ .await;
+ let out =
drain_base_source(reader.base_file_source().await.unwrap()).await;
+
+ assert_eq!(
+ invocations.load(Relaxed),
+ 1,
+ "a mislabelled column the predicate never reads must not cost
pushdown"
+ );
+ assert_eq!(out.num_rows(), 1);
+ }
+
+ /// The unarmed gate. Same mislabelled file and same predicate column, but
gate 1
+ /// reported nothing at risk — the case of every table Spark wrote with
micros.
+ /// The per-file check must not run at all, so pushdown survives.
+ #[tokio::test]
+ async fn base_read_keeps_pushdown_when_no_predicate_column_is_at_risk() {
+ use std::sync::atomic::AtomicUsize;
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let tmp = tempfile::tempdir().unwrap();
+ let base_name = "f1-0_0-1-1_001.parquet";
+ write_straddling_base_file(tmp.path(), base_name,
arrow_schema::TimeUnit::Microsecond);
+
+ let required = straddling_table_schema();
+ let invocations = Arc::new(AtomicUsize::new(0));
+ let builder = nanos_gt_filter_builder("ts", THRESHOLD_NANOS,
invocations.clone());
+
+ let mut reader = test_file_group_reader_with_row_filter(
+ tmp.path(),
+ base_name,
+ required,
+ builder,
+ None,
+ &[], // gate 1 disarmed
+ )
+ .await;
+ let _ =
drain_base_source(reader.base_file_source().await.unwrap()).await;
+
+ assert_eq!(
+ invocations.load(Relaxed),
+ 1,
+ "an empty repair_risk_columns must skip the per-file check
entirely"
+ );
+ }
+
+ /// The table side of gate 2 is the TABLE schema, not the projection. A
pushed
+ /// predicate reads its columns whether or not they were projected,
because the
+ /// `RowFilter` builder derives its own `ProjectionMask` from the parquet
schema.
+ /// Here `ts` is mislabelled and absent from `required_schema`; reading the
+ /// projection instead of the table schema would find nothing and push
anyway.
+ #[tokio::test]
+ async fn base_read_declines_pushdown_for_an_unprojected_predicate_column()
{
+ use std::sync::atomic::AtomicUsize;
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let tmp = tempfile::tempdir().unwrap();
+ let base_name = "f1-0_0-1-1_001.parquet";
+ write_straddling_base_file(tmp.path(), base_name,
arrow_schema::TimeUnit::Microsecond);
+
+ // Projection keeps only the key; `ts` is filtered on but never
returned.
+ let required: SchemaRef =
+ Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
+ "_hoodie_record_key",
+ arrow_schema::DataType::Utf8,
+ true,
+ )]));
+ let invocations = Arc::new(AtomicUsize::new(0));
+ let builder = nanos_gt_filter_builder("ts", THRESHOLD_NANOS,
invocations.clone());
+
+ let mut reader = test_file_group_reader_with_row_filter(
+ tmp.path(),
+ base_name,
+ required,
+ builder,
+ None,
+ &["ts"],
+ )
+ .await;
+ reader.schema_handler.table_schema = Some(straddling_table_schema());
+ let out =
drain_base_source(reader.base_file_source().await.unwrap()).await;
+
+ assert_eq!(
+ invocations.load(Relaxed),
+ 0,
+ "the guard must consult the TABLE schema; a filter column outside
the \
+ projection is still decoded and still misread"
+ );
+ assert_eq!(out.num_rows(), 2);
+ }
+
+ /// A withdrawal takes the row-group selector with it, and is counted on
both
+ /// counters: `row_group_selector_suppressed` so the existing "installed
but
+ /// never passed down" question stays answerable, and
+ /// `pushdown_suppressed_by_repair` so its cause is separable from a
+ /// merge-gate refusal.
+ #[tokio::test]
+ async fn repair_suppression_counts_the_row_group_selector() {
+ use std::sync::atomic::AtomicUsize;
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let tmp = tempfile::tempdir().unwrap();
+ let base_name = "f1-0_0-1-1_001.parquet";
+ write_straddling_base_file(tmp.path(), base_name,
arrow_schema::TimeUnit::Microsecond);
+
+ let invocations = Arc::new(AtomicUsize::new(0));
+ let builder = nanos_gt_filter_builder("ts", THRESHOLD_NANOS,
invocations.clone());
+ let selector_calls = Arc::new(AtomicUsize::new(0));
+ let seen = selector_calls.clone();
+ let selector: RowGroupSelector = Arc::new(move |_| {
+ seen.fetch_add(1, Relaxed);
+ Some(vec![0])
+ });
+
+ let mut reader = test_file_group_reader_with_row_filter(
+ tmp.path(),
+ base_name,
+ straddling_table_schema(),
+ builder,
+ Some(selector),
+ &["ts"],
+ )
+ .await;
+ let volume = reader.storage.read_volume();
+ let out =
drain_base_source(reader.base_file_source().await.unwrap()).await;
+
+ assert_eq!(out.num_rows(), 2);
+ assert_eq!(selector_calls.load(Relaxed), 0, "the selector never ran");
+ assert_eq!(volume.row_group_selector_calls.load(Relaxed), 0);
+ assert_eq!(volume.row_group_selector_suppressed.load(Relaxed), 1);
+ assert_eq!(
+ volume.pushdown_suppressed_by_repair.load(Relaxed),
+ 1,
+ "the cause must be separable from a merge-gate refusal"
+ );
+ }
+
+ /// The row-filter-only case, which `row_group_selector_suppressed`
structurally
+ /// cannot see: no selector was ever installed, so that counter stays zero
while
+ /// pushdown was still withdrawn.
+ #[tokio::test]
+ async fn repair_suppression_is_counted_without_a_row_group_selector() {
+ use std::sync::atomic::AtomicUsize;
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let tmp = tempfile::tempdir().unwrap();
+ let base_name = "f1-0_0-1-1_001.parquet";
+ write_straddling_base_file(tmp.path(), base_name,
arrow_schema::TimeUnit::Microsecond);
+
+ let invocations = Arc::new(AtomicUsize::new(0));
+ let builder = nanos_gt_filter_builder("ts", THRESHOLD_NANOS,
invocations);
+
+ let mut reader = test_file_group_reader_with_row_filter(
+ tmp.path(),
+ base_name,
+ straddling_table_schema(),
+ builder,
+ None,
+ &["ts"],
+ )
+ .await;
+ let volume = reader.storage.read_volume();
+ let _ =
drain_base_source(reader.base_file_source().await.unwrap()).await;
+
+ assert_eq!(
+ volume.row_group_selector_suppressed.load(Relaxed),
+ 0,
+ "no selector was installed, so that counter cannot speak for this
case"
+ );
+ assert_eq!(volume.pushdown_suppressed_by_repair.load(Relaxed), 1);
+ }
+
+ /// And it must stay at zero when pushdown survives, or the counter cannot
+ /// distinguish "a file was withdrawn" from "the scan ran".
+ #[tokio::test]
+ async fn repair_suppression_is_not_counted_when_pushdown_survives() {
+ use std::sync::atomic::AtomicUsize;
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let tmp = tempfile::tempdir().unwrap();
+ let base_name = "f1-0_0-1-1_001.parquet";
+ write_straddling_base_file(tmp.path(), base_name,
arrow_schema::TimeUnit::Millisecond);
+
+ let invocations = Arc::new(AtomicUsize::new(0));
+ let builder = nanos_gt_filter_builder("ts", THRESHOLD_NANOS,
invocations);
+
+ let mut reader = test_file_group_reader_with_row_filter(
+ tmp.path(),
+ base_name,
+ straddling_table_schema(),
+ builder,
+ None,
+ &["ts"],
+ )
+ .await;
+ let volume = reader.storage.read_volume();
+ let _ =
drain_base_source(reader.base_file_source().await.unwrap()).await;
+
+ assert_eq!(volume.pushdown_suppressed_by_repair.load(Relaxed), 0);
+ }
+
+ #[test]
+ fn builder_routes_repair_risk_columns_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_filter_builder(make_row_filter_builder())
+ .with_repair_risk_columns(vec!["ts".to_string()])
+ .build()
+ .unwrap();
+ assert_eq!(
+ reader.reader_context.repair_risk_columns,
+ vec!["ts".to_string()],
+ "with_repair_risk_columns should land on reader_context"
+ );
+ }
+
+ #[test]
+ fn builder_leaves_repair_risk_columns_empty_by_default() {
+ 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_filter_builder(make_row_filter_builder())
+ .build()
+ .unwrap();
+ assert!(
+ reader.reader_context.repair_risk_columns.is_empty(),
+ "unset must leave the guard disarmed, not populated by accident"
+ );
+ }
}
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 883d5ccd..51197f38 100644
--- a/crates/core/src/file_group/reader_v2/reader_context.rs
+++ b/crates/core/src/file_group/reader_v2/reader_context.rs
@@ -180,6 +180,34 @@ pub struct ReaderContext {
/// post-merge evaluation (handled by the caller — Velox/Spark — above
/// the FG reader).
pub mor_pk_safe: bool,
+ /// Predicate columns the TABLE schema declares tz-aware millis, i.e. the
only
+ /// columns the apache/hudi#18132 logical-type repair can reinterpret on
read.
+ ///
+ /// Parquet evaluates a pushed predicate against a file's PHYSICAL values,
+ /// before `project_batch_to_schema` runs. That is sound only while a
physical
+ /// value means what its physical type says, which the repair breaks: a
legacy
+ /// file labels the column micros while the stored i64 is millis, so a
+ /// millis-semantics literal reads those rows as 1970 and the scan drops
rows
+ /// that match. A post-scan filter cannot restore them.
+ ///
+ /// Computed once per scan by whoever supplies
[`Self::row_filter_builder`], from
+ /// that predicate's referenced columns and the table schema, via
+ /// [`crate::schema::batch_evolution::repair_risk_columns`]; routed here by
+ ///
[`HoodieFileGroupReaderBuilder::with_repair_risk_columns`](crate::file_group::reader_v2::HoodieFileGroupReaderBuilder::with_repair_risk_columns).
+ /// **Empty on every table whose predicate touches no tz-aware millis
column,
+ /// which is the common case**; the base read then skips the per-file
footer
+ /// comparison entirely and keeps all pushdown.
+ ///
+ /// Non-empty only arms the check. Whether a given file actually mislabels
one
+ /// of these columns is decided per file against its footer schema, in
+ /// `HoodieFileGroupReader::make_base_file_source`.
+ ///
+ /// Empty is also the default, so a caller that pushes a filter without
setting
+ /// this gets no guard. That is deliberate — the reader cannot derive a
+ /// predicate's columns from an opaque [`RowFilterBuilder`] — and it is why
+ /// `with_repair_risk_columns` is documented as required alongside
+ /// `with_row_filter_builder` rather than as a tuning knob.
+ pub repair_risk_columns: Vec<String>,
/// Gate-3 completed/inflight inputs (completed/inflight/archived sets).
/// Carried here mirroring [`Self::instant_range`]. `Some` only when the
caller holds a
/// timeline *and* the table version is below 8 — see
@@ -230,6 +258,7 @@ impl std::fmt::Debug for ReaderContext {
&self.row_group_selector.as_ref().map(|_| "<closure>"),
)
.field("mor_pk_safe", &self.mor_pk_safe)
+ .field("repair_risk_columns", &self.repair_risk_columns)
.field("completion_gate_inputs", &self.completion_gate_inputs)
.finish()
}
@@ -338,6 +367,7 @@ impl ReaderContext {
row_group_selector: None,
key_predicate: None,
mor_pk_safe: false,
+ repair_risk_columns: Vec::new(),
completion_gate_inputs: None,
}
}
diff --git a/crates/core/src/file_group/reader_v2/resolver.rs
b/crates/core/src/file_group/reader_v2/resolver.rs
index 89076f3a..e053b79f 100644
--- a/crates/core/src/file_group/reader_v2/resolver.rs
+++ b/crates/core/src/file_group/reader_v2/resolver.rs
@@ -103,6 +103,9 @@ pub(crate) fn resolve_reader_context(
row_group_selector: None,
key_predicate: None,
mor_pk_safe: false,
+ // No pushed predicate, so no column can be misread by one and the
+ // per-file repair check never arms.
+ repair_risk_columns: Vec::new(),
// The table-version < 8 completion gate needs a timeline the caller
// has not loaded; leaving it unset keeps the gate a no-op.
completion_gate_inputs: None,
diff --git a/crates/core/src/schema/batch_evolution.rs
b/crates/core/src/schema/batch_evolution.rs
index 6273773d..aaae693c 100644
--- a/crates/core/src/schema/batch_evolution.rs
+++ b/crates/core/src/schema/batch_evolution.rs
@@ -93,6 +93,138 @@ pub(crate) fn index_of_ci(schema: &arrow_schema::Schema,
name: &str) -> Result<O
Ok(found)
}
+/// True when evolving `file` → `table` REINTERPRETS the stored buffer instead
of
+/// preserving what the value means.
+///
+/// Only the apache/hudi#18132 arm below does this: it relabels the i64
because the
+/// value was always millis and only the label was wrong, so for that pairing
alone
+/// the physical value does not mean what the physical type says. Every other
arm
+/// (int widening, decimal rescale, millis→micros cast, the NTZ divide)
denotes the
+/// same logical value before and after.
+///
+/// That distinction is what makes base-read predicate pushdown sound — see
+/// `HoodieFileGroupReader::make_base_file_source`.
+fn is_value_reinterpreting(file: &DataType, table: &DataType) -> bool {
+ matches!(
+ (file, table),
+ (
+ DataType::Timestamp(TimeUnit::Microsecond, Some(_)),
+ DataType::Timestamp(TimeUnit::Millisecond, Some(_)),
+ )
+ )
+}
+
+/// [`is_value_reinterpreting`] lifted through the container arms, mirroring
the
+/// recursion in [`evolve_array`] so a nested affected field is not missed.
+/// Container drift returns `false`; the read itself rejects that pairing.
+fn pair_is_value_reinterpreting(file: &DataType, table: &DataType) -> bool {
+ if is_value_reinterpreting(file, table) {
+ return true;
+ }
+ match (file, table) {
+ (DataType::Struct(ff), DataType::Struct(tf)) => tf.iter().any(|t| {
+ ff.iter()
+ .find(|f| f.name().eq_ignore_ascii_case(t.name()))
+ .is_some_and(|f| pair_is_value_reinterpreting(f.data_type(),
t.data_type()))
+ }),
+ (DataType::List(f), DataType::List(t))
+ | (DataType::LargeList(f), DataType::LargeList(t))
+ | (DataType::Map(f, _), DataType::Map(t, _)) => {
+ pair_is_value_reinterpreting(f.data_type(), t.data_type())
+ }
+ _ => false,
+ }
+}
+
+/// The TABLE half of [`is_value_reinterpreting`], recursed through containers.
+///
+/// The repair arm fires only when the table side is tz-aware millis, so a
column
+/// the table declares as anything else can never carry the #18132 mislabel —
no
+/// matter what any file says. That makes this decidable from the table schema
+/// alone, without opening a file.
+fn is_repair_target(table: &DataType) -> bool {
+ match table {
+ DataType::Timestamp(TimeUnit::Millisecond, Some(_)) => true,
+ DataType::Struct(fields) => fields.iter().any(|f|
is_repair_target(f.data_type())),
+ DataType::List(f) | DataType::LargeList(f) | DataType::Map(f, _) => {
+ is_repair_target(f.data_type())
+ }
+ _ => false,
+ }
+}
+
+/// Which of `predicate_columns` could ever be misread by a pushed predicate,
+/// judged from the TABLE schema alone.
+///
+/// Meant to be computed ONCE per scan by whoever supplies the predicate, and
handed
+/// to the reader via
[`HoodieFileGroupReaderBuilder::with_repair_risk_columns`]
+/// (crate::file_group::reader_v2). Two properties earn it that place:
+///
+/// * **It is usually empty.** Spark's `TimestampType` maps to micros, so a
+/// tz-aware *millis* column is the legacy shape the #18132 repair exists
for.
+/// An empty result means no base read in the scan needs any per-file check,
+/// and no file loses pushdown.
+/// * **It is scoped to the predicate.** A mislabelled column the predicate
never
+/// references cannot make the predicate wrong, so it must not cost pushdown.
+/// `predicate_columns` must therefore be the columns the expression
references,
+/// not every column in the schema the predicate was compiled against.
+///
+/// A name that resolves ambiguously is reported AS at risk rather than
raising:
+/// this decides only whether to push a predicate, and over-reporting costs
+/// pushdown while under-reporting drops rows.
+pub fn repair_risk_columns(
+ table_schema: &arrow_schema::Schema,
+ predicate_columns: &[String],
+) -> Vec<String> {
+ predicate_columns
+ .iter()
+ .filter(|name| match index_of_ci(table_schema, name) {
+ Ok(Some(idx)) =>
is_repair_target(table_schema.fields()[idx].data_type()),
+ Ok(None) => false,
+ Err(_) => true,
+ })
+ .cloned()
+ .collect()
+}
+
+/// Which of `candidates` this file actually mislabels, i.e. evolving it to
+/// `table_schema` would reinterpret the buffer rather than preserve its
meaning.
+///
+/// `candidates` is [`repair_risk_columns`]'s output, so this walks a handful
of
+/// named columns rather than the whole footer schema. A candidate missing from
+/// either schema is skipped: absent from the file there is nothing to misread,
+/// and absent from the table there is no repair to reinterpret it. Names come
+/// back in the FILE's spelling, which is how a pushed predicate addresses the
+/// parquet column.
+///
+/// The table side is deliberately the table schema, not the projected one: a
pushed
+/// predicate reads its columns whether or not they were projected, because the
+/// `RowFilter` builder derives its own `ProjectionMask` from the parquet
schema
+/// rather than from the read's projection.
+pub(crate) fn reinterpreted_columns(
+ file_schema: &arrow_schema::Schema,
+ table_schema: &arrow_schema::Schema,
+ candidates: &[String],
+) -> Result<Vec<String>> {
+ let mut out = Vec::with_capacity(candidates.len());
+ for name in candidates {
+ let (Some(fi), Some(ti)) = (
+ index_of_ci(file_schema, name)?,
+ index_of_ci(table_schema, name)?,
+ ) else {
+ continue;
+ };
+ let file_field = &file_schema.fields()[fi];
+ if pair_is_value_reinterpreting(
+ file_field.data_type(),
+ table_schema.fields()[ti].data_type(),
+ ) {
+ out.push(file_field.name().clone());
+ }
+ }
+ Ok(out)
+}
+
/// True for any nested/container Arrow type the recursion arms care about.
/// Matching variants (List/Struct/Map) are handled by the recursion arms above
/// the guard; this catches everything else (LargeList, FixedSizeList, and any
@@ -1256,4 +1388,352 @@ mod tests {
&[5_000_000_000.0f64]
);
}
+
+ // The two pushdown classifiers. Flagging too little drops rows silently;
+ // flagging too much only costs pushdown. These pin both directions.
+
+ fn ts(unit: TimeUnit, tz: Option<&str>) -> DataType {
+ DataType::Timestamp(unit, tz.map(Into::into))
+ }
+
+ /// Every file column is offered as a candidate, so these pin the per-file
rule
+ /// itself. `repair_risk_columns` decides which candidates a real scan
supplies.
+ fn reinterpreted(file: Vec<Field>, required: Vec<Field>) -> Vec<String> {
+ let file_schema = Schema::new(file);
+ let candidates: Vec<String> = file_schema
+ .fields()
+ .iter()
+ .map(|f| f.name().clone())
+ .collect();
+ super::reinterpreted_columns(&file_schema, &Schema::new(required),
&candidates).unwrap()
+ }
+
+ fn risk(table: Vec<Field>, predicate_columns: &[&str]) -> Vec<String> {
+ let cols: Vec<String> = predicate_columns.iter().map(|c|
c.to_string()).collect();
+ super::repair_risk_columns(&Schema::new(table), &cols)
+ }
+
+ #[test]
+ fn
repair_risk_columns_is_empty_when_the_predicate_touches_no_millis_column() {
+ // THE GATE THAT PAYS FOR ITSELF. A predicate over honest columns arms
+ // nothing, so no base read in the scan opens a footer for this check
and
+ // no file loses pushdown. This is the common case on any table Spark
wrote.
+ assert!(
+ risk(
+ vec![
+ Field::new("id", DataType::Int64, true),
+ Field::new("ts", ts(TimeUnit::Microsecond, Some("UTC")),
true),
+ ],
+ &["id", "ts"],
+ )
+ .is_empty(),
+ "a table declaring micros can never be the TARGET of the #18132
repair"
+ );
+ }
+
+ #[test]
+ fn repair_risk_columns_flags_only_the_predicate_columns_at_risk() {
+ // Scope is the predicate, not the table. `other` is at risk but
unreferenced,
+ // so it must not cost this scan its pushdown.
+ assert_eq!(
+ risk(
+ vec![
+ Field::new("ts", ts(TimeUnit::Millisecond, Some("UTC")),
true),
+ Field::new("other", ts(TimeUnit::Millisecond,
Some("UTC")), true),
+ Field::new("id", DataType::Int64, true),
+ ],
+ &["ts", "id"],
+ ),
+ vec!["ts".to_string()]
+ );
+ }
+
+ #[test]
+ fn repair_risk_columns_ignores_ntz_and_sees_through_containers() {
+ // NTZ millis is not the repair's target (it matches only the tz-aware
+ // logical classes), while a tz-aware millis field nested in a struct
is.
+ assert!(
+ risk(
+ vec![Field::new("ntz", ts(TimeUnit::Millisecond, None), true)],
+ &["ntz"],
+ )
+ .is_empty()
+ );
+ let nested = DataType::Struct(
+ vec![Field::new(
+ "inner",
+ ts(TimeUnit::Millisecond, Some("UTC")),
+ true,
+ )]
+ .into(),
+ );
+ assert_eq!(
+ risk(vec![Field::new("s", nested, true)], &["s"]),
+ vec!["s".to_string()]
+ );
+ }
+
+ #[test]
+ fn repair_risk_columns_matches_names_case_insensitively() {
+ assert_eq!(
+ risk(
+ vec![Field::new(
+ "TS",
+ ts(TimeUnit::Millisecond, Some("UTC")),
+ true
+ )],
+ &["ts"],
+ ),
+ vec!["ts".to_string()]
+ );
+ }
+
+ #[test]
+ fn reinterpreted_columns_checks_only_the_candidates_it_is_given() {
+ // The per-file walk is scoped to the risk set. A mislabelled column
the
+ // predicate never references is not a candidate, so it must not be
+ // reported -- it cannot make the predicate wrong.
+ let file = Schema::new(vec![
+ Field::new("ts", ts(TimeUnit::Microsecond, Some("UTC")), true),
+ Field::new("unreferenced", ts(TimeUnit::Microsecond, Some("UTC")),
true),
+ ]);
+ let required = Schema::new(vec![
+ Field::new("ts", ts(TimeUnit::Millisecond, Some("UTC")), true),
+ Field::new("unreferenced", ts(TimeUnit::Millisecond, Some("UTC")),
true),
+ ]);
+ assert_eq!(
+ super::reinterpreted_columns(&file, &required,
&["ts".to_string()]).unwrap(),
+ vec!["ts".to_string()],
+ "only the candidate is reported, though both columns are
mislabelled"
+ );
+ assert!(
+ super::reinterpreted_columns(&file, &required, &[])
+ .unwrap()
+ .is_empty(),
+ "no candidates means no work and no refusal"
+ );
+ }
+
+ #[test]
+ fn reinterpreted_columns_skips_a_candidate_missing_from_either_schema() {
+ // Absent from the file: nothing to misread. Absent from required:
never
+ // projected, so never repaired. Neither may panic on the index lookup.
+ let file = Schema::new(vec![Field::new(
+ "ts",
+ ts(TimeUnit::Microsecond, Some("UTC")),
+ true,
+ )]);
+ let required = Schema::new(vec![Field::new(
+ "ts",
+ ts(TimeUnit::Millisecond, Some("UTC")),
+ true,
+ )]);
+ let absent = ["nope".to_string()];
+ assert!(
+ super::reinterpreted_columns(&file, &required, &absent)
+ .unwrap()
+ .is_empty()
+ );
+ assert!(
+ super::reinterpreted_columns(&Schema::empty(), &required,
&["ts".to_string()])
+ .unwrap()
+ .is_empty()
+ );
+ }
+
+ #[test]
+ fn reinterpreted_columns_flags_the_hudi_18132_pair() {
+ // File says tz-aware micros, table says tz-aware millis, stored i64
was
+ // millis all along — the only pairing a predicate can misread.
+ assert_eq!(
+ reinterpreted(
+ vec![Field::new(
+ "ts",
+ ts(TimeUnit::Microsecond, Some("UTC")),
+ true
+ )],
+ vec![Field::new(
+ "ts",
+ ts(TimeUnit::Millisecond, Some("UTC")),
+ true
+ )],
+ ),
+ vec!["ts".to_string()]
+ );
+ }
+
+ #[test]
+ fn reinterpreted_columns_flags_the_pair_across_differing_timezones() {
+ // The repair arm accepts a tz mismatch (it warns, then reinterprets,
since
+ // the i64 epoch is instant-preserving). The rule must agree, or a
predicate
+ // would be pushed into a read the repair still rewrites.
+ assert_eq!(
+ reinterpreted(
+ vec![Field::new(
+ "ts",
+ ts(TimeUnit::Microsecond, Some("UTC")),
+ true
+ )],
+ vec![Field::new(
+ "ts",
+ ts(TimeUnit::Millisecond, Some("America/New_York")),
+ true
+ )],
+ ),
+ vec!["ts".to_string()]
+ );
+ }
+
+ #[test]
+ fn reinterpreted_columns_reports_every_affected_column() {
+ // A file can carry more than one affected column; the log line names
them,
+ // so all of them must come back, and unaffected siblings must not.
+ assert_eq!(
+ reinterpreted(
+ vec![
+ Field::new("a", ts(TimeUnit::Microsecond, Some("UTC")),
true),
+ Field::new("ok", DataType::Int32, true),
+ Field::new("b", ts(TimeUnit::Microsecond, Some("UTC")),
true),
+ ],
+ vec![
+ Field::new("a", ts(TimeUnit::Millisecond, Some("UTC")),
true),
+ Field::new("ok", DataType::Int64, true),
+ Field::new("b", ts(TimeUnit::Millisecond, Some("UTC")),
true),
+ ],
+ ),
+ vec!["a".to_string(), "b".to_string()]
+ );
+ }
+
+ #[test]
+ fn reinterpreted_columns_ignores_value_preserving_evolutions() {
+ // Each of these evolves the column but PRESERVES what the value
denotes,
+ // so pushdown stays sound and must not be declined.
+ let cases: Vec<(&str, DataType, DataType)> = vec![
+ // NTZ micros→millis is an arithmetic ÷1000, not a relabel: same
instant.
+ (
+ "ntz_micros_to_millis",
+ ts(TimeUnit::Microsecond, None),
+ ts(TimeUnit::Millisecond, None),
+ ),
+ // The reverse direction is a legitimate widening via arrow_cast
(×1000).
+ (
+ "millis_to_micros",
+ ts(TimeUnit::Millisecond, Some("UTC")),
+ ts(TimeUnit::Microsecond, Some("UTC")),
+ ),
+ // Same unit on both sides: no evolution at all.
+ (
+ "micros_to_micros",
+ ts(TimeUnit::Microsecond, Some("UTC")),
+ ts(TimeUnit::Microsecond, Some("UTC")),
+ ),
+ // Ordinary promotions.
+ ("int_widening", DataType::Int32, DataType::Int64),
+ ("float_widening", DataType::Float32, DataType::Float64),
+ // Mixed tz-awareness is NOT the #18132 shape (Java matches only
the
+ // tz-aware logical classes), so it must not be flagged either way.
+ (
+ "ntz_file_to_tz_table",
+ ts(TimeUnit::Microsecond, None),
+ ts(TimeUnit::Millisecond, Some("UTC")),
+ ),
+ (
+ "tz_file_to_ntz_table",
+ ts(TimeUnit::Microsecond, Some("UTC")),
+ ts(TimeUnit::Millisecond, None),
+ ),
+ // Seconds and nanos are outside the repair entirely.
+ (
+ "seconds_to_millis",
+ ts(TimeUnit::Second, Some("UTC")),
+ ts(TimeUnit::Millisecond, Some("UTC")),
+ ),
+ (
+ "micros_to_nanos",
+ ts(TimeUnit::Microsecond, Some("UTC")),
+ ts(TimeUnit::Nanosecond, Some("UTC")),
+ ),
+ ];
+ for (name, file, required) in cases {
+ assert!(
+ reinterpreted(
+ vec![Field::new("c", file, true)],
+ vec![Field::new("c", required, true)],
+ )
+ .is_empty(),
+ "{name} preserves the value's meaning and must keep its
pushdown"
+ );
+ }
+ }
+
+ #[test]
+ fn reinterpreted_columns_ignores_a_column_absent_from_the_table_schema() {
+ // A file column the table does not ask for is never projected, so it
is
+ // never repaired and cannot be misread.
+ assert!(
+ reinterpreted(
+ vec![Field::new(
+ "ts",
+ ts(TimeUnit::Microsecond, Some("UTC")),
+ true
+ )],
+ vec![Field::new("other", DataType::Int32, true)],
+ )
+ .is_empty()
+ );
+ }
+
+ #[test]
+ fn reinterpreted_columns_matches_names_case_insensitively() {
+ // The projection resolves names case-insensitively (index_of_ci), so
this
+ // must too, or a file spelling the column `TS` would push unsafely.
+ assert_eq!(
+ reinterpreted(
+ vec![Field::new(
+ "TS",
+ ts(TimeUnit::Microsecond, Some("UTC")),
+ true
+ )],
+ vec![Field::new(
+ "ts",
+ ts(TimeUnit::Millisecond, Some("UTC")),
+ true
+ )],
+ ),
+ vec!["TS".to_string()],
+ "the returned name is the FILE's spelling, which is how a
predicate \
+ addresses the parquet column"
+ );
+ }
+
+ #[test]
+ fn reinterpreted_columns_sees_through_containers() {
+ // The repair recurses into structs/lists/maps, so the rule must too --
+ // otherwise an affected field nested one level down keeps its
pushdown and
+ // drops rows silently.
+ let nested = |unit: TimeUnit| {
+ DataType::Struct(vec![Field::new("inner", ts(unit, Some("UTC")),
true)].into())
+ };
+ assert_eq!(
+ reinterpreted(
+ vec![Field::new("s", nested(TimeUnit::Microsecond), true)],
+ vec![Field::new("s", nested(TimeUnit::Millisecond), true)],
+ ),
+ vec!["s".to_string()],
+ "an affected field inside a struct must flag its top-level column"
+ );
+
+ let listed = |unit: TimeUnit| {
+ DataType::List(Arc::new(Field::new("item", ts(unit, Some("UTC")),
true)))
+ };
+ assert_eq!(
+ reinterpreted(
+ vec![Field::new("l", listed(TimeUnit::Microsecond), true)],
+ vec![Field::new("l", listed(TimeUnit::Millisecond), true)],
+ ),
+ vec!["l".to_string()],
+ "and so must one inside a list"
+ );
+ }
}
diff --git a/crates/core/src/storage/mod.rs b/crates/core/src/storage/mod.rs
index a025878f..4dcfb0e4 100644
--- a/crates/core/src/storage/mod.rs
+++ b/crates/core/src/storage/mod.rs
@@ -136,17 +136,35 @@ pub struct ReadVolume {
/// 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.
+ /// Times a selector WAS installed by the caller but a gate refused to
pass it
+ /// down — either the merge-safety gate or a value-reinterpreting
logical-type
+ /// repair on the file.
///
- /// 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:
+ /// Without this a gate silently defeats the counter above: a suppressed
+ /// selector is a third state that also reads zero calls.
+ ///
+ /// Counted regardless of cause, so it stays a faithful answer to "was one
+ /// installed but not passed down"; [`Self::pushdown_suppressed_by_repair`]
+ /// says which cause. Read them 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 a gate refused it; by_repair == 0 means
the
+ /// merge-safety gate (the read merges, and
the
+ /// predicate is not primary-key-safe),
+ /// by_repair > 0 means a repair conflict
/// calls == 0, suppressed == 0 no caller ever installed one
pub row_group_selector_suppressed: AtomicU64,
+ /// Base files where a pushed predicate read a column THIS file mislabels,
so
+ /// every pushdown mechanism was withdrawn for it.
+ ///
+ /// Counted once per such file whether or not a selector was installed, so
+ /// unlike [`Self::row_group_selector_suppressed`] it also covers the
+ /// row-filter side.
+ ///
+ /// Non-zero is expected on a table with legacy `parquet-mr` base files
and a
+ /// predicate over a tz-aware millis column: those files fell back to the
+ /// post-scan filter, they did not lose rows. Rising on a table that
should be
+ /// all-micros is the signal worth chasing.
+ pub pushdown_suppressed_by_repair: AtomicU64,
/// Rows the file contains, from parquet metadata.
pub file_rows: AtomicU64,
/// Rows the stream actually yielded, after any row filter. `file_rows -
@@ -185,6 +203,15 @@ impl ReadVolume {
.fetch_add(1, Ordering::Relaxed);
}
+ /// This base file mislabels a column the pushed predicate reads, so every
+ /// pushdown mechanism was withdrawn for it. Counted separately from
+ /// `record_selector_suppressed` because it fires with no selector
installed
+ /// too, and because the two causes must stay distinguishable.
+ pub(crate) fn record_pushdown_suppressed_by_repair(&self) {
+ self.pushdown_suppressed_by_repair
+ .fetch_add(1, Ordering::Relaxed);
+ }
+
pub(crate) fn add_rows_out(&self, n: u64) {
self.rows_out.fetch_add(n, Ordering::Relaxed);
}
