fix: should not use array contains to implement inlist runtime filter (#18574)

* fix

* add ut

* fix

* fix

Author: SkyFan2002

src/query/service/src/pipelines/processors/transforms/hash_join/runtime_filter/convert.rs

@@ -19,9 +19,11 @@ use databend_common_catalog::runtime_filter_info::RuntimeFilterInfo;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::type_check;
+use databend_common_expression::types::DataType;
 use databend_common_expression::types::NumberDomain;
 use databend_common_expression::types::NumberScalar;
 use databend_common_expression::Column;
+use databend_common_expression::Constant;
 use databend_common_expression::Domain;
 use databend_common_expression::Expr;
 use databend_common_expression::RawExpr;
@@ -71,6 +73,13 @@ pub fn build_runtime_filter_infos(
 }
 
 fn build_inlist_filter(inlist: Column, probe_key: &Expr<String>) -> Result<Expr<String>> {
+    if inlist.len() == 0 {
+        return Ok(Expr::Constant(Constant {
+            span: None,
+            scalar: Scalar::Boolean(false),
+            data_type: DataType::Boolean,
+        }));
+    }
     let probe_key = probe_key.as_column_ref().unwrap();
 
     let raw_probe_key = RawExpr::ColumnRef {
@@ -79,20 +88,47 @@ fn build_inlist_filter(inlist: Column, probe_key: &Expr<String>) -> Result<Expr<
         data_type: probe_key.data_type.clone(),
         display_name: probe_key.display_name.clone(),
     };
-    let array = RawExpr::Constant {
+
+    let mut scalars = inlist.iter();
+
+    let first_scalar = scalars.next().unwrap();
+    let first_constant = RawExpr::Constant {
         span: None,
-        scalar: Scalar::Array(inlist),
+        scalar: first_scalar.to_owned(),
         data_type: None,
     };
-
-    let args = vec![array, raw_probe_key];
-    let contain_func = RawExpr::FunctionCall {
+    let mut or_expr = RawExpr::FunctionCall {
         span: None,
-        name: "contains".to_string(),
+        name: "eq".to_string(),
         params: vec![],
-        args,
+        args: vec![raw_probe_key.clone(), first_constant],
     };
-    let expr = type_check::check(&contain_func, &BUILTIN_FUNCTIONS)?;
+
+    for scalar in scalars {
+        let constant_expr = RawExpr::Constant {
+            span: None,
+            scalar: scalar.to_owned(),
+            data_type: None,
+        };
+
+        let eq_expr = RawExpr::FunctionCall {
+            span: None,
+            name: "eq".to_string(),
+            params: vec![],
+            args: vec![raw_probe_key.clone(), constant_expr],
+        };
+
+        or_expr = RawExpr::FunctionCall {
+            span: None,
+            name: "or".to_string(),
+            params: vec![],
+            args: vec![or_expr, eq_expr],
+        };
+    }
+
+    let final_expr = or_expr;
+
+    let expr = type_check::check(&final_expr, &BUILTIN_FUNCTIONS)?;
     Ok(expr)
 }
 
@@ -188,3 +224,93 @@ fn build_bloom_filter(
     let filter = BinaryFuse16::try_from(&hashes_vec)?;
     Ok((probe_key.id.to_string(), filter))
 }
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+
+    use databend_common_expression::types::DataType;
+    use databend_common_expression::types::NumberDataType;
+    use databend_common_expression::ColumnBuilder;
+    use databend_common_expression::ColumnRef;
+    use databend_common_expression::Constant;
+    use databend_common_expression::ConstantFolder;
+    use databend_common_expression::Domain;
+    use databend_common_expression::Expr;
+    use databend_common_expression::FunctionContext;
+    use databend_common_expression::Scalar;
+    use databend_common_functions::BUILTIN_FUNCTIONS;
+
+    use super::build_inlist_filter;
+
+    #[test]
+    fn test_build_inlist_filter() {
+        let func_ctx = FunctionContext::default();
+
+        // Create test column with values {1, 10}
+        let data_type = DataType::Number(NumberDataType::Int32);
+        let mut builder = ColumnBuilder::with_capacity(&data_type, 2);
+        builder.push(Scalar::Number(1i32.into()).as_ref());
+        builder.push(Scalar::Number(10i32.into()).as_ref());
+        let inlist = builder.build();
+
+        // Create probe key expression: column_a
+        let probe_key = Expr::ColumnRef(ColumnRef {
+            span: None,
+            id: "column_a".to_string(),
+            data_type: data_type.clone(),
+            display_name: "column_a".to_string(),
+        });
+
+        // Build the filter expression
+        let filter_expr = build_inlist_filter(inlist, &probe_key).unwrap();
+
+        // Test with ConstantFolder - case where column_a in [2,10] (can be folded to constant)
+        let mut input_domains = HashMap::new();
+        let domain_value_2_10 = Domain::from_min_max(
+            Scalar::Number(2i32.into()),
+            Scalar::Number(10i32.into()),
+            &data_type,
+        );
+        input_domains.insert("column_a".to_string(), domain_value_2_10);
+
+        let (folded_expr, _) = ConstantFolder::fold_with_domain(
+            &filter_expr,
+            &input_domains,
+            &func_ctx,
+            &BUILTIN_FUNCTIONS,
+        );
+
+        // Verify it's not folded to constant
+        assert!(folded_expr.as_constant().is_none());
+
+        // Test with ConstantFolder - case where column_a in [2,9] (should evaluate to false)
+        let mut input_domains_false = HashMap::new();
+        let domain_value_2_9 = Domain::from_min_max(
+            Scalar::Number(2i32.into()),
+            Scalar::Number(9i32.into()),
+            &data_type,
+        );
+        input_domains_false.insert("column_a".to_string(), domain_value_2_9);
+
+        let (folded_expr_false, _) = ConstantFolder::fold_with_domain(
+            &filter_expr,
+            &input_domains_false,
+            &func_ctx,
+            &BUILTIN_FUNCTIONS,
+        );
+
+        // Range [2,9] does not intersect with {1, 10}, so it should fold to constant false
+        match folded_expr_false {
+            Expr::Constant(Constant {
+                scalar: Scalar::Boolean(false),
+                ..
+            }) => {
+                println!("✓ Test passed: column_a in [2,9] correctly evaluated to false");
+            }
+            _ => {
+                panic!("Expected constant false, got: {:?}", folded_expr_false);
+            }
+        }
+    }
+}

src/query/storages/fuse/src/pruning/expr_runtime_pruner.rs

@@ -57,6 +57,10 @@ impl ExprRuntimePruner {
 
         let part = FuseBlockPartInfo::from_part(part)?;
         let pruned = self.exprs.iter().any(|filter| {
+            // If the filter is a constant false, we can prune the partition.
+            if matches!(filter, Expr::Constant(Constant { scalar: Scalar::Boolean(false), .. })) {
+                return true;
+            }
             let column_refs = filter.column_refs();
             // Currently only support filter with one column (probe key).
             debug_assert!(column_refs.len() == 1);