refactor(optimizer): refactor SelectivityEstimator

src/common/storage/src/statistics.rs

@@ -136,20 +136,11 @@ impl Display for Datum {
 
 impl Datum {
     pub fn type_comparable(&self, other: &Datum) -> bool {
-        matches!(
-            (self, other),
-            (Datum::Bool(_), Datum::Bool(_))
-                | (Datum::Bytes(_), Datum::Bytes(_))
-                | (Datum::Int(_), Datum::UInt(_))
-                | (Datum::Int(_), Datum::Int(_))
-                | (Datum::Int(_), Datum::Float(_))
-                | (Datum::UInt(_), Datum::Int(_))
-                | (Datum::UInt(_), Datum::UInt(_))
-                | (Datum::UInt(_), Datum::Float(_))
-                | (Datum::Float(_), Datum::Float(_))
-                | (Datum::Float(_), Datum::Int(_))
-                | (Datum::Float(_), Datum::UInt(_))
-        )
+        self.is_numeric() && other.is_numeric()
+            || matches!(
+                (self, other),
+                (Datum::Bool(_), Datum::Bool(_)) | (Datum::Bytes(_), Datum::Bytes(_))
+            )
     }
 
     pub fn is_numeric(&self) -> bool {
@@ -192,21 +183,4 @@ impl Datum {
             ))),
         }
     }
-
-    pub fn can_compare(&self, other: &Self) -> bool {
-        matches!(
-            (self, other),
-            (Datum::Bool(_), Datum::Bool(_))
-                | (Datum::Int(_), Datum::Int(_))
-                | (Datum::Int(_), Datum::UInt(_))
-                | (Datum::Int(_), Datum::Float(_))
-                | (Datum::UInt(_), Datum::UInt(_))
-                | (Datum::UInt(_), Datum::Int(_))
-                | (Datum::UInt(_), Datum::Float(_))
-                | (Datum::Float(_), Datum::Float(_))
-                | (Datum::Float(_), Datum::Int(_))
-                | (Datum::Float(_), Datum::UInt(_))
-                | (Datum::Bytes(_), Datum::Bytes(_))
-        )
-    }
 }

src/query/expression/src/expression.rs

@@ -649,6 +649,20 @@ impl<Index: ColumnIndex> Expr<Index> {
         }
     }
 
