fix(optimizer): fix logical scan pk derive

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

@@ -132,8 +132,9 @@ impl LogicalProject {
         let i2o = Self::i2o_col_mapping_inner(input_schema.len(), exprs);
         input_pk
             .iter()
-            .filter_map(|pk_col| i2o.try_map(*pk_col))
-            .collect::<Vec<_>>()
+            .map(|pk_col| i2o.try_map(*pk_col))
+            .collect::<Option<Vec<_>>>()
+            .unwrap_or_default()
     }
 
     pub fn exprs(&self) -> &Vec<ExprImpl> {

src/frontend/src/optimizer/plan_node/logical_scan.rs

@@ -370,7 +370,7 @@ impl ToStream for LogicalScan {
             true => {
                 let mut col_ids = HashSet::new();
 
-                for idx in &self.required_col_idx {
+                for idx in &self.output_col_idx {
                     col_ids.insert(self.table_desc.columns[*idx].column_id);
                 }
                 let mut col_id_to_tb_idx = HashMap::new();

src/frontend/src/optimizer/property/distribution.rs

@@ -124,9 +124,8 @@ impl Distribution {
     /// valid.
     pub fn dist_column_indices(&self) -> &[usize] {
         match self {
-            Distribution::Single => Default::default(),
+            Distribution::Single | Distribution::SomeShard => Default::default(),
             Distribution::HashShard(dists) => dists,
-            _ => unreachable!(),
         }
     }
 }

src/frontend/test_runner/tests/testdata/pk_derive.yaml

@@ -62,6 +62,24 @@
             StreamHashAgg { group_keys: [$0], aggs: [count, min($1)] }
               StreamExchange { dist: HashShard([0]) }
                 StreamTableScan { table: t, columns: [id, v, _row_id], pk_indices: [2] }
+- sql: |
+    create table t (v1 varchar, v2 varchar, v3 varchar);
+    select
+        *
+    from
+        t
+    group by
+        v1,
+        v2,
+        v3;
+  optimized_logical_plan: |
+    LogicalAgg { group_keys: [0, 1, 2], agg_calls: [] }
+      LogicalScan { table: t, columns: [v1, v2, v3] }
+  stream_plan: |
+    StreamMaterialize { columns: [v1, v2, v3, agg#0(hidden)], pk_columns: [v1, v2, v3] }
+      StreamHashAgg { group_keys: [$0, $1, $2], aggs: [count] }
+        StreamExchange { dist: HashShard([0, 1, 2]) }
+          StreamTableScan { table: t, columns: [v1, v2, v3, _row_id], pk_indices: [3] }
 - sql: |
     create table t (v1 varchar, v2 varchar, v3 varchar);
     create materialized view mv as
@@ -74,22 +92,14 @@
         v2,
         v3;
     select
-        max(v1) as v1
+        v1
     from
         mv
     where
         v3 = 'world' or v3 = 'hello';
-  batch_plan: |
-    BatchSimpleAgg { aggs: [max($0)] }
-      BatchExchange { order: [], dist: Single }
-        BatchSimpleAgg { aggs: [max($0)] }
-          BatchProject { exprs: [$0] }
-            BatchFilter { predicate: (($1 = 'world':Varchar) OR ($1 = 'hello':Varchar)) }
-              BatchScan { table: mv, columns: [v1, v3] }
+  optimized_logical_plan: |
+    LogicalScan { table: mv, output_columns: [v1], required_columns: [$0:v1, $2:v3], predicate: (($2 = 'world':Varchar) OR ($2 = 'hello':Varchar)) }
   stream_plan: |
-    StreamMaterialize { columns: [agg#0(hidden), v1], pk_columns: [agg#0, v1] }
-      StreamSimpleAgg { aggs: [count, max($0)] }
-        StreamExchange { dist: Single }
-          StreamProject { exprs: [$0, $1] }
-            StreamFilter { predicate: (($2 = 'world':Varchar) OR ($2 = 'hello':Varchar)) }
-              StreamTableScan { table: mv, columns: [v1, v2, v3], pk_indices: [0, 1, 2] }
+    StreamMaterialize { columns: [v1, v2(hidden), v3(hidden)], pk_columns: [v1, v2, v3] }
+      StreamFilter { predicate: (($2 = 'world':Varchar) OR ($2 = 'hello':Varchar)) }
+        StreamTableScan { table: mv, columns: [v1, v2, v3], pk_indices: [0, 1, 2] }