feat: new table stream graph

e2e_test/streaming/distinct_on.slt

@@ -5,7 +5,7 @@ statement ok
 create table t1 (k int, v int) append only;
 
 statement ok
-create materialized view mv1 as select distinct on (k) k + v as sum from t1;
+create materialized view mv1 as select distinct on (k) k + v as sum from t1 order by k, v;
 
 statement ok
 insert into t1 values (1,11), (2,22), (3,33), (1,111);
@@ -21,7 +21,7 @@ statement ok
 create table t2 (k int, v int);
 
 statement ok
-create materialized view mv2 as select distinct on (k) k + v as sum from t2;
+create materialized view mv2 as select distinct on (k) k + v as sum from t2 order by k, v;
 
 statement ok
 insert into t2 values (1,11), (2,22), (3,33), (1,111);

src/common/src/hash/consistent_hash/vnode.rs

@@ -119,10 +119,18 @@ impl VirtualNode {
         if let Ok(idx) = keys.iter().exactly_one()
             && let ArrayImpl::Serial(serial_array) = &**data_chunk.column_at(*idx)
         {
-            return serial_array
-                .iter()
-                .map(|serial| extract_vnode_id_from_row_id(serial.unwrap().as_row_id()))
-                .collect();
+            return serial_array.iter().enumerate().map(|(idx, serial)| {
+                if let Some(serial) = serial {
+                    extract_vnode_id_from_row_id(serial.as_row_id())
+                } else {
+                    // NOTE: here it will hash the entire row when the `_row_id` is missing,
+                    // which could result in rows from the same chunk being allocated to different chunks.
+                    // This process doesn’t guarantee the order of rows, producing indeterminate results in some cases,
+                    // such as when `distinct on` is used without an `order by`.
+                    let (row, _) = data_chunk.row_at(idx);
+                    row.hash(Crc32FastBuilder).into()
+                }
+            } ).collect();
         }
 
         data_chunk

src/frontend/planner_test/tests/testdata/output/create_source.yaml

@@ -21,10 +21,13 @@
     ) FORMAT PLAIN ENCODE CSV (delimiter = ',', without_header = true);
   explain_output: |
     StreamMaterialize { columns: [v1, v2, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: Overwrite }
-    └─StreamExchange { dist: HashShard(_row_id) }
-      └─StreamRowIdGen { row_id_index: 2 }
-        └─StreamDml { columns: [v1, v2, _row_id] }
-          └─StreamSource { source: s0, columns: [v1, v2, _row_id] }
+    └─StreamRowIdGen { row_id_index: 2 }
+      └─StreamUnion { all: true }
+        ├─StreamExchange [no_shuffle] { dist: SomeShard }
+        │ └─StreamSource { source: s0, columns: [v1, v2, _row_id] }
+        └─StreamExchange { dist: HashShard(_row_id) }
+          └─StreamDml { columns: [v1, v2, _row_id] }
+            └─StreamSource
 - id: csv_delimiter_tab
   sql: |
     explain create table s0 (v1 int, v2 varchar) with (
@@ -35,10 +38,13 @@
     ) FORMAT PLAIN ENCODE CSV (delimiter = E'\t', without_header = true);
   explain_output: |
     StreamMaterialize { columns: [v1, v2, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: Overwrite }
-    └─StreamExchange { dist: HashShard(_row_id) }
-      └─StreamRowIdGen { row_id_index: 2 }
-        └─StreamDml { columns: [v1, v2, _row_id] }
-          └─StreamSource { source: s0, columns: [v1, v2, _row_id] }
+    └─StreamRowIdGen { row_id_index: 2 }
+      └─StreamUnion { all: true }
+        ├─StreamExchange [no_shuffle] { dist: SomeShard }
+        │ └─StreamSource { source: s0, columns: [v1, v2, _row_id] }
+        └─StreamExchange { dist: HashShard(_row_id) }
+          └─StreamDml { columns: [v1, v2, _row_id] }
+            └─StreamSource
 - id: create_source_with_cdc_backfill
   sql: |
     create source mysql_mydb with (
@@ -57,8 +63,11 @@
     ) from mysql_mydb table 'mydb.t1';
   explain_output: |
     StreamMaterialize { columns: [v1, v2], stream_key: [v1], pk_columns: [v1], pk_conflict: Overwrite }
-    └─StreamExchange { dist: HashShard(mydb.t1.v1) }
-      └─StreamDml { columns: [v1, v2] }
-        └─StreamCdcTableScan { table: mydb.t1, columns: [v1, v2] }
+    └─StreamUnion { all: true }
+      ├─StreamExchange { dist: HashShard(mydb.t1.v1) }
+      │ └─StreamCdcTableScan { table: mydb.t1, columns: [v1, v2] }
+      └─StreamExchange { dist: HashShard(v1) }
+        └─StreamDml { columns: [v1, v2] }
+          └─StreamSource
   with_config_map:
     CDC_BACKFILL: 'true'

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

@@ -191,15 +191,19 @@
     explain create table t (v1 int, v2 varchar);
   explain_output: |
     StreamMaterialize { columns: [v1, v2, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: Overwrite }
-    └─StreamExchange { dist: HashShard(_row_id) }
-      └─StreamRowIdGen { row_id_index: 2 }
-        └─StreamDml { columns: [v1, v2, _row_id] }
-          └─StreamSource
+    └─StreamRowIdGen { row_id_index: 2 }
+      └─StreamUnion { all: true }
+        └─StreamExchange { dist: HashShard(_row_id) }
+          └─StreamDml { columns: [v1, v2, _row_id] }
+            └─StreamSource
 - sql: |
     explain create table t (v1 int, v2 varchar) with ( connector = 'kafka', kafka.topic = 'kafka_3_partition_topic', kafka.brokers = '127.0.0.1:1234', kafka.scan.startup.mode='earliest'  ) FORMAT PLAIN ENCODE JSON;
   explain_output: |
     StreamMaterialize { columns: [v1, v2, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: Overwrite }
-    └─StreamExchange { dist: HashShard(_row_id) }
-      └─StreamRowIdGen { row_id_index: 2 }
-        └─StreamDml { columns: [v1, v2, _row_id] }
-          └─StreamSource { source: t, columns: [v1, v2, _row_id] }
+    └─StreamRowIdGen { row_id_index: 2 }
+      └─StreamUnion { all: true }
+        ├─StreamExchange [no_shuffle] { dist: SomeShard }
+        │ └─StreamSource { source: t, columns: [v1, v2, _row_id] }
+        └─StreamExchange { dist: HashShard(_row_id) }
+          └─StreamDml { columns: [v1, v2, _row_id] }
+            └─StreamSource

src/frontend/planner_test/tests/testdata/output/generated_columns.yaml

@@ -4,11 +4,12 @@
     explain create table t1 (v1 int as v2-1, v2 int, v3 int as v2+1);
   explain_output: |
     StreamMaterialize { columns: [v1, v2, v3, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: Overwrite }
-    └─StreamExchange { dist: HashShard(_row_id) }
-      └─StreamRowIdGen { row_id_index: 3 }
-        └─StreamProject { exprs: [(v2 - 1:Int32) as $expr1, v2, (v2 + 1:Int32) as $expr2, _row_id] }
-          └─StreamDml { columns: [v2, _row_id] }
-            └─StreamSource
+    └─StreamRowIdGen { row_id_index: 3 }
+      └─StreamUnion { all: true }
+        └─StreamExchange { dist: HashShard(_row_id) }
+          └─StreamProject { exprs: [(v2 - 1:Int32) as $expr1, v2, (v2 + 1:Int32) as $expr2, _row_id] }
+            └─StreamDml { columns: [v2, _row_id] }
+              └─StreamSource
 - name: source with generated columns
   sql: |
     create source s1 (v1 int as v2-1, v2 int, v3 int as v2+1) with (connector = 'kinesis') FORMAT PLAIN ENCODE JSON;
@@ -30,18 +31,21 @@
     explain create table t1 (proc_time TIMESTAMP AS proctime());
   explain_output: |
     StreamMaterialize { columns: [proc_time, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: Overwrite, watermark_columns: [proc_time] }
-    └─StreamExchange { dist: HashShard(_row_id) }
-      └─StreamRowIdGen { row_id_index: 1 }
-        └─StreamProject { exprs: [AtTimeZone(Proctime, 'UTC':Varchar) as $expr1, _row_id], output_watermarks: [$expr1] }
-          └─StreamDml { columns: [_row_id] }
-            └─StreamSource
+    └─StreamRowIdGen { row_id_index: 1 }
+      └─StreamUnion { all: true, output_watermarks: [$expr1] }
+        └─StreamExchange { dist: HashShard(_row_id) }
+          └─StreamProject { exprs: [AtTimeZone(Proctime, 'UTC':Varchar) as $expr1, _row_id], output_watermarks: [$expr1] }
+            └─StreamDml { columns: [_row_id] }
+              └─StreamSource
 - name: watermark on generated column
   sql: |
     explain create table t (v int, w int as v+1, watermark for w as w) append only
   explain_output: |
     StreamMaterialize { columns: [v, w, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck, watermark_columns: [w] }
     └─StreamRowIdGen { row_id_index: 2 }
       └─StreamWatermarkFilter { watermark_descs: [Desc { column: $expr1, expr: $expr1 }], output_watermarks: [$expr1] }
-        └─StreamProject { exprs: [v, (v + 1:Int32) as $expr1, _row_id] }
-          └─StreamDml { columns: [v, _row_id] }
-            └─StreamSource
+        └─StreamUnion { all: true }
+          └─StreamExchange [no_shuffle] { dist: SomeShard }
+            └─StreamProject { exprs: [v, (v + 1:Int32) as $expr1, _row_id] }
+              └─StreamDml { columns: [v, _row_id] }
+                └─StreamSource

src/frontend/planner_test/tests/testdata/output/union.yaml

@@ -596,7 +596,7 @@
 
     Fragment 4
     StreamProject { exprs: [t4.a, t4.b, t4.c, 3:Int32] }
-    └── StreamTableScan { table: t4, columns: [t4.a, t4.b, t4.c], pk: [t4.b, t4.a], dist: UpstreamHashShard(t4.a, t4.b) }
+    └── StreamTableScan { table: t4, columns: [t4.a, t4.b, t4.c], pk: [t4.b, t4.a], dist: UpstreamHashShard(t4.b, t4.a) }
         ├── state table: 3
         ├── Upstream
         └── BatchPlanNode

src/frontend/planner_test/tests/testdata/output/watermark.yaml

@@ -51,17 +51,23 @@
     StreamMaterialize { columns: [v1, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck, watermark_columns: [v1] }
     └─StreamRowIdGen { row_id_index: 1 }
       └─StreamWatermarkFilter { watermark_descs: [Desc { column: v1, expr: (v1 - '00:00:01':Interval) }], output_watermarks: [v1] }
-        └─StreamDml { columns: [v1, _row_id] }
-          └─StreamSource { source: t, columns: [v1, _row_id] }
+        └─StreamUnion { all: true }
+          ├─StreamExchange [no_shuffle] { dist: SomeShard }
+          │ └─StreamSource { source: t, columns: [v1, _row_id] }
+          └─StreamExchange [no_shuffle] { dist: SomeShard }
+            └─StreamDml { columns: [v1, _row_id] }
+              └─StreamSource
 - name: watermark on append only table without source
   sql: |
     explain create table t (v1 timestamp with time zone, watermark for v1 as v1 - INTERVAL '1' SECOND) append only;
   explain_output: |
     StreamMaterialize { columns: [v1, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck, watermark_columns: [v1] }
     └─StreamRowIdGen { row_id_index: 1 }
       └─StreamWatermarkFilter { watermark_descs: [Desc { column: v1, expr: (v1 - '00:00:01':Interval) }], output_watermarks: [v1] }
-        └─StreamDml { columns: [v1, _row_id] }
-          └─StreamSource
+        └─StreamUnion { all: true }
+          └─StreamExchange [no_shuffle] { dist: SomeShard }
+            └─StreamDml { columns: [v1, _row_id] }
+              └─StreamSource
 - name: hash agg
   sql: |
     create table t (ts timestamp with time zone, v1 int, v2 int, watermark for ts as ts - INTERVAL '1' SECOND) append only;

src/frontend/src/handler/create_table.rs

@@ -698,6 +698,8 @@ fn gen_table_plan_inner(
     let connection_id =
         resolve_privatelink_in_with_option(&mut with_options, &schema_name, &session)?;
 
+    let is_external_source = source_info.is_some();
+
     let source = source_info.map(|source_info| PbSource {
         id: TableId::placeholder().table_id,
         schema_id,
@@ -776,6 +778,7 @@ fn gen_table_plan_inner(
         append_only,
         watermark_descs,
         version,
+        is_external_source,
     )?;
 
     let mut table = materialize.table().to_prost(schema_id, database_id);
@@ -884,8 +887,9 @@ pub(crate) fn gen_create_table_plan_for_cdc_source(
         pk_column_ids,
         None,
         append_only,
-        vec![], // no watermarks
+        vec![],
         Some(col_id_gen.into_version()),
+        true,
     )?;
 
     let mut table = materialize.table().to_prost(schema_id, database_id);