Merge branch 'main' into dev-unescape-string

Author: mergify[bot]

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

@@ -430,6 +430,25 @@ impl PhysicalPlan {
         }
     }
 
+    pub fn name(&self) -> String {
+        match self {
+            PhysicalPlan::TableScan(_) => "TableScan".to_string(),
+            PhysicalPlan::Filter(_) => "Filter".to_string(),
+            PhysicalPlan::Project(_) => "Project".to_string(),
+            PhysicalPlan::EvalScalar(_) => "EvalScalar".to_string(),
+            PhysicalPlan::AggregatePartial(_) => "AggregatePartial".to_string(),
+            PhysicalPlan::AggregateFinal(_) => "AggregateFinal".to_string(),
+            PhysicalPlan::Sort(_) => "Sort".to_string(),
+            PhysicalPlan::Limit(_) => "Limit".to_string(),
+            PhysicalPlan::HashJoin(_) => "HashJoin".to_string(),
+            PhysicalPlan::Exchange(_) => "Exchange".to_string(),
+            PhysicalPlan::UnionAll(_) => "UnionAll".to_string(),
+            PhysicalPlan::DistributedInsertSelect(_) => "DistributedInsertSelect".to_string(),
+            PhysicalPlan::ExchangeSource(_) => "Exchange Source".to_string(),
+            PhysicalPlan::ExchangeSink(_) => "Exchange Sink".to_string(),
+        }
+    }
+
     pub fn children<'a>(&'a self) -> Box<dyn Iterator<Item = &'a PhysicalPlan> + 'a> {
         match self {
             PhysicalPlan::TableScan(_) => Box::new(std::iter::empty()),

src/query/sql/src/executor/physical_plan_builder.rs

@@ -356,7 +356,12 @@ impl PhysicalPlanBuilder {
                                 ..
                             }) => agg.input.output_schema()?,
 
-                            _ => unreachable!(),
+                            _ => {
+                                return Err(ErrorCode::Internal(format!(
+                                    "invalid input physical plan: {}",
+                                    input.name(),
+                                )));
+                            }
                         };
 
                         let agg_funcs: Vec<AggregateFunctionDesc> = agg.aggregate_functions.iter().map(|v| {
@@ -419,10 +424,17 @@ impl PhysicalPlanBuilder {
                                 })
                             }
 
-                            _ => unreachable!(),
+                            _ => {
+                                return Err(ErrorCode::Internal(format!(
+                                    "invalid input physical plan: {}",
+                                    input.name(),
+                                )));
+                            }
                         }
                     }
-                    AggregateMode::Initial => unreachable!(),
+                    AggregateMode::Initial => {
+                        return Err(ErrorCode::Internal("Invalid aggregate mode: Initial"));
+                    }
                 };
 
                 Ok(result)

src/query/sql/src/planner/optimizer/distributed/distributed.rs

@@ -50,6 +50,11 @@ pub fn optimize_distributed_query(ctx: Arc<dyn TableContext>, s_expr: &SExpr) ->
 // Traverse the SExpr tree to find top_k, if find, push down it to Exchange::Merge
 fn push_down_topk_to_merge(s_expr: &mut SExpr, mut top_k: Option<TopK>) -> Result<()> {
     if let RelOperator::Exchange(Exchange::Merge) = s_expr.plan {
+        // A quick fix for Merge child is aggregate.
+        // Todo: consider to push down topk to the above of aggregate.
+        if let RelOperator::Aggregate(_) = s_expr.child(0)?.plan {
+            return Ok(());
+        }
         if let Some(top_k) = top_k {
             let child = &mut s_expr.children[0];
             *child = SExpr::create_unary(top_k.sort.into(), child.clone());

tests/logictest/suites/mode/cluster/04_0002_explain_v2

@@ -131,6 +131,76 @@ EvalScalar
         ├── order by: []
         └── limit: NONE
 
+statement query T
+explain select count(1) as c, count(b) as d, max(a) as e from t1 order by c, e, d limit 10;
+
+----
+Limit
+├── limit: 10
+├── offset: 0
+└── Sort
+    ├── sort keys: [c ASC NULLS LAST, e ASC NULLS LAST, d ASC NULLS LAST]
+    └── EvalScalar
+        ├── expressions: [count(1) (#8), max(a) (#10), count(b) (#9)]
+        └── AggregateFinal
+            ├── group by: []
+            ├── aggregate functions: [count(), count(b), max(a)]
+            └── Exchange
+                ├── exchange type: Merge
+                └── AggregatePartial
+                    ├── group by: []
+                    ├── aggregate functions: [count(), count(b), max(a)]
+                    └── TableScan
+                        ├── table: default.default.t1
+                        ├── read rows: 0
+                        ├── read bytes: 0
+                        ├── partitions total: 0
+                        ├── partitions scanned: 0
+                        └── push downs: [filters: [], limit: NONE]
+
+statement query T
+explain select (t1.a + 1) as c,(t1.b+1) as d, (t2.a+1) as e from t1 join t2 on t1.a = t2.a order by c, d, e limit 10;
+
+----
+Limit
+├── limit: 10
+├── offset: 0
+└── Sort
+    ├── sort keys: [c ASC NULLS LAST, d ASC NULLS LAST, e ASC NULLS LAST]
+    └── Exchange                
+        ├── exchange type: Merge
+        └── Limit
+            ├── limit: 10
+            ├── offset: 0
+            └── Sort
+                ├── sort keys: [c ASC NULLS LAST, d ASC NULLS LAST, e ASC NULLS LAST]
+                └── EvalScalar
+                    ├── expressions: [+(t1.a (#0), 1), +(t1.b (#1), 1), +(t2.a (#2), 1)]
+                    └── HashJoin
+                        ├── join type: INNER
+                        ├── build keys: [t2.a (#2)]
+                        ├── probe keys: [t1.a (#0)]
+                        ├── filters: []
+                        ├── Exchange(Build)
+                        │   ├── exchange type: Hash(t2.a (#2))
+                        │   └── TableScan
+                        │       ├── table: default.default.t2
+                        │       ├── read rows: 0
+                        │       ├── read bytes: 0
+                        │       ├── partitions total: 0
+                        │       ├── partitions scanned: 0
+                        │       ├── push downs: [filters: [], limit: NONE]
+                        │       └── output columns: [0]
+                        └── Exchange(Probe)
+                            ├── exchange type: Hash(t1.a (#0))
+                            └── TableScan
+                                ├── table: default.default.t1
+                                ├── read rows: 0
+                                ├── read bytes: 0
+                                ├── partitions total: 0
+                                ├── partitions scanned: 0
+                                └── push downs: [filters: [], limit: NONE]
+
 statement ok
 set prefer_broadcast_join = 1;