fix(query): prefer proven-empty outer join build side

src/query/sql/src/planner/optimizer/ir/stats/selectivity.rs

@@ -65,6 +65,7 @@ pub struct SelectivityEstimator {
     top_n: TopNSet,
     count_min_sketch: CountMinSketchSet,
     overrides: ColumnStatSet,
+    proven_empty: bool,
 }
 
 impl SelectivityEstimator {
@@ -75,6 +76,7 @@ impl SelectivityEstimator {
             top_n: TopNSet::new(),
             count_min_sketch: CountMinSketchSet::new(),
             overrides: ColumnStatSet::new(),
+            proven_empty: cardinality == StatCardinality::Exact(0),
         }
     }
 
@@ -88,6 +90,13 @@ impl SelectivityEstimator {
         self
     }
 
+    /// Returns true when the predicates deterministically produce no rows.
+    ///
+    /// This is deliberately stronger than an estimated cardinality of zero.
+    pub fn is_proven_empty(&self) -> bool {
+        self.proven_empty
+    }
+
     fn merged_column_stats(&self) -> ColumnStatSet {
         let mut merged = self.column_stats.clone();
         merged.extend(self.overrides.clone());
@@ -139,6 +148,7 @@ impl SelectivityEstimator {
             return match constant_filter_truthiness(&constant.scalar) {
                 Some(true) => Ok(self.cardinality.value()),
                 Some(false) => {
+                    self.proven_empty = true;
                     self.clear_column_stats_for_empty_result();
                     Ok(0.0)
                 }
@@ -159,6 +169,7 @@ impl SelectivityEstimator {
             }) => !domain.has_true,
             _ => false,
         }) {
+            self.proven_empty = true;
             self.clear_column_stats_for_empty_result();
             return Ok(0.0);
         }
