fix(query): fix join subquery scalar rewrite

Author: sundy-li

benchmark/tpcds/load_data.sh

@@ -60,6 +60,7 @@ do
     echo "$t"
     fp="`pwd`/data/$t.csv"
     echo "copy into ${MYSQL_DATABASE}.$t from 'fs://${fp}' file_format = (type = CSV skip_header = 1 field_delimiter = '|' record_delimiter = '\n')" | $BENDSQL_CLIENT_CONNECT
+    echo "analyze table ${MYSQL_DATABASE}.$t" | $BENDSQL_CLIENT_CONNECT
 done
 
 

benchmark/tpch/load_data.sh

@@ -117,4 +117,5 @@ do
     echo "$t"
     fp="`pwd`/data/$t.tbl"
     echo "copy into ${MYSQL_DATABASE}.$t from 'fs://${fp}' file_format = (type = CSV skip_header = 1 field_delimiter = '|' record_delimiter = '\n')" | $BENDSQL_CLIENT_CONNECT
+    echo "analyze table ${MYSQL_DATABASE}.$t" | $BENDSQL_CLIENT_CONNECT
 done

src/common/storage/src/statistics.rs

@@ -182,4 +182,21 @@ impl Datum {
             ))),
         }
     }
+
+    pub fn can_compare(&self, other: &Self) -> bool {
+        match (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(_)) => true,
+            _ => false,
+        }
+    }
 }

src/query/sql/src/planner/optimizer/property/selectivity.rs

@@ -442,6 +442,10 @@ impl<'a> SelectivityEstimator<'a> {
             let lower_bound = bucket.lower_bound();
             let upper_bound = bucket.upper_bound();
 
+            if !const_datum.can_compare(&lower_bound) {
+                return Ok(DEFAULT_SELECTIVITY);
+            }
+
             let const_gte_upper_bound = matches!(
                 const_datum.compare(upper_bound)?,
                 Ordering::Greater | Ordering::Equal

src/query/sql/src/planner/semantic/type_check.rs

@@ -3169,10 +3169,10 @@ impl<'a> TypeChecker<'a> {
         if let SetExpr::Select(select_stmt) = &subquery.body {
             if typ == SubqueryType::Scalar {
                 let select = &select_stmt.select_list[0];
-                if let SelectTarget::AliasedExpr { expr, .. } = select {
+                if matches!(select, SelectTarget::AliasedExpr { .. }) {
                     // Check if contain aggregation function
                     #[derive(Visitor)]
-                    #[visitor(ASTFunctionCall(enter))]
+                    #[visitor(Expr(enter), ASTFunctionCall(enter))]
                     struct AggFuncVisitor {
                         contain_agg: bool,
                     }
@@ -3182,9 +3182,12 @@ impl<'a> TypeChecker<'a> {
                                 || AggregateFunctionFactory::instance()
                                     .contains(func.name.to_string());
                         }
+                        fn enter_expr(&mut self, expr: &Expr) {
+                            self.contain_agg = matches!(expr, Expr::CountAll { window: None, .. });
+                        }
                     }
                     let mut visitor = AggFuncVisitor { contain_agg: false };
-                    expr.drive(&mut visitor);
+                    select.drive(&mut visitor);
                     contain_agg = Some(visitor.contain_agg);
                 }
             }