feat(planner): Use explain verbose for all planner tests (#4359)

* use explain verbose

* Use explain verbose

* minor

* fix

src/frontend/test_runner/src/lib.rs

@@ -18,6 +18,7 @@
 mod resolve_id;
 
 use std::collections::{BTreeMap, HashMap};
+use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
 use anyhow::{anyhow, Result};
@@ -262,6 +263,7 @@ impl TestCase {
         let statements = Parser::parse_sql(sql).unwrap();
         for stmt in statements {
             let context = OptimizerContext::new(session.clone(), Arc::from(sql));
+            context.explain_verbose.store(true, Ordering::Relaxed); // use explain verbose in planner tests
             match stmt.clone() {
                 Statement::Query(_)
                 | Statement::Insert { .. }

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

@@ -6,32 +6,32 @@
     StreamMaterialize { columns: [v1, count(hidden), mx2], pk_columns: [v1] }
       StreamAppendOnlyHashAgg { group_key: [t1.v1], aggs: [count, max(t1.v2)] }
         StreamExchange { dist: HashShard(t1.v1) }
-          StreamTableScan { table: t1, columns: [v1, v2, _row_id] }
+          StreamTableScan { table: t1, columns: [t1.v1, t1.v2, t1._row_id], pk: [t1._row_id], distribution: HashShard(t1._row_id) }
 - sql: |
     create table t1 (v1 int, v2 int) with (appendonly = true);
     create table t2 (v1 int, v3 int) with (appendonly = true);
     select t1.v1 as id, v2, v3 from t1 join t2 on t1.v1=t2.v1;
   stream_plan: |
     StreamMaterialize { columns: [id, v2, v3, t1._row_id(hidden), t2._row_id(hidden)], pk_columns: [t1._row_id, t2._row_id] }
       StreamExchange { dist: HashShard(t1._row_id, t2._row_id) }
-        StreamAppendOnlyHashJoin { type: Inner, predicate: t1.v1 = t2.v1, append_only: true }
+        StreamAppendOnlyHashJoin { type: Inner, predicate: t1.v1 = t2.v1, append_only: true, output: [t1.v1, t1.v2, t2.v3, t1._row_id, t2._row_id] }
           StreamExchange { dist: HashShard(t1.v1) }
-            StreamTableScan { table: t1, columns: [v1, v2, _row_id] }
+            StreamTableScan { table: t1, columns: [t1.v1, t1.v2, t1._row_id], pk: [t1._row_id], distribution: HashShard(t1._row_id) }
           StreamExchange { dist: HashShard(t2.v1) }
-            StreamTableScan { table: t2, columns: [v1, v3, _row_id] }
+            StreamTableScan { table: t2, columns: [t2.v1, t2.v3, t2._row_id], pk: [t2._row_id], distribution: HashShard(t2._row_id) }
 - sql: |
     create table t1 (v1 int, v2 int) with (appendonly = true);
     select v1 from t1 order by v1 limit 3 offset 3;
   stream_plan: |
     StreamMaterialize { columns: [v1, t1._row_id(hidden)], pk_columns: [t1._row_id], order_descs: [v1, t1._row_id] }
       StreamAppendOnlyTopN { order: "[t1.v1 ASC]", limit: 3, offset: 3 }
         StreamExchange { dist: Single }
-          StreamTableScan { table: t1, columns: [v1, _row_id] }
+          StreamTableScan { table: t1, columns: [t1.v1, t1._row_id], pk: [t1._row_id], distribution: HashShard(t1._row_id) }
 - sql: |
     create table t1 (v1 int, v2 int) with (appendonly = true);
     select max(v1) as max_v1 from t1;
   stream_plan: |
     StreamMaterialize { columns: [count(hidden), max_v1], pk_columns: [] }
       StreamAppendOnlyGlobalSimpleAgg { aggs: [count, max(t1.v1)] }
         StreamExchange { dist: Single }
-          StreamTableScan { table: t1, columns: [v1, _row_id] }
+          StreamTableScan { table: t1, columns: [t1.v1, t1._row_id], pk: [t1._row_id], distribution: HashShard(t1._row_id) }

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

@@ -16,11 +16,11 @@
     select (ARRAY[1, v1]) from t;
   logical_plan: |
     LogicalProject { exprs: [Array(1:Int32, t.v1)] }
-      LogicalScan { table: t, columns: [_row_id, v1] }
+      LogicalScan { table: t, columns: [t._row_id, t.v1] }
   batch_plan: |
     BatchExchange { order: [], dist: Single }
       BatchProject { exprs: [Array(1:Int32, t.v1)] }
-        BatchScan { table: t, columns: [v1] }
+        BatchScan { table: t, columns: [t.v1], distribution: SomeShard }
 - sql: |
     select ARRAY[null];
   logical_plan: |

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

@@ -9,10 +9,10 @@
     select * from t;
   batch_plan: |
     BatchExchange { order: [], dist: Single }
-      BatchScan { table: t, columns: [v1, v2] }
+      BatchScan { table: t, columns: [t.v1, t.v2], distribution: SomeShard }
   stream_plan: |
     StreamMaterialize { columns: [v1, v2, t._row_id(hidden)], pk_columns: [t._row_id] }
-      StreamTableScan { table: t, columns: [v1, v2, _row_id] }
+      StreamTableScan { table: t, columns: [t.v1, t.v2, t._row_id], pk: [t._row_id], distribution: HashShard(t._row_id) }
 - sql: |
     create table t (v1 bigint, v2 double precision);
     select t2.* from t;
@@ -23,79 +23,79 @@
   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))) }
