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 93b2013 feat: add inexact NOT filter pushdown (#488)
93b2013 is described below
commit 93b2013f121ffb353c8fc965af798ec8aa95a995
Author: Jiwen liu <[email protected]>
AuthorDate: Thu Jul 9 10:30:29 2026 +0800
feat: add inexact NOT filter pushdown (#488)
---
.../integrations/datafusion/src/filter_pushdown.rs | 207 +++++++++++-------
.../datafusion/src/physical_plan/scan.rs | 5 +
crates/integrations/datafusion/src/table/mod.rs | 28 ++-
.../datafusion/src/variant_pushdown.rs | 6 +-
.../datafusion/tests/sql_context_tests.rs | 49 +++++
crates/paimon/src/predicate_stats.rs | 238 ++++++++++++++++++++-
crates/paimon/src/table/stats_filter.rs | 74 ++++++-
crates/paimon/src/table/table_scan.rs | 49 ++++-
8 files changed, 565 insertions(+), 91 deletions(-)
diff --git a/crates/integrations/datafusion/src/filter_pushdown.rs
b/crates/integrations/datafusion/src/filter_pushdown.rs
index 186c48e..1739a99 100644
--- a/crates/integrations/datafusion/src/filter_pushdown.rs
+++ b/crates/integrations/datafusion/src/filter_pushdown.rs
@@ -25,41 +25,48 @@ use paimon::spec::{DataField, DataType, Datum, Predicate,
PredicateBuilder};
#[derive(Debug)]
struct SingleFilterAnalysis {
translated_predicates: Vec<Predicate>,
- has_untranslated_residual: bool,
+ requires_residual: bool,
}
#[derive(Debug)]
pub(crate) struct FilterPushdownAnalysis {
pub(crate) pushed_predicate: Option<Predicate>,
- pub(crate) has_untranslated_residual: bool,
+ pub(crate) requires_residual: bool,
+}
+
+#[derive(Debug)]
+struct TranslatedPredicate {
+ predicate: Predicate,
+ requires_residual: bool,
}
fn analyze_filter(filter: &Expr, fields: &[DataField]) -> SingleFilterAnalysis
{
let translator = FilterTranslator::new(fields);
- if let Some(predicate) = translator.translate(filter) {
+ if let Some(translated) = translator.translate(filter) {
return SingleFilterAnalysis {
- translated_predicates: vec![predicate],
- has_untranslated_residual: false,
+ translated_predicates: vec![translated.predicate],
+ requires_residual: translated.requires_residual,
};
}
+ let translated = split_conjunction(filter)
+ .into_iter()
+ .filter_map(|expr| translator.translate(expr))
+ .collect::<Vec<_>>();
SingleFilterAnalysis {
- translated_predicates: split_conjunction(filter)
- .into_iter()
- .filter_map(|expr| translator.translate(expr))
- .collect(),
- has_untranslated_residual: true,
+ translated_predicates: translated.iter().map(|t|
t.predicate.clone()).collect(),
+ requires_residual: true,
}
}
pub(crate) fn analyze_filters(filters: &[Expr], fields: &[DataField]) ->
FilterPushdownAnalysis {
let mut translated_predicates = Vec::new();
- let mut has_untranslated_residual = false;
+ let mut requires_residual = false;
for filter in filters {
let analysis = analyze_filter(filter, fields);
translated_predicates.extend(analysis.translated_predicates);
- has_untranslated_residual |= analysis.has_untranslated_residual;
+ requires_residual |= analysis.requires_residual;
}
FilterPushdownAnalysis {
@@ -68,7 +75,7 @@ pub(crate) fn analyze_filters(filters: &[Expr], fields:
&[DataField]) -> FilterP
} else {
Some(Predicate::and(translated_predicates))
},
- has_untranslated_residual,
+ requires_residual,
}
}
@@ -86,8 +93,10 @@ where
F: Fn(&Predicate) -> bool,
{
let translator = FilterTranslator::new(fields);
- if let Some(predicate) = translator.translate(filter) {
- if is_exact_filter_pushdown(&predicate) {
+ if let Some(translated) = translator.translate(filter) {
+ if translated.requires_residual {
+ TableProviderFilterPushDown::Inexact
+ } else if is_exact_filter_pushdown(&translated.predicate) {
TableProviderFilterPushDown::Exact
} else {
TableProviderFilterPushDown::Inexact
@@ -130,21 +139,26 @@ impl<'a> FilterTranslator<'a> {
}
}
- fn translate(&self, expr: &Expr) -> Option<Predicate> {
+ fn translate(&self, expr: &Expr) -> Option<TranslatedPredicate> {
match expr {
Expr::BinaryExpr(binary) => self.translate_binary(binary),
- // NOT is intentionally not translated: Predicate::Not uses
two-valued
- // logic (!bool), which incorrectly returns true when the inner
predicate
- // evaluates NULL to false. Combined with Exact pushdown precision,
- // DataFusion would remove the residual filter, producing wrong
results.
- Expr::Not(_) => None,
+ // Predicate::Not uses Paimon's two-valued predicate semantics, so
+ // translating SQL NOT is only safe as Inexact pushdown: DataFusion
+ // must keep its residual filter for NULL / three-valued semantics.
+ Expr::Not(inner) => {
+ let inner = self.translate(inner.as_ref())?;
+ Some(TranslatedPredicate {
+ predicate: Predicate::negate(inner.predicate),
+ requires_residual: true,
+ })
+ }
Expr::IsNull(inner) => {
let field = self.resolve_field(inner.as_ref())?;
- self.predicate_builder.is_null(field.name()).ok()
+ self.exact(self.predicate_builder.is_null(field.name()).ok()?)
}
Expr::IsNotNull(inner) => {
let field = self.resolve_field(inner.as_ref())?;
- self.predicate_builder.is_not_null(field.name()).ok()
+
self.exact(self.predicate_builder.is_not_null(field.name()).ok()?)
}
Expr::InList(in_list) => self.translate_in_list(in_list),
Expr::Between(between) => self.translate_between(between),
@@ -154,16 +168,21 @@ impl<'a> FilterTranslator<'a> {
}
}
- fn translate_binary(&self, binary: &BinaryExpr) -> Option<Predicate> {
+ fn translate_binary(&self, binary: &BinaryExpr) ->
Option<TranslatedPredicate> {
match binary.op {
- Operator::And => Some(Predicate::and(vec![
- self.translate(binary.left.as_ref())?,
- self.translate(binary.right.as_ref())?,
- ])),
- Operator::Or => Some(Predicate::or(vec![
- self.translate(binary.left.as_ref())?,
- self.translate(binary.right.as_ref())?,
- ])),
+ Operator::And | Operator::Or => {
+ let left = self.translate(binary.left.as_ref())?;
+ let right = self.translate(binary.right.as_ref())?;
+ let predicate = if binary.op == Operator::And {
+ Predicate::and(vec![left.predicate, right.predicate])
+ } else {
+ Predicate::or(vec![left.predicate, right.predicate])
+ };
+ Some(TranslatedPredicate {
+ predicate,
+ requires_residual: left.requires_residual ||
right.requires_residual,
+ })
+ }
Operator::Eq
| Operator::NotEq
| Operator::Lt
@@ -174,21 +193,21 @@ impl<'a> FilterTranslator<'a> {
}
}
- fn translate_comparison(&self, binary: &BinaryExpr) -> Option<Predicate> {
+ fn translate_comparison(&self, binary: &BinaryExpr) ->
Option<TranslatedPredicate> {
if let Some(predicate) = self.translate_column_literal_comparison(
binary.left.as_ref(),
binary.op,
binary.right.as_ref(),
) {
- return Some(predicate);
+ return self.exact(predicate);
}
let reversed = reverse_comparison_operator(binary.op)?;
- self.translate_column_literal_comparison(
+ self.exact(self.translate_column_literal_comparison(
binary.right.as_ref(),
reversed,
binary.left.as_ref(),
- )
+ )?)
}
fn translate_column_literal_comparison(
@@ -221,7 +240,7 @@ impl<'a> FilterTranslator<'a> {
}
}
- fn translate_in_list(&self, in_list: &InList) -> Option<Predicate> {
+ fn translate_in_list(&self, in_list: &InList) ->
Option<TranslatedPredicate> {
let field = self.resolve_field(in_list.expr.as_ref())?;
let literals: Option<Vec<_>> = in_list
.list
@@ -233,16 +252,16 @@ impl<'a> FilterTranslator<'a> {
.collect();
let literals = literals?;
- if in_list.negated {
+ self.exact(if in_list.negated {
self.predicate_builder
.is_not_in(field.name(), literals)
- .ok()
+ .ok()?
} else {
- self.predicate_builder.is_in(field.name(), literals).ok()
- }
+ self.predicate_builder.is_in(field.name(), literals).ok()?
+ })
}
- fn translate_between(&self, between: &Between) -> Option<Predicate> {
+ fn translate_between(&self, between: &Between) ->
Option<TranslatedPredicate> {
let field = self.resolve_field(between.expr.as_ref())?;
let low = scalar_to_datum(
extract_scalar_literal(between.low.as_ref())?,
@@ -259,16 +278,18 @@ impl<'a> FilterTranslator<'a> {
// Parquet row filter all treat a NULL operand as non-matching (SQL
// three-valued logic), and a data-column range stays Inexact so
// DataFusion keeps the residual filter.
- if between.negated {
+ self.exact(if between.negated {
self.predicate_builder
.not_between(field.name(), low, high)
- .ok()
+ .ok()?
} else {
- self.predicate_builder.between(field.name(), low, high).ok()
- }
+ self.predicate_builder
+ .between(field.name(), low, high)
+ .ok()?
+ })
}
- fn translate_scalar_function(&self, func: &ScalarFunction) ->
Option<Predicate> {
+ fn translate_scalar_function(&self, func: &ScalarFunction) ->
Option<TranslatedPredicate> {
// DataFusion built-in UDFs surfaced from `LIKE 'x%' / '%x' / '%x%'`
// rewrites and direct `starts_with(col, 'x') / ends_with / contains`
// calls. Only `(col, literal)` shapes are handled; anything else
@@ -280,21 +301,35 @@ impl<'a> FilterTranslator<'a> {
let scalar = extract_scalar_literal(&func.args[1])?;
let datum = scalar_to_datum(scalar, field.data_type())?;
- match func.name() {
- "starts_with" => self.predicate_builder.starts_with(field.name(),
datum).ok(),
- "ends_with" => self.predicate_builder.ends_with(field.name(),
datum).ok(),
- "contains" => self.predicate_builder.contains(field.name(),
datum).ok(),
- _ => None,
- }
+ let predicate = match func.name() {
+ "starts_with" => self
+ .predicate_builder
+ .starts_with(field.name(), datum)
+ .ok()?,
+ "ends_with" => self.predicate_builder.ends_with(field.name(),
datum).ok()?,
+ "contains" => self.predicate_builder.contains(field.name(),
datum).ok()?,
+ _ => return None,
+ };
+ self.exact(predicate)
}
- fn translate_like(&self, like: &Like) -> Option<Predicate> {
- // Negated and case-insensitive (ILIKE) variants stay unsupported for
- // now: NOT-LIKE has the same NULL-semantics concern as `Expr::Not`;
- // ILIKE has no equivalent in paimon's predicate model.
- if like.negated || like.case_insensitive {
+ fn translate_like(&self, like: &Like) -> Option<TranslatedPredicate> {
+ // ILIKE has no equivalent in Paimon's predicate model.
+ if like.case_insensitive {
return None;
}
+ let predicate = self.translate_positive_like(like)?;
+ if like.negated {
+ Some(TranslatedPredicate {
+ predicate: Predicate::negate(predicate),
+ requires_residual: true,
+ })
+ } else {
+ self.exact(predicate)
+ }
+ }
+
+ fn translate_positive_like(&self, like: &Like) -> Option<Predicate> {
let field = self.resolve_field(like.expr.as_ref())?;
let scalar = extract_scalar_literal(like.pattern.as_ref())?;
let datum = scalar_to_datum(scalar, field.data_type())?;
@@ -305,6 +340,13 @@ impl<'a> FilterTranslator<'a> {
.ok()
}
+ fn exact(&self, predicate: Predicate) -> Option<TranslatedPredicate> {
+ Some(TranslatedPredicate {
+ predicate,
+ requires_residual: false,
+ })
+ }
+
fn resolve_field(&self, expr: &Expr) -> Option<&'a DataField> {
let Expr::Column(Column { name, .. }) = expr else {
return None;
@@ -682,11 +724,11 @@ mod tests {
.to_string(),
"id > 10"
);
- assert!(!analysis.has_untranslated_residual);
+ assert!(!analysis.requires_residual);
}
#[test]
- fn
test_analyze_filters_marks_partial_translation_as_untranslated_residual() {
+ fn test_analyze_filters_pushes_not_and_marks_residual_required() {
let fields = test_fields();
let filters = vec![Expr::Column(Column::from_name("dt"))
.eq(lit("2024-01-01"))
@@ -700,21 +742,27 @@ mod tests {
.pushed_predicate
.expect("supported conjunct should still translate")
.to_string(),
- "dt = '2024-01-01'"
+ "(dt = '2024-01-01' AND NOT (hr = 10))"
);
- assert!(analysis.has_untranslated_residual);
+ assert!(analysis.requires_residual);
}
#[test]
- fn
test_analyze_filters_marks_unsupported_filter_as_untranslated_residual() {
+ fn test_analyze_filters_pushes_not_filter_and_marks_residual_required() {
let fields = test_fields();
let filters = vec![Expr::Not(Box::new(
Expr::Column(Column::from_name("dt")).eq(lit("2024-01-01")),
))];
let analysis = analyze_filters(&filters, &fields);
- assert!(analysis.pushed_predicate.is_none());
- assert!(analysis.has_untranslated_residual);
+ assert_eq!(
+ analysis
+ .pushed_predicate
+ .expect("NOT partition predicate should translate inexactly")
+ .to_string(),
+ "NOT (dt = '2024-01-01')"
+ );
+ assert!(analysis.requires_residual);
}
#[test]
@@ -805,20 +853,22 @@ mod tests {
}
#[test]
- fn test_translate_not_is_not_supported() {
+ fn test_translate_not_pushes_negated_predicate() {
let fields = test_fields();
let filter = Expr::Not(Box::new(
Expr::Column(Column::from_name("dt")).eq(lit("2024-01-01")),
));
- assert!(
- build_pushed_predicate(&[filter], &fields).is_none(),
- "NOT expressions should not translate due to NULL semantics"
+ assert_eq!(
+ build_pushed_predicate(&[filter], &fields)
+ .expect("NOT should translate as an inexact pushed predicate")
+ .to_string(),
+ "NOT (dt = '2024-01-01')"
);
}
#[test]
- fn test_classify_not_filter_as_unsupported() {
+ fn test_classify_not_filter_as_inexact_even_when_partition_only() {
let fields = test_fields();
let filter = Expr::Not(Box::new(
Expr::Column(Column::from_name("dt")).eq(lit("2024-01-01")),
@@ -826,7 +876,7 @@ mod tests {
assert_eq!(
classify_filter_pushdown(&filter, &fields,
is_exact_filter_pushdown),
- TableProviderFilterPushDown::Unsupported
+ TableProviderFilterPushDown::Inexact
);
}
@@ -987,11 +1037,18 @@ mod tests {
}
#[test]
- fn test_translate_negated_like_falls_open() {
+ fn test_translate_negated_like_pushes_inexact_not() {
let fields = test_fields();
- assert!(
- build_pushed_predicate(&[like_filter("a%", true, false)],
&fields).is_none(),
- "NOT LIKE must not translate (NULL semantics)"
+ let predicate = build_pushed_predicate(&[like_filter("a%", true,
false)], &fields)
+ .expect("NOT LIKE should translate as inexact NOT over LIKE");
+ assert_eq!(predicate.to_string(), "NOT (dt STARTS_WITH 'a')");
+ assert_eq!(
+ classify_filter_pushdown(
+ &like_filter("a%", true, false),
+ &fields,
+ is_exact_filter_pushdown
+ ),
+ TableProviderFilterPushDown::Inexact
);
}
diff --git a/crates/integrations/datafusion/src/physical_plan/scan.rs
b/crates/integrations/datafusion/src/physical_plan/scan.rs
index bcde017..a0d9f3c 100644
--- a/crates/integrations/datafusion/src/physical_plan/scan.rs
+++ b/crates/integrations/datafusion/src/physical_plan/scan.rs
@@ -108,6 +108,11 @@ impl PaimonTableScan {
self.pushed_predicate.as_ref()
}
+ #[cfg(test)]
+ pub(crate) fn filter_exact(&self) -> bool {
+ self.filter_exact
+ }
+
pub fn limit(&self) -> Option<usize> {
self.limit
}
diff --git a/crates/integrations/datafusion/src/table/mod.rs
b/crates/integrations/datafusion/src/table/mod.rs
index ba0a119..b10eb26 100644
--- a/crates/integrations/datafusion/src/table/mod.rs
+++ b/crates/integrations/datafusion/src/table/mod.rs
@@ -363,7 +363,7 @@ impl TableProvider for PaimonTableProvider {
if let Some(filter) = filter_analysis.pushed_predicate.clone() {
read_builder.with_filter(filter);
}
- let pushed_limit = limit.filter(|_|
!filter_analysis.has_untranslated_residual);
+ let pushed_limit = limit.filter(|_|
!filter_analysis.requires_residual);
if let Some(limit) = pushed_limit {
read_builder.with_limit(limit);
}
@@ -377,7 +377,7 @@ impl TableProvider for PaimonTableProvider {
.map_err(to_datafusion_error)?;
let target = state.config_options().execution.target_partitions;
- let filter_exact = !filter_analysis.has_untranslated_residual
+ let filter_exact = !filter_analysis.requires_residual
&& filter_analysis
.pushed_predicate
.as_ref()
@@ -780,7 +780,7 @@ mod tests {
}
#[tokio::test]
- async fn
test_scan_partially_translated_filter_keeps_partition_pruning_but_skips_limit_hint()
{
+ async fn
test_scan_partially_translated_not_filter_prunes_partitions_but_skips_limit_hint()
{
let provider = create_provider("multi_partitioned_log_table").await;
let filter = col("dt")
.eq(lit("2024-01-01"))
@@ -794,10 +794,7 @@ mod tests {
assert_eq!(scan.limit(), None);
assert_eq!(
extract_dt_hr_partition_set(scan.planned_partitions()),
- BTreeSet::from([
- ("2024-01-01".to_string(), 10),
- ("2024-01-01".to_string(), 20),
- ]),
+ BTreeSet::from([("2024-01-01".to_string(), 20)]),
);
assert_eq!(
scan.planned_partitions()
@@ -833,6 +830,23 @@ mod tests {
assert_eq!(scan.pushed_predicate(), Some(&expected));
}
+ #[tokio::test]
+ async fn test_scan_pushes_not_as_inexact_and_skips_limit_hint() {
+ let provider = data_evolution_projection_pruning_provider().await;
+ let filter = Expr::Not(Box::new(col("id").eq(lit(1))));
+ let plan = plan_scan(&provider, vec![filter.clone()], Some(1)).await;
+ let scan = plan
+ .downcast_ref::<PaimonTableScan>()
+ .expect("Expected PaimonTableScan");
+
+ let expected = build_pushed_predicate(&[filter],
provider.table().schema().fields())
+ .expect("NOT filter should translate as inexact pushdown");
+
+ assert_eq!(scan.pushed_predicate(), Some(&expected));
+ assert!(!scan.filter_exact());
+ assert_eq!(scan.limit(), None);
+ }
+
#[tokio::test]
async fn test_scan_applies_projection_to_data_evolution_planning() {
let provider = data_evolution_projection_pruning_provider().await;
diff --git a/crates/integrations/datafusion/src/variant_pushdown.rs
b/crates/integrations/datafusion/src/variant_pushdown.rs
index 1aff558..c33339f 100644
--- a/crates/integrations/datafusion/src/variant_pushdown.rs
+++ b/crates/integrations/datafusion/src/variant_pushdown.rs
@@ -214,9 +214,7 @@ impl ExtensionPlanner for VariantExtractionExtensionPlanner
{
if let Some(filter) = filter_analysis.pushed_predicate.clone() {
read_builder.with_filter(filter);
}
- let pushed_limit = node
- .fetch
- .filter(|_| !filter_analysis.has_untranslated_residual);
+ let pushed_limit = node.fetch.filter(|_|
!filter_analysis.requires_residual);
if let Some(limit) = pushed_limit {
read_builder.with_limit(limit);
}
@@ -236,7 +234,7 @@ impl ExtensionPlanner for VariantExtractionExtensionPlanner
{
.map(Arc::from)
.collect()
};
- let filter_exact = !filter_analysis.has_untranslated_residual
+ let filter_exact = !filter_analysis.requires_residual
&& filter_analysis
.pushed_predicate
.as_ref()
diff --git a/crates/integrations/datafusion/tests/sql_context_tests.rs
b/crates/integrations/datafusion/tests/sql_context_tests.rs
index 1d5ea34..ef9d4eb 100644
--- a/crates/integrations/datafusion/tests/sql_context_tests.rs
+++ b/crates/integrations/datafusion/tests/sql_context_tests.rs
@@ -19,6 +19,7 @@
use std::sync::Arc;
+use datafusion::arrow::array::Array;
use datafusion::catalog::CatalogProvider;
use datafusion::datasource::MemTable;
use paimon::catalog::Identifier;
@@ -843,6 +844,54 @@ async fn test_register_temp_table_database_qualified() {
assert_eq!(total_rows, 2);
}
+#[tokio::test]
+async fn test_not_filter_pushdown_keeps_sql_null_semantics() {
+ let (_tmp, catalog) = create_test_env();
+ let ctx = create_sql_context(catalog.clone()).await;
+
+ catalog
+ .create_database("my_db", false, Default::default())
+ .await
+ .unwrap();
+ ctx.sql("CREATE TABLE paimon.my_db.t (id INT, name STRING)")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ ctx.sql("INSERT INTO paimon.my_db.t VALUES (1, 'one'), (2, 'two'), (NULL,
'nil')")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let batches = ctx
+ .sql("SELECT id FROM paimon.my_db.t WHERE NOT (id = 1) ORDER BY id")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let values = batches
+ .iter()
+ .flat_map(|batch| {
+ let ids = batch
+ .column_by_name("id")
+ .and_then(|column| {
+ column
+ .as_any()
+ .downcast_ref::<datafusion::arrow::array::Int32Array>()
+ })
+ .expect("id column");
+ (0..ids.len()).map(|row| ids.value(row)).collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+
+ assert_eq!(values, vec![2]);
+}
+
#[tokio::test]
async fn test_register_temp_table_bare() {
let (_tmp, catalog) = create_test_env();
diff --git a/crates/paimon/src/predicate_stats.rs
b/crates/paimon/src/predicate_stats.rs
index d293d5c..846970b 100644
--- a/crates/paimon/src/predicate_stats.rs
+++ b/crates/paimon/src/predicate_stats.rs
@@ -207,6 +207,104 @@ pub(crate) fn data_leaf_may_match<T: StatsAccessor>(
}
}
+pub(crate) fn data_leaf_must_match<T: StatsAccessor>(
+ index: usize,
+ stats_data_type: &DataType,
+ predicate_data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+ stats: &T,
+) -> bool {
+ let row_count = stats.row_count();
+ if row_count <= 0 {
+ return false;
+ }
+
+ let null_count = stats.null_count(index);
+ match op {
+ PredicateOperator::IsNull => return null_count == Some(row_count),
+ PredicateOperator::IsNotNull => return null_count == Some(0),
+ _ => {
+ if null_count != Some(0) {
+ return false;
+ }
+ }
+ }
+
+ let min_value = match stats
+ .min_value(index, stats_data_type)
+ .and_then(|datum| coerce_stats_datum_for_predicate(datum,
predicate_data_type))
+ {
+ Some(value) => value,
+ None => return false,
+ };
+ let max_value = match stats
+ .max_value(index, stats_data_type)
+ .and_then(|datum| coerce_stats_datum_for_predicate(datum,
predicate_data_type))
+ {
+ Some(value) => value,
+ None => return false,
+ };
+ if !matches!(
+ min_value.partial_cmp(&max_value),
+ Some(Ordering::Less | Ordering::Equal)
+ ) {
+ return false;
+ }
+
+ match op {
+ PredicateOperator::Eq => literals
+ .first()
+ .is_some_and(|literal| min_value == *literal && max_value ==
*literal),
+ PredicateOperator::NotEq => literals.first().is_some_and(|literal| {
+ matches!(literal.partial_cmp(&min_value), Some(Ordering::Less))
+ || matches!(literal.partial_cmp(&max_value),
Some(Ordering::Greater))
+ }),
+ PredicateOperator::Lt => literals
+ .first()
+ .is_some_and(|literal| matches!(max_value.partial_cmp(literal),
Some(Ordering::Less))),
+ PredicateOperator::LtEq => literals.first().is_some_and(|literal| {
+ matches!(
+ max_value.partial_cmp(literal),
+ Some(Ordering::Less | Ordering::Equal)
+ )
+ }),
+ PredicateOperator::Gt => literals.first().is_some_and(|literal| {
+ matches!(min_value.partial_cmp(literal), Some(Ordering::Greater))
+ }),
+ PredicateOperator::GtEq => literals.first().is_some_and(|literal| {
+ matches!(
+ min_value.partial_cmp(literal),
+ Some(Ordering::Greater | Ordering::Equal)
+ )
+ }),
+ PredicateOperator::In => min_value == max_value &&
literals.contains(&min_value),
+ PredicateOperator::Between => {
+ let (Some(low), Some(high)) = (literals.first(), literals.get(1))
else {
+ return false;
+ };
+ let min_ge_low = !matches!(min_value.partial_cmp(low),
Some(Ordering::Less));
+ let max_le_high = !matches!(max_value.partial_cmp(high),
Some(Ordering::Greater));
+ min_ge_low && max_le_high
+ }
+ PredicateOperator::NotBetween => {
+ let (Some(low), Some(high)) = (literals.first(), literals.get(1))
else {
+ return false;
+ };
+ let max_lt_low = matches!(max_value.partial_cmp(low),
Some(Ordering::Less));
+ let min_gt_high = matches!(min_value.partial_cmp(high),
Some(Ordering::Greater));
+ max_lt_low || min_gt_high
+ }
+ PredicateOperator::IsNull
+ | PredicateOperator::IsNotNull
+ | PredicateOperator::NotIn
+ | PredicateOperator::StartsWith
+ | PredicateOperator::EndsWith
+ | PredicateOperator::Contains
+ | PredicateOperator::Like => false,
+ }
+}
+
/// Return the literal prefix of a SQL LIKE pattern up to the first unescaped
/// `%` or `_`. A backslash escapes the next character (which is appended
/// literally, mirroring arrow's `like` kernel); a trailing backslash is a
@@ -252,6 +350,10 @@ pub(crate) fn missing_field_may_match(op:
PredicateOperator, row_count: i64) ->
matches!(op, PredicateOperator::IsNull)
}
+pub(crate) fn missing_field_must_match(op: PredicateOperator, row_count: i64)
-> bool {
+ row_count > 0 && matches!(op, PredicateOperator::IsNull)
+}
+
/// Stats-prune `field BETWEEN low AND high` (and its negation) by treating it
/// as the conjunction `field >= low AND field <= high`:
/// * `Between` may match iff the file's `[min, max]` overlaps `[low, high]`.
@@ -321,7 +423,9 @@ fn predicate_may_match_with_schema<T: StatsAccessor>(
Predicate::Or(children) => children
.iter()
.any(|child| predicate_may_match_with_schema(child, stats,
field_mapping, file_fields)),
- Predicate::Not(_) => true,
+ Predicate::Not(inner) => {
+ !predicate_must_match_with_schema(inner, stats, field_mapping,
file_fields)
+ }
Predicate::Leaf {
index,
data_type,
@@ -347,6 +451,49 @@ fn predicate_may_match_with_schema<T: StatsAccessor>(
}
}
+fn predicate_must_match_with_schema<T: StatsAccessor>(
+ predicate: &Predicate,
+ stats: &T,
+ field_mapping: &[Option<usize>],
+ file_fields: &[DataField],
+) -> bool {
+ match predicate {
+ Predicate::AlwaysTrue => stats.row_count() > 0,
+ Predicate::AlwaysFalse => false,
+ Predicate::And(children) => children.iter().all(|child| {
+ predicate_must_match_with_schema(child, stats, field_mapping,
file_fields)
+ }),
+ Predicate::Or(children) => children.iter().any(|child| {
+ predicate_must_match_with_schema(child, stats, field_mapping,
file_fields)
+ }),
+ Predicate::Not(inner) => {
+ !predicate_may_match_with_schema(inner, stats, field_mapping,
file_fields)
+ }
+ Predicate::Leaf {
+ index,
+ data_type,
+ op,
+ literals,
+ ..
+ } => match field_mapping.get(*index).copied().flatten() {
+ Some(file_index) => {
+ let Some(file_field) = file_fields.get(file_index) else {
+ return false;
+ };
+ data_leaf_must_match(
+ file_index,
+ file_field.data_type(),
+ data_type,
+ *op,
+ literals,
+ stats,
+ )
+ }
+ None => missing_field_must_match(*op, stats.row_count()),
+ },
+ }
+}
+
fn coerce_stats_datum_for_predicate(datum: Datum, predicate_data_type:
&DataType) -> Option<Datum> {
match (datum, predicate_data_type) {
(datum @ Datum::Bool(_), DataType::Boolean(_))
@@ -613,6 +760,95 @@ mod tests {
));
}
+ #[test]
+ fn not_predicate_prunes_only_when_inner_must_match() {
+ let dt = DataType::Int(IntType::new());
+ let fields = vec![DataField::new(0, "id".to_string(), dt.clone())];
+ let mapping = vec![Some(0)];
+ let stats = MockStats {
+ row_count: 5,
+ null_count: Some(0),
+ min: Some(Datum::Int(10)),
+ max: Some(Datum::Int(10)),
+ };
+ let predicate = Predicate::negate(Predicate::Leaf {
+ column: "id".to_string(),
+ index: 0,
+ data_type: dt,
+ op: PredicateOperator::Eq,
+ literals: vec![Datum::Int(10)],
+ });
+
+ assert!(!predicates_may_match_with_schema(
+ &[predicate],
+ &stats,
+ &mapping,
+ &fields,
+ ));
+ }
+
+ #[test]
+ fn not_predicate_fails_open_when_nulls_or_stats_make_inner_uncertain() {
+ let dt = DataType::Int(IntType::new());
+ let fields = vec![DataField::new(0, "id".to_string(), dt.clone())];
+ let mapping = vec![Some(0)];
+ let predicate = Predicate::negate(Predicate::Leaf {
+ column: "id".to_string(),
+ index: 0,
+ data_type: dt,
+ op: PredicateOperator::Gt,
+ literals: vec![Datum::Int(5)],
+ });
+
+ let with_nulls = MockStats {
+ row_count: 5,
+ null_count: Some(1),
+ min: Some(Datum::Int(10)),
+ max: Some(Datum::Int(20)),
+ };
+ assert!(predicates_may_match_with_schema(
+ std::slice::from_ref(&predicate),
+ &with_nulls,
+ &mapping,
+ &fields,
+ ));
+
+ let missing_stats = MockStats {
+ row_count: 5,
+ null_count: Some(0),
+ min: None,
+ max: Some(Datum::Int(20)),
+ };
+ assert!(predicates_may_match_with_schema(
+ &[predicate],
+ &missing_stats,
+ &mapping,
+ &fields,
+ ));
+ }
+
+ #[test]
+ fn not_predicate_prunes_inner_range_that_must_match() {
+ let dt = DataType::Int(IntType::new());
+ let fields = vec![DataField::new(0, "id".to_string(), dt.clone())];
+ let mapping = vec![Some(0)];
+ let stats = int_stats(10, 20);
+ let predicate = Predicate::negate(Predicate::Leaf {
+ column: "id".to_string(),
+ index: 0,
+ data_type: dt,
+ op: PredicateOperator::Between,
+ literals: vec![Datum::Int(0), Datum::Int(100)],
+ });
+
+ assert!(!predicates_may_match_with_schema(
+ &[predicate],
+ &stats,
+ &mapping,
+ &fields,
+ ));
+ }
+
#[test]
fn between_with_all_null_file_is_pruned() {
let dt = DataType::Int(IntType::new());
diff --git a/crates/paimon/src/table/stats_filter.rs
b/crates/paimon/src/table/stats_filter.rs
index 63cf520..2eb18e5 100644
--- a/crates/paimon/src/table/stats_filter.rs
+++ b/crates/paimon/src/table/stats_filter.rs
@@ -20,7 +20,8 @@
use super::Table;
use crate::arrow::schema_evolution::create_index_mapping;
use crate::predicate_stats::{
- data_leaf_may_match, missing_field_may_match,
predicates_may_match_with_schema, StatsAccessor,
+ data_leaf_may_match, data_leaf_must_match, missing_field_may_match,
missing_field_must_match,
+ predicates_may_match_with_schema, StatsAccessor,
};
use crate::spec::{extract_datum, BinaryRow, DataField, DataFileMeta, DataType,
Datum, Predicate};
use std::collections::HashMap;
@@ -391,7 +392,13 @@ fn data_evolution_predicate_may_match(
row_count,
)
}),
- Predicate::Not(_) => true,
+ Predicate::Not(inner) => !data_evolution_predicate_must_match(
+ inner,
+ table_fields,
+ field_sources,
+ file_stats,
+ row_count,
+ ),
Predicate::Leaf {
index,
data_type,
@@ -420,6 +427,69 @@ fn data_evolution_predicate_may_match(
}
}
+fn data_evolution_predicate_must_match(
+ predicate: &Predicate,
+ table_fields: &[DataField],
+ field_sources: &[Option<(usize, usize)>],
+ file_stats: &[FileStatsRows],
+ row_count: i64,
+) -> bool {
+ match predicate {
+ Predicate::AlwaysTrue => row_count > 0,
+ Predicate::AlwaysFalse => false,
+ Predicate::And(children) => children.iter().all(|child| {
+ data_evolution_predicate_must_match(
+ child,
+ table_fields,
+ field_sources,
+ file_stats,
+ row_count,
+ )
+ }),
+ Predicate::Or(children) => children.iter().any(|child| {
+ data_evolution_predicate_must_match(
+ child,
+ table_fields,
+ field_sources,
+ file_stats,
+ row_count,
+ )
+ }),
+ Predicate::Not(inner) => !data_evolution_predicate_may_match(
+ inner,
+ table_fields,
+ field_sources,
+ file_stats,
+ row_count,
+ ),
+ Predicate::Leaf {
+ index,
+ data_type,
+ op,
+ literals,
+ ..
+ } => {
+ let Some(source) = field_sources.get(*index).copied().flatten()
else {
+ return missing_field_must_match(*op, row_count);
+ };
+ let (file_idx, field_index) = source;
+ let stats = &file_stats[file_idx];
+ let stats_data_type = table_fields
+ .get(*index)
+ .map(|f| f.data_type())
+ .unwrap_or(data_type);
+ data_leaf_must_match(
+ field_index,
+ stats_data_type,
+ data_type,
+ *op,
+ literals,
+ stats,
+ )
+ }
+ }
+}
+
/// Groups data files by overlapping `row_id_range` for data evolution.
///
/// Files are sorted by `(first_row_id, -max_sequence_number)`. Files whose
row ID ranges
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 71ad467..08b225f 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -2632,7 +2632,30 @@ mod tests {
}
#[test]
- fn test_data_file_matches_not_fails_open() {
+ fn test_data_file_matches_not_prunes_when_inner_must_match() {
+ let fields = int_field();
+ let file = test_data_file_meta(
+ int_stats_row(Some(10)),
+ int_stats_row(Some(10)),
+ vec![Some(0)],
+ 5,
+ );
+ let predicate = Predicate::negate(
+ PredicateBuilder::new(&fields)
+ .equal("id", Datum::Int(10))
+ .unwrap(),
+ );
+
+ assert!(!data_file_matches_predicates(
+ &file,
+ &[predicate],
+ TEST_SCHEMA_ID,
+ &test_schema_fields(),
+ ));
+ }
+
+ #[test]
+ fn test_data_file_matches_not_fails_open_when_inner_not_certain() {
let fields = int_field();
let file = test_data_file_meta(
int_stats_row(Some(10)),
@@ -2642,7 +2665,7 @@ mod tests {
);
let predicate = Predicate::negate(
PredicateBuilder::new(&fields)
- .less_than("id", Datum::Int(5))
+ .equal("id", Datum::Int(10))
.unwrap(),
);
@@ -2698,6 +2721,28 @@ mod tests {
));
}
+ #[test]
+ fn test_data_evolution_group_matches_not_prunes_when_inner_must_match() {
+ let fields = int_field();
+ let file = test_data_file_meta(
+ int_stats_row(Some(10)),
+ int_stats_row(Some(10)),
+ vec![Some(0)],
+ 5,
+ );
+ let predicate = Predicate::negate(
+ PredicateBuilder::new(&fields)
+ .equal("id", Datum::Int(10))
+ .unwrap(),
+ );
+
+ assert!(!data_evolution_group_matches_predicates(
+ &[file],
+ &[predicate],
+ &fields,
+ ));
+ }
+
#[test]
fn test_data_file_matches_corrupt_stats_fails_open() {
let fields = int_field();
