Merge 'core/incremental: Implement "is null" and "is not null" tests for view filter' from Glauber Costa

Just overlook on our side that they were not generated before.

Closes #3603

core/incremental/compiler.rs

@@ -2149,6 +2149,31 @@ impl DbspCompiler {
                     ))
                 }
             }
+            LogicalExpr::IsNull { expr, negated } => {
+                // Extract column index from the inner expression
+                if let LogicalExpr::Column(col) = expr.as_ref() {
+                    let column_idx = schema
+                        .columns
+                        .iter()
+                        .position(|c| c.name == col.name)
+                        .ok_or_else(|| {
+                            LimboError::ParseError(format!(
+                                "Column '{}' not found in schema for IS NULL filter",
+                                col.name
+                            ))
+                        })?;
+
+                    if *negated {
+                        Ok(FilterPredicate::IsNotNull { column_idx })
+                    } else {
+                        Ok(FilterPredicate::IsNull { column_idx })
+                    }
+                } else {
+                    Err(LimboError::ParseError(
+                        "IS NULL/IS NOT NULL expects a column reference".to_string(),
+                    ))
+                }
+            }
             _ => Err(LimboError::ParseError(format!(
                 "Unsupported filter expression: {expr:?}"
             ))),

core/incremental/filter_operator.rs

@@ -39,6 +39,11 @@ pub enum FilterPredicate {
     /// Column <= Column comparisons
     ColumnLessThanOrEqual { left_idx: usize, right_idx: usize },
 
+    /// Column IS NULL check
+    IsNull { column_idx: usize },
+    /// Column IS NOT NULL check
+    IsNotNull { column_idx: usize },
+
     /// Logical AND of two predicates
     And(Box<FilterPredicate>, Box<FilterPredicate>),
     /// Logical OR of two predicates
@@ -214,6 +219,18 @@ impl FilterOperator {
                 }
                 false
             }
+            FilterPredicate::IsNull { column_idx } => {
+                if let Some(v) = values.get(*column_idx) {
+                    return matches!(v, Value::Null);
+                }
+                false
+            }
+            FilterPredicate::IsNotNull { column_idx } => {
+                if let Some(v) = values.get(*column_idx) {
+                    return !matches!(v, Value::Null);
+                }
+                false
+            }
         }
     }
 }
