fix(binder): correct join statistics propagation (#20122)

* x

* fix(query): invalidate stats when DPhyp replaces joins

* fix(query): replace DPhyp join tree directly

* x

* x

* x

Author: forsaken628

src/query/sql/src/planner/binder/bind_context.rs

@@ -205,7 +205,7 @@ pub struct CteContext {
     pub cte_map: Box<IndexMap<String, CteInfo>>,
     /// Pending output columns generated by materialized CTE virtual-column rewrite.
     /// `init_cte` consumes this map and stores the per-CTE result in `CteInfo`.
-    pub virtual_column_outputs: HashMap<String, HashSet<String>>,
+    pub virtual_column_outputs: HashMap<String, BTreeSet<String>>,
 }
 
 impl CteContext {
@@ -241,7 +241,7 @@ impl CteContext {
 #[derive(Clone, Debug)]
 pub struct CteInfo {
     pub columns_alias: Vec<String>,
-    pub virtual_column_outputs: HashSet<String>,
+    pub virtual_column_outputs: BTreeSet<String>,
     pub query: Query,
     pub recursive: bool,
     pub logical_recursive_cte_id: Option<u32>,

src/query/sql/src/planner/binder/virtual_column.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::collections::BTreeMap;
+use std::collections::BTreeSet;
 use std::collections::HashMap;
 use std::collections::HashSet;
 
@@ -291,7 +292,7 @@ impl Binder {
         bind_context: &BindContext,
         with: &mut With,
         body: &mut SetExpr,
-    ) -> HashMap<String, HashSet<String>> {
+    ) -> HashMap<String, BTreeSet<String>> {
         if !bind_context.allow_virtual_column || with.recursive {
             return HashMap::new();
         }

src/query/sql/src/planner/format/display.rs

@@ -192,7 +192,10 @@ where
             ));
         }
 
-        if s_expr.plan.is_join() {
+        if let RelOperator::MaterializedCTERef(_) = op {
+            // MaterializedCTERef definitions are represented by their producer MaterializedCTE
+            // in the surrounding Sequence.
+        } else if s_expr.plan.is_join() {
             tree.children
                 .push(self.humanize_s_expr(s_expr.build_side_child())?);
             tree.children

src/query/sql/src/planner/format/display_rel_operator.rs

@@ -26,6 +26,8 @@ use crate::plans::Exchange;
 use crate::plans::Filter;
 use crate::plans::Join;
 use crate::plans::Limit;
+use crate::plans::MaterializedCTE;
+use crate::plans::MaterializedCTERef;
 use crate::plans::Mutation;
 use crate::plans::Operator;
 use crate::plans::RelOperator;
@@ -63,6 +65,10 @@ fn to_format_tree<I: IdHumanizer>(id_humanizer: &I, op: &RelOperator) -> FormatT
         RelOperator::ConstantTableScan(op) => constant_scan_to_format_tree(id_humanizer, op),
         RelOperator::UnionAll(op) => union_all_to_format_tree(id_humanizer, op),
         RelOperator::Mutation(op) => merge_into_to_format_tree(id_humanizer, op),
+        RelOperator::MaterializedCTE(op) => materialized_cte_to_format_tree(op),
+        RelOperator::MaterializedCTERef(op) => {
+            materialized_cte_ref_to_format_tree(id_humanizer, op)
+        }
         _ => FormatTreeNode::with_children(format!("{:?}", op), vec![]),
     }
 }
@@ -501,6 +507,44 @@ fn union_all_to_format_tree<I: IdHumanizer>(id_humanizer: &I, op: &UnionAll) ->
     FormatTreeNode::with_children(format!("{:?}", op.rel_op()), children)
 }
 
+fn materialized_cte_to_format_tree(op: &MaterializedCTE) -> FormatTreeNode {
+    FormatTreeNode::with_children("MaterializedCTE".to_string(), vec![
+        FormatTreeNode::new(format!("cte_name: {}", op.cte_name)),
+        FormatTreeNode::new(format!("ref_count: {}", op.ref_count)),
+        FormatTreeNode::new(format!("channel_size: {:?}", op.channel_size)),
+    ])
+}
+
+fn materialized_cte_ref_to_format_tree<I: IdHumanizer>(
+    id_humanizer: &I,
+    op: &MaterializedCTERef,
+) -> FormatTreeNode {
+    let output_columns = op
+        .output_columns
+        .iter()
+        .map(|idx| id_humanizer.humanize_column_id(*idx))
+        .join(", ");
+
+    let column_mapping = op
+        .column_mapping
+        .iter()
+        .sorted_by_key(|(from, to)| (**from, **to))
+        .map(|(from, to)| {
+            format!(
+                "{} -> {}",
+                id_humanizer.humanize_column_id(*from),
+                id_humanizer.humanize_column_id(*to)
+            )
+        })
+        .join(", ");
+
+    FormatTreeNode::with_children("MaterializedCTERef".to_string(), vec![
+        FormatTreeNode::new(format!("cte_name: {}", op.cte_name)),
+        FormatTreeNode::new(format!("output columns: [{}]", output_columns)),
+        FormatTreeNode::new(format!("column mapping: [{}]", column_mapping)),
+    ])
+}
+
 fn merge_into_to_format_tree<I: IdHumanizer>(
     id_humanizer: &I,
     merge_into: &Mutation,

src/query/sql/src/planner/optimizer/ir/expr/mod.rs

@@ -15,6 +15,7 @@
 mod extract;
 mod m_expr;
 mod s_expr;
+mod type_validator;
 mod visitor;
 
 pub use extract::Matcher;