This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 1fe06d2 Apply exact predicate filtering on the primary-key merge read
path (#463)
1fe06d2 is described below
commit 1fe06d2a886f859bdd29852a8ceb7c76bb68955e
Author: Junrui Lee <[email protected]>
AuthorDate: Tue Jul 7 21:07:03 2026 +0800
Apply exact predicate filtering on the primary-key merge read path (#463)
---
crates/paimon/src/table/kv_file_reader.rs | 852 +++++++++++++++++++++++++++++-
crates/paimon/src/table/read_builder.rs | 9 +-
crates/paimon/src/table/table_read.rs | 6 +-
crates/paimon/src/table/table_scan.rs | 215 +++++++-
4 files changed, 1041 insertions(+), 41 deletions(-)
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index 6d3e4e5..2a6224b 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<Predicate>,
}
/// Configuration for [`KeyValueFileReader`], grouping table schema and
@@ -65,37 +72,49 @@ pub(crate) struct KeyValueReadConfig {
pub sequence_fields: Vec<String>,
}
+/// 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<Predicate> {
+ let pk_set: std::collections::HashSet<&str> = primary_keys.iter().map(|s|
s.as_str()).collect();
+ let mapping: Vec<Option<usize>> = 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<Option<usize>> = 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<DataField> = 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,731 @@ 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<i32>, values: Vec<Option<i32>>) -> 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()
+ }
+
+ fn evo_batch(ids: Vec<i32>, values: Vec<Option<i32>>, scores:
Vec<Option<i32>>) -> 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();
+ 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<Predicate>,
+ ) -> Vec<RecordBatch> {
+ let mut rb = table.new_read_builder();
+ if let Some(cols) = projection {
+ rb.with_projection(cols).unwrap();
+ }
+ 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<i32> {
+ batches
+ .iter()
+ .flat_map(|b| {
+ let idx = b.schema().index_of(name).unwrap();
+ let arr =
b.column(idx).as_any().downcast_ref::<Int32Array>().unwrap();
+ (0..arr.len()).map(|i| arr.value(i)).collect::<Vec<_>>()
+ })
+ .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());
+
+ // Constant predicates reference no columns and must survive the PK
+ // 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));
+ 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
+ /// 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]);
+ }
+
+ /// 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<i32>, 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<String> = batches
+ .iter()
+ .flat_map(|b| {
+ let idx = b.schema().index_of("name").unwrap();
+ let arr = b
+ .column(idx)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ (0..arr.len())
+ .map(|i| arr.value(i).to_string())
+ .collect::<Vec<_>>()
+ })
+ .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<i32>, a: Vec<Option<i32>>, b: Vec<Option<i32>>| {
+ 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]);
+ }
+
+ /// 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();
+ 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");
+ }
+ }
+}
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index 1d54b8d..c3c2789 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 c2f9627..216fbd6 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
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index b69097c..115fb01 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,
@@ -856,9 +857,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<DataField> = if
self.bucket_predicate.is_none() {
@@ -894,7 +895,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(),
@@ -914,6 +915,44 @@ 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, 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<Predicate> {
+ let has_primary_keys = !self.table.schema().primary_keys().is_empty();
+ 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(),
+ &self.table.schema().trimmed_primary_keys(),
+ )
+ } else {
+ self.data_predicates.clone()
+ }
+ }
+
async fn plan_snapshot(
&self,
snapshot: Snapshot,
@@ -944,7 +983,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();
@@ -966,7 +1006,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
@@ -2152,6 +2192,171 @@ 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<i32>, value: Option<i32>) -> Vec<u8> {
+ 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:?}"
+ );
+ }
+
+ /// `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"
+ );
+ }
+
#[tokio::test]
async fn
test_plan_with_trace_records_limit_early_stop_during_split_construction() {
let table_path =
