chore: refine cte profile (#18467)

* chore: refine cte profile

* chore: add setting

* make lint

Author: SkyFan2002

src/query/settings/src/settings_default.rs

@@ -228,6 +228,13 @@ impl DefaultSettings {
                     scope: SettingScope::Both,
                     range: Some(SettingRange::Numeric(1..=1024)),
                 }),
+                ("grouping_sets_channel_size", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(2),
+                    desc: "Sets the channel size for grouping sets to union transformation.",
+                    mode: SettingMode::Both,
+                    scope: SettingScope::Both,
+                    range: Some(SettingRange::Numeric(1..=1024)),
+                }),
                 ("max_storage_io_requests", DefaultSettingValue {
                     value: UserSettingValue::UInt64(default_max_storage_io_requests),
                     desc: "Sets the maximum number of concurrent storage I/O requests.",

src/query/settings/src/settings_getter_setter.rs

@@ -877,6 +877,16 @@ impl Settings {
         self.try_get_u64("max_spill_io_requests")
     }
 
+    // Get grouping_sets_channel_size.
+    pub fn get_grouping_sets_channel_size(&self) -> Result<u64> {
+        self.try_get_u64("grouping_sets_channel_size")
+    }
+
+    // Set grouping_sets_channel_size.
+    pub fn set_grouping_sets_channel_size(&self, val: u64) -> Result<()> {
+        self.try_set_u64("grouping_sets_channel_size", val)
+    }
+
     pub fn get_short_sql_max_length(&self) -> Result<u64> {
         self.try_get_u64("short_sql_max_length")
     }

src/query/sql/src/executor/physical_plan.rs

@@ -1018,6 +1018,9 @@ impl PhysicalPlan {
                 .iter()
                 .map(|x| x.display_name.clone())
                 .join(", "),
+            PhysicalPlan::MaterializedCTE(v) => format!("CTE({})", v.cte_name),
+            PhysicalPlan::MaterializeCTERef(v) => format!("CTE_REF({})", v.cte_name),
+
             _ => String::new(),
         })
     }

src/query/sql/src/planner/optimizer/optimizers/rule/agg_rules/rule_grouping_sets_to_union.rs

@@ -27,6 +27,7 @@ use crate::optimizer::ir::SExpr;
 use crate::optimizer::optimizers::rule::Rule;
 use crate::optimizer::optimizers::rule::RuleID;
 use crate::optimizer::optimizers::rule::TransformResult;
+use crate::optimizer::OptimizerContext;
 use crate::plans::walk_expr_mut;
 use crate::plans::Aggregate;
 use crate::plans::AggregateMode;
@@ -62,10 +63,11 @@ const ID: RuleID = RuleID::GroupingSetsToUnion;
 pub struct RuleGroupingSetsToUnion {
     id: RuleID,
     matchers: Vec<Matcher>,
+    ctx: Arc<OptimizerContext>,
 }
 
 impl RuleGroupingSetsToUnion {
-    pub fn new() -> Self {
+    pub fn new(ctx: Arc<OptimizerContext>) -> Self {
         Self {
             id: ID,
             //  Aggregate
@@ -78,6 +80,7 @@ impl RuleGroupingSetsToUnion {
                     children: vec![Matcher::Leaf],
                 }],
             }],
+            ctx,
         }
     }
 }
@@ -112,8 +115,15 @@ impl Rule for RuleGroupingSetsToUnion {
                 let hash = hasher.finish();
                 let temp_cte_name = format!("cte_groupingsets_{hash}");
 
+                let channel_size = self
+                    .ctx
+                    .get_table_ctx()
+                    .get_settings()
+                    .get_grouping_sets_channel_size()
+                    .unwrap_or(2);
+
                 let cte_materialized_sexpr = SExpr::create_unary(
-                    MaterializedCTE::new(temp_cte_name.clone(), None, Some(1)),
+                    MaterializedCTE::new(temp_cte_name.clone(), None, Some(channel_size as usize)),
                     agg_input.clone(),
                 );
 
@@ -211,12 +221,6 @@ impl Rule for RuleGroupingSetsToUnion {
     }
 }
 
-impl Default for RuleGroupingSetsToUnion {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
 struct ReplaceColumnForGroupingSetsVisitor {
     group_indexes: Vec<IndexType>,
     exclude_group_indexes: Vec<IndexType>,

src/query/sql/src/planner/optimizer/optimizers/rule/factory.rs

@@ -102,7 +102,7 @@ impl RuleFactory {
             RuleID::MergeEvalScalar => Ok(Box::new(RuleMergeEvalScalar::new())),
             RuleID::MergeFilter => Ok(Box::new(RuleMergeFilter::new())),
             RuleID::NormalizeScalarFilter => Ok(Box::new(RuleNormalizeScalarFilter::new())),
-            RuleID::GroupingSetsToUnion => Ok(Box::new(RuleGroupingSetsToUnion::new())),
+            RuleID::GroupingSetsToUnion => Ok(Box::new(RuleGroupingSetsToUnion::new(ctx))),
             RuleID::SplitAggregate => Ok(Box::new(RuleSplitAggregate::new())),
             RuleID::FoldCountAggregate => Ok(Box::new(RuleFoldCountAggregate::new())),
             RuleID::CommuteJoin => Ok(Box::new(RuleCommuteJoin::new())),