-        BatchScan { table: t, columns: [] }
+        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))) }
-        StreamTableScan { table: t, columns: [_row_id] }
+        StreamTableScan { table: t, columns: [t._row_id], pk: [t._row_id], distribution: HashShard(t._row_id) }
 - sql: |
     create table t (v1 int);
     select * from t where v1<1;
   batch_plan: |
     BatchExchange { order: [], dist: Single }
       BatchFilter { predicate: (t.v1 < 1:Int32) }
-        BatchScan { table: t, columns: [v1] }
+        BatchScan { table: t, columns: [t.v1], distribution: SomeShard }
   stream_plan: |
     StreamMaterialize { columns: [v1, t._row_id(hidden)], pk_columns: [t._row_id] }
       StreamFilter { predicate: (t.v1 < 1:Int32) }
-        StreamTableScan { table: t, columns: [v1, _row_id] }
+        StreamTableScan { table: t, columns: [t.v1, t._row_id], pk: [t._row_id], distribution: HashShard(t._row_id) }
 - sql: |
     /* test boolean expression common factor extraction */
     create table t (v1 Boolean, v2 Boolean, v3 Boolean);
     select * from t where v1 AND v2 AND ((v1 AND v2) OR (v2 AND v3));
   batch_plan: |
     BatchExchange { order: [], dist: Single }
       BatchFilter { predicate: t.v1 AND t.v2 AND (t.v1 OR t.v3) }
-        BatchScan { table: t, columns: [v1, v2, v3] }
+        BatchScan { table: t, columns: [t.v1, t.v2, t.v3], distribution: SomeShard }
 - sql: |
     /* test boolean expression simplification */
     create table t (v1 Boolean, v2 Boolean, v3 Boolean);
     select * from t where v1 AND NOT(v1 OR v2 Or NOT(v1 AND v2 AND true));
   batch_plan: |
     BatchExchange { order: [], dist: Single }
       BatchFilter { predicate: t.v1 AND Not(t.v1) AND Not(t.v2) AND t.v2 }
-        BatchScan { table: t, columns: [v1, v2, v3] }
+        BatchScan { table: t, columns: [t.v1, t.v2, t.v3], distribution: SomeShard }
 - sql: |
     /* test boolean expression simplification */
     create table t (v1 Boolean, v2 Boolean);
     select * from t where (v1 AND v2) OR (v1 AND v2);
   batch_plan: |
     BatchExchange { order: [], dist: Single }
       BatchFilter { predicate: t.v1 AND t.v2 }
-        BatchScan { table: t, columns: [v1, v2] }
+        BatchScan { table: t, columns: [t.v1, t.v2], distribution: SomeShard }
 - sql: |
     /* constant folding for IS TRUE, IS FALSE, IS NULL*/
     create table t(a Boolean);
     select * from t where (NULL IS NULL) IS TRUE AND FALSE IS FALSE AND a;
   logical_plan: |
     LogicalProject { exprs: [t.a] }
       LogicalFilter { predicate: t.a }
-        LogicalScan { table: t, columns: [_row_id, a] }
+        LogicalScan { table: t, columns: [t._row_id, t.a] }
 - sql: |
     /* constant folding for IS NOT TRUE, IS NOT FALSE */
     create table t(a Boolean);
     select * from t where (NULL IS NOT TRUE) IS NOT FALSE AND a IS NOT TRUE;
   logical_plan: |
     LogicalProject { exprs: [t.a] }
       LogicalFilter { predicate: IsNotTrue(t.a) }
