fix(query): match nullable scalar correlation keys

Author: youngsofun

src/query/sql/src/planner/optimizer/optimizers/hyper_dp/dphyp.rs

@@ -27,7 +27,6 @@ use super::algorithm::JoinEdgeRef;
 use super::algorithm::JoinNode;
 use super::algorithm::JoinOrderModel;
 use crate::IndexType;
-use crate::ScalarExpr;
 use crate::optimizer::Optimizer;
 use crate::optimizer::OptimizerContext;
 use crate::optimizer::ir::RelExpr;
@@ -59,7 +58,7 @@ pub struct DPhpyOptimizer {
 
 struct DPhypJoinOrderModel<'a> {
     join_relations: &'a [JoinRelation],
-    join_conditions: &'a [(ScalarExpr, ScalarExpr)],
+    join_conditions: &'a [JoinEquiCondition],
 }
 
 impl DPhypJoinOrderModel<'_> {
@@ -71,17 +70,14 @@ impl DPhypJoinOrderModel<'_> {
     ) -> SExpr {
         let left_expr = left.state().clone();
         let right_expr = right.state().clone();
-        let mut left_conditions = Vec::with_capacity(edge_refs.len());
-        let mut right_conditions = Vec::with_capacity(edge_refs.len());
+        let mut conditions = Vec::with_capacity(edge_refs.len());
 
         for edge_ref in edge_refs {
-            let (mut left_condition, mut right_condition) =
-                self.join_conditions[edge_ref.id].clone();
+            let mut condition = self.join_conditions[edge_ref.id].clone();
             if edge_ref.reversed {
-                std::mem::swap(&mut left_condition, &mut right_condition);
+                std::mem::swap(&mut condition.left, &mut condition.right);
             }
-            left_conditions.push(left_condition);
-            right_conditions.push(right_condition);
+            conditions.push(condition);
         }
 
         let join_type = if edge_refs.is_empty() {
@@ -90,11 +86,7 @@ impl DPhypJoinOrderModel<'_> {
             JoinType::Inner
         };
         let rel_op = RelOperator::Join(Join {
-            equi_conditions: JoinEquiCondition::new_conditions(
-                left_conditions,
-                right_conditions,
-                vec![],
-            ),
+            equi_conditions: conditions,
             non_equi_conditions: vec![],
             join_type,
             marker_index: None,
@@ -269,7 +261,7 @@ impl DPhpyOptimizer {
     async fn process_join_node(
         &mut self,
         s_expr: &SExpr,
-        join_conditions: &mut Vec<(ScalarExpr, ScalarExpr)>,
+        join_conditions: &mut Vec<JoinEquiCondition>,
     ) -> Result<(Arc<SExpr>, bool)> {
         let op = match s_expr.plan() {
             RelOperator::Join(op) => op,
@@ -304,7 +296,7 @@ impl DPhpyOptimizer {
                 break;
             }
 
-            join_conditions.push((condition.left.clone(), condition.right.clone()));
+            join_conditions.push(condition.clone());
         }
 
         // Add non-equi conditions to filters
@@ -571,7 +563,7 @@ impl DPhpyOptimizer {
     async fn process_unary_node(
         &mut self,
         s_expr: &SExpr,
-        join_conditions: &mut Vec<(ScalarExpr, ScalarExpr)>,
+        join_conditions: &mut Vec<JoinEquiCondition>,
         join_child: bool,
         join_relation: Option<&SExpr>,
     ) -> Result<(Arc<SExpr>, bool)> {
@@ -611,7 +603,7 @@ impl DPhpyOptimizer {
     async fn get_base_relations(
         &mut self,
         s_expr: &SExpr,
-        join_conditions: &mut Vec<(ScalarExpr, ScalarExpr)>,
+        join_conditions: &mut Vec<JoinEquiCondition>,
         join_child: bool,
         join_relation: Option<&SExpr>,
         is_subquery: bool,
@@ -673,7 +665,6 @@ impl DPhpyOptimizer {
         }
 
         // Firstly, we need to extract all join conditions and base tables
-        // `join_condition` is pair, left is left_condition, right is right_condition
         let mut join_conditions = vec![];
         let (s_expr, optimized) = self
             .get_base_relations(s_expr, &mut join_conditions, false, None, false)
@@ -715,18 +706,18 @@ impl DPhpyOptimizer {
     fn build_join_order_edges(
         &self,
         hyper_dp: &mut HyperDp<'_, DPhypJoinOrderModel<'_>>,
-        join_conditions: &[(ScalarExpr, ScalarExpr)],
+        join_conditions: &[JoinEquiCondition],
     ) -> Result<bool> {
-        for (edge_id, (left_condition, right_condition)) in join_conditions.iter().enumerate() {
+        for (edge_id, condition) in join_conditions.iter().enumerate() {
             let mut left_relation_set = HashSet::new();
             let mut right_relation_set = HashSet::new();
 
-            let left_used_tables = left_condition.used_tables()?;
+            let left_used_tables = condition.left.used_tables()?;
             for table in left_used_tables.iter() {
                 left_relation_set.insert(self.table_index_map[table]);
             }
 
-            let right_used_tables = right_condition.used_tables()?;
+            let right_used_tables = condition.right.used_tables()?;
             for table in right_used_tables.iter() {
                 right_relation_set.insert(self.table_index_map[table]);
             }
@@ -904,7 +895,7 @@ mod tests {
     use crate::plans::MaterializedCTERef;
     use crate::plans::Sequence;
 
-    fn bool_constant(value: bool) -> ScalarExpr {
+    fn bool_constant(value: bool) -> crate::ScalarExpr {
         ConstantExpr {
             span: None,
             value: Scalar::Boolean(value),

src/query/sql/src/planner/optimizer/optimizers/operator/decorrelate/decorrelate.rs

@@ -275,6 +275,12 @@ impl SubqueryDecorrelatorOptimizer {
                     &mut left_conditions,
                 )?;
 
+                // These conditions reconnect the flattened subquery result to the outer row.
+                // They are internal correlation keys rather than user-written equality
+                // predicates, so NULL correlation groups must match each other.
+                let is_null_equal =
+                    Self::nullable_condition_indexes(&left_conditions, &right_conditions);
+
                 let join_type = if matches!(subquery.contain_agg, Some(true)) && {
                     let rel_expr = RelExpr::with_s_expr(&subquery.subquery);
                     rel_expr
@@ -292,7 +298,7 @@ impl SubqueryDecorrelatorOptimizer {
                     equi_conditions: JoinEquiCondition::new_conditions(
                         left_conditions,
                         right_conditions,
-                        vec![],
+                        is_null_equal,
                     ),
                     non_equi_conditions: vec![],
                     join_type,
@@ -339,16 +345,8 @@ impl SubqueryDecorrelatorOptimizer {
                     &mut left_conditions,
                     &mut right_conditions,
                 )?;
-                let mut is_null_equal = Vec::new();
-                for (i, (l, r)) in left_conditions
-                    .iter()
-                    .zip(right_conditions.iter())
-                    .enumerate()
-                {
-                    if l.data_type().is_nullable() || r.data_type().is_nullable() {
-                        is_null_equal.push(i);
-                    }
-                }
+                let is_null_equal =
+                    Self::nullable_condition_indexes(&left_conditions, &right_conditions);
 
                 let marker_index = if let Some(idx) = subquery.projection_index {
                     idx
@@ -404,16 +402,8 @@ impl SubqueryDecorrelatorOptimizer {
                     &mut right_conditions,
                 )?;
 
-                let mut is_null_equal = Vec::new();
-                for (i, (l, r)) in left_conditions
-                    .iter()
-                    .zip(right_conditions.iter())
-                    .enumerate()
-                {
-                    if l.data_type().is_nullable() || r.data_type().is_nullable() {
-                        is_null_equal.push(i);
-                    }
-                }
+                let is_null_equal =
+                    Self::nullable_condition_indexes(&left_conditions, &right_conditions);
 
                 let output_column = subquery.output_column.clone();
                 let column_name = format!("subquery_{}", output_column.index);
@@ -522,6 +512,20 @@ impl SubqueryDecorrelatorOptimizer {
         Ok(())
     }
 
+    pub(crate) fn nullable_condition_indexes(
+        left_conditions: &[ScalarExpr],
+        right_conditions: &[ScalarExpr],
+    ) -> Vec<usize> {
+        left_conditions
+            .iter()
+            .zip(right_conditions)
+            .enumerate()
+            .filter_map(|(index, (left, right))| {
+                (left.data_type().is_nullable() || right.data_type().is_nullable()).then_some(index)
+            })
+            .collect()
+    }
+
     // Check if need to join outer and inner table
     // If correlated_columns only occur in equi-conditions, such as `where t1.a = t.a and t1.b = t.b`(t1 is outer table)
     // Then we won't join outer and inner table.

src/query/sql/src/planner/optimizer/optimizers/operator/decorrelate/subquery_decorrelator.rs

@@ -736,16 +736,8 @@ impl SubqueryDecorrelatorOptimizer {
                     )
                 };
 
-                let mut is_null_equal = Vec::new();
-                for (i, (l, r)) in left_conditions
-                    .iter()
-                    .zip(right_conditions.iter())
-                    .enumerate()
-                {
-                    if l.data_type().is_nullable() || r.data_type().is_nullable() {
-                        is_null_equal.push(i);
-                    }
-                }
+                let is_null_equal =
+                    Self::nullable_condition_indexes(&left_conditions, &right_conditions);
 
                 // Consider the sql: select * from t1 where t1.a = any(select t2.a from t2);
                 // Will be transferred to:select t1.a, t2.a, marker_index from t1, t2 where t2.a = t1.a;