fix(explain): add missed fields in explain streaming dist plan (risingwavelabs#8544)

Author: st1page

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

@@ -668,12 +668,12 @@
         Upstream
         BatchPlanNode
 
-     Table 0 { columns: [t_src, t_dst, t_src_0, t_dst_0, t__row_id, t__row_id_0], primary key: [$2 ASC, $0 ASC, $4 ASC, $5 ASC, $1 ASC], value indices: [0, 1, 2, 3, 4, 5], distribution key: [0] }
-     Table 1 { columns: [t_src, t_src_0, t__row_id, t__row_id_0, t_dst, _degree], primary key: [$0 ASC, $1 ASC, $2 ASC, $3 ASC, $4 ASC], value indices: [5], distribution key: [1] }
-     Table 2 { columns: [t_dst, t__row_id], primary key: [$0 ASC, $0 ASC, $1 ASC], value indices: [0, 1], distribution key: [0] }
-     Table 3 { columns: [t_dst, t_dst_0, t__row_id, _degree], primary key: [$0 ASC, $1 ASC, $2 ASC], value indices: [3], distribution key: [1] }
-     Table 4 { columns: [t_src, t_dst, t__row_id], primary key: [$1 ASC, $2 ASC], value indices: [0, 1, 2], distribution key: [1] }
-     Table 5 { columns: [t_dst, t__row_id, _degree], primary key: [$0 ASC, $1 ASC], value indices: [2], distribution key: [0] }
-     Table 6 { columns: [t_src, t_dst, t__row_id], primary key: [$0 ASC, $2 ASC], value indices: [0, 1, 2], distribution key: [0] }
-     Table 7 { columns: [t_src, t__row_id, _degree], primary key: [$0 ASC, $1 ASC], value indices: [2], distribution key: [0] }
-     Table 4294967294 { columns: [p1, p2, p3, t._row_id, t._row_id#1, t.src, t._row_id#2], primary key: [$3 ASC, $4 ASC, $1 ASC, $6 ASC, $5 ASC, $0 ASC], value indices: [0, 1, 2, 3, 4, 5, 6], distribution key: [0] }
+     Table 0 { columns: [t_src, t_dst, t_src_0, t_dst_0, t__row_id, t__row_id_0], primary key: [$2 ASC, $0 ASC, $4 ASC, $5 ASC, $1 ASC], value indices: [0, 1, 2, 3, 4, 5], distribution key: [0], read pk prefix len hint: 2 }
+     Table 1 { columns: [t_src, t_src_0, t__row_id, t__row_id_0, t_dst, _degree], primary key: [$0 ASC, $1 ASC, $2 ASC, $3 ASC, $4 ASC], value indices: [5], distribution key: [1], read pk prefix len hint: 2 }
+     Table 2 { columns: [t_dst, t__row_id], primary key: [$0 ASC, $0 ASC, $1 ASC], value indices: [0, 1], distribution key: [0], read pk prefix len hint: 2 }
+     Table 3 { columns: [t_dst, t_dst_0, t__row_id, _degree], primary key: [$0 ASC, $1 ASC, $2 ASC], value indices: [3], distribution key: [1], read pk prefix len hint: 2 }
+     Table 4 { columns: [t_src, t_dst, t__row_id], primary key: [$1 ASC, $2 ASC], value indices: [0, 1, 2], distribution key: [1], read pk prefix len hint: 1 }
+     Table 5 { columns: [t_dst, t__row_id, _degree], primary key: [$0 ASC, $1 ASC], value indices: [2], distribution key: [0], read pk prefix len hint: 1 }
+     Table 6 { columns: [t_src, t_dst, t__row_id], primary key: [$0 ASC, $2 ASC], value indices: [0, 1, 2], distribution key: [0], read pk prefix len hint: 1 }
+     Table 7 { columns: [t_src, t__row_id, _degree], primary key: [$0 ASC, $1 ASC], value indices: [2], distribution key: [0], read pk prefix len hint: 1 }
+     Table 4294967294 { columns: [p1, p2, p3, t._row_id, t._row_id#1, t.src, t._row_id#2], primary key: [$3 ASC, $4 ASC, $1 ASC, $6 ASC, $5 ASC, $0 ASC], value indices: [0, 1, 2, 3, 4, 5, 6], distribution key: [0], read pk prefix len hint: 6 }

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

@@ -56,3 +56,34 @@
       └─StreamExchange { dist: Broadcast }
         └─StreamProject { exprs: [(AtTimeZone((AtTimeZone(now, 'UTC':Varchar) - '00:00:00':Interval), 'UTC':Varchar) - '01:00:00':Interval) as $expr2], output_watermarks: [$expr2] }
           └─StreamNow { output: [now] }
+  stream_dist_plan: |
+    Fragment 0
+      StreamMaterialize { columns: [ts, t1._row_id(hidden)], pk_columns: [t1._row_id], pk_conflict: "no check" }
+          materialized table: 4294967294
+        StreamDynamicFilter { predicate: (t1.ts < $expr2), output: [t1.ts, t1._row_id] }
+            left table: 0, right table 1
+          StreamDynamicFilter { predicate: (t1.ts >= $expr1), output_watermarks: [t1.ts], output: [t1.ts, t1._row_id] }
+              left table: 2, right table 3
+            Chain { table: t1, columns: [t1.ts, t1._row_id], pk: [t1._row_id], dist: UpstreamHashShard(t1._row_id) }
+              Upstream
+              BatchPlanNode
+            StreamExchange Broadcast from 1
+          StreamExchange Broadcast from 2
+
+    Fragment 1
+      StreamProject { exprs: [(AtTimeZone((AtTimeZone(now, 'UTC':Varchar) - '00:00:00':Interval), 'UTC':Varchar) - '02:00:00':Interval) as $expr1], output_watermarks: [$expr1] }
+        StreamNow { output: [now] }
+            state table: 4
+
+    Fragment 2
+      StreamProject { exprs: [(AtTimeZone((AtTimeZone(now, 'UTC':Varchar) - '00:00:00':Interval), 'UTC':Varchar) - '01:00:00':Interval) as $expr2], output_watermarks: [$expr2] }
+        StreamNow { output: [now] }
+            state table: 5
+
+     Table 0 { columns: [t1_ts, t1__row_id], primary key: [$0 ASC, $1 ASC], value indices: [0, 1], distribution key: [1], read pk prefix len hint: 0 }
+     Table 1 { columns: [$expr2], primary key: [], value indices: [0], distribution key: [], read pk prefix len hint: 0 }
+     Table 2 { columns: [t1_ts, t1__row_id], primary key: [$0 ASC, $1 ASC], value indices: [0, 1], distribution key: [1], read pk prefix len hint: 0 }
+     Table 3 { columns: [$expr1], primary key: [], value indices: [0], distribution key: [], read pk prefix len hint: 0 }
+     Table 4 { columns: [now], primary key: [], value indices: [0], distribution key: [], read pk prefix len hint: 0 }
+     Table 5 { columns: [now], primary key: [], value indices: [0], distribution key: [], read pk prefix len hint: 0 }
+     Table 4294967294 { columns: [ts, t1._row_id], primary key: [$1 ASC], value indices: [0, 1], distribution key: [1], read pk prefix len hint: 1 }