update planner test

Signed-off-by: Richard Chien <[email protected]>
risingwavelabs · Oct 24, 2023 · c010792 · c010792
1 parent 6b35e9c
commit c010792
Show file tree

Hide file tree

Showing 2 changed files with 100 additions and 11 deletions.
diff --git a/src/frontend/planner_test/tests/testdata/input/agg.yaml b/src/frontend/planner_test/tests/testdata/input/agg.yaml
@@ -912,12 +912,12 @@
   expected_outputs:
   - batch_plan
   - stream_plan
+  - stream_dist_plan # check the state table schema
 - sql: |
     create table t (x int, y int);
     select first_value(distinct x order by x asc) from t;
   expected_outputs:
-  - batch_plan
-  - stream_plan
+  - stream_dist_plan # check the state table schema
 - sql: |
     create table t (x int, y int);
     select last_value(x order by y desc nulls last) from t;

diff --git a/src/frontend/planner_test/tests/testdata/output/agg.yaml b/src/frontend/planner_test/tests/testdata/output/agg.yaml
@@ -1613,19 +1613,108 @@
       └─StreamSimpleAgg { aggs: [first_value(t.x order_by(t.y ASC)), count] }
         └─StreamExchange { dist: Single }
           └─StreamTableScan { table: t, columns: [t.x, t.y, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
+  stream_dist_plan: |+
+    Fragment 0
+    StreamMaterialize { columns: [first_value], stream_key: [], pk_columns: [], pk_conflict: NoCheck }
+    ├── materialized table: 4294967294
+    └── StreamProject { exprs: [first_value(t.x order_by(t.y ASC))] }
+        └── StreamSimpleAgg { aggs: [first_value(t.x order_by(t.y ASC)), count] }
+            ├── intermediate state table: 1
+            ├── state tables: [ 0 ]
+            ├── distinct tables: []
+            └── StreamExchange Single from 1
+
+    Fragment 1
+    Chain { table: t, columns: [t.x, t.y, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
+    ├── state table: 2
+    ├── Upstream
+    └── BatchPlanNode
+
+    Table 0
+    ├── columns: [ t_y, t__row_id, t_x ]
+    ├── primary key: [ $0 ASC, $1 ASC ]
+    ├── value indices: [ 0, 1, 2 ]
+    ├── distribution key: []
+    └── read pk prefix len hint: 0
+
+    Table 1
+    ├── columns: [ first_value(t_x order_by(t_y ASC)), count ]
+    ├── primary key: []
+    ├── value indices: [ 0, 1 ]
+    ├── distribution key: []
+    └── read pk prefix len hint: 0
+
+    Table 2
+    ├── columns: [ vnode, _row_id, t_backfill_finished, t_row_count ]
+    ├── primary key: [ $0 ASC ]
+    ├── value indices: [ 1, 2, 3 ]
+    ├── distribution key: [ 0 ]
+    ├── read pk prefix len hint: 1
+    └── vnode column idx: 0
+
+    Table 4294967294
+    ├── columns: [ first_value ]
+    ├── primary key: []
+    ├── value indices: [ 0 ]
+    ├── distribution key: []
+    └── read pk prefix len hint: 0
+
 - sql: |
     create table t (x int, y int);
     select first_value(distinct x order by x asc) from t;
-  batch_plan: |-
-    BatchSimpleAgg { aggs: [first_value(distinct t.x order_by(t.x ASC))] }
-    └─BatchExchange { order: [], dist: Single }
-      └─BatchScan { table: t, columns: [t.x], distribution: SomeShard }
-  stream_plan: |-
+  stream_dist_plan: |+
+    Fragment 0
     StreamMaterialize { columns: [first_value], stream_key: [], pk_columns: [], pk_conflict: NoCheck }
-    └─StreamProject { exprs: [first_value(distinct t.x order_by(t.x ASC))] }
-      └─StreamSimpleAgg { aggs: [first_value(distinct t.x order_by(t.x ASC)), count] }
-        └─StreamExchange { dist: Single }
-          └─StreamTableScan { table: t, columns: [t.x, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
+    ├── materialized table: 4294967294
+    └── StreamProject { exprs: [first_value(distinct t.x order_by(t.x ASC))] }
+        └── StreamSimpleAgg { aggs: [first_value(distinct t.x order_by(t.x ASC)), count] }
+            ├── intermediate state table: 1
+            ├── state tables: [ 0 ]
+            ├── distinct tables: [ (distinct key: t.x, table id: 2) ]
+            └── StreamExchange Single from 1
+
+    Fragment 1
+    Chain { table: t, columns: [t.x, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
+    ├── state table: 3
+    ├── Upstream
+    └── BatchPlanNode
+
+    Table 0
+    ├── columns: [ t_x ]
+    ├── primary key: [ $0 ASC ]
+    ├── value indices: [ 0 ]
+    ├── distribution key: []
+    └── read pk prefix len hint: 0
+
+    Table 1
+    ├── columns: [ first_value(distinct t_x order_by(t_x ASC)), count ]
+    ├── primary key: []
+    ├── value indices: [ 0, 1 ]
+    ├── distribution key: []
+    └── read pk prefix len hint: 0
+
+    Table 2
+    ├── columns: [ t_x, count_for_agg_call_0 ]
+    ├── primary key: [ $0 ASC ]
+    ├── value indices: [ 1 ]
+    ├── distribution key: []
+    └── read pk prefix len hint: 1
+
+    Table 3
+    ├── columns: [ vnode, _row_id, t_backfill_finished, t_row_count ]
+    ├── primary key: [ $0 ASC ]
+    ├── value indices: [ 1, 2, 3 ]
+    ├── distribution key: [ 0 ]
+    ├── read pk prefix len hint: 1
+    └── vnode column idx: 0
+
+    Table 4294967294
+    ├── columns: [ first_value ]
+    ├── primary key: []
+    ├── value indices: [ 0 ]
+    ├── distribution key: []
+    └── read pk prefix len hint: 0
+
 - sql: |
     create table t (x int, y int);
     select last_value(x order by y desc nulls last) from t;