From bcdff0c70d77663b6993b25be19f6f3634ed08bd Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Mon, 6 Jul 2026 17:19:25 +0800 Subject: [PATCH 1/8] fix(core): apply exact post-merge residual filtering on the PK merge read path --- crates/paimon/src/table/kv_file_reader.rs | 326 ++++++++++++++++++++-- 1 file changed, 297 insertions(+), 29 deletions(-) diff --git a/crates/paimon/src/table/kv_file_reader.rs b/crates/paimon/src/table/kv_file_reader.rs index 6d3e4e5a..dc5eeaf8 100644 --- a/crates/paimon/src/table/kv_file_reader.rs +++ b/crates/paimon/src/table/kv_file_reader.rs @@ -20,6 +20,8 @@ //! Each data file in a split is read as a separate sorted stream. The streams //! are merged by primary key using a LoserTree, and rows with the same key are //! deduplicated by keeping the one with the highest `_SEQUENCE_NUMBER`. +//! Non-primary-key predicate conjuncts are enforced by an exact post-merge +//! residual filter; only primary-key conjuncts are pushed below the merge. //! //! Reference: Java Paimon `SortMergeReaderWithMinHeap`. @@ -48,6 +50,11 @@ use std::collections::HashMap; pub(crate) struct KeyValueFileReader { file_io: FileIO, config: KeyValueReadConfig, + /// PK-only conjuncts pushed down to the per-file readers before merge. + /// Non-PK conjuncts must not run pre-merge (they can change which version + /// of a key survives); they are enforced by the post-merge residual + /// filter using the full `config.predicates` instead. + pushdown_predicates: Vec, } /// Configuration for [`KeyValueFileReader`], grouping table schema and @@ -65,37 +72,49 @@ pub(crate) struct KeyValueReadConfig { pub sequence_fields: Vec, } +/// Keep only the conjuncts of `predicates` that reference primary-key columns, +/// preserving table-schema field indices. Mixed `AND`s keep their PK children; +/// `OR`/`NOT` require every child to be PK-only (see +/// [`Predicate::project_field_index_inclusive`]). +/// +/// Used for pre-merge pushdown in [`KeyValueFileReader`] and for per-file +/// stats pruning of primary-key tables in scan planning: a key's versions all +/// share the key columns, so key conjuncts can never drop one version of a +/// key while keeping another — non-key conjuncts can, which corrupts merge. +pub(super) fn retain_primary_key_conjuncts( + predicates: &[Predicate], + table_fields: &[DataField], + primary_keys: &[String], +) -> Vec { + let pk_set: std::collections::HashSet<&str> = primary_keys.iter().map(|s| s.as_str()).collect(); + let mapping: Vec> = table_fields + .iter() + .enumerate() + .map(|(i, f)| { + if pk_set.contains(f.name()) { + Some(i) + } else { + None + } + }) + .collect(); + predicates + .iter() + .filter_map(|p| p.project_field_index_inclusive(&mapping)) + .collect() +} + impl KeyValueFileReader { pub(crate) fn new(file_io: FileIO, config: KeyValueReadConfig) -> Self { - // Only keep predicates that reference primary key columns. - // Non-PK predicates applied before merge can cause incorrect results. - // Use project_field_index_inclusive: AND keeps PK children, OR requires all PK. - let pk_set: std::collections::HashSet<&str> = - config.primary_keys.iter().map(|s| s.as_str()).collect(); - let mapping: Vec> = config - .table_fields - .iter() - .enumerate() - .map(|(i, f)| { - if pk_set.contains(f.name()) { - Some(i) - } else { - None - } - }) - .collect(); - let pk_predicates = config - .predicates - .into_iter() - .filter_map(|p| p.project_field_index_inclusive(&mapping)) - .collect(); - + let pushdown_predicates = retain_primary_key_conjuncts( + &config.predicates, + &config.table_fields, + &config.primary_keys, + ); Self { file_io, - config: KeyValueReadConfig { - predicates: pk_predicates, - ..config - }, + config, + pushdown_predicates, } } @@ -193,6 +212,22 @@ impl KeyValueFileReader { } } + // Widen with predicate columns not already read so the post-merge + // residual filter can evaluate every leaf (predicate leaf indices are + // table-schema positions). Extras ride through the merge as ordinary + // value columns — partial-update/aggregation apply their configured + // per-field semantics to them, so the residual sees properly MERGED + // values — and the read_type reorder below drops them from the output. + let residual_file_predicates = + (!self.config.predicates.is_empty()).then(|| crate::arrow::format::FilePredicates { + predicates: self.config.predicates.clone(), + file_fields: self.config.table_fields.clone(), + }); + let user_fields = crate::arrow::residual::widen_scan_fields( + &user_fields, + residual_file_predicates.as_ref(), + ); + // Internal read type: [_SEQ, _VK, user_fields...] let mut internal_read_type: Vec = Vec::new(); internal_read_type.push(seq_field); @@ -274,7 +309,8 @@ impl KeyValueFileReader { let table_fields = self.config.table_fields; let table_name = self.config.table_name; let table_options = self.config.table_options; - let predicates = self.config.predicates; + let pushdown_predicates = self.pushdown_predicates; + let residual_predicates = self.config.predicates; let primary_keys = self.config.primary_keys; let sequence_fields = self.config.sequence_fields; @@ -313,7 +349,7 @@ impl KeyValueFileReader { table_schema_id, table_fields.clone(), internal_read_type.clone(), - predicates.clone(), + pushdown_predicates.clone(), ); let stream = reader.read_single_file_stream( @@ -355,6 +391,36 @@ impl KeyValueFileReader { while let Some(batch) = merge_stream.next().await { let batch = batch?; + // Post-merge residual: enforce the FULL data predicate on + // merged rows. PK conjuncts are also in this set (they were + // already pushed down pre-merge); re-evaluating them on + // already-matching rows is a no-op and keeps one shared + // evaluator instead of deriving a non-PK subset. Runs on + // the merge-output batch (keys + values, including widened + // predicate columns); the reorder below projects the + // output back to read_type. + let batch = if residual_predicates.is_empty() { + batch + } else { + match crate::arrow::residual::evaluate_predicates_mask( + &batch, + &residual_predicates, + &table_fields, + &merge_output_fields, + )? { + Some(mask) => { + arrow_select::filter::filter_record_batch(&batch, &mask).map_err( + |e| Error::DataInvalid { + message: format!( + "Failed to filter merged batch by predicates: {e}" + ), + source: Some(Box::new(e)), + }, + )? + } + None => batch, + } + }; // Reorder columns from [keys..., values...] to read_type order. let columns: Vec<_> = reorder_map .iter() @@ -377,3 +443,205 @@ impl KeyValueFileReader { .boxed()) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::catalog::Identifier; + use crate::io::FileIOBuilder; + use crate::spec::{DataType, Datum, IntType, PredicateBuilder, Schema, TableSchema}; + use crate::table::table_commit::TableCommit; + use crate::table::{Table, TableWrite}; + use arrow_array::{Array, Int32Array}; + use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema}; + use std::sync::Arc; + + fn test_file_io() -> FileIO { + FileIOBuilder::new("memory").build().unwrap() + } + + fn pk_table(file_io: &FileIO, table_path: &str, options: &[(&str, &str)]) -> Table { + let mut builder = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("bucket", "1"); + for (key, value) in options { + builder = builder.option(*key, *value); + } + Table::new( + file_io.clone(), + Identifier::new("default", "kv_residual_t"), + table_path.to_string(), + TableSchema::new(0, &builder.build().unwrap()), + None, + ) + } + + async fn setup_dirs(file_io: &FileIO, table_path: &str) { + file_io + .mkdirs(&format!("{table_path}/snapshot/")) + .await + .unwrap(); + file_io + .mkdirs(&format!("{table_path}/manifest/")) + .await + .unwrap(); + } + + fn int_batch(ids: Vec, values: Vec>) -> RecordBatch { + let schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int32, false), + ArrowField::new("value", ArrowDataType::Int32, true), + ])); + RecordBatch::try_new( + schema, + vec![ + Arc::new(Int32Array::from(ids)), + Arc::new(Int32Array::from(values)), + ], + ) + .unwrap() + } + + async fn write_commit(table: &Table, batch: &RecordBatch) { + let mut tw = TableWrite::new(table, "test-user".to_string()).unwrap(); + tw.write_arrow_batch(batch).await.unwrap(); + let msgs = tw.prepare_commit().await.unwrap(); + TableCommit::new(table.clone(), "test-user".to_string()) + .commit(msgs) + .await + .unwrap(); + } + + async fn read_rows( + table: &Table, + projection: Option<&[&str]>, + filter: Option, + ) -> Vec { + let mut rb = table.new_read_builder(); + if let Some(cols) = projection { + rb.with_projection(cols); + } + if let Some(f) = filter { + rb.with_filter(f); + } + let plan = rb.new_scan().plan().await.unwrap(); + let read = rb.new_read().unwrap(); + futures::TryStreamExt::try_collect(read.to_arrow(plan.splits()).unwrap()) + .await + .unwrap() + } + + fn int_column(batches: &[RecordBatch], name: &str) -> Vec { + batches + .iter() + .flat_map(|b| { + let idx = b.schema().index_of(name).unwrap(); + let arr = b.column(idx).as_any().downcast_ref::().unwrap(); + (0..arr.len()).map(|i| arr.value(i)).collect::>() + }) + .collect() + } + + /// Non-PK equality filter on a dedup PK table read through the sort-merge + /// path must return only matching rows. Before the post-merge residual, + /// the non-PK conjunct was silently dropped and all rows came back. + #[tokio::test] + async fn kv_read_applies_non_pk_filter_exactly() { + let file_io = test_file_io(); + let table_path = "memory:/kv_residual_eq"; + setup_dirs(&file_io, table_path).await; + let table = pk_table(&file_io, table_path, &[]); + + // Overlapping keys across two commits -> split is not raw convertible + // -> forced through KeyValueFileReader. + write_commit( + &table, + &int_batch(vec![1, 2, 3], vec![Some(10), Some(20), Some(30)]), + ) + .await; + write_commit( + &table, + &int_batch(vec![1, 2, 3], vec![Some(11), Some(21), Some(31)]), + ) + .await; + + let fields = table.schema().fields().to_vec(); + let filter = PredicateBuilder::new(&fields) + .equal("value", Datum::Int(21)) + .unwrap(); + let batches = read_rows(&table, None, Some(filter)).await; + + assert_eq!(int_column(&batches, "id"), vec![2]); + assert_eq!(int_column(&batches, "value"), vec![21]); + } + + /// Gap-A: the predicate column is NOT in the projection. The merge read + /// must widen internally, filter, then project back — output schema must + /// contain only the projected column. + #[tokio::test] + async fn kv_read_filters_on_unprojected_column() { + let file_io = test_file_io(); + let table_path = "memory:/kv_residual_gap_a"; + setup_dirs(&file_io, table_path).await; + let table = pk_table(&file_io, table_path, &[]); + + write_commit( + &table, + &int_batch(vec![1, 2, 3], vec![Some(10), Some(20), Some(30)]), + ) + .await; + write_commit( + &table, + &int_batch(vec![1, 2, 3], vec![Some(11), Some(21), Some(31)]), + ) + .await; + + let fields = table.schema().fields().to_vec(); + let filter = PredicateBuilder::new(&fields) + .equal("value", Datum::Int(21)) + .unwrap(); + let batches = read_rows(&table, Some(&["id"]), Some(filter)).await; + + assert_eq!(int_column(&batches, "id"), vec![2]); + for batch in &batches { + assert_eq!( + batch.num_columns(), + 1, + "widened predicate column must not leak into the output" + ); + assert_eq!(batch.schema().field(0).name(), "id"); + } + } + + /// Regression: PK-column filters were already exact (pushed down pre-merge + /// AND now re-checked in the residual). Must stay exact. + #[tokio::test] + async fn kv_read_pk_filter_still_exact() { + let file_io = test_file_io(); + let table_path = "memory:/kv_residual_pk"; + setup_dirs(&file_io, table_path).await; + let table = pk_table(&file_io, table_path, &[]); + + write_commit( + &table, + &int_batch(vec![1, 2, 3], vec![Some(10), Some(20), Some(30)]), + ) + .await; + write_commit( + &table, + &int_batch(vec![1, 2, 3], vec![Some(11), Some(21), Some(31)]), + ) + .await; + + let fields = table.schema().fields().to_vec(); + let filter = PredicateBuilder::new(&fields) + .equal("id", Datum::Int(2)) + .unwrap(); + let batches = read_rows(&table, None, Some(filter)).await; + + assert_eq!(int_column(&batches, "id"), vec![2]); + assert_eq!(int_column(&batches, "value"), vec![21]); + } +} From e0324cd8fc5b78f0226f311ed2993ff52f458ece Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Mon, 6 Jul 2026 17:40:33 +0800 Subject: [PATCH 2/8] test(core): lock post-merge residual semantics across KV merge engines --- crates/paimon/src/table/kv_file_reader.rs | 304 ++++++++++++++++++++++ 1 file changed, 304 insertions(+) diff --git a/crates/paimon/src/table/kv_file_reader.rs b/crates/paimon/src/table/kv_file_reader.rs index dc5eeaf8..9bc8e8dc 100644 --- a/crates/paimon/src/table/kv_file_reader.rs +++ b/crates/paimon/src/table/kv_file_reader.rs @@ -644,4 +644,308 @@ mod tests { assert_eq!(int_column(&batches, "id"), vec![2]); assert_eq!(int_column(&batches, "value"), vec![21]); } + + /// A filter matching only a superseded version must return nothing: the + /// newer version wins the merge first, THEN the filter runs. If the full + /// predicate leaked below the merge, the stale (2, 20) row would survive + /// its file's scan, win against nothing, and leak into the output. + #[tokio::test] + async fn kv_read_filter_on_superseded_value_returns_nothing() { + let file_io = test_file_io(); + let table_path = "memory:/kv_residual_superseded"; + setup_dirs(&file_io, table_path).await; + let table = pk_table(&file_io, table_path, &[]); + + write_commit( + &table, + &int_batch(vec![1, 2, 3], vec![Some(10), Some(20), Some(30)]), + ) + .await; + write_commit( + &table, + &int_batch(vec![1, 2, 3], vec![Some(11), Some(21), Some(31)]), + ) + .await; + + let fields = table.schema().fields().to_vec(); + let filter = PredicateBuilder::new(&fields) + .equal("value", Datum::Int(20)) + .unwrap(); + let batches = read_rows(&table, None, Some(filter)).await; + + let total: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!( + total, 0, + "superseded value must not resurrect through the filter" + ); + } + + /// Compound residual `value > 15 AND value < 25` on merged values. + #[tokio::test] + async fn kv_read_applies_compound_range_filter() { + let file_io = test_file_io(); + let table_path = "memory:/kv_residual_range"; + setup_dirs(&file_io, table_path).await; + let table = pk_table(&file_io, table_path, &[]); + + write_commit( + &table, + &int_batch(vec![1, 2, 3], vec![Some(10), Some(20), Some(30)]), + ) + .await; + write_commit( + &table, + &int_batch(vec![1, 2, 3], vec![Some(11), Some(21), Some(31)]), + ) + .await; + + let fields = table.schema().fields().to_vec(); + let pb = PredicateBuilder::new(&fields); + let filter = Predicate::and(vec![ + pb.greater_than("value", Datum::Int(15)).unwrap(), + pb.less_than("value", Datum::Int(25)).unwrap(), + ]); + let batches = read_rows(&table, None, Some(filter)).await; + + assert_eq!(int_column(&batches, "id"), vec![2]); + assert_eq!(int_column(&batches, "value"), vec![21]); + } + + /// COUNT(*)-style read: empty projection + non-PK filter. The residual + /// runs on the pre-reorder merge batch (which still has columns), and the + /// zero-column output batch must carry the filtered row count. + #[tokio::test] + async fn kv_read_empty_projection_with_filter_keeps_row_count() { + let file_io = test_file_io(); + let table_path = "memory:/kv_residual_count"; + setup_dirs(&file_io, table_path).await; + let table = pk_table(&file_io, table_path, &[]); + + write_commit( + &table, + &int_batch(vec![1, 2, 3], vec![Some(10), Some(20), Some(30)]), + ) + .await; + write_commit( + &table, + &int_batch(vec![1, 2, 3], vec![Some(11), Some(21), Some(31)]), + ) + .await; + + let fields = table.schema().fields().to_vec(); + let filter = PredicateBuilder::new(&fields) + .greater_than("value", Datum::Int(15)) + .unwrap(); + let batches = read_rows(&table, Some(&[] as &[&str]), Some(filter)).await; + + let total: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!(total, 2, "only merged rows with value > 15 (21, 31) count"); + for batch in &batches { + assert_eq!(batch.num_columns(), 0); + } + } + + /// String residual op (starts_with) on a value column — exercises the + /// residual string kernel on the KV path. + #[tokio::test] + async fn kv_read_applies_string_starts_with_filter() { + use crate::spec::VarCharType; + use arrow_array::StringArray; + + let file_io = test_file_io(); + let table_path = "memory:/kv_residual_string"; + setup_dirs(&file_io, table_path).await; + + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("name", DataType::VarChar(VarCharType::string_type())) + .primary_key(["id"]) + .option("bucket", "1") + .build() + .unwrap(); + let table = Table::new( + file_io.clone(), + Identifier::new("default", "kv_residual_string_t"), + table_path.to_string(), + TableSchema::new(0, &schema), + None, + ); + + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int32, false), + ArrowField::new("name", ArrowDataType::Utf8, true), + ])); + let make = |ids: Vec, names: Vec<&str>| { + RecordBatch::try_new( + arrow_schema.clone(), + vec![ + Arc::new(Int32Array::from(ids)), + Arc::new(StringArray::from(names)), + ], + ) + .unwrap() + }; + + write_commit( + &table, + &make(vec![1, 2, 3], vec!["apple", "banana", "apricot"]), + ) + .await; + write_commit(&table, &make(vec![2], vec!["avocado"])).await; + + let fields = table.schema().fields().to_vec(); + let filter = PredicateBuilder::new(&fields) + .starts_with("name", Datum::String("a".to_string())) + .unwrap(); + let batches = read_rows(&table, None, Some(filter)).await; + + // Merged rows: (1, apple), (2, avocado), (3, apricot) — all start with 'a'. + // The overwritten (2, banana) must not resurrect; if the filter ran + // pre-merge it would also be wrong the other way (banana dropped, but + // then avocado wins anyway — so also assert the merged VALUE). + let mut ids = int_column(&batches, "id"); + ids.sort_unstable(); + assert_eq!(ids, vec![1, 2, 3]); + let names: Vec = batches + .iter() + .flat_map(|b| { + let idx = b.schema().index_of("name").unwrap(); + let arr = b + .column(idx) + .as_any() + .downcast_ref::() + .unwrap(); + (0..arr.len()) + .map(|i| arr.value(i).to_string()) + .collect::>() + }) + .collect(); + assert!(names.contains(&"avocado".to_string())); + assert!(!names.contains(&"banana".to_string())); + } + + /// Aggregation (sum): inputs 10 + 20 merge to 30. `value = 30` must match + /// the merged row (a pre-merge filter would drop both inputs); + /// `value = 10` must match nothing (a pre-merge filter would keep the + /// 10-input and leak it). + #[tokio::test] + async fn kv_read_aggregation_filters_on_merged_value() { + let file_io = test_file_io(); + let table_path = "memory:/kv_residual_agg"; + setup_dirs(&file_io, table_path).await; + let table = pk_table( + &file_io, + table_path, + &[ + ("merge-engine", "aggregation"), + ("fields.value.aggregate-function", "sum"), + ], + ); + + write_commit(&table, &int_batch(vec![1], vec![Some(10)])).await; + write_commit(&table, &int_batch(vec![1], vec![Some(20)])).await; + + let fields = table.schema().fields().to_vec(); + + let match_merged = PredicateBuilder::new(&fields) + .equal("value", Datum::Int(30)) + .unwrap(); + let batches = read_rows(&table, None, Some(match_merged)).await; + assert_eq!(int_column(&batches, "id"), vec![1]); + assert_eq!(int_column(&batches, "value"), vec![30]); + + let match_input = PredicateBuilder::new(&fields) + .equal("value", Datum::Int(10)) + .unwrap(); + let batches = read_rows(&table, None, Some(match_input)).await; + let total: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!(total, 0, "pre-merge input value must not leak through"); + } + + /// Aggregation + Gap-A: the aggregated predicate column is unprojected. + /// The widened column must be aggregated with its configured function + /// (sum), not treated as a plain latest-value column. + #[tokio::test] + async fn kv_read_aggregation_filters_merged_value_unprojected() { + let file_io = test_file_io(); + let table_path = "memory:/kv_residual_agg_gap_a"; + setup_dirs(&file_io, table_path).await; + let table = pk_table( + &file_io, + table_path, + &[ + ("merge-engine", "aggregation"), + ("fields.value.aggregate-function", "sum"), + ], + ); + + write_commit(&table, &int_batch(vec![1], vec![Some(10)])).await; + write_commit(&table, &int_batch(vec![1], vec![Some(20)])).await; + + let fields = table.schema().fields().to_vec(); + let filter = PredicateBuilder::new(&fields) + .equal("value", Datum::Int(30)) + .unwrap(); + let batches = read_rows(&table, Some(&["id"]), Some(filter)).await; + assert_eq!(int_column(&batches, "id"), vec![1]); + } + + /// Partial-update: (1, a=5, b=NULL) then (1, a=NULL, b=7) merge to + /// (1, 5, 7). A conjunction over both columns only matches the MERGED row + /// — no single input row satisfies it. + #[tokio::test] + async fn kv_read_partial_update_filters_on_merged_row() { + let file_io = test_file_io(); + let table_path = "memory:/kv_residual_pu"; + setup_dirs(&file_io, table_path).await; + + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("a", DataType::Int(IntType::new())) + .column("b", DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("bucket", "1") + .option("merge-engine", "partial-update") + .build() + .unwrap(); + let table = Table::new( + file_io.clone(), + Identifier::new("default", "kv_residual_pu_t"), + table_path.to_string(), + TableSchema::new(0, &schema), + None, + ); + + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int32, false), + ArrowField::new("a", ArrowDataType::Int32, true), + ArrowField::new("b", ArrowDataType::Int32, true), + ])); + let make = |ids: Vec, a: Vec>, b: Vec>| { + RecordBatch::try_new( + arrow_schema.clone(), + vec![ + Arc::new(Int32Array::from(ids)), + Arc::new(Int32Array::from(a)), + Arc::new(Int32Array::from(b)), + ], + ) + .unwrap() + }; + + write_commit(&table, &make(vec![1], vec![Some(5)], vec![None])).await; + write_commit(&table, &make(vec![1], vec![None], vec![Some(7)])).await; + + let fields = table.schema().fields().to_vec(); + let pb = PredicateBuilder::new(&fields); + let filter = Predicate::and(vec![ + pb.equal("a", Datum::Int(5)).unwrap(), + pb.equal("b", Datum::Int(7)).unwrap(), + ]); + let batches = read_rows(&table, None, Some(filter)).await; + + assert_eq!(int_column(&batches, "id"), vec![1]); + assert_eq!(int_column(&batches, "a"), vec![5]); + assert_eq!(int_column(&batches, "b"), vec![7]); + } } From 27786b66cfec03d8d7c70d9052f6bce9af4cfc7e Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Mon, 6 Jul 2026 18:00:27 +0800 Subject: [PATCH 3/8] fix(core): restrict PK-table file stats pruning to key conjuncts --- crates/paimon/src/table/kv_file_reader.rs | 37 ++++++ crates/paimon/src/table/table_scan.rs | 136 +++++++++++++++++++++- 2 files changed, 168 insertions(+), 5 deletions(-) diff --git a/crates/paimon/src/table/kv_file_reader.rs b/crates/paimon/src/table/kv_file_reader.rs index 9bc8e8dc..6fc144af 100644 --- a/crates/paimon/src/table/kv_file_reader.rs +++ b/crates/paimon/src/table/kv_file_reader.rs @@ -544,6 +544,43 @@ mod tests { .collect() } + #[test] + fn retain_primary_key_conjuncts_semantics() { + let fields = vec![ + DataField::new(0, "id".to_string(), PaimonDataType::Int(IntType::new())), + DataField::new(1, "value".to_string(), PaimonDataType::Int(IntType::new())), + ]; + let pks = vec!["id".to_string()]; + let pb = PredicateBuilder::new(&fields); + + // Plain PK leaf: kept. Plain non-PK leaf: dropped. + let kept = + retain_primary_key_conjuncts(&[pb.equal("id", Datum::Int(1)).unwrap()], &fields, &pks); + assert_eq!(kept.len(), 1); + let dropped = retain_primary_key_conjuncts( + &[pb.equal("value", Datum::Int(1)).unwrap()], + &fields, + &pks, + ); + assert!(dropped.is_empty()); + + // Mixed AND keeps the PK child only. + let mixed = Predicate::and(vec![ + pb.equal("id", Datum::Int(1)).unwrap(), + pb.equal("value", Datum::Int(2)).unwrap(), + ]); + let kept = retain_primary_key_conjuncts(&[mixed], &fields, &pks); + assert_eq!(kept.len(), 1); + assert!(matches!(&kept[0], Predicate::Leaf { index: 0, .. })); + + // OR with a non-PK child: dropped entirely (cannot be tightened). + let or = Predicate::or(vec![ + pb.equal("id", Datum::Int(1)).unwrap(), + pb.equal("value", Datum::Int(2)).unwrap(), + ]); + assert!(retain_primary_key_conjuncts(&[or], &fields, &pks).is_empty()); + } + /// Non-PK equality filter on a dedup PK table read through the sort-merge /// path must return only matching rows. Before the post-merge residual, /// the non-PK conjunct was silently dropped and all rows came back. diff --git a/crates/paimon/src/table/table_scan.rs b/crates/paimon/src/table/table_scan.rs index cab47814..c7fabe6a 100644 --- a/crates/paimon/src/table/table_scan.rs +++ b/crates/paimon/src/table/table_scan.rs @@ -21,6 +21,7 @@ //! and [FullStartingScanner](https://github.com/apache/paimon/blob/release-1.3/paimon-python/pypaimon/read/scanner/full_starting_scanner.py). use super::bucket_filter::compute_target_buckets; +use super::kv_file_reader::retain_primary_key_conjuncts; use super::partition_filter::PartitionFilter; use super::stats_filter::{ data_evolution_group_matches_predicates, data_file_matches_predicates, @@ -808,9 +809,9 @@ impl<'a> TableScan<'a> { let partition_fields = self.table.schema().partition_fields(); let pushdown_data_predicates = if data_evolution_enabled { - &[][..] + Vec::new() } else { - self.data_predicates.as_slice() + self.stats_pruning_predicates() }; let bucket_key_fields: Vec = if self.bucket_predicate.is_none() { @@ -846,7 +847,7 @@ impl<'a> TableScan<'a> { has_primary_keys, self.partition_filter.as_ref(), &partition_fields, - pushdown_data_predicates, + &pushdown_data_predicates, self.table.schema().id(), self.table.schema().fields(), self.bucket_predicate.as_ref(), @@ -866,6 +867,30 @@ impl<'a> TableScan<'a> { can_push_down_limit_hint_for_scan(&self.data_predicates, row_ranges) } + /// The predicate set that may prune WHOLE FILES by their stats. + /// + /// For primary-key tables read by merging (everything except DV tables, + /// which read raw with per-row deletion masks), only key conjuncts are + /// safe: a key's versions agree on the key columns but not on value + /// columns, so a value conjunct could prune the file holding the newest + /// version and resurrect an older one from a surviving file. The dropped + /// conjuncts are still enforced exactly by the post-merge residual filter + /// in `KeyValueFileReader`. + fn stats_pruning_predicates(&self) -> Vec { + let has_primary_keys = !self.table.schema().primary_keys().is_empty(); + let deletion_vectors_enabled = + CoreOptions::new(self.table.schema().options()).deletion_vectors_enabled(); + if has_primary_keys && !deletion_vectors_enabled { + retain_primary_key_conjuncts( + &self.data_predicates, + self.table.schema().fields(), + &self.table.schema().trimmed_primary_keys(), + ) + } else { + self.data_predicates.clone() + } + } + async fn plan_snapshot( &self, snapshot: Snapshot, @@ -896,7 +921,8 @@ impl<'a> TableScan<'a> { // For non-data-evolution tables, cross-schema files were kept (fail-open) // by the pushdown. Apply the full schema-aware filter for those files. - let entries = if self.data_predicates.is_empty() || data_evolution_enabled { + let stats_pruning_predicates = self.stats_pruning_predicates(); + let entries = if stats_pruning_predicates.is_empty() || data_evolution_enabled { entries } else { let current_schema_id = self.table.schema().id(); @@ -918,7 +944,7 @@ impl<'a> TableScan<'a> { || data_file_matches_predicates_for_table( self.table, entry.file(), - &self.data_predicates, + &stats_pruning_predicates, &mut schema_cache, ) .await @@ -2019,6 +2045,106 @@ mod tests { ); } + fn pk_stats_gate_table(table_path: &str) -> Table { + let file_io = FileIOBuilder::new("memory").build().unwrap(); + let schema = PaimonSchema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("bucket", "1") + .build() + .unwrap(); + Table::new( + file_io, + Identifier::new("test_db", "pk_stats_gate"), + table_path.to_string(), + TableSchema::new(0, &schema), + None, + ) + } + + fn two_int_stats_row(id: Option, value: Option) -> Vec { + let mut builder = BinaryRowBuilder::new(2); + match id { + Some(id) => builder.write_int(0, id), + None => builder.set_null_at(0), + } + match value { + Some(value) => builder.write_int(1, value), + None => builder.set_null_at(1), + } + builder.build_serialized() + } + + fn pk_stats_file(name: &str, id_range: (i32, i32), value_range: (i32, i32)) -> DataFileMeta { + let mut file = test_data_file_meta( + two_int_stats_row(Some(id_range.0), Some(value_range.0)), + two_int_stats_row(Some(id_range.1), Some(value_range.1)), + vec![Some(0), Some(0)], + 2, + ); + file.file_name = name.to_string(); + file + } + + /// Merge reads combine versions of a key across files, so scan planning + /// must not prune a PK table's files by NON-key conjuncts: dropping the + /// file that holds the newest version resurrects an older version from a + /// surviving file — an error no post-merge residual can repair. Key + /// conjuncts stay safe (every version of a key shares the key columns) + /// and must still prune. + #[tokio::test] + async fn test_pk_table_stats_pruning_ignores_non_key_conjuncts() { + let table_path = "memory:/test_pk_stats_gate"; + let table = pk_stats_gate_table(table_path); + setup_scan_trace_dirs(&table).await; + + // Both files cover key id=1; the newer version's value (50) falls + // outside the value predicate while the older one (150) matches. + TableCommit::new(table.clone(), "pk-gate-test".to_string()) + .commit(vec![CommitMessage::new( + BinaryRowBuilder::new(0).build_serialized(), + 0, + vec![ + pk_stats_file("old-version.parquet", (1, 5), (100, 200)), + pk_stats_file("new-version.parquet", (1, 5), (10, 60)), + ], + )]) + .await + .unwrap(); + + let fields = vec![ + DataField::new(0, "id".to_string(), DataType::Int(IntType::new())), + DataField::new(1, "value".to_string(), DataType::Int(IntType::new())), + ]; + let pb = PredicateBuilder::new(&fields); + + // Non-key conjunct: must NOT prune any file of a PK table. + let value_filter = pb.greater_than("value", Datum::Int(90)).unwrap(); + let mut reader = table.new_read_builder(); + reader.with_filter(value_filter); + let (plan, trace) = reader.new_scan().plan_with_trace().await.unwrap(); + assert_eq!( + trace.manifest_entries_pruned_by_data_stats, 0, + "non-key conjuncts must not file-prune a PK table: {trace:?}" + ); + let planned_files: usize = plan.splits().iter().map(|s| s.data_files().len()).sum(); + assert_eq!( + planned_files, 2, + "both versions must reach the merge reader" + ); + + // Key conjunct: still prunes (id=9 outside both files' key range). + let key_filter = pb.equal("id", Datum::Int(9)).unwrap(); + let mut reader = table.new_read_builder(); + reader.with_filter(key_filter); + let (_plan, trace) = reader.new_scan().plan_with_trace().await.unwrap(); + assert!( + trace.manifest_entries_pruned_by_data_stats >= 2, + "key conjuncts must still prune PK-table files: {trace:?}" + ); + } + #[test] fn test_data_file_matches_in_prunes_when_all_literals_out_of_range() { let fields = int_field(); From e4c5dd0c778fe2c4d2a4e7bdacc945be536c2734 Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Mon, 6 Jul 2026 18:14:35 +0800 Subject: [PATCH 4/8] docs(core): reflect exact PK-merge residual filtering in read API docs --- crates/paimon/src/table/read_builder.rs | 9 +++++---- crates/paimon/src/table/table_read.rs | 6 +++--- 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/crates/paimon/src/table/read_builder.rs b/crates/paimon/src/table/read_builder.rs index 1d54b8d8..c3c2789a 100644 --- a/crates/paimon/src/table/read_builder.rs +++ b/crates/paimon/src/table/read_builder.rs @@ -155,11 +155,12 @@ impl<'a> ReadBuilder<'a> { /// /// [`TableRead`] may use supported non-partition data predicates on formats /// with reader pruning for conservative row-group pruning. Parquet may also - /// use native row filtering. Row-level exactness is enforced on append and - /// data-evolution read paths: format readers apply an exact residual filter + /// use native row filtering. Row-level exactness is enforced on all read + /// paths: format readers apply an exact residual filter on append reads /// (see `FormatFileReader::read_batch_stream` for per-format exceptions), - /// and data-evolution reads filter batches exactly before yielding. - /// Primary-key merge reads currently apply only primary-key conjuncts. + /// data-evolution reads filter batches exactly before yielding, and + /// primary-key merge reads push key conjuncts below the merge and enforce + /// the full predicate with an exact post-merge residual filter. pub fn with_filter(&mut self, filter: Predicate) -> &mut Self { self.filter = normalize_filter(self.table, filter); self.try_extract_row_id_ranges(); diff --git a/crates/paimon/src/table/table_read.rs b/crates/paimon/src/table/table_read.rs index c2f96279..216fbd67 100644 --- a/crates/paimon/src/table/table_read.rs +++ b/crates/paimon/src/table/table_read.rs @@ -63,10 +63,10 @@ impl<'a> TableRead<'a> { } /// Set a filter predicate. Used conservatively for read-side pruning and - /// enforced exactly by the residual filter on append and data-evolution - /// read paths (see [`ReadBuilder::with_filter`](crate::table::ReadBuilder::with_filter) + /// enforced exactly by residual filtering on append, data-evolution, and + /// primary-key merge read paths (see + /// [`ReadBuilder::with_filter`](crate::table::ReadBuilder::with_filter) /// for per-format exceptions). - /// Primary-key merge reads currently apply only primary-key conjuncts. pub fn with_filter(mut self, filter: Predicate) -> Self { let (_, data_predicates) = split_scan_predicates(self.table, filter); // Keep the FULL data predicate (including `And`/`Or`/`Not`). Native From 0f9ac3694cd6e1574af2f6d746650237f07b9287 Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Mon, 6 Jul 2026 21:31:41 +0800 Subject: [PATCH 5/8] test(core): lock AlwaysFalse and schema-evolution semantics on the KV residual path --- crates/paimon/src/table/kv_file_reader.rs | 184 ++++++++++++++++++++++ 1 file changed, 184 insertions(+) diff --git a/crates/paimon/src/table/kv_file_reader.rs b/crates/paimon/src/table/kv_file_reader.rs index 6fc144af..a487b8b0 100644 --- a/crates/paimon/src/table/kv_file_reader.rs +++ b/crates/paimon/src/table/kv_file_reader.rs @@ -504,6 +504,59 @@ mod tests { .unwrap() } + fn evo_batch(ids: Vec, values: Vec>, scores: Vec>) -> RecordBatch { + let schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int32, false), + ArrowField::new("value", ArrowDataType::Int32, true), + ArrowField::new("score", ArrowDataType::Int32, true), + ])); + RecordBatch::try_new( + schema, + vec![ + Arc::new(Int32Array::from(ids)), + Arc::new(Int32Array::from(values)), + Arc::new(Int32Array::from(scores)), + ], + ) + .unwrap() + } + + /// User schema for the evolution fixture: `id INT pk, value INT` at + /// version 0, plus `score INT` (new field id 2) at version 1. Field ids + /// line up across versions exactly as a real ADD COLUMN produces. + fn evo_user_schema(with_score: bool) -> Schema { + let mut builder = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())); + if with_score { + builder = builder.column("score", DataType::Int(IntType::new())); + } + builder + .primary_key(["id"]) + .option("bucket", "1") + .build() + .unwrap() + } + + /// Persist a schema version as `{table_path}/schema/schema-{id}` JSON so + /// `SchemaManager::schema` can resolve old-file schemas at read time. The + /// write path only stamps `DataFileMeta.schema_id`; schema files are + /// normally written by the catalog, which these fixtures bypass. Follows + /// the `write_schema_file` pattern from the table_scan tests. + async fn write_schema_file(table: &Table, schema: &TableSchema) { + let path = table.schema_manager().schema_path(schema.id()); + let dir = path.rsplit_once('/').map(|(dir, _)| dir).unwrap(); + table.file_io().mkdirs(dir).await.unwrap(); + let json = serde_json::to_vec(schema).unwrap(); + table + .file_io() + .new_output(&path) + .unwrap() + .write(bytes::Bytes::from(json)) + .await + .unwrap(); + } + async fn write_commit(table: &Table, batch: &RecordBatch) { let mut tw = TableWrite::new(table, "test-user".to_string()).unwrap(); tw.write_arrow_batch(batch).await.unwrap(); @@ -579,6 +632,17 @@ mod tests { pb.equal("value", Datum::Int(2)).unwrap(), ]); assert!(retain_primary_key_conjuncts(&[or], &fields, &pks).is_empty()); + + // Constant predicates reference no columns and must survive the PK + // trim verbatim. Dropping AlwaysFalse here would be catastrophic: + // scan-side stats pruning treats it as prune-everything, so losing it + // would flip "return no rows" into "return every row". + let kept = retain_primary_key_conjuncts(&[Predicate::AlwaysFalse], &fields, &pks); + assert_eq!(kept.len(), 1); + assert!(matches!(&kept[0], Predicate::AlwaysFalse)); + let kept = retain_primary_key_conjuncts(&[Predicate::AlwaysTrue], &fields, &pks); + assert_eq!(kept.len(), 1); + assert!(matches!(&kept[0], Predicate::AlwaysTrue)); } /// Non-PK equality filter on a dedup PK table read through the sort-merge @@ -985,4 +1049,124 @@ mod tests { assert_eq!(int_column(&batches, "a"), vec![5]); assert_eq!(int_column(&batches, "b"), vec![7]); } + + /// An AlwaysFalse filter on a PK table must return nothing. AlwaysFalse + /// references no columns, so a PK-conjunct trim that dropped it (instead + /// of preserving it verbatim) would remove the only thing stopping the + /// read — stats pruning treats AlwaysFalse as prune-everything, and the + /// residual masks every row to false — and every row would come back. + #[tokio::test] + async fn kv_read_always_false_filter_returns_nothing() { + let file_io = test_file_io(); + let table_path = "memory:/kv_residual_always_false"; + setup_dirs(&file_io, table_path).await; + let table = pk_table(&file_io, table_path, &[]); + + // Overlapping keys across two commits -> split is not raw convertible + // -> forced through KeyValueFileReader. + write_commit( + &table, + &int_batch(vec![1, 2, 3], vec![Some(10), Some(20), Some(30)]), + ) + .await; + write_commit( + &table, + &int_batch(vec![1, 2, 3], vec![Some(11), Some(21), Some(31)]), + ) + .await; + + let batches = read_rows(&table, None, Some(Predicate::AlwaysFalse)).await; + + let total: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!(total, 0, "AlwaysFalse must return no rows on a PK table"); + } + + /// Schema evolution on the KV residual path: a predicate column that is + /// MISSING from an old-schema file is null-filled pre-merge by + /// DataFileReader; the post-merge residual must treat those NULLs as + /// non-matching (comparison mask NULL -> false), and `is_null` must match + /// exactly them. Locks the null-fill -> merge -> residual composition; the + /// shared evaluator's semantics are already locked on the data-evolution + /// path (`test_evolution_read_null_filled_predicate_column_semantics`). + /// + /// Setup: commit 1 goes through a Table at schema 0 (id, value), stamping + /// schema_id 0 into its files; commit 2 goes through a Table at the same + /// path at schema 1 (id, value, score — new field id). Both schema JSONs + /// are persisted so `SchemaManager::schema(0)` resolves at read time. Keys + /// overlap across commits so the split is not raw convertible and routes + /// through the KV merge reader. + #[tokio::test] + async fn kv_read_schema_evolution_null_filled_predicate_semantics() { + let file_io = test_file_io(); + let table_path = "memory:/kv_residual_schema_evolution"; + setup_dirs(&file_io, table_path).await; + + let schema0 = TableSchema::new(0, &evo_user_schema(false)); + let schema1 = TableSchema::new(1, &evo_user_schema(true)); + let table_v0 = Table::new( + file_io.clone(), + Identifier::new("default", "kv_residual_evo_t"), + table_path.to_string(), + schema0.clone(), + None, + ); + let table_v1 = Table::new( + file_io.clone(), + Identifier::new("default", "kv_residual_evo_t"), + table_path.to_string(), + schema1.clone(), + None, + ); + write_schema_file(&table_v1, &schema0).await; + write_schema_file(&table_v1, &schema1).await; + + // Commit 1 at schema 0: files carry schema_id 0. Commit 2 at schema 1 + // overwrites key 3, so file_meta.schema_id != table_schema_id holds for + // the old files when reading through table_v1, forcing the null-fill + // remap in read() -> DataFileReader::read_single_file_stream. + write_commit( + &table_v0, + &int_batch(vec![1, 2, 3], vec![Some(10), Some(20), Some(30)]), + ) + .await; + write_commit( + &table_v1, + &evo_batch(vec![3], vec![Some(31)], vec![Some(300)]), + ) + .await; + + // Merged rows: (1, 10, NULL), (2, 20, NULL), (3, 31, 300). + let fields = table_v1.schema().fields().to_vec(); + let pb = PredicateBuilder::new(&fields); + + // Comparison: score = 300 matches only id 3; old rows' null-filled + // score must collapse to false, not match or error. + let filter = pb.equal("score", Datum::Int(300)).unwrap(); + let batches = read_rows(&table_v1, None, Some(filter)).await; + assert_eq!(int_column(&batches, "id"), vec![3]); + assert_eq!(int_column(&batches, "value"), vec![31]); + assert_eq!(int_column(&batches, "score"), vec![300]); + + // IS NULL: matches exactly the null-filled old rows (ids 1, 2). + let filter = pb.is_null("score").unwrap(); + let batches = read_rows(&table_v1, None, Some(filter)).await; + let mut ids = int_column(&batches, "id"); + ids.sort_unstable(); + assert_eq!(ids, vec![1, 2]); + + // Gap-A on an evolution column: score is filtered but not projected. + // The merge read must widen internally (null-filling score for the old + // files), filter, then project back to just "id". + let filter = pb.equal("score", Datum::Int(300)).unwrap(); + let batches = read_rows(&table_v1, Some(&["id"]), Some(filter)).await; + assert_eq!(int_column(&batches, "id"), vec![3]); + for batch in &batches { + assert_eq!( + batch.num_columns(), + 1, + "widened evolution column must not leak into the output" + ); + assert_eq!(batch.schema().field(0).name(), "id"); + } + } } From dd9bff5f6a7fb247332c4d0842456417bcd6d574 Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Mon, 6 Jul 2026 22:13:41 +0800 Subject: [PATCH 6/8] test(core): correct AlwaysFalse doc comments to reflect layered enforcement --- crates/paimon/src/table/kv_file_reader.rs | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/crates/paimon/src/table/kv_file_reader.rs b/crates/paimon/src/table/kv_file_reader.rs index a487b8b0..38324850 100644 --- a/crates/paimon/src/table/kv_file_reader.rs +++ b/crates/paimon/src/table/kv_file_reader.rs @@ -634,9 +634,10 @@ mod tests { assert!(retain_primary_key_conjuncts(&[or], &fields, &pks).is_empty()); // Constant predicates reference no columns and must survive the PK - // trim verbatim. Dropping AlwaysFalse here would be catastrophic: - // scan-side stats pruning treats it as prune-everything, so losing it - // would flip "return no rows" into "return every row". + // trim verbatim. The post-merge residual (full predicate set) would + // still mask every row to false if AlwaysFalse were dropped here, but + // the scan/pushdown layers would lose their prune-everything fast + // path (stats_filter treats any AlwaysFalse as prune-all). let kept = retain_primary_key_conjuncts(&[Predicate::AlwaysFalse], &fields, &pks); assert_eq!(kept.len(), 1); assert!(matches!(&kept[0], Predicate::AlwaysFalse)); @@ -1050,11 +1051,11 @@ mod tests { assert_eq!(int_column(&batches, "b"), vec![7]); } - /// An AlwaysFalse filter on a PK table must return nothing. AlwaysFalse - /// references no columns, so a PK-conjunct trim that dropped it (instead - /// of preserving it verbatim) would remove the only thing stopping the - /// read — stats pruning treats AlwaysFalse as prune-everything, and the - /// residual masks every row to false — and every row would come back. + /// An AlwaysFalse filter on a PK table must return nothing, end to end. + /// Two layers enforce it: scan-side stats pruning treats AlwaysFalse as + /// prune-everything (plans no files), and the post-merge residual masks + /// every row to false. This locks the composed contract regardless of + /// which layer short-circuits first. #[tokio::test] async fn kv_read_always_false_filter_returns_nothing() { let file_io = test_file_io(); From b0064211b7b7c5c93023aa9d1a88cac7e8513d6c Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Mon, 6 Jul 2026 22:58:14 +0800 Subject: [PATCH 7/8] test(core): adapt to fallible with_projection from main --- crates/paimon/src/table/kv_file_reader.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/paimon/src/table/kv_file_reader.rs b/crates/paimon/src/table/kv_file_reader.rs index 38324850..2a6224b8 100644 --- a/crates/paimon/src/table/kv_file_reader.rs +++ b/crates/paimon/src/table/kv_file_reader.rs @@ -574,7 +574,7 @@ mod tests { ) -> Vec { let mut rb = table.new_read_builder(); if let Some(cols) = projection { - rb.with_projection(cols); + rb.with_projection(cols).unwrap(); } if let Some(f) = filter { rb.with_filter(f); From 81fadc7eb7ec85fb993d97d47748dcf40e2b4100 Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Tue, 7 Jul 2026 15:06:27 +0800 Subject: [PATCH 8/8] fix(core): exempt first-row tables from key-only stats pruning First-row PK tables read raw (skip_level_zero planning, DataFileReader), so file-level pruning by non-key conjuncts cannot resurrect stale versions; keep full-predicate stats pruning for them, matching the split-generation gate. --- crates/paimon/src/table/table_scan.rs | 99 ++++++++++++++++++++++++--- 1 file changed, 89 insertions(+), 10 deletions(-) diff --git a/crates/paimon/src/table/table_scan.rs b/crates/paimon/src/table/table_scan.rs index c7fabe6a..266c6cb1 100644 --- a/crates/paimon/src/table/table_scan.rs +++ b/crates/paimon/src/table/table_scan.rs @@ -869,18 +869,32 @@ impl<'a> TableScan<'a> { /// The predicate set that may prune WHOLE FILES by their stats. /// - /// For primary-key tables read by merging (everything except DV tables, - /// which read raw with per-row deletion masks), only key conjuncts are - /// safe: a key's versions agree on the key columns but not on value - /// columns, so a value conjunct could prune the file holding the newest - /// version and resurrect an older one from a surviving file. The dropped - /// conjuncts are still enforced exactly by the post-merge residual filter - /// in `KeyValueFileReader`. + /// For primary-key tables read by merging, only key conjuncts are safe: a + /// key's versions agree on the key columns but not on value columns, so a + /// value conjunct could prune the file holding the newest version and + /// resurrect an older one from a surviving file. The dropped conjuncts + /// are still enforced exactly by the post-merge residual filter in + /// `KeyValueFileReader`. + /// + /// Exempt (full predicates kept): + /// - Deletion-vector tables: they read raw with per-row masks, stats are + /// a superset of live rows, full pruning stays safe. + /// - `merge-engine=first-row`: planned with `skip_level_zero` and read + /// via `DataFileReader` (see `TableRead::to_arrow`), no merge on the + /// read path — pruning a file drops exactly the rows the raw path's + /// exact residual filter would drop anyway. If first-row ever gains a + /// merge read path, this exemption must be revisited. fn stats_pruning_predicates(&self) -> Vec { let has_primary_keys = !self.table.schema().primary_keys().is_empty(); - let deletion_vectors_enabled = - CoreOptions::new(self.table.schema().options()).deletion_vectors_enabled(); - if has_primary_keys && !deletion_vectors_enabled { + let core_options = CoreOptions::new(self.table.schema().options()); + let deletion_vectors_enabled = core_options.deletion_vectors_enabled(); + // An unknown merge engine stays conservative (key-only pruning); the + // read side fails on it anyway before returning rows. + let first_row = matches!( + core_options.merge_engine(), + Ok(crate::spec::MergeEngine::FirstRow) + ); + if has_primary_keys && !deletion_vectors_enabled && !first_row { retain_primary_key_conjuncts( &self.data_predicates, self.table.schema().fields(), @@ -2145,6 +2159,71 @@ mod tests { ); } + /// `merge-engine=first-row` PK tables read raw (no merge on the read + /// path: planned with `skip_level_zero`, read via `DataFileReader`), so + /// pruning a file by a non-key conjunct cannot resurrect anything — it + /// drops exactly the rows the raw path's exact residual filter would + /// drop. The key-only gate must exempt first-row and keep full-predicate + /// stats pruning, matching the split-generation path. + #[tokio::test] + async fn test_first_row_table_stats_pruning_keeps_non_key_conjuncts() { + let table_path = "memory:/test_first_row_stats_gate"; + let file_io = FileIOBuilder::new("memory").build().unwrap(); + let schema = PaimonSchema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("bucket", "1") + .option("merge-engine", "first-row") + .build() + .unwrap(); + let table = Table::new( + file_io, + Identifier::new("test_db", "first_row_stats_gate"), + table_path.to_string(), + TableSchema::new(0, &schema), + None, + ); + setup_scan_trace_dirs(&table).await; + + // Compacted (level 1) files: first-row planning skips level 0, so the + // fixture files must sit above it to be planned at all. Distinct key + // ranges; only file A's value range can match `value > 90`. + TableCommit::new(table.clone(), "first-row-gate-test".to_string()) + .commit(vec![CommitMessage::new( + BinaryRowBuilder::new(0).build_serialized(), + 0, + vec![ + pk_stats_file("file-a.parquet", (1, 5), (100, 200)), + pk_stats_file("file-b.parquet", (6, 9), (10, 60)), + ], + )]) + .await + .unwrap(); + + let fields = vec![ + DataField::new(0, "id".to_string(), DataType::Int(IntType::new())), + DataField::new(1, "value".to_string(), DataType::Int(IntType::new())), + ]; + let pb = PredicateBuilder::new(&fields); + + // Non-key conjunct: first-row reads raw, so full-predicate pruning + // stays enabled — file-b (value stats [10, 60]) must be pruned. + let value_filter = pb.greater_than("value", Datum::Int(90)).unwrap(); + let mut reader = table.new_read_builder(); + reader.with_filter(value_filter); + let (plan, trace) = reader.new_scan().plan_with_trace().await.unwrap(); + assert!( + trace.manifest_entries_pruned_by_data_stats >= 1, + "first-row tables must keep full-predicate stats pruning: {trace:?}" + ); + let planned_files: usize = plan.splits().iter().map(|s| s.data_files().len()).sum(); + assert_eq!( + planned_files, 1, + "only the value-matching file should be planned on first-row" + ); + } + #[test] fn test_data_file_matches_in_prunes_when_all_literals_out_of_range() { let fields = int_field();