@@ -293,3 +310,202 @@ impl IncrementalOperator for FilterOperator {
         self.tracker = Some(tracker);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::types::Text;
+
+    #[test]
+    fn test_is_null_predicate() {
+        let predicate = FilterPredicate::IsNull { column_idx: 1 };
+        let filter = FilterOperator::new(predicate);
+
+        // Test with NULL value
+        let values_with_null = vec![
+            Value::Integer(1),
+            Value::Null,
+            Value::Text(Text::from("test")),
+        ];
+        assert!(filter.evaluate_predicate(&values_with_null));
+
+        // Test with non-NULL value
+        let values_without_null = vec![
+            Value::Integer(1),
+            Value::Integer(42),
+            Value::Text(Text::from("test")),
+        ];
+        assert!(!filter.evaluate_predicate(&values_without_null));
+
+        // Test with different non-NULL types
+        let values_with_text = vec![
+            Value::Integer(1),
+            Value::Text(Text::from("not null")),
+            Value::Text(Text::from("test")),
+        ];
+        assert!(!filter.evaluate_predicate(&values_with_text));
+
+        let values_with_blob = vec![
+            Value::Integer(1),
+            Value::Blob(vec![1, 2, 3]),
+            Value::Text(Text::from("test")),
+        ];
+        assert!(!filter.evaluate_predicate(&values_with_blob));
+    }
+
+    #[test]
+    fn test_is_not_null_predicate() {
+        let predicate = FilterPredicate::IsNotNull { column_idx: 1 };
+        let filter = FilterOperator::new(predicate);
+
+        // Test with NULL value
+        let values_with_null = vec![
+            Value::Integer(1),
+            Value::Null,
+            Value::Text(Text::from("test")),
+        ];
+        assert!(!filter.evaluate_predicate(&values_with_null));
+
+        // Test with non-NULL value (Integer)
+        let values_with_integer = vec![
+            Value::Integer(1),
+            Value::Integer(42),
+            Value::Text(Text::from("test")),
+        ];
+        assert!(filter.evaluate_predicate(&values_with_integer));
+
+        // Test with non-NULL value (Text)
+        let values_with_text = vec![
+            Value::Integer(1),
+            Value::Text(Text::from("not null")),
+            Value::Text(Text::from("test")),
+        ];
+        assert!(filter.evaluate_predicate(&values_with_text));
+
+        // Test with non-NULL value (Blob)
+        let values_with_blob = vec![
+            Value::Integer(1),
+            Value::Blob(vec![1, 2, 3]),
+            Value::Text(Text::from("test")),
+        ];
+        assert!(filter.evaluate_predicate(&values_with_blob));
+    }
+
+    #[test]
+    fn test_is_null_with_and() {
+        // Test: column_0 = 1 AND column_1 IS NULL
+        let predicate = FilterPredicate::And(
+            Box::new(FilterPredicate::Equals {
+                column_idx: 0,
+                value: Value::Integer(1),
+            }),
+            Box::new(FilterPredicate::IsNull { column_idx: 1 }),
+        );
+        let filter = FilterOperator::new(predicate);
+
+        // Should match: column_0 = 1 AND column_1 IS NULL
+        let values_match = vec![
+            Value::Integer(1),
+            Value::Null,
+            Value::Text(Text::from("test")),
+        ];
+        assert!(filter.evaluate_predicate(&values_match));
+
+        // Should not match: column_0 = 2 AND column_1 IS NULL
+        let values_wrong_first = vec![
+            Value::Integer(2),
+            Value::Null,
+            Value::Text(Text::from("test")),
+        ];
+        assert!(!filter.evaluate_predicate(&values_wrong_first));
+
+        // Should not match: column_0 = 1 AND column_1 IS NOT NULL
+        let values_not_null = vec![
+            Value::Integer(1),
+            Value::Integer(42),
+            Value::Text(Text::from("test")),
+        ];
+        assert!(!filter.evaluate_predicate(&values_not_null));
+    }
+
+    #[test]
+    fn test_is_not_null_with_or() {
+        // Test: column_0 = 1 OR column_1 IS NOT NULL
+        let predicate = FilterPredicate::Or(
+            Box::new(FilterPredicate::Equals {
+                column_idx: 0,
+                value: Value::Integer(1),
+            }),
+            Box::new(FilterPredicate::IsNotNull { column_idx: 1 }),
+        );
+        let filter = FilterOperator::new(predicate);
+
+        // Should match: column_0 = 1 (regardless of column_1)
+        let values_first_matches = vec![
+            Value::Integer(1),
+            Value::Null,
+            Value::Text(Text::from("test")),
+        ];
+        assert!(filter.evaluate_predicate(&values_first_matches));
+
+        // Should match: column_1 IS NOT NULL (regardless of column_0)
+        let values_second_matches = vec![
+            Value::Integer(2),
+            Value::Integer(42),
+            Value::Text(Text::from("test")),
+        ];
+        assert!(filter.evaluate_predicate(&values_second_matches));
+
+        // Should not match: column_0 != 1 AND column_1 IS NULL
+        let values_no_match = vec![
+            Value::Integer(2),
+            Value::Null,
+            Value::Text(Text::from("test")),
+        ];
+        assert!(!filter.evaluate_predicate(&values_no_match));
+    }
+
+    #[test]
+    fn test_complex_null_predicates() {
+        // Test: (column_0 IS NULL OR column_1 IS NOT NULL) AND column_2 = 'test'
+        let predicate = FilterPredicate::And(
+            Box::new(FilterPredicate::Or(
+                Box::new(FilterPredicate::IsNull { column_idx: 0 }),
+                Box::new(FilterPredicate::IsNotNull { column_idx: 1 }),
+            )),
+            Box::new(FilterPredicate::Equals {
+                column_idx: 2,
+                value: Value::Text(Text::from("test")),
+            }),
+        );
+        let filter = FilterOperator::new(predicate);
+
+        // Should match: column_0 IS NULL, column_2 = 'test'
+        let values1 = vec![Value::Null, Value::Null, Value::Text(Text::from("test"))];
+        assert!(filter.evaluate_predicate(&values1));
+
+        // Should match: column_1 IS NOT NULL, column_2 = 'test'
+        let values2 = vec![
+            Value::Integer(1),
+            Value::Integer(42),
+            Value::Text(Text::from("test")),
+        ];
+        assert!(filter.evaluate_predicate(&values2));
+
+        // Should not match: column_2 != 'test'
+        let values3 = vec![
+            Value::Null,
+            Value::Integer(42),
+            Value::Text(Text::from("other")),
+        ];
+        assert!(!filter.evaluate_predicate(&values3));
+
+        // Should not match: column_0 IS NOT NULL AND column_1 IS NULL AND column_2 = 'test'
+        let values4 = vec![
+            Value::Integer(1),
+            Value::Null,
+            Value::Text(Text::from("test")),
+        ];
+        assert!(!filter.evaluate_predicate(&values4));
+    }
+}