Merge branch 'main' into rc/rename-main1-slt

Cargo.toml

@@ -241,5 +241,6 @@ getrandom = { git = "https://github.com/madsim-rs/getrandom.git", rev = "8daf97e
 tokio-stream = { git = "https://github.com/madsim-rs/tokio.git", rev = "fe39bb8e" }
 tokio-retry = { git = "https://github.com/madsim-rs/rust-tokio-retry.git", rev = "95e2fd3" }
 tokio-postgres = { git = "https://github.com/madsim-rs/rust-postgres.git", rev = "ac00d88" }
+futures-timer = { git = "https://github.com/madsim-rs/futures-timer.git", rev = "05b33b4" }
 # patch: unlimit 4MB message size for grpc client
 etcd-client = { git = "https://github.com/risingwavelabs/etcd-client.git", rev = "4e84d40" }

ci/scripts/deterministic-it-test.sh

@@ -16,7 +16,6 @@ mv target/ci-sim target/sim
 echo "--- Run integration tests in deterministic simulation mode"
 seq $TEST_NUM | parallel MADSIM_TEST_SEED={} NEXTEST_PROFILE=ci-sim \
  cargo nextest run \
- --no-capture \
  --no-fail-fast \
  --cargo-metadata target/nextest/cargo-metadata.json \
  --binaries-metadata target/nextest/binaries-metadata.json \

ci/scripts/e2e-source-test.sh

@@ -66,6 +66,11 @@ sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.load.slt'
 sleep 10
 sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.check.slt'
 
+# cdc share stream test cases
+export MYSQL_HOST=mysql MYSQL_TCP_PORT=3306 MYSQL_PWD=123456
+sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.share_stream.slt'
+
+
 # kill cluster and the connector node
 cargo make kill
 echo "cluster killed "

e2e_test/batch/over_window/generated

@@ -0,0 +1 @@
+../../over_window/generated/batch/

e2e_test/batch/over_window/main.slt.part

@@ -1,5 +1 @@
-statement ok
-SET RW_IMPLICIT_FLUSH TO true;
-
-include ./special_cases/mod.slt.part
-include ./over_window/mod.slt.part
+include ./generated/main.slt.part

e2e_test/over_window/gen.py

@@ -25,19 +25,27 @@
     shutil.rmtree(mode_dir)
     os.makedirs(mode_dir, exist_ok=True)
 
-for file in os.listdir(templates_dir):
-    if not file.endswith(".slt") and not file.endswith(".slt.part"):
-        continue
 
-    print(f"Generating `{file}`...")
+def render(filepath: str):
+    relpath = path.relpath(filepath, templates_dir)
+    print(f"Rendering `{relpath}`...")
 
-    with open(path.join(templates_dir, file), "r") as f:
+    with open(path.join(templates_dir, relpath), "r") as f:
         tpl = Template(f.read())
 
     for mode, context in contexts.items():
-        out_file = path.join(generated_dir, mode, file)
+        out_file = path.join(generated_dir, mode, relpath)
+        os.makedirs(path.dirname(out_file), exist_ok=True)
         with open(out_file, "w") as f:
             f.write(file_head + "\n\n")
             f.write(tpl.safe_substitute(context))
 
+
+for dirpath, dirnames, filenames in os.walk(templates_dir):
+    for filename in filenames:
+        if not filename.endswith(".slt") and not filename.endswith(".slt.part"):
+            continue
+        render(path.join(dirpath, filename))
+
+
 print("Done.")

e2e_test/over_window/generated/batch/agg_in_win_func/mod.slt.part

@@ -0,0 +1,80 @@
+# This file is generated by `gen.py`. Do not edit it manually!
+
+# Test aggregate function calls as window function args/PARTITION BY/ORDER BY.
+
+statement ok
+create table t (
+      id int
+    , p1 int
+    , p2 int
+    , time int
+    , v1 int
+    , v2 int
+);
+
+statement ok
+create view v as
+select
+    p1, p2
+    , row_number() over (partition by p1 order by p2) as out1
+    , sum(sum(v2)) over (partition by p1, avg(time) order by max(v1), p2) as out2
+from t
+group by p1, p2;
+
+statement ok
+insert into t values
+  (100001, 100, 200, 1, 701, 805)
+, (100002, 100, 200, 2, 700, 806)
+, (100003, 100, 208, 2, 723, 807)
+, (100004, 103, 200, 2, 702, 808);
+
+query iiii
+select * from v order by p1, p2;
+----
+100  200  1  1611
+100  208  2   807
+103  200  1   808
+
+statement ok
+insert into t values
+  (100005, 100, 200, 3, 717, 810)
+, (100006, 105, 204, 5, 703, 828);
+
+query iiii
+select * from v order by p1, p2;
+----
+100  200  1  2421
+100  208  2  3228
+103  200  1   808
+105  204  1   828
+
+statement ok
+update t set v1 = 799 where id = 100002; -- value change
+
+statement ok
+update t set p2 = 200 where id = 100003; -- partition change
+
+statement ok
+update t set "time" = 1 where id = 100005; -- order change
+
+query iiiiiii
+select * from v order by p1, p2;
+----
+100  200  1  3228
+103  200  1   808
+105  204  1   828
+
+statement ok
+delete from t where time = 2;
+
+query iiii
+select * from v order by p1, p2;
+----
+100  200  1  1615
+105  204  1   828
+
+statement ok
+drop view v;
+
+statement ok
+drop table t;

e2e_test/over_window/generated/batch/mod.slt.part → e2e_test/over_window/generated/batch/basic/mod.slt.part

@@ -1,6 +1,8 @@
 # This file is generated by `gen.py`. Do not edit it manually!
 
-include ./create.slt.part
+# Test basic functionality of general batch and streaming over window.
+
+include ./setup.slt.part
 
 statement ok
 insert into t values
@@ -33,21 +35,6 @@ select * from v_c order by id;
 100003  100  208  2  723  807  723  NULL  NULL  NULL  NULL
 100004  103  200  2  702  808  702  NULL  NULL  NULL  NULL
 
-query II
-select * from v_d order by id;
-----
-100001  100  200  1  701  805  1  2
-100002  100  200  2  700  806  2  3
-100003  100  208  2  723  807  3  1
-100004  103  200  2  702  808  1  1
-
-query iiii
-select * from v_e order by p1;
-----
-100  200  1  1611
-100  208  2   807
-103  200  1   808
-
 include ./cross_check.slt.part
 
 statement ok
@@ -85,24 +72,6 @@ select * from v_c order by id;
 100005  100  200  3  717  810  717   700   700  NULL  NULL
 100006  105  204  5  703  828  703  NULL  NULL  NULL  NULL
 
-query II
-select * from v_d order by id;
-----
-100001  100  200  1  701  805  1  2
-100002  100  200  2  700  806  2  3
-100003  100  208  2  723  807  3  1
-100004  103  200  2  702  808  1  1
-100005  100  200  3  717  810  4  4
-100006  105  204  5  703  828  1  1
-
-query iiii
-select * from v_e order by p1, p2;
-----
-100  200  1  2421
-100  208  2  3228
-103  200  1   808
-105  204  1   828
-
 include ./cross_check.slt.part
 
 statement ok
@@ -144,33 +113,6 @@ select * from v_c order by id;
 100005  100  200  1  717  810  717   723   701   806   806
 100006  105  204  5  703  828  703  NULL  NULL  NULL  NULL
 
-query iiiiiii
-select * from v_d order by id;
-----
-100001  100  200  1  701  805  1  1
-100002  100  200  2  799  806  3  2
-100003  100  200  2  723  807  4  3
-100004  103  200  2  702  808  1  1
-100005  100  200  1  717  810  2  4
-100006  105  204  5  703  828  1  1
-
-query iiiiiii
-select * from v_e order by p1;
-----
-100  200  1  3228
-103  200  1   808
-105  204  1   828
-
-query iiiiiiiiii
-select * from v_expr order by id;
-----
-100001 100 200 1 701 805 805 0 701 NULL 1402
-100002 100 200 2 799 806 806 0 701 703 1446
-100003 100 200 2 723 807 807 0 701 801 1446
-100004 103 200 2 702 808 808 0 702 NULL 1404
-100005 100 200 1 717 810 810 0 701 725 1434
-100006 105 204 5 703 828 828 0 703 NULL 1406
-
 include ./cross_check.slt.part
 
 statement ok
@@ -197,26 +139,6 @@ select * from v_c order by id;
 100005  100  200  1  717  810  717   701   701  NULL  NULL
 100006  105  204  5  703  828  703  NULL  NULL  NULL  NULL
 
-query iiiiiii
-select * from v_d order by id;
-----
-100001  100  200  1  701  805  1  1
-100005  100  200  1  717  810  2  2
-100006  105  204  5  703  828  1  1
-
-query iiii
-select * from v_e order by p1;
-----
-100  200  1  1615
-105  204  1   828
-
-query iiiiiiiiii
-select * from v_expr order by id;
-----
-100001 100 200 1 701 805 805 0 701 NULL 1402
-100005 100 200 1 717 810 810 0 701 703 1434
-100006 105 204 5 703 828 828 0 703 NULL 1406
-
 include ./cross_check.slt.part
 
-include ./drop.slt.part
+include ./teardown.slt.part

e2e_test/over_window/generated/batch/create.slt.part → e2e_test/over_window/generated/batch/basic/setup.slt.part

@@ -40,25 +40,6 @@ select
     , lead(v2, 2) over (partition by p1, p2 order by v1, v2) as out9
 from t;
 
-# row_number
-statement ok
-create view v_d as
-select
-    *
-    , row_number() over (partition by p1 order by time, id) as out10
-    , row_number() over (partition by p1 order by p2 desc, id) as out11
-from t;
-
-# over + agg
-statement ok
-create view v_e as
-select
-    p1, p2
-    , row_number() over (partition by p1 order by p2) as out12
-    , sum(sum(v2)) over (partition by p1, avg(time) order by max(v1), p2) as out13
-from t
-group by p1, p2;
-
 statement ok
 create view v_a_b as
 select
@@ -103,14 +84,3 @@ select
     , lead(v2, 1) over (partition by p1, p2 order by time, id) as out8
     , lead(v2, 2) over (partition by p1, p2 order by v1, v2) as out9
 from t;
-
-statement ok
-create view v_expr as
-select
-    *
-    , t.v2 as out1
-    , 0 as out2
-    , first_value(v1) over (partition by p1, p2 order by time, id rows 3 preceding) as out3
-    , lag(v1 + 2, 0 + 1) over (partition by p1 - 1 order by id) as out4
-    , min(v1 * 2) over (partition by p1, p2 order by time + 1, id rows between current row and unbounded following) as out5
-from t;

e2e_test/over_window/generated/batch/drop.slt.part → e2e_test/over_window/generated/batch/basic/teardown.slt.part

@@ -9,12 +9,6 @@ drop view v_b;
 statement ok
 drop view v_c;
 
-statement ok
-drop view v_d;
-
-statement ok
-drop view v_e;
-
 statement ok
 drop view v_a_b;
 
@@ -27,8 +21,5 @@ drop view v_a_c;
 statement ok
 drop view v_a_b_c;
 
-statement ok
-drop view v_expr;
-
 statement ok
 drop table t;