-        LogicalScan { table: t, columns: [_row_id, a] }
+        LogicalScan { table: t, columns: [t._row_id, t.a] }
 - sql: |
     /* constant folding IS NOT NULL */
     create table t(a double precision);
     select * from t where (a IS NOT NULL AND 3.14 IS NOT NULL) OR (NULL IS NOT NULL);
   logical_plan: |
     LogicalProject { exprs: [t.a] }
       LogicalFilter { predicate: IsNotNull(t.a) }
-        LogicalScan { table: t, columns: [_row_id, a] }
+        LogicalScan { table: t, columns: [t._row_id, t.a] }
 - sql: |
     create table t (v1 int, v2 int);
     select v1 from t;
   batch_plan: |
     BatchExchange { order: [], dist: Single }
-      BatchScan { table: t, columns: [v1] }
+      BatchScan { table: t, columns: [t.v1], distribution: SomeShard }
   stream_plan: |
     StreamMaterialize { columns: [v1, t._row_id(hidden)], pk_columns: [t._row_id] }
-      StreamTableScan { table: t, columns: [v1, _row_id] }
+      StreamTableScan { table: t, columns: [t.v1, t._row_id], pk: [t._row_id], distribution: HashShard(t._row_id) }
 - sql: select 1
   batch_plan: |
     BatchProject { exprs: [1:Int32] }
@@ -105,20 +105,20 @@
     select a from t as t2(a);
   batch_plan: |
     BatchExchange { order: [], dist: Single }
-      BatchScan { table: t, columns: [v1] }
+      BatchScan { table: t, columns: [t.v1], distribution: SomeShard }
 - sql: |
     create table t (v1 int, v2 int);
     delete from t;
   batch_plan: |
     BatchDelete { table: t }
-      BatchScan { table: t, columns: [_row_id, v1, v2] }
+      BatchScan { table: t, columns: [t._row_id, t.v1, t.v2], distribution: HashShard(t._row_id) }
 - sql: |
     create table t (v1 int, v2 int);
     delete from t where v1 = 1;
   batch_plan: |
     BatchDelete { table: t }
       BatchFilter { predicate: (t.v1 = 1:Int32) }
-        BatchScan { table: t, columns: [_row_id, v1, v2] }
+        BatchScan { table: t, columns: [t._row_id, t.v1, t.v2], distribution: HashShard(t._row_id) }
 - sql: |
     select * from generate_series('2'::INT,'10'::INT,'2'::INT);
   batch_plan: |

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

@@ -4,33 +4,33 @@
     select v1 from t
   logical_plan: |
     LogicalProject { exprs: [t.v1] }
-      LogicalScan { table: t, columns: [_row_id, v1, v2] }
+      LogicalScan { table: t, columns: [t._row_id, t.v1, t.v2] }
   optimized_logical_plan: |
-    LogicalScan { table: t, columns: [v1] }
+    LogicalScan { table: t, columns: [t.v1] }
 - sql: |
     /* filter */
     create table t (v1 bigint, v2 double precision, v3 int);
     select v1 from t where v2 > 2
   logical_plan: |
     LogicalProject { exprs: [t.v1] }
       LogicalFilter { predicate: (t.v2 > 2:Int32) }
-        LogicalScan { table: t, columns: [_row_id, v1, v2, v3] }
+        LogicalScan { table: t, columns: [t._row_id, t.v1, t.v2, t.v3] }
   optimized_logical_plan: |
-    LogicalScan { table: t, output_columns: [v1], required_columns: [v1, v2], predicate: (t.v2 > 2:Int32) }
+    LogicalScan { table: t, output_columns: [t.v1], required_columns: [v1, v2], predicate: (t.v2 > 2:Int32) }
 - sql: |
     /* join */
     create table t1 (v1 int not null, v2 int not null, v3 int);
     create table t2 (v1 int not null, v2 int not null, v3 int);
     select t1.v1, t2.v1 from t1 join t2 on t1.v2 = t2.v2;
   logical_plan: |
     LogicalProject { exprs: [t1.v1, t2.v1] }
-      LogicalJoin { type: Inner, on: (t1.v2 = t2.v2) }
-        LogicalScan { table: t1, columns: [_row_id, v1, v2, v3] }
-        LogicalScan { table: t2, columns: [_row_id, v1, v2, v3] }
+      LogicalJoin { type: Inner, on: (t1.v2 = t2.v2), output: all }
+        LogicalScan { table: t1, columns: [t1._row_id, t1.v1, t1.v2, t1.v3] }
+        LogicalScan { table: t2, columns: [t2._row_id, t2.v1, t2.v2, t2.v3] }
   optimized_logical_plan: |
