fix(query): Prevent invalid filter expression generation in InferFilterOptimizer (#17929)

b41sh · web-flow · commit 937894fbb317 · 2025-05-15T12:04:26.000+08:00
fix(query): fix infer filter transitive equality with different column types
diff --git a/src/query/service/tests/it/sql/planner/optimizer/optimizers/operator/filter/infer_filter_test.rs b/src/query/service/tests/it/sql/planner/optimizer/optimizers/operator/filter/infer_filter_test.rs
@@ -970,6 +970,7 @@ fn test_different_data_types() -> Result<()> {
         "",
         0,
     );
+    let col_variant = builder.column("variant", 3, "variant", DataType::Variant, "", 0);
     let _col_a = builder.column("A", 3, "A", DataType::Number(NumberDataType::Int64), "", 0);
 
     // Test: int8 column with values at type boundaries
@@ -1023,6 +1024,19 @@ fn test_different_data_types() -> Result<()> {
         );
     }
 
+    // Test: mixing integer and variant types
+    {
+        let const_5_int = builder.int(5);
+
+        // Test: variant = int8 AND variant = 5
+        let pred_variant_eq_int = builder.eq(col_variant.clone(), col_int8.clone());
+        let pred_variant_eq_5 = builder.eq(col_variant.clone(), const_5_int.clone());
+
+        let result = run_optimizer(vec![pred_variant_eq_int, pred_variant_eq_5])?;
+
+        assert_eq!(result.len(), 2, "Shouldn't add transitive equality");
+    }
+
     // Different data type not work yet, need fix.
     // Test: mixing integer and float types
     // {
