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

Author: dantengsky

outer-join-empty-cardinality-memo.md

@@ -0,0 +1,75 @@
+# Outer Join 空基数规划问题备忘录
+
+## 范围与脱敏说明
+
+本文记录一个位于 `UNION ALL` 与顶层 `LIMIT` 下方的 Outer Join 优化器回归。文中不包含原始
+SQL、Schema 名称、业务字段、数据规模和性能测量数据。
+
+## 问题现象
+
+两个版本生成了语义等价、执行成本却不同的 Hash Join：
+
+- 快计划把 `LEFT OUTER JOIN` 交换成 `RIGHT OUTER JOIN`，将能够低成本确定为空的输入放在
+  Hash Build 侧。
+- 慢计划保留 `LEFT OUTER JOIN`，将昂贵输入放在 Hash Build 侧；执行器完成 Build 后，才发现
+  Preserve/Probe 输入为空。
+
+Hash Build 是阻塞阶段，因此外层 `LIMIT` 无法提前终止这部分工作。
+
+## 证据与根因
+
+两个版本的 Join 交换规则本身没有实质差异。规则按左右子树的输出基数决定 Build 方向；回归发生
+在两侧基数同时坍缩为零时。
+
+新版选择率估算器在内部已经区分：
+
+- 由常量、布尔规则或列域矛盾确定的空集；
+- 普通数值估算得到的零。
+
+但生成算子统计信息时，这个区别丢失了，两者都只剩 `cardinality = 0.0`。`CommuteJoin` 因而看到
+无法区分的平局，并保留了不利的 Build 方向。
+
+诊断还发现，缺少频率统计时，严重倾斜的等值谓词可能被当作近似唯一值。这会进一步恶化基数，
+但不能通过特殊处理某个业务值或假设统一选择率来安全修复。
+
+## 修复设计
+
+使用 `precise_cardinality = Some(0)` 保留“确定为空”的信息：
+
+1. `SelectivityEstimator` 暴露零是否来自确定性规则，例如精确空输入、常量 False 或列域矛盾。
+2. `Filter` 和 `Scan` 将确定空集保存为精确零；普通估算零继续保持非精确。
+3. `Join` 按 Join 语义传播精确零。
+4. 普通 `LEFT/RIGHT OUTER JOIN` 两侧同为零、且仅一侧为精确零时，`CommuteJoin` 将精确空侧放到
+   Build 侧。
+
+非零估算、两个非精确零和两个精确零仍保持原行为。
+
+## 未采用的方案
+
+- 把 `<` 改为 `<=`：会翻转所有零基数平局，可能让镜像场景从 Build 空小侧退化为 Build 大侧。
+- 比较底表总行数：无法反映过滤和裁剪后的成本，可能选择相反方向。
+- 特殊处理哨兵值：依赖业务数据，无法泛化。
+- 要求立即分析整张大表：操作成本高，而且不能保证缺失或陈旧统计下的健壮性。
+
+## 测试覆盖
+
+- 确定性 False 谓词会标记精确空集。
+- 数值估算零不会被提升为精确空集。
+- Outer Join 只从 Preserve 侧传播精确空集。
+- 零基数平局时，精确空侧被放到 Build。
+- 精确空侧本来就在 Build 时不交换。
+- 原有零值 canonicalization 保持不变。
+
+验证结果：
+
+- `databend-common-sql` 单元测试全部通过。
+- SQL crate 轻量 planner/integration 测试全部通过。
+- SQL crate `clippy -D warnings` 通过。
+- 本次修改涉及的 Rust 文件通过格式和 diff 检查。
+
+## 后续方向
+
+- 为推导 NDV 保留置信度和来源。
+- 为倾斜列维护 Top-N/频率统计。
+- 将裁剪后的扫描工作量纳入 Join Cost，而非只依赖输出基数。
+- 研究 Preserve 输入可能低成本为空时的自适应 Outer Join 执行。

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(())
+    }
+}