refactor(frontend): separate post-agg Project from LogicalAgg::create (…

…#2266)

src/frontend/src/binder/select.rs

@@ -38,36 +38,6 @@ pub struct BoundSelect {
 }
 
 impl BoundSelect {
-    pub fn create(
-        binder: &Binder,
-        distinct: bool,
-        select_items: Vec<ExprImpl>,
-        aliases: Vec<Option<String>>,
-        from: Option<Relation>,
-        where_clause: Option<ExprImpl>,
-        group_by: Vec<ExprImpl>,
-    ) -> Result<Self> {
-        // Store field from `ExprImpl` to support binding `field_desc` in `subquery`.
-        let fields = select_items
-            .iter()
-            .zip_eq(aliases.iter())
-            .map(|(s, a)| {
-                let name = a.clone().unwrap_or_else(|| UNNAMED_COLUMN.to_string());
-                binder.expr_to_field(s, name)
-            })
-            .collect::<Result<Vec<Field>>>()?;
-
-        Ok(Self {
-            distinct,
-            select_items,
-            aliases,
-            from,
-            where_clause,
-            group_by,
-            schema: Schema { fields },
-        })
-    }
-
     /// The schema returned by this [`BoundSelect`].
     pub fn schema(&self) -> &Schema {
         &self.schema
@@ -116,15 +86,25 @@ impl Binder {
         // Bind SELECT clause.
         let (select_items, aliases) = self.bind_project(select.projection)?;
 
-        BoundSelect::create(
-            self,
-            select.distinct,
+        // Store field from `ExprImpl` to support binding `field_desc` in `subquery`.
+        let fields = select_items
+            .iter()
+            .zip_eq(aliases.iter())
+            .map(|(s, a)| {
+                let name = a.clone().unwrap_or_else(|| UNNAMED_COLUMN.to_string());
+                self.expr_to_field(s, name)
+            })
+            .collect::<Result<Vec<Field>>>()?;
+
+        Ok(BoundSelect {
+            distinct: select.distinct,
             select_items,
             aliases,
             from,
-            selection,
+            where_clause: selection,
             group_by,
-        )
+            schema: Schema { fields },
+        })
     }
 
     pub fn bind_project(

src/frontend/src/optimizer/plan_node/logical_agg.rs

@@ -317,14 +317,15 @@ impl LogicalAgg {
     /// `create` will analyze the select exprs and group exprs, and construct a plan like
     ///
     /// ```text
-    /// LogicalProject -> LogicalAgg -> LogicalProject -> input
+    /// LogicalAgg -> LogicalProject -> input
     /// ```
+    ///
+    /// It also returns the rewritten select exprs that reference into the aggregated results.
     pub fn create(
         select_exprs: Vec<ExprImpl>,
-        select_alias: Vec<Option<String>>,
         group_exprs: Vec<ExprImpl>,
         input: PlanRef,
-    ) -> Result<PlanRef> {
+    ) -> Result<(PlanRef, Vec<ExprImpl>)> {
         let group_keys = (0..group_exprs.len()).collect();
         let mut expr_handler = ExprHandler::new(group_exprs)?;
 
@@ -346,13 +347,7 @@ impl LogicalAgg {
         // This LogicalAgg focuses on calculating the aggregates and grouping.
         let logical_agg = LogicalAgg::new(expr_handler.agg_calls, group_keys, logical_project);
 
-        // This LogicalProject focus on transforming the aggregates and grouping columns to
-        // InputRef.
-        Ok(LogicalProject::create(
-            logical_agg.into(),
-            rewritten_select_exprs,
-            select_alias,
-        ))
+        Ok((logical_agg.into(), rewritten_select_exprs))
     }
 
     /// Get a reference to the logical agg's agg calls.
@@ -574,18 +569,14 @@ mod tests {
         let gen_internal_value = |select_exprs: Vec<ExprImpl>,
                                   group_exprs|
          -> (Vec<ExprImpl>, Vec<PlanAggCall>, Vec<usize>) {
-            let select_alias = vec![None; select_exprs.len()];
-            let plan =
-                LogicalAgg::create(select_exprs, select_alias, group_exprs, input.clone()).unwrap();
-            let logical_project = plan.as_logical_project().unwrap();
-            let exprs = logical_project.exprs();
+            let (plan, exprs) =
+                LogicalAgg::create(select_exprs, group_exprs, input.clone()).unwrap();
 
-            let plan = logical_project.input();
             let logical_agg = plan.as_logical_agg().unwrap();
             let agg_calls = logical_agg.agg_calls().to_vec();
             let group_keys = logical_agg.group_keys().to_vec();
 
-            (exprs.clone(), agg_calls, group_keys)
+            (exprs, agg_calls, group_keys)
         };
 
         // Test case: select v1 from test group by v1;

src/frontend/src/planner/select.rs

@@ -53,13 +53,13 @@ impl Planner {
         // TODO: select-agg, group-by, having can also contain subquery exprs.
         let has_agg_call = select_items.iter().any(|expr| expr.has_agg_call());
         if !group_by.is_empty() || has_agg_call {
-            LogicalAgg::create(select_items, aliases, group_by, root)
-        } else {
-            if select_items.iter().any(|e| e.has_subquery()) {
-                (root, select_items) = self.substitute_subqueries(root, select_items)?;
-            }
-            Ok(LogicalProject::create(root, select_items, aliases))
+            (root, select_items) = LogicalAgg::create(select_items, group_by, root)?;
+        }
+
+        if select_items.iter().any(|e| e.has_subquery()) {
+            (root, select_items) = self.substitute_subqueries(root, select_items)?;
         }
+        Ok(LogicalProject::create(root, select_items, aliases))
     }
 
     /// Helper to create a dummy node as child of [`LogicalProject`].