feat(frontend): apply SessionTimezone and ConstEvalRewriter expr rewriters to during gen_{batch,stream}_plan

e2e_test/batch/functions/now.slt.part

@@ -11,11 +11,11 @@ statement ok
 insert into t values(now());
 
 # constant eval of now in batch plan
-# query T
-# explain select now() + interval '1 hour' = now() + interval '30 minutes' + interval '30 minutes' true;
-# ----
-# BatchProject { exprs: [true:Boolean] }
-#     └─BatchValues { rows: [[]] }
+query T
+explain select now() + interval '1 hour' = now() + interval '30 minutes' + interval '30 minutes' true;
+----
+BatchProject { exprs: [true:Boolean] }
+    └─BatchValues { rows: [[]] }
 
 statement ok
 drop table tz

src/common/src/types/interval.rs

@@ -686,11 +686,17 @@ impl Display for IntervalUnit {
             write(format_args!("{days} days"))?;
         }
         if self.ms != 0 || self.months == 0 && self.days == 0 {
-            let hours = self.ms / 1000 / 3600;
-            let minutes = (self.ms / 1000 / 60) % 60;
-            let seconds = self.ms % 60000 / 1000;
-            let secs_fract = self.ms % 1000;
-            write(format_args!("{hours:0>2}:{minutes:0>2}:{seconds:0>2}"))?;
+            let ms = self.ms.abs();
+            let hours = ms / 1000 / 3600;
+            let minutes = (ms / 1000 / 60) % 60;
+            let seconds = ms % 60000 / 1000;
+            let secs_fract = ms % 1000;
+
+            if self.ms < 0 {
+                write(format_args!("-{hours:0>2}:{minutes:0>2}:{seconds:0>2}"))?;
+            } else {
+                write(format_args!("{hours:0>2}:{minutes:0>2}:{seconds:0>2}"))?;
+            }
             if secs_fract != 0 {
                 let mut buf = [0u8; 4];
                 write!(buf.as_mut_slice(), ".{:03}", secs_fract).unwrap();
@@ -1122,6 +1128,15 @@ mod tests {
             "-1 years -2 mons 3 days"
         );
         assert_eq!(IntervalUnit::default().to_string(), "00:00:00");
+        assert_eq!(
+            IntervalUnit::new(
+                -14,
+                3,
+                -(11 * 3600 * 1000 + 45 * 60 * 1000 + 14 * 1000 + 233)
+            )
+            .to_string(),
+            "-1 years -2 mons 3 days -11:45:14.233"
+        );
     }
 
     #[test]

src/expr/src/expr/expr_field.rs

@@ -17,7 +17,7 @@ use std::convert::TryFrom;
 use anyhow::anyhow;
 use risingwave_common::array::{ArrayImpl, ArrayRef, DataChunk};
 use risingwave_common::row::OwnedRow;
-use risingwave_common::types::{DataType, Datum};
+use risingwave_common::types::{DataType, Datum, ScalarImpl};
 use risingwave_common::util::value_encoding::deserialize_datum;
 use risingwave_pb::expr::expr_node::{RexNode, Type};
 use risingwave_pb::expr::ExprNode;
@@ -47,8 +47,15 @@ impl Expression for FieldExpression {
         }
     }
 
-    fn eval_row(&self, _input: &OwnedRow) -> Result<Datum> {
-        Err(anyhow!("expects a struct array ref").into())
+    fn eval_row(&self, input: &OwnedRow) -> Result<Datum> {
+        let struct_datum = self.input.eval_row(input)?;
+        struct_datum
+            .map(|s| match s {
+                ScalarImpl::Struct(v) => Ok(v.fields()[self.index].clone()),
+                _ => Err(anyhow!("expects a struct array ref").into()),
+            })
+            .transpose()
+            .map(|x| x.flatten())
     }
 }
 

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

@@ -4,13 +4,13 @@
   logical_plan: |
     LogicalValues { rows: [[Array('foo':Varchar, 'bar':Varchar)]], schema: Schema { fields: [*VALUES*_0.column_0:List { datatype: Varchar }] } }
   batch_plan: |
