[core] Negate predicates like Java so NULL rows stay excluded

crates/integrations/datafusion/src/filter_pushdown.rs

@@ -1459,7 +1459,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);
     }
@@ -1477,7 +1477,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);
     }
@@ -1639,7 +1639,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'"
         );
     }
 

crates/integrations/datafusion/src/physical_plan/scan.rs

@@ -493,11 +493,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| {
@@ -517,12 +516,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,
     }

crates/paimon/src/arrow/filtering.rs

@@ -120,19 +120,66 @@ fn remap_predicate(predicate: &Predicate, mapping: &[Option<usize>]) -> Predicat
                 }
             }
         }
-        Predicate::Not(inner) => {
-            let remapped = remap_predicate(inner, mapping);
-            match remapped {
-                Predicate::AlwaysTrue => Predicate::AlwaysFalse,
-                Predicate::AlwaysFalse => Predicate::AlwaysTrue,
-                other => Predicate::Not(Box::new(other)),
-            }
-        }
+        Predicate::Not(inner) => remap_negated(inner, mapping),
         Predicate::AlwaysTrue => Predicate::AlwaysTrue,
         Predicate::AlwaysFalse => Predicate::AlwaysFalse,
     }
 }
 
+/// Remaps `NOT predicate`. A field missing from the file is NULL, so a leaf on
+/// it is unknown and stays unknown under the NOT: it keeps no row, except for
+/// the null checks. The NOT is pushed down before missing leaves are folded,
+/// because folding first would turn the unknown into `AlwaysFalse` and the NOT
+/// would flip it to `AlwaysTrue`.
+fn remap_negated(predicate: &Predicate, mapping: &[Option<usize>]) -> Predicate {
+    let missing_under_not = |index: usize, op: &PredicateOperator| {
+        if mapping.get(index).copied().flatten().is_some() {
+            return None;
+        }
+        Some(match op {
+            PredicateOperator::IsNotNull => Predicate::AlwaysTrue,
+            _ => Predicate::AlwaysFalse,
+        })
+    };
+    match predicate {
+        Predicate::Leaf {
+            column, index, op, ..
+        } if !is_row_tracking_column(column) => {
+            if let Some(folded) = missing_under_not(*index, op) {
+                return folded;
+            }
+        }
+        Predicate::NestedLeaf { index, op, .. } => {
+            if let Some(folded) = missing_under_not(*index, op) {
+                return folded;
+            }
+        }
+        Predicate::And(children) => {
+            return remap_predicate(
+                &Predicate::Or(children.iter().map(negated_child).collect()),
+                mapping,
+            );
+        }
+        Predicate::Or(children) => {
+            return remap_predicate(
+                &Predicate::And(children.iter().map(negated_child).collect()),
+                mapping,
+            );
+        }
+        Predicate::Not(inner) => return remap_predicate(inner, mapping),
+        _ => {}
+    }
+    match remap_predicate(predicate, mapping) {
+        Predicate::AlwaysTrue => Predicate::AlwaysFalse,
+        Predicate::AlwaysFalse => Predicate::AlwaysTrue,
+        other => Predicate::Not(Box::new(other)),
+    }
+}
+
+fn negated_child(child: &Predicate) -> Predicate {
+    Predicate::Not(Box::new(child.clone()))
+}
+
 pub(crate) fn build_field_mapping(
     table_fields: &[DataField],
     file_fields: &[DataField],

crates/paimon/src/arrow/format/mosaic.rs

@@ -2031,6 +2031,58 @@ mod tests {
         assert_eq!(collect_i32_column(&batches, 0), vec![1, 2, 10, 11, 20, 21]);
     }
 
+    #[tokio::test]
+    async fn test_read_predicate_missing_column_under_not_stays_null() {
+        // The file omits `new_name`, so it is NULL in every row. A NOT over a
+        // leaf on it is still NULL and keeps no row; null checks still flip.
+        let mut fields = data_fields();
+        let missing_field = field(
+            3,
+            "new_name",
+            DataType::VarChar(VarCharType::with_nullable(true, 20).unwrap()),
+        );
+        fields.push(missing_field.clone());
+        let pb = PredicateBuilder::new(&fields);
+        let starts = pb
+            .starts_with("new_name", Datum::String("a".to_string()))
+            .unwrap();
+        let id_is_1 = pb.equal("id", Datum::Int(1)).unwrap();
+        let all = vec![1, 2, 10, 11, 20, 21];
+        for (predicate, expected) in [
+            (crate::spec::Predicate::negate(starts.clone()), vec![]),
+            (
+                crate::spec::Predicate::negate(crate::spec::Predicate::and(vec![
+                    starts.clone(),
+                    id_is_1.clone(),
+                ])),
+                vec![2, 10, 11, 20, 21],
+            ),
+            (
+                crate::spec::Predicate::negate(crate::spec::Predicate::or(vec![
+                    starts.clone(),
+                    id_is_1,
+                ])),
+                vec![],
+            ),
+            (
+                crate::spec::Predicate::negate(pb.is_null("new_name").unwrap()),
+                vec![],
+            ),
+            (
+                crate::spec::Predicate::negate(pb.is_not_null("new_name").unwrap()),
+                all,
+            ),
+        ] {
+            let predicates = predicate_file_predicates(fields.clone(), vec![predicate.clone()]);
+            let projected = vec![fields[0].clone(), missing_field.clone()];
+            let data = multi_row_group_mosaic(vec!["id".to_string()]);
+            let batches = read_batches_with_predicates(data, &projected, Some(&predicates), None)
+                .await
+                .unwrap();
+            assert_eq!(collect_i32_column(&batches, 0), expected, "{predicate:?}");
+        }
+    }
+
     #[tokio::test]
     async fn test_read_projection_with_missing_column() {
         let fields = data_fields();

crates/paimon/src/arrow/residual.rs

@@ -161,15 +161,7 @@ pub(crate) fn evaluate_predicate_mask(
             }
             Ok(Some(combined))
         }
-        Predicate::Not(inner) => {
-            let Some(mask) = evaluate_predicate_mask(batch, inner, file_fields, scan_fields)?
-            else {
-                return Ok(None);
-            };
-            Ok(Some(boolean_mask_from_predicate(mask.len(), |row_index| {
-                !mask.value(row_index)
-            })))
-        }
+        Predicate::Not(inner) => evaluate_negated_mask(batch, inner, file_fields, scan_fields),
         Predicate::Leaf {
             column,
             index,
@@ -258,6 +250,63 @@ pub(crate) fn evaluate_predicate_mask(
 
 /// Two [`DataField`]s refer to the same logical column when their IDs match, or
 /// (for schemas without stable IDs) their names match.
+/// Evaluates `NOT predicate`. The `NOT` is pushed down to the leaves with
+/// De Morgan first: a leaf masks NULL as false, so negating a combined mask
+/// would turn UNKNOWN into true.
+fn evaluate_negated_mask(
+    batch: &RecordBatch,
+    predicate: &Predicate,
+    file_fields: &[DataField],
+    scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+    match predicate {
+        // NOT(a AND b) = NOT a OR NOT b, evaluated like the `Or` arm.
+        Predicate::And(children) => {
+            let mut combined = BooleanArray::from(vec![false; batch.num_rows()]);
+            for child in children {
+                let Some(mask) = evaluate_negated_mask(batch, child, file_fields, scan_fields)?
+                else {
+                    return Ok(None);
+                };
+                combined = combine_filter_masks(&combined, &mask, true);
+            }
+            Ok(Some(combined))
+        }
+        // NOT(a OR b) = NOT a AND NOT b, evaluated like the `And` arm.
+        Predicate::Or(children) => {
+            let mut combined = None;
+            for child in children {
+                let Some(mask) = evaluate_negated_mask(batch, child, file_fields, scan_fields)?
+                else {
+                    continue;
+                };
+                combined = Some(match combined {
+                    Some(existing) => combine_filter_masks(&existing, &mask, false),
+                    None => mask,
+                });
+            }
+            Ok(combined)
+        }
+        Predicate::Not(inner) => evaluate_predicate_mask(batch, inner, file_fields, scan_fields),
+        _ => {
+            let Some(mask) = evaluate_predicate_mask(batch, predicate, file_fields, scan_fields)?
+            else {
+                return Ok(None);
+            };
+            let negated =
+                boolean_mask_from_predicate(mask.len(), |row_index| !mask.value(row_index));
+            if let Some(guard) = predicate.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))
+        }
+    }
+}
+
 pub(crate) fn same_data_field(left: &DataField, right: &DataField) -> bool {
     left.id() == right.id() || left.name() == right.name()
 }
@@ -2125,6 +2174,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).

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,