feat(frontend): define ConstantEvalRewriter, impl ExprRewritable for batch

e2e_test/batch/functions/now.slt.part

@@ -10,6 +10,13 @@ create table t (a timestamp);
 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: [[]] }
+
 statement ok
 drop table tz
 

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,7 +22,7 @@
     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] }
@@ -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,11 @@
     LogicalProject { exprs: [1:Int32] }
     └─LogicalValues { rows: [[]], schema: Schema { fields: [] } }
 
+    Const eval exprs:
+
+    LogicalProject { exprs: [1:Int32] }
+    └─LogicalValues { rows: [[]], schema: Schema { fields: [] } }
+
     To Batch Physical Plan:
 
     BatchProject { exprs: [1:Int32] }
@@ -51,7 +56,7 @@
       "stages": {
         "0": {
           "root": {
-            "plan_node_id": 28,
+            "plan_node_id": 34,
             "plan_node_type": "BatchProject",
             "schema": [
               {
@@ -64,7 +69,7 @@
             ],
             "children": [
               {
-                "plan_node_id": 26,
+                "plan_node_id": 32,
                 "plan_node_type": "BatchValues",
                 "schema": [],
                 "children": [],
@@ -165,6 +170,12 @@
     ├─LogicalScan { table: t1, columns: [v1] }
     └─LogicalScan { table: t2, columns: [v2] }
 
+    Const eval exprs:
+
+    LogicalJoin { type: Inner, on: (t1.v1 = t2.v2) }
+    ├─LogicalScan { table: t1, columns: [v1] }
+    └─LogicalScan { table: t2, columns: [v2] }
+
 - sql: |
     explain (logical) create table t1(v1 int);
   explain_output: |