+    pub fn data_type_remove_generics(&self) -> DataType {
+        match self {
+            Expr::Constant(Constant { data_type, .. }) => data_type.clone(),
+            Expr::ColumnRef(ColumnRef { data_type, .. }) => data_type.clone(),
+            Expr::Cast(Cast { dest_type, .. }) => dest_type.clone(),
+            Expr::FunctionCall(FunctionCall {
+                return_type,
+                generics,
+                ..
+            }) => return_type.remove_generics(generics),
+            Expr::LambdaFunctionCall(LambdaFunctionCall { return_type, .. }) => return_type.clone(),
+        }
+    }
+
     pub fn column_refs(&self) -> HashMap<Index, DataType> {
         struct ColumnRefs<I>(HashMap<I, DataType>);
         impl<I: ColumnIndex> ExprVisitor<I> for ColumnRefs<I> {

src/query/expression/src/filter/filter_executor.rs

@@ -54,7 +54,7 @@ impl FilterExecutor {
         fn_registry: &'static FunctionRegistry,
         keep_order: bool,
     ) -> Self {
-        let (select_expr, has_or) = SelectExprBuilder::new().build(&expr).into();
+        let (select_expr, has_or) = SelectExprBuilder::new(fn_registry).build(&expr).into();
 
         let true_selection = vec![0; max_block_size];
         let false_selection = if has_or {

src/query/expression/src/filter/select_expr.rs

@@ -20,6 +20,7 @@ use itertools::Itertools;
 use crate::Expr;
 use crate::Function;
 use crate::FunctionID;
+use crate::FunctionRegistry;
 use crate::LikePattern;
 use crate::Scalar;
 use crate::expr::*;
@@ -48,14 +49,32 @@ pub enum SelectExpr {
     BooleanScalar((Scalar, DataType)),
 }
 
-#[derive(Default)]
 pub struct SelectExprBuilder {
-    not_function: Option<(FunctionID, Arc<Function>)>,
+    not_function: (FunctionID, Arc<Function>),
+    nullable_not_function: (FunctionID, Arc<Function>),
 }
 
 impl SelectExprBuilder {
-    pub fn new() -> Self {
-        Self::default()
+    pub fn new(fn_registry: &'static FunctionRegistry) -> Self {
+        let funcs =
+            fn_registry.search_candidates("not", &[], &[Expr::<usize>::Constant(Constant {
+                span: None,
+                scalar: Scalar::Boolean(true),
+                data_type: DataType::Boolean,
+            })]);
+
+        SelectExprBuilder {
+            not_function: funcs
+                .iter()
+                .find(|(id, func)| func.signature.return_type.is_boolean())
+                .unwrap()
+                .clone(),
+            nullable_not_function: funcs
+                .iter()
+                .find(|(id, func)| func.signature.return_type.is_nullable())
+                .unwrap()
+                .clone(),
+        }
     }
 
     pub fn build(&mut self, expr: &Expr) -> SelectExprBuildResult {
@@ -139,7 +158,6 @@ impl SelectExprBuilder {
                             .can_reorder(can_reorder)
                         }
                         "not" => {
-                            self.not_function = Some((*id.clone(), function.clone()));
                             let result = self.build_select_expr(&args[0], not ^ true);
                             if result.can_push_down_not {
                                 result
@@ -215,9 +233,11 @@ impl SelectExprBuilder {
                     .can_push_down_not(false)
             }
             Expr::Constant(Constant {
-                scalar, data_type, ..
-            }) if matches!(data_type, &DataType::Boolean | &DataType::Nullable(box DataType::Boolean)) =>
-            {
+                scalar,
+                data_type:
+                    data_type @ (DataType::Boolean | DataType::Nullable(box DataType::Boolean)),
+                ..
+            }) => {
                 let scalar = if not {
                     match scalar {
                         Scalar::Null => Scalar::Null,
@@ -259,24 +279,31 @@ impl SelectExprBuilder {
 
     fn other_select_expr(&self, expr: &Expr, not: bool) -> SelectExprBuildResult {
         let can_push_down_not = !not
-            || matches!(expr.data_type(), DataType::Boolean | DataType::Nullable(box DataType::Boolean));
-        let expr = if not && can_push_down_not {
-            self.wrap_not(expr)
+            || matches!(expr.data_type_remove_generics(), DataType::Boolean | DataType::Nullable(box DataType::Boolean));
+
+        let expr = SelectExpr::Others(if not && can_push_down_not {
+            self.wrap_not(expr.clone())
         } else {
             expr.clone()
-        };
-        SelectExprBuildResult::new(SelectExpr::Others(expr)).can_push_down_not(can_push_down_not)
+        });
+
+        SelectExprBuildResult::new(expr).can_push_down_not(can_push_down_not)
     }
 
-    fn wrap_not(&self, expr: &Expr) -> Expr {
-        let (id, function) = self.not_function.as_ref().unwrap();
+    fn wrap_not(&self, expr: Expr) -> Expr {
+        let data_type = expr.data_type_remove_generics();
+        let (id, function) = if data_type.is_nullable() {
+            self.nullable_not_function.clone()
+        } else {
+            self.not_function.clone()
+        };
         FunctionCall {
             span: None,
-            id: Box::new(id.clone()),
-            function: function.clone(),
+            id: Box::new(id),
+            return_type: function.signature.return_type.clone(),
+            function,
             generics: vec![],
-            args: vec![expr.clone()],
-            return_type: expr.data_type().clone(),
+            args: vec![expr],
         }
         .into()
     }

src/query/expression/src/filter/selector.rs

@@ -81,7 +81,6 @@ impl<'a> Selector<'a> {
     }
 
     // Process `SelectExpr`.
-    #[allow(clippy::too_many_arguments)]
     fn process_select_expr(
         &self,
         select_expr: &mut SelectExpr,
@@ -403,7 +402,6 @@ impl<'a> Selector<'a> {
     }
 
     // Process SelectExpr::Others.
-    #[allow(clippy::too_many_arguments)]
     fn process_others(
         &self,
         expr: &Expr,
@@ -444,7 +442,6 @@ impl<'a> Selector<'a> {
         self.select_value(Value::Scalar(constant), data_type, buffers, has_false)
     }
 
-    #[allow(clippy::too_many_arguments)]
     fn process_expr(
         &self,
         expr: &Expr,

src/query/expression/src/types.rs

@@ -198,6 +198,7 @@ impl DataType {
 
     pub fn has_generic(&self) -> bool {
         match self {
+            DataType::Generic(_) => true,
             DataType::Null
             | DataType::EmptyArray
             | DataType::EmptyMap
@@ -214,13 +215,44 @@ impl DataType {
             | DataType::Variant
             | DataType::Geometry
             | DataType::Geography
-            | DataType::Vector(_) => false,
+            | DataType::Vector(_)
+            | DataType::Opaque(_)
+            | DataType::StageLocation => false,
             DataType::Nullable(ty) => ty.has_generic(),
             DataType::Array(ty) => ty.has_generic(),
             DataType::Map(ty) => ty.has_generic(),
             DataType::Tuple(tys) => tys.iter().any(|ty| ty.has_generic()),
-            DataType::Generic(_) => true,
-            DataType::Opaque(_) | DataType::StageLocation => false,
+        }
+    }
+
+    pub fn remove_generics(&self, generics: &[DataType]) -> DataType {
+        match self {
+            DataType::Generic(i) => generics[*i].clone(),
+            DataType::Null
+            | DataType::EmptyArray
+            | DataType::EmptyMap
+            | DataType::Boolean
+            | DataType::Binary
+            | DataType::String
+            | DataType::Number(_)
+            | DataType::Decimal(_)
+            | DataType::Timestamp
+            | DataType::TimestampTz
+            | DataType::Date
+            | DataType::Interval
+            | DataType::Bitmap
+            | DataType::Variant
+            | DataType::Geometry
+            | DataType::Geography
+            | DataType::Vector(_)
+            | DataType::Opaque(_)
+            | DataType::StageLocation => self.clone(),
+            DataType::Nullable(ty) => DataType::Nullable(Box::new(ty.remove_generics(generics))),
+            DataType::Array(ty) => DataType::Array(Box::new(ty.remove_generics(generics))),
+            DataType::Map(ty) => DataType::Map(Box::new(ty.remove_generics(generics))),
+            DataType::Tuple(tys) => {
+                DataType::Tuple(tys.iter().map(|ty| ty.remove_generics(generics)).collect())
+            }
         }
     }
 

src/query/expression/src/utils/display.rs

@@ -1027,7 +1027,35 @@ impl<Index: ColumnIndex> Expr<Index> {
                     function, args, id, ..
                 }) => match (function.signature.name.as_str(), args.as_slice()) {
                     ("and", [lhs, rhs]) => write_binary_op("AND", lhs, rhs, 10, min_precedence),
+                    ("and_filters", args) if !args.is_empty() => {
+                        let precedence = 10;
+                        let str = args
+                            .iter()
+                            .map(|arg| write_expr(arg, precedence))
+                            .collect::<Vec<_>>()
+                            .join(" and ");
+
+                        if precedence < min_precedence {
+                            format!("({str})")
+                        } else {
+                            str
+                        }
+                    }
                     ("or", [lhs, rhs]) => write_binary_op("OR", lhs, rhs, 5, min_precedence),
+                    ("or_filters", args) if !args.is_empty() => {
+                        let precedence = 5;
+                        let str = args
+                            .iter()
+                            .map(|arg| write_expr(arg, precedence))
+                            .collect::<Vec<_>>()
+                            .join(" or ");
+
+                        if precedence < min_precedence {
+                            format!("({str})")
+                        } else {
+                            str
+                        }
+                    }
                     ("not", [expr]) => write_unary_op("NOT", expr, 15, min_precedence),
                     ("gte", [lhs, rhs]) => write_binary_op(">=", lhs, rhs, 20, min_precedence),
                     ("gt", [lhs, rhs]) => write_binary_op(">", lhs, rhs, 20, min_precedence),

src/query/expression/src/values.rs

@@ -424,6 +424,13 @@ impl Value<AnyType> {
     pub fn is_scalar_null(&self) -> bool {
         *self == Value::Scalar(Scalar::Null)
     }
+
+    pub fn is_value_of_type(&self, data_type: &DataType) -> bool {
+        match self {
+            Value::Scalar(scalar) => scalar.as_ref().is_value_of_type(data_type),
+            Value::Column(column) => column.data_type() == *data_type,
+        }
+    }
 }
 
 impl Scalar {

src/query/service/src/physical_plans/format/format_table_scan.rs

@@ -139,7 +139,7 @@ impl<'a> PhysicalFormat for TableScanFormatter<'a> {
         if let Some(agg_index) = agg_index {
             let (_, agg_index_sql, _) = ctx
                 .metadata
-                .get_agg_indexes(&table_name)
+                .get_agg_indices(&table_name)
                 .unwrap()
                 .iter()
                 .find(|(index, _, _)| *index == agg_index.index_id)

src/query/service/tests/it/sql/planner/optimizer/optimizers/rule/agg_rules/agg_index_query_rewrite.rs

@@ -381,7 +381,7 @@ async fn test_query_rewrite_impl(format: &str) -> Result<()> {
         let (mut query, _, metadata) = plan_sql(ctx.clone(), suite.query, true).await?;
         {
             let mut metadata = metadata.write();
-            metadata.add_agg_indexes("default.default.t".to_string(), vec![(
+            metadata.add_agg_indices("default.default.t".to_string(), vec![(
                 0,
                 suite.index.to_string(),
                 index,

src/query/sql/Cargo.toml

@@ -75,6 +75,7 @@ unicase = { workspace = true }
 url = { workspace = true }
 
 [dev-dependencies]
+goldenfile = { workspace = true }
 
 [lints]
 workspace = true

src/query/sql/src/planner/binder/ddl/index.rs

@@ -191,7 +191,7 @@ impl Binder {
                 let full_table_name = format!("{catalog}.{database}.{table_name}");
                 metadata
                     .write()
-                    .add_agg_indexes(full_table_name, agg_indexes);
+                    .add_agg_indices(full_table_name, agg_indexes);
             }
         }
 

src/query/sql/src/planner/binder/scalar_common.rs

@@ -134,7 +134,7 @@ fn fold_or_arguments(iter: impl Iterator<Item = ScalarExpr>) -> ScalarExpr {
                 span: None,
                 func_name: "or".to_string(),
                 params: vec![],
-                arguments: vec![acc, arg.clone()],
+                arguments: vec![acc, arg],
             }
             .into()
         },

src/query/sql/src/planner/binder/table.rs

@@ -88,7 +88,6 @@ use crate::plans::RecursiveCteScan;
 use crate::plans::RelOperator;
 use crate::plans::Scan;
 use crate::plans::SecureFilter;
-use crate::plans::Statistics;
 
 impl Binder {
     pub fn bind_dummy_table(
@@ -443,7 +442,6 @@ impl Binder {
             Scan {
                 table_index,
                 columns: columns.into_iter().map(|col| col.index()).collect(),
-                statistics: Arc::new(Statistics::default()),
                 change_type,
                 sample: sample.clone(),
                 scan_id,