@@ -263,6 +274,7 @@ impl SelectivityEstimator {
             Selectivity::Unknown => DEFAULT_SELECTIVITY,
             Selectivity::LowerBound => UNKNOWN_COL_STATS_FILTER_SEL_LOWER_BOUND,
             Selectivity::Zero => {
+                self.proven_empty = true;
                 self.clear_column_stats_for_empty_result();
                 return 0.0;
             }

src/query/sql/src/planner/optimizer/optimizers/rule/join_rules/rule_commute_join.rs

@@ -19,6 +19,7 @@ use databend_common_exception::Result;
 use crate::optimizer::ir::Matcher;
 use crate::optimizer::ir::RelExpr;
 use crate::optimizer::ir::SExpr;
+use crate::optimizer::ir::StatInfo;
 use crate::optimizer::optimizers::rule::Rule;
 use crate::optimizer::optimizers::rule::RuleID;
 use crate::optimizer::optimizers::rule::TransformResult;
@@ -32,6 +33,45 @@ fn contains_recursive_cte(expr: &SExpr) -> bool {
         || expr.children().any(contains_recursive_cte)
 }
 
+fn should_commute(join_type: JoinType, left: &StatInfo, right: &StatInfo) -> bool {
+    if left.cardinality < right.cardinality {
+        return matches!(
+            join_type,
+            JoinType::Inner
+                | JoinType::Cross
+                | JoinType::Left
+                | JoinType::Right
+                | JoinType::LeftSingle
+                | JoinType::RightSingle
+                | JoinType::LeftSemi
+                | JoinType::RightSemi
+                | JoinType::LeftAnti
+                | JoinType::RightAnti
+                | JoinType::LeftMark
+                | JoinType::RightMark
+        );
+    }
+
+    if left.cardinality != right.cardinality {
+        return false;
+    }
+
+    if left.cardinality == 0.0 && matches!(join_type, JoinType::Left | JoinType::Right) {
+        let left_proven_empty = left.statistics.precise_cardinality == Some(0);
+        let right_proven_empty = right.statistics.precise_cardinality == Some(0);
+        if left_proven_empty != right_proven_empty {
+            // The right child is the hash-build side. Prefer the input that is
+            // known to be empty over one whose zero cardinality is only estimated.
+            return left_proven_empty;
+        }
+    }
+
+    matches!(
+        join_type,
+        JoinType::Right | JoinType::RightSingle | JoinType::RightSemi | JoinType::RightAnti
+    )
+}
+
 /// Rule to apply commutativity of join operator.
 /// Since we will always use the right child as build side, this
 /// rule will help us measure which child is the better one.
@@ -79,33 +119,9 @@ impl Rule for RuleCommuteJoin {
 
         let left_rel_expr = RelExpr::with_s_expr(left_child);
         let right_rel_expr = RelExpr::with_s_expr(right_child);
-        let left_card = left_rel_expr.derive_cardinality()?.cardinality;
-        let right_card = right_rel_expr.derive_cardinality()?.cardinality;
-
-        let need_commute = if left_card < right_card {
-            matches!(
-                join.join_type,
-                JoinType::Inner
-                    | JoinType::Cross
-                    | JoinType::Left
-                    | JoinType::Right
-                    | JoinType::LeftSingle
-                    | JoinType::RightSingle
-                    | JoinType::LeftSemi
-                    | JoinType::RightSemi
-                    | JoinType::LeftAnti
-                    | JoinType::RightAnti
-                    | JoinType::LeftMark
-                    | JoinType::RightMark
-            )
-        } else if left_card == right_card {
-            matches!(
-                join.join_type,
-                JoinType::Right | JoinType::RightSingle | JoinType::RightSemi | JoinType::RightAnti
-            )
-        } else {
-            false
-        };
+        let left_stat = left_rel_expr.derive_cardinality()?;
+        let right_stat = right_rel_expr.derive_cardinality()?;
+        let need_commute = should_commute(join.join_type, &left_stat, &right_stat);
         if need_commute {
             // Swap the join conditions side
             for condition in join.equi_conditions.iter_mut() {
@@ -135,3 +151,123 @@ impl Default for RuleCommuteJoin {
         Self::new()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use databend_common_expression::Scalar;
+
+    use super::*;
+    use crate::optimizer::ir::Statistics;
+    use crate::plans::ConstantExpr;
+    use crate::plans::DummyTableScan;
+    use crate::plans::Filter;
+    use crate::plans::MutationSource;
+    use crate::plans::ScalarExpr;
+
+    fn empty_stat(precise: bool) -> StatInfo {
+        StatInfo {
+            cardinality: 0.0,
+            statistics: Statistics {
+                precise_cardinality: precise.then_some(0),
+                column_stats: Default::default(),
+                top_n: Default::default(),
+                count_min_sketch: Default::default(),
+            },
+        }
+    }
+
+    fn proven_empty_expr() -> SExpr {
+        SExpr::create_unary(
+            Filter {
+                predicates: vec![ScalarExpr::ConstantExpr(ConstantExpr {
+                    span: None,
+                    value: Scalar::Boolean(false),
+                })],
+            },
+            SExpr::create_leaf(DummyTableScan::default()),
+        )
+    }
+
+    #[test]
+    fn test_outer_join_zero_tie_prefers_proven_empty_build_side() {
+        let proven_empty = empty_stat(true);
+        let estimated_empty = empty_stat(false);
+
+        assert!(should_commute(
+            JoinType::Left,
+            &proven_empty,
+            &estimated_empty
+        ));
+        assert!(!should_commute(
+            JoinType::Left,
+            &estimated_empty,
+            &proven_empty
+        ));
+        assert!(should_commute(
+            JoinType::Right,
+            &proven_empty,
+            &estimated_empty
+        ));
+        assert!(!should_commute(
+            JoinType::Right,
+            &estimated_empty,
+            &proven_empty
+        ));
+    }
+
+    #[test]
+    fn test_outer_join_zero_tie_preserves_existing_canonicalization() {
+        let left = empty_stat(false);
+        let right = empty_stat(false);
+
+        assert!(!should_commute(JoinType::Left, &left, &right));
+        assert!(should_commute(JoinType::Right, &left, &right));
+    }
+
+    #[test]
+    fn test_commute_join_builds_proven_empty_input() -> Result<()> {
+        let join = Join {
+            join_type: JoinType::Left,
+            ..Default::default()
+        };
+        let expr = SExpr::create_binary(
+            join,
+            proven_empty_expr(),
+            SExpr::create_leaf(MutationSource::default()),
+        );
+        let mut state = TransformResult::new();
+
+        RuleCommuteJoin::new().apply(&expr, &mut state)?;
+
+        assert_eq!(state.results().len(), 1);
+        let result_join: Join = state.results()[0].plan().clone().try_into()?;
+        assert_eq!(result_join.join_type, JoinType::Right);
+        assert_eq!(
+            RelExpr::with_s_expr(state.results()[0].child(1)?)
+                .derive_cardinality()?
+                .statistics
+                .precise_cardinality,
+            Some(0)
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn test_commute_join_keeps_proven_empty_build_input() -> Result<()> {
+        let join = Join {
+            join_type: JoinType::Left,
+            ..Default::default()
+        };
+        let expr = SExpr::create_binary(
+            join,
+            SExpr::create_leaf(MutationSource::default()),
+            proven_empty_expr(),
+        );
+        let mut state = TransformResult::new();
+
+        RuleCommuteJoin::new().apply(&expr, &mut state)?;
+
+        assert!(state.results().is_empty());
+        Ok(())
+    }
+}

src/query/sql/src/planner/plans/filter.rs

@@ -98,6 +98,7 @@ impl Operator for Filter {
                 .with_top_n(stat_info.statistics.top_n.clone())
                 .with_count_min_sketch(stat_info.statistics.count_min_sketch.clone());
         let cardinality = sb.apply(&self.predicates)?;
+        let precise_cardinality = sb.is_proven_empty().then_some(0);
         // Derive column statistics
         let column_stats = if cardinality == 0.0 {
             HashMap::new()
@@ -107,11 +108,54 @@ impl Operator for Filter {
         Ok(Arc::new(StatInfo {
             cardinality,
             statistics: Statistics {
-                precise_cardinality: None,
+                precise_cardinality,
                 column_stats,
                 top_n: Default::default(),
                 count_min_sketch: Default::default(),
             },
         }))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use databend_common_expression::Scalar;
+
+    use super::*;
+    use crate::optimizer::ir::SExpr;
+    use crate::plans::ConstantExpr;
+    use crate::plans::DummyTableScan;
+    use crate::plans::MutationSource;
+
+    fn constant_filter(value: bool, input: SExpr) -> SExpr {
+        SExpr::create_unary(
+            Filter {
+                predicates: vec![ScalarExpr::ConstantExpr(ConstantExpr {
+                    span: None,
+                    value: Scalar::Boolean(value),
+                })],
+            },
+            input,
+        )
+    }
+
+    #[test]
+    fn test_filter_preserves_proven_empty_cardinality() -> Result<()> {
+        let expr = constant_filter(false, SExpr::create_leaf(DummyTableScan::default()));
+        let stat = RelExpr::with_s_expr(&expr).derive_cardinality()?;
+
+        assert_eq!(stat.cardinality, 0.0);
+        assert_eq!(stat.statistics.precise_cardinality, Some(0));
+        Ok(())
+    }
+
+    #[test]
+    fn test_filter_does_not_promote_estimated_zero_to_precise() -> Result<()> {
+        let expr = constant_filter(true, SExpr::create_leaf(MutationSource::default()));
+        let stat = RelExpr::with_s_expr(&expr).derive_cardinality()?;
+
+        assert_eq!(stat.cardinality, 0.0);
+        assert_eq!(stat.statistics.precise_cardinality, None);
+        Ok(())
+    }
+}