-    BatchValues { rows: [[Array('foo':Varchar, 'bar':Varchar)]] }
+    BatchValues { rows: [[ARRAY[foo, bar]:List { datatype: Varchar }]] }
 - sql: |
     values (ARRAY[1, 2+3, 4*5+1]);
   logical_plan: |
     LogicalValues { rows: [[Array(1:Int32, (2:Int32 + 3:Int32), ((4:Int32 * 5:Int32) + 1:Int32))]], schema: Schema { fields: [*VALUES*_0.column_0:List { datatype: Int32 }] } }
   batch_plan: |
-    BatchValues { rows: [[Array(1:Int32, (2:Int32 + 3:Int32), ((4:Int32 * 5:Int32) + 1:Int32))]] }
+    BatchValues { rows: [[ARRAY[1, 5, 21]:List { datatype: Int32 }]] }
 - sql: |
     create table t (v1 int);
     select (ARRAY[1, v1]) from t;
@@ -52,23 +52,23 @@
     LogicalProject { exprs: [ArrayCat(Array(66:Int32), Array(123:Int32))] }
     └─LogicalValues { rows: [[]], schema: Schema { fields: [] } }
   batch_plan: |
-    BatchProject { exprs: [ArrayCat(Array(66:Int32), Array(123:Int32))] }
+    BatchProject { exprs: [ARRAY[66, 123]:List { datatype: Int32 }] }
     └─BatchValues { rows: [[]] }
 - sql: |
     select array_cat(array[array[66]], array[233]);
   logical_plan: |
     LogicalProject { exprs: [ArrayCat(Array(Array(66:Int32)), Array(233:Int32))] }
     └─LogicalValues { rows: [[]], schema: Schema { fields: [] } }
   batch_plan: |
-    BatchProject { exprs: [ArrayCat(Array(Array(66:Int32)), Array(233:Int32))] }
+    BatchProject { exprs: [ARRAY[{66}, {233}]:List { datatype: List { datatype: Int32 } }] }
     └─BatchValues { rows: [[]] }
 - sql: |
     select array_cat(array[233], array[array[66]]);
   logical_plan: |
     LogicalProject { exprs: [ArrayCat(Array(233:Int32), Array(Array(66:Int32)))] }
     └─LogicalValues { rows: [[]], schema: Schema { fields: [] } }
   batch_plan: |
-    BatchProject { exprs: [ArrayCat(Array(233:Int32), Array(Array(66:Int32)))] }
+    BatchProject { exprs: [ARRAY[{233}, {66}]:List { datatype: List { datatype: Int32 } }] }
     └─BatchValues { rows: [[]] }
 - sql: |
     select array_cat(array[233], array[array[array[66]]]);
@@ -85,7 +85,7 @@
     LogicalProject { exprs: [ArrayAppend(Array(66:Int32), 123:Int32)] }
     └─LogicalValues { rows: [[]], schema: Schema { fields: [] } }
   batch_plan: |
-    BatchProject { exprs: [ArrayAppend(Array(66:Int32), 123:Int32)] }
+    BatchProject { exprs: [ARRAY[66, 123]:List { datatype: Int32 }] }
     └─BatchValues { rows: [[]] }
 - sql: |
     select array_append(123, 234);
@@ -102,7 +102,7 @@
     LogicalProject { exprs: [ArrayPrepend(123:Int32, Array(66:Int32))] }
     └─LogicalValues { rows: [[]], schema: Schema { fields: [] } }
   batch_plan: |
-    BatchProject { exprs: [ArrayPrepend(123:Int32, Array(66:Int32))] }
+    BatchProject { exprs: [ARRAY[123, 66]:List { datatype: Int32 }] }
     └─BatchValues { rows: [[]] }
 - sql: |
     select array_prepend(123, 234);

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

@@ -1,7 +1,7 @@
 # This file is automatically generated. See `src/frontend/planner_test/README.md` for more information.
 - sql: values (11, 22), (33+(1+2), 44);
   batch_plan: |
-    BatchValues { rows: [[11:Int32, 22:Int32], [(33:Int32 + (1:Int32 + 2:Int32)), 44:Int32]] }
+    BatchValues { rows: [[11:Int32, 22:Int32], [36:Int32, 44:Int32]] }
 - sql: select * from t
   binder_error: 'Catalog error: table or source not found: t'
 - sql: |