-    LogicalJoin { type: Inner, on: (t1.v2 = t2.v2) }
-      LogicalScan { table: t1, columns: [v1, v2] }
-      LogicalScan { table: t2, columns: [v1, v2] }
+    LogicalJoin { type: Inner, on: (t1.v2 = t2.v2), output: [t1.v1, t2.v1] }
+      LogicalScan { table: t1, columns: [t1.v1, t1.v2] }
+      LogicalScan { table: t2, columns: [t2.v1, t2.v2] }
 - sql: |
     /* agg */
     create table t (v1 bigint, v2 double precision, v3 int);
@@ -40,10 +40,10 @@
       LogicalAgg { aggs: [count(t.v1)] }
         LogicalProject { exprs: [t.v1] }
           LogicalFilter { predicate: (t.v2 > 2:Int32) }
-            LogicalScan { table: t, columns: [_row_id, v1, v2, v3] }
+            LogicalScan { table: t, columns: [t._row_id, t.v1, t.v2, t.v3] }
   optimized_logical_plan: |
     LogicalAgg { aggs: [count(t.v1)] }
-      LogicalScan { table: t, output_columns: [v1], required_columns: [v1, v2], predicate: (t.v2 > 2:Int32) }
+      LogicalScan { table: t, output_columns: [t.v1], required_columns: [v1, v2], predicate: (t.v2 > 2:Int32) }
 - sql: |
     /* top n */
     create table t (v1 int, v2 int, v3 int);
@@ -52,18 +52,18 @@
     LogicalProject { exprs: [t.v3] }
       LogicalTopN { order: "[t.v3 ASC, t.v2 ASC]", limit: 2, offset: 0 }
         LogicalProject { exprs: [t.v1, t.v2, t.v3] }
-          LogicalScan { table: t, columns: [_row_id, v1, v2, v3] }
+          LogicalScan { table: t, columns: [t._row_id, t.v1, t.v2, t.v3] }
   optimized_logical_plan: |
     LogicalProject { exprs: [t.v3] }
       LogicalTopN { order: "[t.v3 ASC, t.v2 ASC]", limit: 2, offset: 0 }
-        LogicalScan { table: t, columns: [v2, v3] }
+        LogicalScan { table: t, columns: [t.v2, t.v3] }
 - sql: |
     /* constant */
     create table t (v1 bigint, v2 double precision, v3 int);
     select 1 from t
   logical_plan: |
     LogicalProject { exprs: [1:Int32] }
-      LogicalScan { table: t, columns: [_row_id, v1, v2, v3] }
+      LogicalScan { table: t, columns: [t._row_id, t.v1, t.v2, t.v3] }
   optimized_logical_plan: |
     LogicalProject { exprs: [1:Int32] }
       LogicalScan { table: t, columns: [] }
@@ -74,7 +74,7 @@
   logical_plan: |
     LogicalProject { exprs: [1:Int32] }
       LogicalFilter { predicate: (t.v2 > 1:Int32) }
-        LogicalScan { table: t, columns: [_row_id, v1, v2, v3] }
+        LogicalScan { table: t, columns: [t._row_id, t.v1, t.v2, t.v3] }
   optimized_logical_plan: |
     LogicalProject { exprs: [1:Int32] }
       LogicalScan { table: t, output_columns: [], required_columns: [v2], predicate: (t.v2 > 1:Int32) }
@@ -86,7 +86,7 @@
     LogicalProject { exprs: [count(1:Int32)] }
       LogicalAgg { aggs: [count(1:Int32)] }
         LogicalProject { exprs: [1:Int32] }
-          LogicalScan { table: t, columns: [_row_id, v1, v2, v3] }
+          LogicalScan { table: t, columns: [t._row_id, t.v1, t.v2, t.v3] }
   optimized_logical_plan: |
     LogicalAgg { aggs: [count(1:Int32)] }
       LogicalProject { exprs: [1:Int32] }
@@ -100,7 +100,7 @@
       LogicalAgg { aggs: [count(1:Int32)] }
         LogicalProject { exprs: [1:Int32] }
           LogicalFilter { predicate: (t.v2 > 1:Int32) }
