fix(optimizer): impl ExprRewritable for StreamValues

e2e_test/streaming/values.slt

@@ -75,3 +75,14 @@ NULL
 
 statement ok
 drop materialized view mv;
+
+statement ok
+create materialized view mv as select '2020-01-01'::timestamptz;;
+
+query I
+select * from mv;
+----
+2020-01-01 00:00:00+00:00
+
+statement ok
+drop materialized view mv;

src/frontend/planner_test/tests/testdata/output/array.yaml

@@ -56,7 +56,7 @@
   batch_plan: 'BatchValues { rows: [[ARRAY[]:List(Timestamptz)]] }'
   stream_plan: |-
     StreamMaterialize { columns: [array, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
-    └─StreamValues { rows: [[Array, 0:Int64]] }
+    └─StreamValues { rows: [[ARRAY[]:List(Timestamptz), 0:Int64]] }
 - sql: |
     select ARRAY[]::STRUCT<f1 INT>[];
   logical_plan: |-
@@ -65,7 +65,7 @@
   batch_plan: 'BatchValues { rows: [[ARRAY[]:List(Struct(StructType { field_names: ["f1"], field_types: [Int32] }))]] }'
   stream_plan: |-
     StreamMaterialize { columns: [array, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
-    └─StreamValues { rows: [[Array, 0:Int64]] }
+    └─StreamValues { rows: [[ARRAY[]:List(Struct(StructType { field_names: ["f1"], field_types: [Int32] })), 0:Int64]] }
 - sql: |
     select array_cat(array[66], array[123]);
   logical_plan: |-

src/frontend/planner_test/tests/testdata/output/basic_query.yaml

@@ -3,7 +3,7 @@
   batch_plan: 'BatchValues { rows: [[11:Int32, 22:Int32], [36:Int32, 44:Int32]] }'
   stream_plan: |-
     StreamMaterialize { columns: [*VALUES*_0.column_0, *VALUES*_0.column_1, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
-    └─StreamValues { rows: [[11:Int32, 22:Int32, 0:Int64], [(33:Int32 + (1:Int32 + 2:Int32)), 44:Int32, 1:Int64]] }
+    └─StreamValues { rows: [[11:Int32, 22:Int32, 0:Int64], [36:Int32, 44:Int32, 1:Int64]] }
 - sql: select * from t
   binder_error: 'Catalog error: table or source not found: t'
 - sql: |

src/frontend/planner_test/tests/testdata/output/expr.yaml

@@ -600,7 +600,7 @@
   batch_plan: 'BatchValues { rows: [[1:Int32]] }'
   stream_plan: |-
     StreamMaterialize { columns: [abs, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
-    └─StreamValues { rows: [[Abs(-1:Int32), 0:Int64]] }
+    └─StreamValues { rows: [[1:Int32, 0:Int64]] }
 - sql: |
     select * from range(1,2);
   batch_plan: |-

src/frontend/planner_test/tests/testdata/output/with_ordinality.yaml

@@ -190,7 +190,7 @@
   batch_plan: 'BatchValues { rows: [[1:Int32, 1:Int64]] }'
   stream_plan: |-
     StreamMaterialize { columns: [abs, ordinality, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
-    └─StreamValues { rows: [[Abs(1:Int32), 1:Int64, 0:Int64]] }
+    └─StreamValues { rows: [[1:Int32, 1:Int64, 0:Int64]] }
 - sql: |
     create table t(x int , arr int[]);
     select * from t, abs(x) WITH ORDINALITY;

src/frontend/src/optimizer/plan_node/stream_values.rs

@@ -88,4 +88,19 @@ impl StreamNode for StreamValues {
     }
 }
 
-impl ExprRewritable for StreamValues {}
+impl ExprRewritable for StreamValues {
+    fn has_rewritable_expr(&self) -> bool {
+        true
+    }
+
+    fn rewrite_exprs(&self, r: &mut dyn crate::expr::ExprRewriter) -> crate::PlanRef {
+        Self::new(
+            self.logical
+                .rewrite_exprs(r)
+                .as_logical_values()
+                .unwrap()
+                .clone(),
+        )
+        .into()
+    }
+}