Merge branch 'main' into xx/buf

e2e_test/streaming/aggregate/jsonb_agg.slt

@@ -0,0 +1,46 @@
+statement ok
+SET RW_IMPLICIT_FLUSH TO true;
+
+statement ok
+create table t(v1 boolean, v2 int, v3 varchar, v4 jsonb);
+
+statement ok
+create materialized view mv_tmp as
+select jsonb_agg(v1) as j1 from t;
+
+statement ok
+drop materialized view mv_tmp;
+
+statement ok
+create materialized view mv1 as
+select
+    jsonb_agg(v1 order by v2) as j1,
+    jsonb_agg(v2 order by v2) as j2,
+    jsonb_object_agg(v3, v4) as j3
+from t;
+
+statement ok
+insert into t values
+    (null, 2, 'bbb', null),
+    (false, 1, 'ccc', 'null');
+
+query TTT
+select * from mv1;
+----
+[false, null]  [1, 2]  {"bbb": null, "ccc": null}
+
+statement ok
+insert into t values
+    (true, 0, 'bbb', '999'),
+    (true, 8, 'ddd', '{"foo": "bar"}');
+
+query TTT
+select * from mv1;
+----
+[true, false, null, true]  [0, 1, 2, 8]  {"bbb": 999, "ccc": null, "ddd": {"foo": "bar"}}
+
+statement ok
+drop materialized view mv1;
+
+statement ok
+drop table t;

src/expr/core/src/aggregate/def.rs

@@ -308,11 +308,7 @@ pub mod agg_kinds {
     #[macro_export]
     macro_rules! unimplemented_in_stream {
         () => {
-            AggKind::JsonbAgg
-                | AggKind::JsonbObjectAgg
-                | AggKind::PercentileCont
-                | AggKind::PercentileDisc
-                | AggKind::Mode
+            AggKind::PercentileCont | AggKind::PercentileDisc | AggKind::Mode
         };
     }
     pub use unimplemented_in_stream;

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

@@ -53,3 +53,45 @@
   expected_outputs:
   - batch_plan
   - optimized_logical_plan_for_batch
+- name: test merged union stream key (2 columns, row_id + src_col)
+  sql: |
+    create table t1 (a int, b numeric, c bigint);
+    create table t2 (a int, b numeric, c bigint);
+    create table t3 (a int, b numeric, c bigint);
+    create table t4 (a int, b numeric, c bigint);
+    create table t5 (a int, b numeric, c bigint);
+    select * from t1 union all select * from t2 union all select * from t3 union all select * from t4 union all select * from t5;
+  expected_outputs:
+    - batch_plan
+    - stream_plan
+    - stream_dist_plan
+- name: test merged union stream key (5 columns, row_id + src_col + a + b + c)
+  sql: |
+    create table t1 (a int, b numeric, c bigint, primary key (a));
+    create table t2 (a int, b numeric, c bigint, primary key (b));
+    create table t3 (a int, b numeric, c bigint, primary key (c));
+    create table t4 (a int, b numeric, c bigint);
+    create table t5 (a int, b numeric, c bigint, primary key (a, b));
+    select * from t1 union all select * from t2 union all select * from t3 union all select * from t4 union all select * from t5;
+  expected_outputs:
+    - stream_dist_plan
+- name: test merged union stream key (4 columns, row_id + src_col + a + b)
+  sql: |
+    create table t1 (a int, b numeric, c bigint, primary key (a));
+    create table t2 (a int, b numeric, c bigint, primary key (b));
+    create table t3 (a int, b numeric, c bigint);
+    create table t4 (a int, b numeric, c bigint);
+    create table t5 (a int, b numeric, c bigint, primary key (a, b));
+    select * from t1 union all select * from t2 union all select * from t3 union all select * from t4 union all select * from t5;
+  expected_outputs:
+    - stream_dist_plan
+- name: test merged union stream key (3 columns, src_col + a + b)
+  sql: |
+    create table t1 (a int, b numeric, c bigint, primary key (a));
+    create table t2 (a int, b numeric, c bigint, primary key (b));
+    create table t3 (a int, b numeric, c bigint, primary key (b));
+    create table t4 (a int, b numeric, c bigint, primary key (b, a));
+    create table t5 (a int, b numeric, c bigint, primary key (a, b));
+    select * from t1 union all select * from t2 union all select * from t3 union all select * from t4 union all select * from t5;
+  expected_outputs:
+    - stream_dist_plan

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

@@ -150,7 +150,7 @@
     create table t(a int, b int);
     with cte as (select count(*) from t) select * from cte union all select * from cte;
   stream_plan: |-
-    StreamMaterialize { columns: [count, 0:Int32(hidden)], stream_key: [0:Int32], pk_columns: [0:Int32], pk_conflict: NoCheck }
+    StreamMaterialize { columns: [count, $src(hidden)], stream_key: [$src], pk_columns: [$src], pk_conflict: NoCheck }
     └─StreamUnion { all: true }
       ├─StreamExchange { dist: HashShard(0:Int32) }
       │ └─StreamProject { exprs: [sum0(count), 0:Int32] }
@@ -173,7 +173,7 @@
     create table t(a int, b int);
     with cte as (select count(*) from t) select * from cte union all select * from cte;
   stream_plan: |-
-    StreamMaterialize { columns: [count, 0:Int32(hidden)], stream_key: [0:Int32], pk_columns: [0:Int32], pk_conflict: NoCheck }
+    StreamMaterialize { columns: [count, $src(hidden)], stream_key: [$src], pk_columns: [$src], pk_conflict: NoCheck }
     └─StreamUnion { all: true }
       ├─StreamExchange { dist: HashShard(0:Int32) }
       │ └─StreamProject { exprs: [sum0(count), 0:Int32] }