refactor(optimizer): LogicalProject::with_mapping should return `Se…

…lf` (#2566)

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

@@ -528,6 +528,7 @@ impl ColPrunable for LogicalAgg {
                 agg.into(),
                 ColIndexMapping::with_remaining_columns(&output_required_cols, src_size),
             )
+            .into()
         }
     }
 }

src/frontend/src/optimizer/plan_node/logical_filter.rs

@@ -138,6 +138,7 @@ impl ColPrunable for LogicalFilter {
                 filter.into(),
                 ColIndexMapping::with_remaining_columns(&output_required_cols, src_size),
             )
+            .into()
         }
     }
 }

src/frontend/src/optimizer/plan_node/logical_hop_window.rs

@@ -204,6 +204,7 @@ impl ColPrunable for LogicalHopWindow {
             new_hop.into(),
             ColIndexMapping::with_remaining_columns(&required_output_cols, self.schema().len()),
         )
+        .into()
     }
 }
 

src/frontend/src/optimizer/plan_node/logical_join.rs

@@ -348,6 +348,7 @@ impl ColPrunable for LogicalJoin {
                 join.into(),
                 ColIndexMapping::with_remaining_columns(&remaining_columns, self.schema().len()),
             )
+            .into()
         }
     }
 }

src/frontend/src/optimizer/plan_node/logical_project.rs

@@ -87,11 +87,11 @@ impl LogicalProject {
     ///
     /// This is useful in column pruning when we want to add a project to ensure the output schema
     /// is correct.
-    pub fn with_mapping(input: PlanRef, mapping: ColIndexMapping) -> PlanRef {
+    pub fn with_mapping(input: PlanRef, mapping: ColIndexMapping) -> Self {
         if mapping.target_size() == 0 {
             // The mapping is empty, so the parent actually doesn't need the output of the input.
             // This can happen when the parent node only selects constant expressions.
-            return input;
+            return LogicalProject::new(input, vec![]);
         };
         let mut input_refs = vec![None; mapping.target_size()];
         for (src, tar) in mapping.mapping_pairs() {
@@ -105,7 +105,7 @@ impl LogicalProject {
             .map(|i| InputRef::new(i, input_schema.fields()[i].data_type()).into())
             .collect();
 
-        LogicalProject::new(input, exprs).into()
+        LogicalProject::new(input, exprs)
     }
 
     fn derive_schema(exprs: &[ExprImpl], input_schema: &Schema) -> Schema {

src/frontend/src/optimizer/plan_node/logical_source.rs

@@ -82,7 +82,7 @@ impl fmt::Display for LogicalSource {
 impl ColPrunable for LogicalSource {
     fn prune_col(&self, required_cols: &[usize]) -> PlanRef {
         let mapping = ColIndexMapping::with_remaining_columns(required_cols, self.schema().len());
-        LogicalProject::with_mapping(self.clone().into(), mapping)
+        LogicalProject::with_mapping(self.clone().into(), mapping).into()
     }
 }
 

src/frontend/src/optimizer/plan_node/logical_topn.rs

@@ -154,6 +154,7 @@ impl ColPrunable for LogicalTopN {
                 top_n,
                 ColIndexMapping::with_remaining_columns(required_cols, src_size),
             )
+            .into()
         }
     }
 }