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

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'"
         );
     }
 

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,
     }

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,