refine code and fix tests

Author: xudong963

src/query/service/src/sql/planner/binder/join.rs

@@ -27,7 +27,7 @@ use common_exception::Result;
 use crate::sessions::TableContext;
 use crate::sql::binder::scalar_common::split_conjunctions;
 use crate::sql::binder::scalar_common::split_equivalent_predicate;
-use crate::sql::binder::scalar_common::wrap_cast_if_needed;
+use crate::sql::binder::wrap_cast;
 use crate::sql::binder::Visibility;
 use crate::sql::normalize_identifier;
 use crate::sql::optimizer::ColumnSet;
@@ -450,8 +450,10 @@ impl<'a> JoinConditionResolver<'a> {
         let right_type = right.data_type();
         if left_type.ne(&right_type) {
             let least_super_type = compare_coercion(&left_type, &right_type)?;
-            left = wrap_cast_if_needed(left, &least_super_type);
-            right = wrap_cast_if_needed(right, &least_super_type);
+            // Wrap cast for both left and right, `cast` can change the physical type of the data block
+            // Related issue: https://github.com/datafuselabs/databend/issues/7650
+            left = wrap_cast(left, &least_super_type);
+            right = wrap_cast(right, &least_super_type);
         }
 
         if left_used_columns.is_subset(&left_columns)

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

@@ -86,6 +86,19 @@ pub fn split_equivalent_predicate(scalar: &Scalar) -> Option<(Scalar, Scalar)> {
 }
 
 pub fn wrap_cast_if_needed(scalar: Scalar, target_type: &DataTypeImpl) -> Scalar {
+    if scalar.data_type() != *target_type {
+        let cast = CastExpr {
+            from_type: Box::new(scalar.data_type()),
+            argument: Box::new(scalar),
+            target_type: Box::new(target_type.clone()),
+        };
+        cast.into()
+    } else {
+        scalar
+    }
+}
+
+pub fn wrap_cast(scalar: Scalar, target_type: &DataTypeImpl) -> Scalar {
     CastExpr {
         from_type: Box::new(scalar.data_type()),
         argument: Box::new(scalar),

src/query/service/src/sql/planner/optimizer/heuristic/decorrelate.rs

@@ -21,7 +21,7 @@ use common_datavalues::NullableType;
 use common_exception::ErrorCode;
 use common_exception::Result;
 
-use crate::sql::binder::wrap_cast_if_needed;
+use crate::sql::binder::wrap_cast;
 use crate::sql::binder::JoinPredicate;
 use crate::sql::binder::Visibility;
 use crate::sql::optimizer::heuristic::subquery_rewriter::FlattenInfo;
@@ -172,9 +172,14 @@ impl SubqueryRewriter {
                 }
 
                 JoinPredicate::Both { left, right } => {
+                    if left.data_type().eq(&right.data_type()) {
+                        left_conditions.push(left.clone());
+                        right_conditions.push(right.clone());
+                        continue;
+                    }
                     let join_type = compare_coercion(&left.data_type(), &right.data_type())?;
-                    let left = wrap_cast_if_needed(left.clone(), &join_type);
-                    let right = wrap_cast_if_needed(right.clone(), &join_type);
+                    let left = wrap_cast(left.clone(), &join_type);
+                    let right = wrap_cast(right.clone(), &join_type);
                     left_conditions.push(left);
                     right_conditions.push(right);
                 }

src/query/service/src/sql/planner/optimizer/rule/rewrite/rule_push_down_filter_join.rs

@@ -15,7 +15,7 @@
 use common_datavalues::type_coercion::compare_coercion;
 use common_exception::Result;
 
-use crate::sql::binder::wrap_cast_if_needed;
+use crate::sql::binder::wrap_cast;
 use crate::sql::binder::JoinPredicate;
 use crate::sql::optimizer::rule::Rule;
 use crate::sql::optimizer::rule::TransformState;
@@ -117,10 +117,15 @@ impl Rule for RulePushDownFilterJoin {
                         if join.join_type == JoinType::Cross {
                             join.join_type = JoinType::Inner;
                         }
-                        let left = wrap_cast_if_needed(left.clone(), &join_key_type);
-                        let right = wrap_cast_if_needed(right.clone(), &join_key_type);
-                        join.left_conditions.push(left);
-                        join.right_conditions.push(right);
+                        if left.data_type().ne(&right.data_type()) {
+                            let left = wrap_cast(left.clone(), &join_key_type);
+                            let right = wrap_cast(right.clone(), &join_key_type);
+                            join.left_conditions.push(left);
+                            join.right_conditions.push(right);
+                        } else {
+                            join.left_conditions.push(left.clone());
+                            join.right_conditions.push(right.clone());
+                        }
                         need_push = true;
                     } else {
                         original_predicates.push(predicate);

tests/logictest/suites/mode/cluster/exchange.test

@@ -129,7 +129,7 @@ Exchange
 └── HashJoin
     ├── join type: INNER
     ├── build keys: [CAST(numbers.number (#4) AS BIGINT UNSIGNED NULL)]
-    ├── probe keys: [number (#1)]
+    ├── probe keys: [CAST(number (#1) AS BIGINT UNSIGNED NULL)]
     ├── filters: []
     ├── Exchange(Build)
     │   ├── exchange type: Hash(CAST(numbers.number (#4) AS BIGINT UNSIGNED NULL))

tests/logictest/suites/mode/standalone/explain/limit.test

@@ -148,7 +148,7 @@ Limit
     ├── sort keys: [c1 ASC NULLS LAST]
     └── HashJoin
         ├── join type: LEFT OUTER
-        ├── build keys: [c (#5)]
+        ├── build keys: [CAST(c (#5) AS BIGINT UNSIGNED NULL)]
         ├── probe keys: [CAST(c1 (#1) AS BIGINT UNSIGNED NULL)]
         ├── filters: []
         ├── Project(Build)

tests/logictest/suites/mode/standalone/explain/select_limit_offset.test

@@ -141,7 +141,7 @@ Limit
 ├── offset: 1
 └── HashJoin
     ├── join type: LEFT OUTER
-    ├── build keys: [t1.a (#1)]
+    ├── build keys: [CAST(t1.a (#1) AS INT NULL)]
     ├── probe keys: [CAST(t.a (#0) AS INT NULL)]
     ├── filters: []
     ├── Limit(Build)