From 4a54784bd7e751afbf4234a0886cc63acddb0aca Mon Sep 17 00:00:00 2001 From: JeonDaehong Date: Sat, 17 Oct 2026 14:46:29 +0900 Subject: [PATCH 1/3] [core] Negate predicates like Java so NULL rows stay excluded Predicate::negate wrapped leaves in Not, which was evaluated as !inner, so NOT(p = 'a') selected rows where p is NULL. Reads returned them and delete_by_predicate deleted them. Flip the leaf operator and push through AND/OR like Java Predicate.negate. Leaves without a negated operator stay as Not, which now evaluates to false for NULL input. Closes #1143 --- .../datafusion/src/filter_pushdown.rs | 6 +- .../datafusion/src/physical_plan/scan.rs | 16 +- crates/paimon/src/arrow/residual.rs | 29 ++- crates/paimon/src/predicate_stats.rs | 5 +- crates/paimon/src/spec/predicate.rs | 210 +++++++++++++++--- .../src/spec/predicate/field_ref_tests.rs | 35 +++ crates/paimon/src/table/data_file_reader.rs | 4 +- .../src/table/global_index_scanner/tests.rs | 2 +- .../tests/table_predicate_delete_test.rs | 119 ++++++++++ 9 files changed, 372 insertions(+), 54 deletions(-) diff --git a/crates/integrations/datafusion/src/filter_pushdown.rs b/crates/integrations/datafusion/src/filter_pushdown.rs index cdbe6460d..43044eca8 100644 --- a/crates/integrations/datafusion/src/filter_pushdown.rs +++ b/crates/integrations/datafusion/src/filter_pushdown.rs @@ -1350,7 +1350,7 @@ mod tests { .pushed_predicate .expect("supported conjunct should still translate") .to_string(), - "(dt = '2024-01-01' AND NOT (hr = 10))" + "(dt = '2024-01-01' AND hr != 10)" ); assert!(analysis.requires_residual); } @@ -1368,7 +1368,7 @@ mod tests { .pushed_predicate .expect("NOT partition predicate should translate inexactly") .to_string(), - "NOT (dt = '2024-01-01')" + "dt != '2024-01-01'" ); assert!(analysis.requires_residual); } @@ -1530,7 +1530,7 @@ mod tests { build_pushed_predicate(&[filter], &fields) .expect("NOT should translate as an inexact pushed predicate") .to_string(), - "NOT (dt = '2024-01-01')" + "dt != '2024-01-01'" ); } diff --git a/crates/integrations/datafusion/src/physical_plan/scan.rs b/crates/integrations/datafusion/src/physical_plan/scan.rs index e7edacbaa..a2eefe46b 100644 --- a/crates/integrations/datafusion/src/physical_plan/scan.rs +++ b/crates/integrations/datafusion/src/physical_plan/scan.rs @@ -452,11 +452,10 @@ fn predicate_covers_candidate(predicate: &Predicate, candidate: &Predicate) -> b op: PredicateOperator::NotBetween, literals, }, - Predicate::Not(inner), + // `Predicate::negate` turns NOT(x >= lo AND x <= hi) into + // x < lo OR x > hi. + Predicate::Or(children), ) if literals.len() == 2 => { - let Predicate::And(children) = inner.as_ref() else { - return false; - }; let matches_bound = |candidate: &Predicate, expected_op: PredicateOperator, expected_literal: &Datum| { @@ -476,12 +475,13 @@ fn predicate_covers_candidate(predicate: &Predicate, candidate: &Predicate) -> b && candidate_literals[0] == *expected_literal ) }; - children - .iter() - .any(|child| matches_bound(child, PredicateOperator::GtEq, &literals[0])) + children.len() == 2 + && children + .iter() + .any(|child| matches_bound(child, PredicateOperator::Lt, &literals[0])) && children .iter() - .any(|child| matches_bound(child, PredicateOperator::LtEq, &literals[1])) + .any(|child| matches_bound(child, PredicateOperator::Gt, &literals[1])) } _ => false, } diff --git a/crates/paimon/src/arrow/residual.rs b/crates/paimon/src/arrow/residual.rs index 71211a860..e86082547 100644 --- a/crates/paimon/src/arrow/residual.rs +++ b/crates/paimon/src/arrow/residual.rs @@ -166,9 +166,16 @@ pub(crate) fn evaluate_predicate_mask( else { return Ok(None); }; - Ok(Some(boolean_mask_from_predicate(mask.len(), |row_index| { - !mask.value(row_index) - }))) + let negated = + boolean_mask_from_predicate(mask.len(), |row_index| !mask.value(row_index)); + if let Some(guard) = inner.not_null_guard() { + let Some(guard) = evaluate_predicate_mask(batch, &guard, file_fields, scan_fields)? + else { + return Ok(None); + }; + return Ok(Some(combine_filter_masks(&negated, &guard, false))); + } + Ok(Some(negated)) } Predicate::Leaf { column, @@ -2125,6 +2132,22 @@ mod tests { assert_eq!(int_values(&out), vec![10, 20]); } + #[test] + fn test_not_excludes_null_row() { + // NOT(age > 25) on [10, null, 30] -> [10]; NOT(unknown) is unknown. + let f = int_field(0, "age"); + let b = int_batch("age", vec![Some(10), None, Some(30)]); + let pred = Predicate::Not(Box::new(leaf( + 0, + DataType::Int(IntType::new()), + PredicateOperator::Gt, + vec![Datum::Int(25)], + ))); + let fp = file_predicates(vec![pred], vec![f.clone()]); + let out = filter_record_batch_by_predicates(b, &fp, &[f]).unwrap(); + assert_eq!(int_values(&out), vec![10]); + } + #[test] fn test_null_row_excluded() { // age = [10, null, 30]; age > 5 -> [10, 30] (null → false). diff --git a/crates/paimon/src/predicate_stats.rs b/crates/paimon/src/predicate_stats.rs index 871705e04..77c83bdcc 100644 --- a/crates/paimon/src/predicate_stats.rs +++ b/crates/paimon/src/predicate_stats.rs @@ -1054,13 +1054,14 @@ mod tests { 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 { + // Built directly: `Predicate::negate` would flip the leaf to `<=`. + let predicate = Predicate::Not(Box::new(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, diff --git a/crates/paimon/src/spec/predicate.rs b/crates/paimon/src/spec/predicate.rs index dab8bb18e..a210718d3 100644 --- a/crates/paimon/src/spec/predicate.rs +++ b/crates/paimon/src/spec/predicate.rs @@ -292,6 +292,40 @@ pub enum PredicateOperator { NotBetween, } +impl PredicateOperator { + /// The operator selecting the complement of the non-NULL rows, matching + /// Java `LeafFunction.negate`. `None` when there is no such operator. + fn negate(self) -> Option { + Some(match self { + Self::IsNull => Self::IsNotNull, + Self::IsNotNull => Self::IsNull, + Self::Eq => Self::NotEq, + Self::NotEq => Self::Eq, + Self::Lt => Self::GtEq, + Self::GtEq => Self::Lt, + Self::LtEq => Self::Gt, + Self::Gt => Self::LtEq, + Self::In => Self::NotIn, + Self::NotIn => Self::In, + Self::Between => Self::NotBetween, + Self::NotBetween => Self::Between, + Self::StartsWith + | Self::EndsWith + | Self::Contains + | Self::ArrayContains + | Self::ArraysOverlap + | Self::ArrayContainsAll + | Self::Like => return None, + }) + } +} + +/// The operator [`Predicate::negate`] flips `op` to on a field of `data_type`. +fn negated_op(op: PredicateOperator, data_type: &DataType) -> Option { + let _ = data_type; + op.negate() +} + impl fmt::Display for PredicateOperator { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { @@ -470,58 +504,112 @@ impl Predicate { /// Negate a predicate with simplification. /// + /// Like Java `Predicate.negate`, leaf operators are flipped (`=` / `!=`, + /// `<` / `>=`, `IS NULL` / `IS NOT NULL`, `IN` / `NOT IN`, ...) and AND/OR + /// are pushed down with De Morgan, so a NULL input still evaluates to false. + /// A plain `!p` would select NULL rows for `NOT(p = 'a')`. + /// /// - `NOT(NOT(p))` → `p` (double negation elimination) /// - `NOT(AlwaysTrue)` → `AlwaysFalse` /// - `NOT(AlwaysFalse)` → `AlwaysTrue` + /// - Leaves without a negated operator stay wrapped in `Not`. pub fn negate(predicate: Predicate) -> Predicate { match predicate { Predicate::Not(inner) => *inner, Predicate::AlwaysTrue => Predicate::AlwaysFalse, Predicate::AlwaysFalse => Predicate::AlwaysTrue, + Predicate::And(children) => { + Predicate::or(children.into_iter().map(Predicate::negate).collect()) + } + Predicate::Or(children) => { + Predicate::and(children.into_iter().map(Predicate::negate).collect()) + } + Predicate::Leaf { + column, + index, + data_type, + op, + literals, + } => match negated_op(op, &data_type) { + Some(op) => Predicate::Leaf { + column, + index, + data_type, + op, + literals, + }, + None => Predicate::Not(Box::new(Predicate::Leaf { + column, + index, + data_type, + op, + literals, + })), + }, Predicate::NestedLeaf { root, index, path, op, literals, - } if literals.is_empty() - && matches!(op, PredicateOperator::In | PredicateOperator::NotIn) => - { - Predicate::NestedLeaf { + } => match negated_op(op, path[path.len() - 1].data_type()) { + Some(op) => Predicate::NestedLeaf { root, index, path, - op: if op == PredicateOperator::In { - PredicateOperator::NotIn - } else { - PredicateOperator::In - }, + op, literals, - } - } + }, + None => Predicate::Not(Box::new(Predicate::NestedLeaf { + root, + index, + path, + op, + literals, + })), + }, + } + } + + /// `IS NOT NULL` on the field of a leaf kept under `Not` by [`negate`]. + /// Like Java `NotStartsWith`, `NotLike`, ..., that `Not` is false for NULL + /// input, so it is evaluated as `guard AND NOT leaf`. `None` for other + /// predicates and for `IS [NOT] NULL` leaves. + /// + /// [`negate`]: Predicate::negate + pub(crate) fn not_null_guard(&self) -> Option { + match self { Predicate::Leaf { column, index, data_type, op, - literals, - } if literals.is_empty() - && matches!(op, PredicateOperator::In | PredicateOperator::NotIn) => - { - // Java negates the leaf function, retaining its NULL exclusion. - Predicate::Leaf { - column, - index, - data_type, - op: if op == PredicateOperator::In { - PredicateOperator::NotIn - } else { - PredicateOperator::In - }, - literals, - } + .. + } if !matches!(op, PredicateOperator::IsNull | PredicateOperator::IsNotNull) => { + Some(Predicate::Leaf { + column: column.clone(), + index: *index, + data_type: data_type.clone(), + op: PredicateOperator::IsNotNull, + literals: Vec::new(), + }) } - other => Predicate::Not(Box::new(other)), + Predicate::NestedLeaf { + root, + index, + path, + op, + .. + } if !matches!(op, PredicateOperator::IsNull | PredicateOperator::IsNotNull) => { + Some(Predicate::NestedLeaf { + root: root.clone(), + index: *index, + path: path.clone(), + op: PredicateOperator::IsNotNull, + literals: Vec::new(), + }) + } + _ => None, } } @@ -2453,7 +2541,10 @@ pub(crate) fn eval_row(predicate: &Predicate, row: &BinaryRow) -> Result { } Ok(false) } - Predicate::Not(inner) => Ok(!eval_row(inner, row)?), + Predicate::Not(inner) => match inner.not_null_guard() { + Some(guard) => Ok(eval_row(&guard, row)? && !eval_row(inner, row)?), + None => Ok(!eval_row(inner, row)?), + }, Predicate::Leaf { index, data_type, @@ -3172,13 +3263,39 @@ mod tests { #[test] fn test_not() { let pb = PredicateBuilder::new(&test_fields()); - let p = pb.equal("id", Datum::Int(1)).unwrap(); - let negated = Predicate::negate(p.clone()); + let eq = pb.equal("id", Datum::Int(1)).unwrap(); + let gt = pb.greater_than("hr", Datum::Int(2)).unwrap(); - match &negated { - Predicate::Not(inner) => assert_eq!(inner.as_ref(), &p), - other => panic!("expected Not, got {other:?}"), - } + // Like Java, leaves flip their operator instead of becoming `Not`. + assert_eq!( + Predicate::negate(eq.clone()), + pb.not_equal("id", Datum::Int(1)).unwrap() + ); + assert_eq!( + Predicate::negate(Predicate::and(vec![eq.clone(), gt.clone()])), + Predicate::or(vec![ + pb.not_equal("id", Datum::Int(1)).unwrap(), + pb.less_or_equal("hr", Datum::Int(2)).unwrap(), + ]) + ); + assert_eq!( + Predicate::negate(Predicate::or(vec![eq.clone(), gt.clone()])), + Predicate::and(vec![ + pb.not_equal("id", Datum::Int(1)).unwrap(), + pb.less_or_equal("hr", Datum::Int(2)).unwrap(), + ]) + ); + assert_eq!( + Predicate::negate(pb.is_null("id").unwrap()), + pb.is_not_null("id").unwrap() + ); + + // No negated operator: stays `Not`. + let starts = pb.starts_with("name", Datum::String("a".into())).unwrap(); + assert_eq!( + Predicate::negate(starts.clone()), + Predicate::Not(Box::new(starts)) + ); } // ======================== field_idx_to_partition_idx ======================== @@ -3257,7 +3374,7 @@ mod tests { fn test_display_not() { let pb = PredicateBuilder::new(&test_fields()); let p = pb.equal("id", Datum::Int(1)).unwrap(); - let negated = Predicate::negate(p); + let negated = Predicate::Not(Box::new(p)); assert_eq!(format!("{negated}"), "NOT (id = 1)"); } @@ -3677,6 +3794,27 @@ mod tests { &row ) .unwrap()); + // ... and so is its negation. + assert!(!eval_row( + &Predicate::negate(make_leaf( + "x", + 0, + PredicateOperator::Eq, + vec![Datum::Int(42)] + )), + &row + ) + .unwrap()); + assert!(!eval_row( + &Predicate::Not(Box::new(make_leaf( + "x", + 0, + PredicateOperator::Gt, + vec![Datum::Int(42)] + ))), + &row + ) + .unwrap()); // IsNull / IsNotNull assert!(eval_row(&make_leaf("x", 0, PredicateOperator::IsNull, vec![]), &row).unwrap()); assert!(!eval_row( diff --git a/crates/paimon/src/spec/predicate/field_ref_tests.rs b/crates/paimon/src/spec/predicate/field_ref_tests.rs index 15578a10d..f3fb7bd00 100644 --- a/crates/paimon/src/spec/predicate/field_ref_tests.rs +++ b/crates/paimon/src/spec/predicate/field_ref_tests.rs @@ -378,3 +378,38 @@ fn nested_field_resolution_uses_components_and_canonical_names() { .nested_field("profile", &["score"]) .is_err()); } + +#[test] +fn negated_nested_leaves_flip_like_top_level_leaves() { + let builder = PredicateBuilder::new(&schema()); + let score = builder + .nested_field("profile", &["detail", "score"]) + .unwrap(); + assert_nested( + Predicate::negate(builder.equal(&score, Datum::Int(7)).unwrap()), + PredicateOperator::NotEq, + &[Datum::Int(7)], + ); + assert_nested( + Predicate::negate(builder.less_than(&score, Datum::Int(7)).unwrap()), + PredicateOperator::GtEq, + &[Datum::Int(7)], + ); + + // No negated operator: stays `Not`, guarded so a NULL label is excluded. + let label = builder + .nested_field("profile", &["detail", "label"]) + .unwrap(); + let starts = builder + .starts_with(&label, Datum::String("a".into())) + .unwrap(); + assert_eq!( + Predicate::negate(starts.clone()), + Predicate::Not(Box::new(starts.clone())) + ); + assert_nested( + starts.not_null_guard().unwrap(), + PredicateOperator::IsNotNull, + &[], + ); +} diff --git a/crates/paimon/src/table/data_file_reader.rs b/crates/paimon/src/table/data_file_reader.rs index 4ea4c9bd4..f4e4f6dc4 100644 --- a/crates/paimon/src/table/data_file_reader.rs +++ b/crates/paimon/src/table/data_file_reader.rs @@ -4059,7 +4059,9 @@ mod tests { .try_collect::>() .await .unwrap(); - assert_eq!(collect_ids(&batches), vec![1, 2], "enabled={enabled}"); + // `added` is NULL in the old file, so for id 2 the predicate is + // NOT(true AND unknown), which is unknown. Java filters it out too. + assert_eq!(collect_ids(&batches), vec![1], "enabled={enabled}"); } } diff --git a/crates/paimon/src/table/global_index_scanner/tests.rs b/crates/paimon/src/table/global_index_scanner/tests.rs index c00c99673..5ca34c6a6 100644 --- a/crates/paimon/src/table/global_index_scanner/tests.rs +++ b/crates/paimon/src/table/global_index_scanner/tests.rs @@ -3949,7 +3949,7 @@ async fn scalar_exactness_keeps_string_candidates_and_dropped_conjuncts_inexact( builder .equal("other", Datum::String("needle".to_string())) .unwrap(), - Predicate::negate(exact.clone()), + Predicate::Not(Box::new(exact.clone())), ] { let filter = Predicate::and(vec![exact.clone(), candidate]); let (ranges, is_exact) = scanner diff --git a/crates/paimon/tests/table_predicate_delete_test.rs b/crates/paimon/tests/table_predicate_delete_test.rs index cc30f3447..b14341438 100644 --- a/crates/paimon/tests/table_predicate_delete_test.rs +++ b/crates/paimon/tests/table_predicate_delete_test.rs @@ -571,3 +571,122 @@ async fn partition_deletes_support_append_pk_and_pending_files_without_row_track } } } + +type NegateCase = (fn(&PredicateBuilder) -> Predicate, Vec); + +#[tokio::test] +async fn negated_predicates_keep_null_rows() { + // Seeded id 3 has p = NULL. NOT(p = 'a') is unknown for it, so it must + // survive, as in Java where the negation is `p <> 'a'`. + let cases: Vec = vec![ + ( + |pb| Predicate::negate(pb.equal("p", Datum::String("a".into())).unwrap()), + vec![0, 1, 3], + ), + ( + |pb| { + Predicate::negate(Predicate::and(vec![ + pb.equal("q", Datum::Int(2)).unwrap(), + pb.equal("p", Datum::String("b".into())).unwrap(), + ])) + }, + vec![2, 3], + ), + ( + |pb| Predicate::negate(pb.not_equal("p", Datum::String("a".into())).unwrap()), + vec![2, 3], + ), + ]; + for (predicate, expected) in cases { + // Partition-only predicates drop files; adding a data conjunct goes + // through deletion vectors. + for data_conjunct in [false, true] { + let table = table(&[]).await; + seed(&table).await; + let pb = PredicateBuilder::new(&table.schema().fields_with_read_metadata()); + let mut predicate = predicate(&pb); + if data_conjunct { + predicate = Predicate::and(vec![ + predicate, + pb.greater_than("v", Datum::Int(0)).unwrap(), + ]); + } + let update = table.new_write_builder().new_update().unwrap(); + commit( + &table, + update + .delete_by_predicate(Some(predicate.clone())) + .await + .unwrap(), + ) + .await; + assert_eq!( + ids(&table).await, + expected, + "{predicate} data={data_conjunct}" + ); + } + } +} + +#[tokio::test] +async fn negated_data_predicate_keeps_null_rows() { + let table = table(&[]).await; + write_batch( + &table, + &RecordBatch::try_from_iter([ + ( + "id", + Arc::new(Int32Array::from(vec![0, 1, 2, 3])) as ArrayRef, + ), + ( + "p", + Arc::new(StringArray::from(vec![Some("a"); 4])) as ArrayRef, + ), + ( + "v", + Arc::new(Int32Array::from(vec![Some(10), Some(20), Some(30), None])) as ArrayRef, + ), + ("q", Arc::new(Int32Array::from(vec![1; 4])) as ArrayRef), + ]) + .unwrap(), + ) + .await; + let pb = PredicateBuilder::new(&table.schema().fields_with_read_metadata()); + let predicate = Predicate::negate(pb.equal("v", Datum::Int(10)).unwrap()); + + let mut read = table.new_read_builder(); + read.with_projection(&["id"]).unwrap(); + read.with_filter(predicate.clone()); + let plan = read.new_scan().plan().await.unwrap(); + let batches: Vec = read + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + let mut read_ids: Vec = batches + .iter() + .flat_map(|batch| { + batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_vec() + }) + .collect(); + read_ids.sort_unstable(); + assert_eq!(read_ids, vec![1, 2]); + + let update = table.new_write_builder().new_update().unwrap(); + commit( + &table, + update.delete_by_predicate(Some(predicate)).await.unwrap(), + ) + .await; + assert_eq!(ids(&table).await, vec![0, 3]); +} From 1053c4f7010a1590d5cc4857e7c4969073a83a5b Mon Sep 17 00:00:00 2001 From: JeonDaehong Date: Sun, 18 Oct 2026 17:18:03 +0900 Subject: [PATCH 2/3] [core] Keep NaN rows under negation and refine floating bitmap candidates Negating a FLOAT/DOUBLE comparison keeps Not, since NaN makes the flipped operator non-complementary. Scalar index exactness treats FLOAT/DOUBLE leaves that the bitmap index answers with every non-null row as inexact, so filtered Top-K refines them before ranking. --- crates/paimon/src/spec/predicate.rs | 11 +- .../paimon/src/table/de_vector_scan/tests.rs | 106 +++++++++++++++++ .../paimon/src/table/global_index_scanner.rs | 12 +- .../tests/table_predicate_delete_test.rs | 110 +++++++++++++----- 4 files changed, 206 insertions(+), 33 deletions(-) diff --git a/crates/paimon/src/spec/predicate.rs b/crates/paimon/src/spec/predicate.rs index a210718d3..995c37b05 100644 --- a/crates/paimon/src/spec/predicate.rs +++ b/crates/paimon/src/spec/predicate.rs @@ -322,7 +322,13 @@ impl PredicateOperator { /// The operator [`Predicate::negate`] flips `op` to on a field of `data_type`. fn negated_op(op: PredicateOperator, data_type: &DataType) -> Option { - let _ = data_type; + // `NaN < 1.0` and `NaN >= 1.0` are both false, so a flipped comparison + // would drop NaN. `Not` keeps it, as Java's NaN-last ordering does. + if matches!(data_type, DataType::Float(_) | DataType::Double(_)) + && !matches!(op, PredicateOperator::IsNull | PredicateOperator::IsNotNull) + { + return None; + } op.negate() } @@ -512,7 +518,8 @@ impl Predicate { /// - `NOT(NOT(p))` → `p` (double negation elimination) /// - `NOT(AlwaysTrue)` → `AlwaysFalse` /// - `NOT(AlwaysFalse)` → `AlwaysTrue` - /// - Leaves without a negated operator stay wrapped in `Not`. + /// - Leaves without a negated operator stay wrapped in `Not`, and so do + /// FLOAT/DOUBLE comparisons, which NaN makes non-complementary. pub fn negate(predicate: Predicate) -> Predicate { match predicate { Predicate::Not(inner) => *inner, diff --git a/crates/paimon/src/table/de_vector_scan/tests.rs b/crates/paimon/src/table/de_vector_scan/tests.rs index 0e16c0452..478021274 100644 --- a/crates/paimon/src/table/de_vector_scan/tests.rs +++ b/crates/paimon/src/table/de_vector_scan/tests.rs @@ -643,3 +643,109 @@ async fn scalar_coverage_gaps_follow_scalar_and_vector_modes() { } } } + +#[tokio::test] +async fn floating_bitmap_candidates_are_refined_before_top_k() { + // A bitmap index answers `!=` and range filters on FLOAT/DOUBLE with all + // non-null rows. Row 0 is closest to [1, 0] but fails every filter, so + // ranking those candidates unrefined would return it. + use crate::spec::{ArrayType, DataType, Datum, DoubleType, FloatType, IntType}; + use arrow_array::types::Float32Type; + use arrow_array::{Float64Array, ListArray}; + + let schema = crate::spec::Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("f", DataType::Double(DoubleType::new())) + .column( + "embedding", + DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))), + ) + .options(HashMap::from([ + ("bucket".into(), "-1".into()), + ("row-tracking.enabled".into(), "true".into()), + ("data-evolution.enabled".into(), "true".into()), + ("global-index.enabled".into(), "true".into()), + ])) + .build() + .unwrap(); + let table = Table::new( + crate::io::FileIOBuilder::new("memory").build().unwrap(), + crate::catalog::Identifier::new("default", "float_bitmap"), + format!("memory:/float-bitmap-{}", uuid::Uuid::new_v4()), + crate::spec::TableSchema::new(0, &schema), + None, + ); + let columns: Vec = vec![ + Arc::new(Int32Array::from(vec![0, 1])), + Arc::new(Float64Array::from(vec![1.0, 2.0])), + Arc::new(ListArray::from_iter_primitive::(vec![ + Some(vec![Some(1.0), Some(0.0)]), + Some(vec![Some(0.0), Some(1.0)]), + ])), + ]; + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int32, true), + ArrowField::new("f", ArrowDataType::Float64, true), + ArrowField::new("embedding", columns[2].data_type().clone(), true), + ])); + let mut writer = TableWrite::new(&table, "test".to_string()).unwrap(); + writer + .write_arrow_batch(&RecordBatch::try_new(arrow_schema, columns).unwrap()) + .await + .unwrap(); + TableCommit::new(table.clone(), "test".to_string()) + .commit(writer.prepare_commit().await.unwrap()) + .await + .unwrap(); + let mut builder = table.new_global_index_build_builder(); + builder.with_index_type("bitmap").with_index_column("f"); + builder.execute().await.unwrap(); + let mut builder = table.new_global_index_build_builder(); + builder + .with_index_type("ivf-flat") + .with_index_column("embedding") + .with_options(HashMap::from([ + ("ivf-flat.dimension".into(), "2".into()), + ("ivf-flat.nlist".into(), "1".into()), + ("ivf-flat.distance.metric".into(), "l2".into()), + ])); + builder.execute().await.unwrap(); + + let pb = crate::spec::PredicateBuilder::new(table.schema().fields()); + for filter in [ + Predicate::negate(pb.equal("f", Datum::Double(1.0)).unwrap()), + pb.not_equal("f", Datum::Double(1.0)).unwrap(), + pb.greater_than("f", Datum::Double(1.5)).unwrap(), + ] { + let batches: Vec = table + .new_vector_search_builder() + .with_vector_column("embedding") + .with_query_vector(vec![1.0, 0.0]) + .with_limit(1) + .with_filter(filter.clone()) + .execute() + .await + .unwrap() + .new_read_builder() + .read() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + let ids: Vec = batches + .iter() + .flat_map(|batch| { + batch + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_vec() + }) + .collect(); + assert_eq!(ids, vec![1], "{filter}"); + } +} diff --git a/crates/paimon/src/table/global_index_scanner.rs b/crates/paimon/src/table/global_index_scanner.rs index 397d37669..ca3320745 100644 --- a/crates/paimon/src/table/global_index_scanner.rs +++ b/crates/paimon/src/table/global_index_scanner.rs @@ -385,11 +385,19 @@ impl GlobalIndexScanner { predicate: &Predicate, evaluated_fields: &std::collections::HashSet, ) -> bool { - use crate::spec::PredicateOperator; + use crate::spec::{DataType, PredicateOperator}; match predicate { Predicate::NestedLeaf { .. } => false, - Predicate::Leaf { column, op, .. } => { + Predicate::Leaf { + column, + op, + data_type, + .. + } => { predicates::is_sorted_global_index_supported_op(*op) + // Bitmap answers these with every non-null row. + && !(matches!(data_type, DataType::Float(_) | DataType::Double(_)) + && super::bitmap_global_index_format::is_bitmap_floating_residual_sensitive_op(*op)) && !matches!( op, PredicateOperator::Contains diff --git a/crates/paimon/tests/table_predicate_delete_test.rs b/crates/paimon/tests/table_predicate_delete_test.rs index b14341438..7789ba735 100644 --- a/crates/paimon/tests/table_predicate_delete_test.rs +++ b/crates/paimon/tests/table_predicate_delete_test.rs @@ -20,12 +20,12 @@ mod common; use std::collections::HashMap; use std::sync::Arc; -use arrow_array::{ArrayRef, Int32Array, Int64Array, RecordBatch, StringArray}; +use arrow_array::{ArrayRef, Float64Array, Int32Array, Int64Array, RecordBatch, StringArray}; use common::incremental_helpers::{memory_table, persist_table_schema, setup_dirs, write_batch}; use futures::TryStreamExt; use paimon::spec::{ - DataType, Datum, IndexManifest, IntType, Predicate, PredicateBuilder, Schema, TableSchema, - VarCharType, + DataType, Datum, DoubleType, IndexManifest, IntType, Predicate, PredicateBuilder, Schema, + TableSchema, VarCharType, }; use paimon::table::{CommitMessage, Table}; @@ -129,6 +129,35 @@ async fn ids(table: &Table) -> Vec { ids } +async fn read_ids(table: &Table, filter: Predicate) -> Vec { + let mut read = table.new_read_builder(); + read.with_projection(&["id"]).unwrap(); + read.with_filter(filter); + let plan = read.new_scan().plan().await.unwrap(); + let batches: Vec = read + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + let mut ids: Vec = batches + .iter() + .flat_map(|batch| { + batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_vec() + }) + .collect(); + ids.sort_unstable(); + ids +} + async fn snapshot_id(table: &Table) -> i64 { table .snapshot_manager() @@ -655,32 +684,7 @@ async fn negated_data_predicate_keeps_null_rows() { let pb = PredicateBuilder::new(&table.schema().fields_with_read_metadata()); let predicate = Predicate::negate(pb.equal("v", Datum::Int(10)).unwrap()); - let mut read = table.new_read_builder(); - read.with_projection(&["id"]).unwrap(); - read.with_filter(predicate.clone()); - let plan = read.new_scan().plan().await.unwrap(); - let batches: Vec = read - .new_read() - .unwrap() - .to_arrow(plan.splits()) - .unwrap() - .try_collect() - .await - .unwrap(); - let mut read_ids: Vec = batches - .iter() - .flat_map(|batch| { - batch - .column(0) - .as_any() - .downcast_ref::() - .unwrap() - .values() - .to_vec() - }) - .collect(); - read_ids.sort_unstable(); - assert_eq!(read_ids, vec![1, 2]); + assert_eq!(read_ids(&table, predicate.clone()).await, vec![1, 2]); let update = table.new_write_builder().new_update().unwrap(); commit( @@ -690,3 +694,51 @@ async fn negated_data_predicate_keeps_null_rows() { .await; assert_eq!(ids(&table).await, vec![0, 3]); } + +#[tokio::test] +async fn negated_float_comparison_keeps_nan_rows() { + // `NaN < 1.0` and `NaN >= 1.0` are both false. Java orders NaN above + // every number, so NOT(f < 1.0) selects it, partitioned by `f` or not. + for partitioned in [false, true] { + let mut schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("f", DataType::Double(DoubleType::new())) + .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") + .option("deletion-vectors.enabled", "true"); + if partitioned { + schema = schema.partition_keys(["f"]); + } + let path = "memory:/negated_float"; + let (io, table) = memory_table(path, TableSchema::new(0, &schema.build().unwrap())); + setup_dirs(&io, path).await; + persist_table_schema(&io, path, table.schema()).await; + write_batch( + &table, + &RecordBatch::try_from_iter([ + ("id", Arc::new(Int32Array::from(vec![0, 1, 2])) as ArrayRef), + ( + "f", + Arc::new(Float64Array::from(vec![0.0, f64::NAN, 2.0])) as ArrayRef, + ), + ]) + .unwrap(), + ) + .await; + let pb = PredicateBuilder::new(&table.schema().fields_with_read_metadata()); + let predicate = Predicate::negate(pb.less_than("f", Datum::Double(1.0)).unwrap()); + + assert_eq!( + read_ids(&table, predicate.clone()).await, + vec![1, 2], + "partitioned={partitioned}" + ); + let update = table.new_write_builder().new_update().unwrap(); + commit( + &table, + update.delete_by_predicate(Some(predicate)).await.unwrap(), + ) + .await; + assert_eq!(ids(&table).await, vec![0], "partitioned={partitioned}"); + } +} From ae3660ebc403174578db8d10afe3b8498dee3671 Mon Sep 17 00:00:00 2001 From: JeonDaehong Date: Mon, 19 Oct 2026 21:46:59 +0900 Subject: [PATCH 3/3] [core] Keep exact BTree answers for floating filters Only a bitmap answers floating ranges and negations with every non-null row, so mark a leaf inexact only when a bitmap entry answered it. --- .../paimon/src/table/de_vector_scan/tests.rs | 194 ++++++++++-------- .../paimon/src/table/global_index_scanner.rs | 12 +- .../table/global_index_scanner/evaluator.rs | 26 ++- 3 files changed, 129 insertions(+), 103 deletions(-) diff --git a/crates/paimon/src/table/de_vector_scan/tests.rs b/crates/paimon/src/table/de_vector_scan/tests.rs index 478021274..ad38f61dd 100644 --- a/crates/paimon/src/table/de_vector_scan/tests.rs +++ b/crates/paimon/src/table/de_vector_scan/tests.rs @@ -648,104 +648,116 @@ async fn scalar_coverage_gaps_follow_scalar_and_vector_modes() { async fn floating_bitmap_candidates_are_refined_before_top_k() { // A bitmap index answers `!=` and range filters on FLOAT/DOUBLE with all // non-null rows. Row 0 is closest to [1, 0] but fails every filter, so - // ranking those candidates unrefined would return it. + // ranking those candidates unrefined would return it. A BTree answers the + // same filters exactly, so its candidates need no refinement. use crate::spec::{ArrayType, DataType, Datum, DoubleType, FloatType, IntType}; use arrow_array::types::Float32Type; use arrow_array::{Float64Array, ListArray}; - let schema = crate::spec::Schema::builder() - .column("id", DataType::Int(IntType::new())) - .column("f", DataType::Double(DoubleType::new())) - .column( - "embedding", - DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))), - ) - .options(HashMap::from([ - ("bucket".into(), "-1".into()), - ("row-tracking.enabled".into(), "true".into()), - ("data-evolution.enabled".into(), "true".into()), - ("global-index.enabled".into(), "true".into()), - ])) - .build() - .unwrap(); - let table = Table::new( - crate::io::FileIOBuilder::new("memory").build().unwrap(), - crate::catalog::Identifier::new("default", "float_bitmap"), - format!("memory:/float-bitmap-{}", uuid::Uuid::new_v4()), - crate::spec::TableSchema::new(0, &schema), - None, - ); - let columns: Vec = vec![ - Arc::new(Int32Array::from(vec![0, 1])), - Arc::new(Float64Array::from(vec![1.0, 2.0])), - Arc::new(ListArray::from_iter_primitive::(vec![ - Some(vec![Some(1.0), Some(0.0)]), - Some(vec![Some(0.0), Some(1.0)]), - ])), - ]; - let arrow_schema = Arc::new(ArrowSchema::new(vec![ - ArrowField::new("id", ArrowDataType::Int32, true), - ArrowField::new("f", ArrowDataType::Float64, true), - ArrowField::new("embedding", columns[2].data_type().clone(), true), - ])); - let mut writer = TableWrite::new(&table, "test".to_string()).unwrap(); - writer - .write_arrow_batch(&RecordBatch::try_new(arrow_schema, columns).unwrap()) - .await - .unwrap(); - TableCommit::new(table.clone(), "test".to_string()) - .commit(writer.prepare_commit().await.unwrap()) - .await - .unwrap(); - let mut builder = table.new_global_index_build_builder(); - builder.with_index_type("bitmap").with_index_column("f"); - builder.execute().await.unwrap(); - let mut builder = table.new_global_index_build_builder(); - builder - .with_index_type("ivf-flat") - .with_index_column("embedding") - .with_options(HashMap::from([ - ("ivf-flat.dimension".into(), "2".into()), - ("ivf-flat.nlist".into(), "1".into()), - ("ivf-flat.distance.metric".into(), "l2".into()), + for (index_type, refine_from_data) in + [("bitmap", None), ("btree", None), ("btree", Some("false"))] + { + let schema = crate::spec::Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("f", DataType::Double(DoubleType::new())) + .column( + "embedding", + DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))), + ) + .options(HashMap::from([ + ("bucket".into(), "-1".into()), + ("row-tracking.enabled".into(), "true".into()), + ("data-evolution.enabled".into(), "true".into()), + ("global-index.enabled".into(), "true".into()), + ])) + .build() + .unwrap(); + let table = Table::new( + crate::io::FileIOBuilder::new("memory").build().unwrap(), + crate::catalog::Identifier::new("default", "float_bitmap"), + format!("memory:/float-bitmap-{}", uuid::Uuid::new_v4()), + crate::spec::TableSchema::new(0, &schema), + None, + ); + let columns: Vec = vec![ + Arc::new(Int32Array::from(vec![0, 1])), + Arc::new(Float64Array::from(vec![1.0, 2.0])), + Arc::new(ListArray::from_iter_primitive::(vec![ + Some(vec![Some(1.0), Some(0.0)]), + Some(vec![Some(0.0), Some(1.0)]), + ])), + ]; + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int32, true), + ArrowField::new("f", ArrowDataType::Float64, true), + ArrowField::new("embedding", columns[2].data_type().clone(), true), ])); - builder.execute().await.unwrap(); - - let pb = crate::spec::PredicateBuilder::new(table.schema().fields()); - for filter in [ - Predicate::negate(pb.equal("f", Datum::Double(1.0)).unwrap()), - pb.not_equal("f", Datum::Double(1.0)).unwrap(), - pb.greater_than("f", Datum::Double(1.5)).unwrap(), - ] { - let batches: Vec = table - .new_vector_search_builder() - .with_vector_column("embedding") - .with_query_vector(vec![1.0, 0.0]) - .with_limit(1) - .with_filter(filter.clone()) - .execute() + let mut writer = TableWrite::new(&table, "test".to_string()).unwrap(); + writer + .write_arrow_batch(&RecordBatch::try_new(arrow_schema, columns).unwrap()) .await - .unwrap() - .new_read_builder() - .read() - .await - .unwrap() - .try_collect() + .unwrap(); + TableCommit::new(table.clone(), "test".to_string()) + .commit(writer.prepare_commit().await.unwrap()) .await .unwrap(); - let ids: Vec = batches - .iter() - .flat_map(|batch| { - batch - .column_by_name("id") - .unwrap() - .as_any() - .downcast_ref::() - .unwrap() - .values() - .to_vec() - }) - .collect(); - assert_eq!(ids, vec![1], "{filter}"); + let mut builder = table.new_global_index_build_builder(); + builder.with_index_type(index_type).with_index_column("f"); + builder.execute().await.unwrap(); + let mut builder = table.new_global_index_build_builder(); + builder + .with_index_type("ivf-flat") + .with_index_column("embedding") + .with_options(HashMap::from([ + ("ivf-flat.dimension".into(), "2".into()), + ("ivf-flat.nlist".into(), "1".into()), + ("ivf-flat.distance.metric".into(), "l2".into()), + ])); + builder.execute().await.unwrap(); + + let table = match refine_from_data { + Some(refine) => table.copy_with_options(HashMap::from([( + "global-index.filter.refine-from-data".to_string(), + refine.to_string(), + )])), + None => table, + }; + let pb = crate::spec::PredicateBuilder::new(table.schema().fields()); + for filter in [ + Predicate::negate(pb.equal("f", Datum::Double(1.0)).unwrap()), + pb.not_equal("f", Datum::Double(1.0)).unwrap(), + pb.greater_than("f", Datum::Double(1.5)).unwrap(), + ] { + let batches: Vec = table + .new_vector_search_builder() + .with_vector_column("embedding") + .with_query_vector(vec![1.0, 0.0]) + .with_limit(1) + .with_filter(filter.clone()) + .execute() + .await + .unwrap() + .new_read_builder() + .read() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + let ids: Vec = batches + .iter() + .flat_map(|batch| { + batch + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_vec() + }) + .collect(); + assert_eq!(ids, vec![1], "{index_type} {refine_from_data:?} {filter}"); + } } } diff --git a/crates/paimon/src/table/global_index_scanner.rs b/crates/paimon/src/table/global_index_scanner.rs index ca3320745..397d37669 100644 --- a/crates/paimon/src/table/global_index_scanner.rs +++ b/crates/paimon/src/table/global_index_scanner.rs @@ -385,19 +385,11 @@ impl GlobalIndexScanner { predicate: &Predicate, evaluated_fields: &std::collections::HashSet, ) -> bool { - use crate::spec::{DataType, PredicateOperator}; + use crate::spec::PredicateOperator; match predicate { Predicate::NestedLeaf { .. } => false, - Predicate::Leaf { - column, - op, - data_type, - .. - } => { + Predicate::Leaf { column, op, .. } => { predicates::is_sorted_global_index_supported_op(*op) - // Bitmap answers these with every non-null row. - && !(matches!(data_type, DataType::Float(_) | DataType::Double(_)) - && super::bitmap_global_index_format::is_bitmap_floating_residual_sensitive_op(*op)) && !matches!( op, PredicateOperator::Contains diff --git a/crates/paimon/src/table/global_index_scanner/evaluator.rs b/crates/paimon/src/table/global_index_scanner/evaluator.rs index 8a4530d5a..1dc48ccde 100644 --- a/crates/paimon/src/table/global_index_scanner/evaluator.rs +++ b/crates/paimon/src/table/global_index_scanner/evaluator.rs @@ -19,8 +19,8 @@ use super::all_match::all_matching_entries; use super::entry::{ - bitmap_meta_may_match, bitmap_meta_may_match_between, multivalue_meta_may_match, - sorted_entry_meta, GlobalIndexEntry, GlobalIndexFileKind, + bitmap_meta_may_match, bitmap_meta_may_match_between, is_floating_point, + multivalue_meta_may_match, sorted_entry_meta, GlobalIndexEntry, GlobalIndexFileKind, }; use super::predicates::{ entries_support_predicate, is_multivalue_predicate, is_sorted_global_index_supported_op, @@ -39,6 +39,7 @@ use crate::btree::{make_key_comparator, serialize_datum}; #[cfg(test)] use crate::spec::GlobalIndexSearchMode; use crate::spec::{DataType, Datum, Predicate, PredicateOperator}; +use crate::table::bitmap_global_index_format::is_bitmap_floating_residual_sensitive_op; use crate::table::bitmap_global_index_format::{ make_bitmap_key_comparator, serialize_bitmap_datum, }; @@ -386,6 +387,27 @@ impl GlobalIndexScanner { &self, entries: &[&GlobalIndexEntry], predicates: &[(PredicateOperator, &[Datum], &DataType)], + ) -> Result, Vec, bool)>> { + // A bitmap answers floating ranges and negations with every non-null + // row (see `bitmap_meta_may_match`), so those results are candidates + // only. Sorted indexes answer them exactly. + let bitmap_superset = entries + .iter() + .any(|entry| matches!(entry.index_type, GlobalIndexFileKind::Bitmap)) + && predicates.iter().any(|(op, _, data_type)| { + is_floating_point(data_type) && is_bitmap_floating_residual_sensitive_op(*op) + }); + Ok(self.evaluate_leaf_entries(entries, predicates).await?.map( + |(row_ranges, coverage, fully_evaluated)| { + (row_ranges, coverage, fully_evaluated && !bitmap_superset) + }, + )) + } + + async fn evaluate_leaf_entries( + &self, + entries: &[&GlobalIndexEntry], + predicates: &[(PredicateOperator, &[Datum], &DataType)], ) -> Result, Vec, bool)>> { let Some(normalized_predicates) = predicates .iter()