@@ -22,11 +22,11 @@
     select * from t where 1>2 and 1=1 and 3<1 and 4<>1 or 1=1 and 2>=1 and 1<=2;
   batch_plan: |
     BatchExchange { order: [], dist: Single }
-    └─BatchFilter { predicate: (1:Int32 = 1:Int32) AND ((((1:Int32 > 2:Int32) AND (3:Int32 < 1:Int32)) AND (4:Int32 <> 1:Int32)) OR ((2:Int32 >= 1:Int32) AND (1:Int32 <= 2:Int32))) }
+    └─BatchFilter { predicate: true:Boolean AND true:Boolean }
       └─BatchScan { table: t, columns: [], distribution: SomeShard }
   stream_plan: |
     StreamMaterialize { columns: [t._row_id(hidden)], pk_columns: [t._row_id] }
-    └─StreamFilter { predicate: (1:Int32 = 1:Int32) AND ((((1:Int32 > 2:Int32) AND (3:Int32 < 1:Int32)) AND (4:Int32 <> 1:Int32)) OR ((2:Int32 >= 1:Int32) AND (1:Int32 <= 2:Int32))) }
+    └─StreamFilter { predicate: true:Boolean AND true:Boolean }
       └─StreamTableScan { table: t, columns: [t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
 - sql: |
     create table t (v1 int);
@@ -130,11 +130,11 @@
 - sql: |
     select * from unnest(Array[1,2,3]);
   batch_plan: |
-    BatchTableFunction { Unnest(Array(1:Int32, 2:Int32, 3:Int32)) }
+    BatchTableFunction { Unnest(ARRAY[1, 2, 3]:List { datatype: Int32 }) }
 - sql: |
     select * from unnest(Array[Array[1,2,3], Array[4,5,6]]);
   batch_plan: |
-    BatchTableFunction { Unnest(Array(Array(1:Int32, 2:Int32, 3:Int32), Array(4:Int32, 5:Int32, 6:Int32))) }
+    BatchTableFunction { Unnest(ARRAY[{1,2,3}, {4,5,6}]:List { datatype: List { datatype: Int32 } }) }
 - sql: |
     create table t1 (x int);
     select * from t1 where EXISTS(select * where t1.x=1);

src/frontend/planner_test/tests/testdata/cast.yaml

@@ -18,7 +18,7 @@
   sql: |
     select case when NULL then 1 end;
   batch_plan: |
-    BatchProject { exprs: [Case(null:Boolean, 1:Int32)] }
+    BatchProject { exprs: [null:Int32] }
     └─BatchValues { rows: [[]] }
 - name: implicit cast boolean (JOIN ON NULL)
   sql: |
@@ -65,7 +65,7 @@
   sql: |
     select case when 'y' then 1 end;
   batch_plan: |
-    BatchProject { exprs: [Case(true:Boolean, 1:Int32)] }
+    BatchProject { exprs: [1:Int32] }
     └─BatchValues { rows: [[]] }
 - name: implicit cast boolean (JOIN ON with literal 'y' of unknown type)
   sql: |

src/frontend/planner_test/tests/testdata/explain.yaml

@@ -36,6 +36,16 @@
     LogicalProject { exprs: [1:Int32] }
     └─LogicalValues { rows: [[]], schema: Schema { fields: [] } }
 
+    Const eval exprs:
+
+    BatchProject { exprs: [1:Int32] }
+    └─BatchValues { rows: [[]] }
+
+    Const eval exprs:
+
+    BatchProject { exprs: [1:Int32] }
+    └─BatchValues { rows: [[]] }
+
     To Batch Physical Plan:
 
     BatchProject { exprs: [1:Int32] }
@@ -51,7 +61,7 @@
       "stages": {
         "0": {
           "root": {
-            "plan_node_id": 28,
+            "plan_node_id": 40,
             "plan_node_type": "BatchProject",
             "schema": [
               {
@@ -64,7 +74,7 @@
             ],
             "children": [
               {
-                "plan_node_id": 26,
+                "plan_node_id": 38,
                 "plan_node_type": "BatchValues",
                 "schema": [],
                 "children": [],