chore(query): check schema mismatch in projection render result (#17654)

* chore(query): check schema mismatch in projection render result

* chore(query): check schema mismatch in projection render result

* chore(query): add ScalarExpr::TypedConstantExpr

* chore(query): add ScalarExpr::TypedConstantExpr

* update

* update

Author: sundy-li

src/query/expression/src/expression.rs

@@ -42,6 +42,7 @@ pub enum RawExpr<Index: ColumnIndex = usize> {
     Constant {
         span: Span,
         scalar: Scalar,
+        data_type: Option<DataType>,
     },
     ColumnRef {
         span: Span,
@@ -361,9 +362,14 @@ impl<Index: ColumnIndex> RawExpr<Index> {
         f: impl Fn(&Index) -> ToIndex + Copy,
     ) -> RawExpr<ToIndex> {
         match self {
-            RawExpr::Constant { span, scalar } => RawExpr::Constant {
+            RawExpr::Constant {
+                span,
+                scalar,
+                data_type,
+            } => RawExpr::Constant {
                 span: *span,
                 scalar: scalar.clone(),
+                data_type: data_type.clone(),
             },
             RawExpr::ColumnRef {
                 span,

src/query/expression/src/type_check.rs

@@ -44,10 +44,17 @@ pub fn check<Index: ColumnIndex>(
     fn_registry: &FunctionRegistry,
 ) -> Result<Expr<Index>> {
     match expr {
-        RawExpr::Constant { span, scalar } => Ok(Expr::Constant {
+        RawExpr::Constant {
+            span,
+            scalar,
+            data_type,
+        } => Ok(Expr::Constant {
             span: *span,
             scalar: scalar.clone(),
-            data_type: scalar.as_ref().infer_data_type(),
+            data_type: data_type
+                .as_ref()
+                .cloned()
+                .unwrap_or_else(|| scalar.as_ref().infer_data_type()),
         }),
         RawExpr::ColumnRef {
             span,

src/query/expression/src/utils/mod.rs

@@ -93,7 +93,11 @@ pub fn cast_scalar(
     let raw_expr = RawExpr::Cast {
         span,
         is_try: false,
-        expr: Box::new(RawExpr::Constant { span, scalar }),
+        expr: Box::new(RawExpr::Constant {
+            span,
+            scalar,
+            data_type: None,
+        }),
         dest_type,
     };
     let expr = crate::type_check::check(&raw_expr, fn_registry)?;

src/query/functions/src/test_utils.rs

@@ -113,6 +113,7 @@ fn transform_expr(ast: AExpr, columns: &[(&str, DataType)]) -> RawExpr {
         AExpr::Literal { span, value } => RawExpr::Constant {
             span,
             scalar: transform_literal(value),
+            data_type: None,
         },
         AExpr::ColumnRef {
             span,
@@ -374,6 +375,7 @@ fn transform_expr(ast: AExpr, columns: &[(&str, DataType)]) -> RawExpr {
                     RawExpr::Constant {
                         span,
                         scalar: Scalar::String(key.name),
+                        data_type: Some(DataType::String),
                     },
                 ]),
                 MapAccessor::DotNumber { key } => {

src/query/service/src/pipelines/builders/builder_project.rs

@@ -41,21 +41,19 @@ impl PipelineBuilder {
         }
 
         let mut projections = Vec::with_capacity(result_columns.len());
-
         for column_binding in result_columns {
             let index = column_binding.index;
             projections.push(input_schema.index_of(index.to_string().as_str())?);
 
-            // TODO: open this check after fix grouping sets datatype mismatch
-            // #[cfg(debug_assertions)]
-            // {
-            //     let f = input_schema.field_with_name(index.to_string().as_str())?;
-            //     assert_eq!(
-            //         f.data_type(),
-            //         column_binding.data_type.as_ref(),
-            //         "Result projection schema mismatch"
-            //     );
-            // }
+            #[cfg(debug_assertions)]
+            {
+                let f = input_schema.field_with_name(index.to_string().as_str())?;
+                assert_eq!(
+                    f.data_type(),
+                    column_binding.data_type.as_ref(),
+                    "Result projection schema mismatch"
+                );
+            }
         }
         let num_input_columns = input_schema.num_fields();
         pipeline.add_transformer(|| {

src/query/service/src/pipelines/processors/transforms/hash_join/util.rs

@@ -77,6 +77,7 @@ pub(crate) fn inlist_filter(
         let array = RawExpr::Constant {
             span: None,
             scalar: build_column.as_scalar().unwrap().to_owned(),
+            data_type: None,
         };
 
         let args = vec![array, raw_probe_key];
@@ -120,6 +121,7 @@ pub(crate) fn dedup_build_key_column(
         list.push(RawExpr::Constant {
             span: None,
             scalar: value.to_owned(),
+            data_type: None,
         })
     }
     let array = RawExpr::FunctionCall {
@@ -245,10 +247,12 @@ pub(crate) fn min_max_filter(
         let min = RawExpr::Constant {
             span: None,
             scalar: min,
+            data_type: None,
         };
         let max = RawExpr::Constant {
             span: None,
             scalar: max,
+            data_type: None,
         };
         // Make gte and lte function
         let gte_func = RawExpr::FunctionCall {

src/query/service/tests/it/storages/fuse/statistics.rs

@@ -385,6 +385,7 @@ async fn test_ft_cluster_stats_with_stats() -> databend_common_exception::Result
             RawExpr::Constant {
                 span: None,
                 scalar: Scalar::Number(NumberScalar::UInt64(1)),
+                data_type: None,
             },
         ],
     };

src/query/sql/src/executor/physical_plans/physical_window.rs

@@ -220,6 +220,7 @@ impl PhysicalPlanBuilder {
                     expr: Box::new(RawExpr::Constant {
                         span: w.span,
                         scalar: scalar.clone(),
+                        data_type: None,
                     }),
                     dest_type: common_ty.clone(),
                 };

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

@@ -135,7 +135,7 @@ impl Binder {
                 ..
             }) = item.scalar.clone()
             {
-                if typ == SubqueryType::Any || typ == SubqueryType::Exists {
+                if typ == SubqueryType::Any {
                     ScalarExpr::SubqueryExpr(SubqueryExpr {
                         span,
                         typ,
@@ -172,14 +172,19 @@ impl Binder {
         child: SExpr,
     ) -> Result<SExpr> {
         bind_context.set_expr_context(ExprContext::SelectClause);
-
+        let mut columns = columns.to_vec();
         let mut scalars = scalars
             .iter()
             .map(|(_, item)| {
                 if bind_context.in_grouping {
                     let mut scalar = item.scalar.clone();
                     let mut grouping_checker = GroupingChecker::new(bind_context);
                     grouping_checker.visit(&mut scalar)?;
+
+                    if let Some(x) = columns.iter_mut().find(|x| x.index == item.index) {
+                        x.data_type = Box::new(scalar.data_type()?);
+                    }
+
                     Ok(ScalarItem {
                         scalar,
                         index: item.index,
@@ -198,12 +203,9 @@ impl Binder {
 
         scalars.sort_by_key(|s| s.index);
         let eval_scalar = EvalScalar { items: scalars };
-
         let new_expr = SExpr::create_unary(Arc::new(eval_scalar.into()), Arc::new(child));
-
         // Set output columns
-        bind_context.columns = columns.to_vec();
-
+        bind_context.columns = columns;
         Ok(new_expr)
     }
 

src/query/sql/src/planner/format/display.rs

@@ -243,6 +243,7 @@ pub fn format_scalar(scalar: &ScalarExpr) -> String {
             }
         }
         ScalarExpr::ConstantExpr(constant) => constant.value.to_string(),
+        ScalarExpr::TypedConstantExpr(constant, _) => constant.value.to_string(),
         ScalarExpr::WindowFunction(win) => win.display_name.clone(),
         ScalarExpr::AggregateFunction(agg) => agg.display_name.clone(),
         ScalarExpr::LambdaFunction(lambda) => {