-            LogicalScan { table: t, columns: [_row_id, v1, v2, v3] }
+            LogicalScan { table: t, columns: [t._row_id, t.v1, t.v2, t.v3] }
   optimized_logical_plan: |
     LogicalAgg { aggs: [count(1:Int32)] }
       LogicalProject { exprs: [1:Int32] }
@@ -113,13 +113,13 @@
   logical_plan: |
     LogicalProject { exprs: [t1.v1, t2.v1] }
       LogicalFilter { predicate: (t1.v3 < 1:Int32) }
-        LogicalJoin { type: Inner, on: (t1.v2 = t2.v2) }
-          LogicalScan { table: t1, columns: [_row_id, v1, v2, v3] }
-          LogicalScan { table: t2, columns: [_row_id, v1, v2, v3] }
+        LogicalJoin { type: Inner, on: (t1.v2 = t2.v2), output: all }
+          LogicalScan { table: t1, columns: [t1._row_id, t1.v1, t1.v2, t1.v3] }
+          LogicalScan { table: t2, columns: [t2._row_id, t2.v1, t2.v2, t2.v3] }
   optimized_logical_plan: |
-    LogicalJoin { type: Inner, on: (t1.v2 = t2.v2) }
-      LogicalScan { table: t1, output_columns: [v1, v2], required_columns: [v1, v2, v3], predicate: (t1.v3 < 1:Int32) }
-      LogicalScan { table: t2, columns: [v1, v2] }
+    LogicalJoin { type: Inner, on: (t1.v2 = t2.v2), output: [t1.v1, t2.v1] }
+      LogicalScan { table: t1, output_columns: [t1.v1, t1.v2], required_columns: [v1, v2, v3], predicate: (t1.v3 < 1:Int32) }
+      LogicalScan { table: t2, columns: [t2.v1, t2.v2] }
 - sql: |
     /* mixed */
     create table t (v1 bigint, v2 double precision, v3 int);
@@ -129,27 +129,27 @@
       LogicalAgg { aggs: [count(1:Int32), count(t.v1)] }
         LogicalProject { exprs: [1:Int32, t.v1] }
           LogicalFilter { predicate: (t.v2 > 1:Int32) }
-            LogicalScan { table: t, columns: [_row_id, v1, v2, v3] }
+            LogicalScan { table: t, columns: [t._row_id, t.v1, t.v2, t.v3] }
   optimized_logical_plan: |
     LogicalAgg { aggs: [count(1:Int32), count(t.v1)] }
       LogicalProject { exprs: [1:Int32, t.v1] }
-        LogicalScan { table: t, output_columns: [v1], required_columns: [v1, v2], predicate: (t.v2 > 1:Int32) }
+        LogicalScan { table: t, output_columns: [t.v1], required_columns: [v1, v2], predicate: (t.v2 > 1:Int32) }
 - sql: |
     /* hop window, time_col not selected */
     create table t1 (a int, b int, created_at timestamp);
     select a, window_end from hop(t1, created_at, interval '15' minute, interval '30' minute)
   logical_plan: |
     LogicalProject { exprs: [t1.a, window_end] }
       LogicalHopWindow { time_col: t1.created_at, slide: 00:15:00, size: 00:30:00, output: all }
-        LogicalScan { table: t1, columns: [_row_id, a, b, created_at] }
+        LogicalScan { table: t1, columns: [t1._row_id, t1.a, t1.b, t1.created_at] }
   optimized_logical_plan: |
     LogicalHopWindow { time_col: t1.created_at, slide: 00:15:00, size: 00:30:00, output: [t1.a, window_end] }
-      LogicalScan { table: t1, columns: [a, created_at] }
+      LogicalScan { table: t1, columns: [t1.a, t1.created_at] }
   batch_plan: |
     BatchHopWindow { time_col: t1.created_at, slide: 00:15:00, size: 00:30:00, output: [t1.a, window_end] }
       BatchExchange { order: [], dist: Single }
-        BatchScan { table: t1, columns: [a, created_at] }
+        BatchScan { table: t1, columns: [t1.a, t1.created_at], distribution: SomeShard }
   stream_plan: |
     StreamMaterialize { columns: [a, window_end, t1._row_id(hidden)], pk_columns: [t1._row_id, window_end] }
       StreamHopWindow { time_col: t1.created_at, slide: 00:15:00, size: 00:30:00, output: [t1.a, window_end, t1._row_id] }
-        StreamTableScan { table: t1, columns: [a, created_at, _row_id] }
+        StreamTableScan { table: t1, columns: [t1.a, t1.created_at, t1._row_id], pk: [t1._row_id], distribution: HashShard(t1._row_id) }