diff --git a/src/query/sql/src/planner/optimizer/optimizers/operator/filter/infer_filter.rs b/src/query/sql/src/planner/optimizer/optimizers/operator/filter/infer_filter.rs
@@ -98,7 +98,9 @@ impl<'a> InferFilterOptimizer<'a> {
                     ) {
                         (true, true) => {
                             if op == ComparisonOp::Equal {
-                                self.add_equal_expr(&func.arguments[0], &func.arguments[1]);
+                                if !self.add_equal_expr(&func.arguments[0], &func.arguments[1]) {
+                                    remaining_predicates.push(predicate);
+                                }
                             } else {
                                 remaining_predicates.push(predicate);
                             }
@@ -178,7 +180,16 @@ impl<'a> InferFilterOptimizer<'a> {
         self.expr_equal_to.push(expr_equal_to);
     }
 
-    pub fn add_equal_expr(&mut self, left: &ScalarExpr, right: &ScalarExpr) {
+    pub fn add_equal_expr(&mut self, left: &ScalarExpr, right: &ScalarExpr) -> bool {
+        let Ok(left_ty) = left.data_type() else {
+            return false;
+        };
+        let Ok(right_ty) = right.data_type() else {
+            return false;
+        };
+        if !Self::check_equal_expr_type(&left_ty, &right_ty) {
+            return false;
+        }
         match self.expr_index.get(left) {
             Some(index) => self.expr_equal_to[*index].push(right.clone()),
             None => self.add_expr(left, vec![], vec![right.clone()]),
@@ -188,6 +199,28 @@ impl<'a> InferFilterOptimizer<'a> {
             Some(index) => self.expr_equal_to[*index].push(left.clone()),
             None => self.add_expr(right, vec![], vec![left.clone()]),
         };
+
+        true
+    }
+
+    // equal expr must have the same type, otherwise the function may fail on execution.
+    fn check_equal_expr_type(left_ty: &DataType, right_ty: &DataType) -> bool {
+        match (left_ty.remove_nullable(), right_ty.remove_nullable()) {
+            (DataType::Number(l), DataType::Number(r)) => {
+                (l.is_integer() && r.is_integer()) || (l.is_float() && r.is_float())
+            }
+            (DataType::Decimal(_), DataType::Decimal(_)) => true,
+            (DataType::Array(box l), DataType::Array(box r)) => Self::check_equal_expr_type(&l, &r),
+            (DataType::Map(box l), DataType::Map(box r)) => Self::check_equal_expr_type(&l, &r),
+            (DataType::Tuple(l_tys), DataType::Tuple(r_tys)) => {
+                l_tys.len() == r_tys.len()
+                    && l_tys
+                        .iter()
+                        .zip(r_tys.iter())
+                        .all(|(l_ty, r_ty)| Self::check_equal_expr_type(l_ty, r_ty))
+            }
+            (_, _) => left_ty.eq(right_ty),
+        }
     }
 
     fn add_expr_predicate(&mut self, expr: &ScalarExpr, new_predicate: Predicate) -> Result<()> {
diff --git a/tests/sqllogictests/suites/mode/standalone/explain/infer_filter.test b/tests/sqllogictests/suites/mode/standalone/explain/infer_filter.test
@@ -7,14 +7,20 @@ drop table if exists t2;
 statement ok
 drop table if exists t3;
 
+statement ok
+drop table if exists t4;
+
 statement ok
 create table t1(a int not null, b int not null);
 
 statement ok
 create table t2(a int not null, b int not null);
 
 statement ok
-create table t3(a int not null, b int not null);
+create table t3(a int not null, b int null);
+
+statement ok
+create table t4(a int not null, b variant null);
 
 # a = 1
 query T
@@ -753,9 +759,66 @@ HashJoin
     ├── apply join filters: [#1]
     └── estimated rows: 0.00
 
+query T
+explain select * from t4 where a = b and strip_null_value(b) is not null;
+----
+Filter
+├── output columns: [t4.a (#0), t4.b (#1)]
+├── filters: [is_true(CAST(t4.a (#0) AS Int32 NULL) = TRY_CAST(t4.b (#1) AS Int32 NULL)), is_not_null(strip_null_value(t4.b (#1)))]
+├── estimated rows: 0.00
+└── TableScan
+    ├── table: default.default.t4
+    ├── output columns: [a (#0), b (#1)]
+    ├── read rows: 0
+    ├── read size: 0
+    ├── partitions total: 0
+    ├── partitions scanned: 0
+    ├── push downs: [filters: [and_filters(CAST(t4.a (#0) AS Int32 NULL) = TRY_CAST(t4.b (#1) AS Int32 NULL), is_not_null(strip_null_value(t4.b (#1))))], limit: NONE]
+    └── estimated rows: 0.00
+
+query T
+explain select * from t3 join t4 on t3.b = t4.b where strip_null_value(t4.b) is not null;
+----
+HashJoin
+├── output columns: [t3.a (#0), t3.b (#1), t4.a (#2), t4.b (#3)]
+├── join type: INNER
+├── build keys: [CAST(t4.b (#3) AS Int32 NULL)]
+├── probe keys: [t3.b (#1)]
+├── keys is null equal: [false]
+├── filters: []
+├── build join filters:
+│   └── filter id:0, build key:CAST(t4.b (#3) AS Int32 NULL), probe key:t3.b (#1), filter type:bloom,inlist,min_max
+├── estimated rows: 0.00
+├── Filter(Build)
+│   ├── output columns: [t4.a (#2), t4.b (#3)]
+│   ├── filters: [is_not_null(strip_null_value(t4.b (#3)))]
+│   ├── estimated rows: 0.00
+│   └── TableScan
+│       ├── table: default.default.t4
+│       ├── output columns: [a (#2), b (#3)]
+│       ├── read rows: 0
+│       ├── read size: 0
+│       ├── partitions total: 0
+│       ├── partitions scanned: 0
+│       ├── push downs: [filters: [is_not_null(strip_null_value(t4.b (#3)))], limit: NONE]
+│       └── estimated rows: 0.00
+└── TableScan(Probe)
+    ├── table: default.default.t3
+    ├── output columns: [a (#0), b (#1)]
+    ├── read rows: 0
+    ├── read size: 0
+    ├── partitions total: 0
+    ├── partitions scanned: 0
+    ├── push downs: [filters: [], limit: NONE]
+    ├── apply join filters: [#0]
+    └── estimated rows: 0.00
+
 statement ok
 drop table if exists t3;
 
+statement ok
+drop table if exists t4;
+
 # merge predicates with different data types.
 statement ok
 create or replace table t1(id BIGINT NOT NULL);
@@ -815,3 +878,4 @@ drop table if exists t1;
 
 statement ok
 drop table if exists t2;
+
diff --git a/tests/sqllogictests/suites/mode/standalone/explain_native/infer_filter.test b/tests/sqllogictests/suites/mode/standalone/explain_native/infer_filter.test
@@ -7,14 +7,20 @@ drop table if exists t2;
 statement ok
 drop table if exists t3;
 
+statement ok
+drop table if exists t4;
+
 statement ok
 create table t1(a int not null, b int not null);
 
 statement ok
 create table t2(a int not null, b int not null);
 
 statement ok
-create table t3(a int not null, b int not null);
+create table t3(a int not null, b int null);
+
+statement ok
+create table t4(a int not null, b variant null);
 
 # a = 1
 query T
@@ -621,6 +627,52 @@ HashJoin
     ├── apply join filters: [#1]
     └── estimated rows: 0.00
 
+query T
+explain select * from t4 where a = b and strip_null_value(b) is not null;
+----
+TableScan
+├── table: default.default.t4
+├── output columns: [a (#0), b (#1)]
+├── read rows: 0
+├── read size: 0
+├── partitions total: 0
+├── partitions scanned: 0
+├── push downs: [filters: [and_filters(CAST(t4.a (#0) AS Int32 NULL) = TRY_CAST(t4.b (#1) AS Int32 NULL), is_not_null(strip_null_value(t4.b (#1))))], limit: NONE]
+└── estimated rows: 0.00
+
+query T
+explain select * from t3 join t4 on t3.b = t4.b where strip_null_value(t4.b) is not null;
+----
+HashJoin
+├── output columns: [t3.a (#0), t3.b (#1), t4.a (#2), t4.b (#3)]
+├── join type: INNER
+├── build keys: [CAST(t4.b (#3) AS Int32 NULL)]
+├── probe keys: [t3.b (#1)]
+├── keys is null equal: [false]
+├── filters: []
+├── build join filters:
+│   └── filter id:0, build key:CAST(t4.b (#3) AS Int32 NULL), probe key:t3.b (#1), filter type:bloom,inlist,min_max
+├── estimated rows: 0.00
+├── TableScan(Build)
+│   ├── table: default.default.t4
+│   ├── output columns: [a (#2), b (#3)]
+│   ├── read rows: 0
+│   ├── read size: 0
+│   ├── partitions total: 0
+│   ├── partitions scanned: 0
+│   ├── push downs: [filters: [is_not_null(strip_null_value(t4.b (#3)))], limit: NONE]
+│   └── estimated rows: 0.00
+└── TableScan(Probe)
+    ├── table: default.default.t3
+    ├── output columns: [a (#0), b (#1)]
+    ├── read rows: 0
+    ├── read size: 0
+    ├── partitions total: 0
+    ├── partitions scanned: 0
+    ├── push downs: [filters: [], limit: NONE]
+    ├── apply join filters: [#0]
+    └── estimated rows: 0.00
+
 statement ok
 drop table if exists t1;
 
@@ -630,13 +682,10 @@ drop table if exists t2;
 statement ok
 drop table if exists t3;
 
-# merge predicates with different data types.
 statement ok
-drop table if exists t1;
-
-statement ok
-drop table if exists t2;
+drop table if exists t4;
 
+# merge predicates with different data types.
 statement ok
 create table t1(id BIGINT NOT NULL);
 
@@ -687,3 +736,4 @@ drop table if exists t1;
 
 statement ok
 drop table if exists t2;
+
