Merge branch 'main' into yiming/clear-with-version-id

Cargo.toml

@@ -133,6 +133,7 @@ tonic = { package = "madsim-tonic", version = "0.4.1" }
 tonic-build = { package = "madsim-tonic-build", version = "0.4.2" }
 otlp-embedded = { git = "https://github.com/risingwavelabs/otlp-embedded", rev = "492c244e0be91feb659c0cd48a624bbd96045a33" }
 prost = { version = "0.12" }
+prost-build = { version = "0.12" }
 icelake = { git = "https://github.com/icelake-io/icelake", rev = "07d53893d7788b4e41fc11efad8a6be828405c31", features = [
     "prometheus",
 ] }

ci/scripts/sql/nexmark/q0-temporal-filter.drop.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q0_temporal_filter;

ci/scripts/sql/nexmark/q0-temporal-filter.sql

@@ -0,0 +1,7 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q0_temporal_filter
+AS
+SELECT auction, bidder, price, date_time
+FROM bid_filtered
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q0.sql

@@ -4,4 +4,4 @@ CREATE SINK nexmark_q0
 AS
 SELECT auction, bidder, price, date_time
 FROM bid
-WITH ( connector = 'blackhole', type = 'append-only');
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q1-temporal-filter.drop.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q1_temporal_filter;

ci/scripts/sql/nexmark/q1-temporal-filter.sql

@@ -0,0 +1,10 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q1_temporal_filter
+AS
+SELECT auction,
+       bidder,
+       0.908 * price as price,
+       date_time
+FROM bid_filtered
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q1.sql

@@ -7,4 +7,4 @@ SELECT auction,
        0.908 * price as price,
        date_time
 FROM bid
-WITH ( connector = 'blackhole', type = 'append-only');
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q10-temporal-filter.drop.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q10_temporal_filter;

ci/scripts/sql/nexmark/q10-temporal-filter.sql

@@ -0,0 +1,11 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q10_temporal_filter AS
+SELECT auction,
+       bidder,
+       price,
+       date_time,
+       TO_CHAR(date_time, 'YYYY-MM-DD') as date,
+       TO_CHAR(date_time, 'HH:MI')      as time
+FROM bid_filtered
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q10.sql

@@ -8,4 +8,4 @@ SELECT auction,
        TO_CHAR(date_time, 'YYYY-MM-DD') as date,
        TO_CHAR(date_time, 'HH:MI')      as time
 FROM bid
-WITH ( connector = 'blackhole', type = 'append-only');
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q101-temporal-filter.drop.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q101_temporal_filter;

ci/scripts/sql/nexmark/q101-temporal-filter.sql

@@ -0,0 +1,16 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q101_temporal_filter AS
+SELECT
+    a.id AS auction_id,
+    a.item_name AS auction_item_name,
+    b.max_price AS current_highest_bid
+FROM auction a
+LEFT OUTER JOIN (
+    SELECT
+        b1.auction,
+        MAX(b1.price) max_price
+    FROM bid_filtered b1
+    GROUP BY b1.auction
+) b ON a.id = b.auction
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q102-temporal-filter.drop.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q102_temporal_filter;

ci/scripts/sql/nexmark/q102-temporal-filter.sql

@@ -0,0 +1,14 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q102_temporal_filter AS
+SELECT
+    a.id AS auction_id,
+    a.item_name AS auction_item_name,
+    COUNT(b.auction) AS bid_count
+FROM auction a
+JOIN bid_filtered b ON a.id = b.auction
+GROUP BY a.id, a.item_name
+HAVING COUNT(b.auction) >= (
+    SELECT COUNT(*) / COUNT(DISTINCT auction) FROM bid_filtered
+)
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q103-temporal-filter.drop.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q103_temporal_filter;

ci/scripts/sql/nexmark/q103-temporal-filter.sql

@@ -0,0 +1,13 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q103_temporal_filter AS
+SELECT
+    a.id AS auction_id,
+    a.item_name AS auction_item_name
+FROM auction a
+WHERE a.id IN (
+    SELECT b.auction FROM bid_filtered b
+    GROUP BY b.auction
+    HAVING COUNT(*) >= 20
+)
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q104-temporal-filter.drop.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q104_temporal_filter;

ci/scripts/sql/nexmark/q104-temporal-filter.sql

@@ -0,0 +1,13 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q104_temporal_filter AS
+SELECT
+    a.id AS auction_id,
+    a.item_name AS auction_item_name
+FROM auction a
+WHERE a.id NOT IN (
+    SELECT b.auction FROM bid_filtered b
+    GROUP BY b.auction
+    HAVING COUNT(*) < 20
+)
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q105-temporal-filter.drop.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q105_temporal_filter;

ci/scripts/sql/nexmark/q105-temporal-filter.sql

@@ -0,0 +1,13 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q105_temporal_filter AS
+SELECT
+    a.id AS auction_id,
+    a.item_name AS auction_item_name,
+    COUNT(b.auction) AS bid_count
+FROM auction a
+JOIN bid_filtered b ON a.id = b.auction
+GROUP BY a.id, a.item_name
+ORDER BY bid_count DESC
+LIMIT 1000
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q105-without-limit-temporal-filter.drop.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q105_without_limit_temporal_filter;

ci/scripts/sql/nexmark/q105-without-limit-temporal-filter.sql

@@ -0,0 +1,11 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q105_without_limit_temporal_filter AS
+SELECT
+    a.id AS auction_id,
+    a.item_name AS auction_item_name,
+    COUNT(b.auction) AS bid_count
+FROM auction a
+JOIN bid_filtered b ON a.id = b.auction
+GROUP BY a.id, a.item_name
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q105-without-limit.drop.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q105_without_limit;

ci/scripts/sql/nexmark/q105-without-limit.sql

@@ -0,0 +1,11 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q105_without_limit AS
+SELECT
+    a.id AS auction_id,
+    a.item_name AS auction_item_name,
+    COUNT(b.auction) AS bid_count
+FROM auction a
+JOIN bid b ON a.id = b.auction
+GROUP BY a.id, a.item_name
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q106.drop.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q106;

ci/scripts/sql/nexmark/q106.sql

@@ -0,0 +1,21 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q106
+AS
+SELECT
+    MIN(final) AS min_final
+FROM
+    (
+        SELECT
+            auction.id,
+            MAX(price) AS final
+        FROM
+            auction,
+            bid
+        WHERE
+            bid.auction = auction.id
+            AND bid.date_time BETWEEN auction.date_time AND auction.expires
+        GROUP BY
+            auction.id
+    )
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q107.drop.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q107;

ci/scripts/sql/nexmark/q107.sql

@@ -0,0 +1,11 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q107 AS
+SELECT
+    approx_percentile(0.01, 0.01) within group (order by price) as p01,
+    approx_percentile(0.1, 0.01) within group (order by price) as p10,
+    approx_percentile(0.5, 0.01) within group (order by price) as p50,
+    approx_percentile(0.9, 0.01) within group (order by price) as p90,
+    approx_percentile(0.99, 0.01) within group (order by price) as p99
+FROM bid
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q108.drop.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q108;

ci/scripts/sql/nexmark/q108.sql

@@ -0,0 +1,13 @@
+-- test two-phase simple approx percentile and merge
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q108 AS
+SELECT
+    approx_percentile(0.01, 0.01) within group (order by price) as p01,
+    approx_percentile(0.1, 0.01) within group (order by price) as p10,
+    approx_percentile(0.5, 0.01) within group (order by price) as p50,
+    approx_percentile(0.9, 0.01) within group (order by price) as p90,
+    approx_percentile(0.99, 0.01) within group (order by price) as p99
+FROM bid
+GROUP BY auction
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q12-temporal-filter.drop.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q12_temporal_filter;

ci/scripts/sql/nexmark/q12-temporal-filter.sql

@@ -0,0 +1,7 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q12_temporal_filter AS
+SELECT bidder, count(*) as bid_count, window_start, window_end
+FROM TUMBLE(bid_filtered, p_time, INTERVAL '10' SECOND)
+GROUP BY bidder, window_start, window_end
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q13-by-row-id.drop.sql

@@ -0,0 +1,4 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q13_by_row_id;
+DROP TABLE side_input;

ci/scripts/sql/nexmark/q13-by-row-id.sql

@@ -0,0 +1,12 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE TABLE side_input(
+    key BIGINT PRIMARY KEY,
+    value VARCHAR
+);
+INSERT INTO side_input SELECT v, v::varchar FROM generate_series(0, ${BENCHMARK_NEXMARK_RISINGWAVE_Q13_SIDE_INPUT_ROW_COUNT} - 1) AS s(v);
+
+CREATE SINK nexmark_q13_by_row_id AS
+SELECT B.auction, B.bidder, B.price, B.date_time, S.value
+FROM bid B join side_input FOR SYSTEM_TIME AS OF PROCTIME() S on mod(B._row_id::bigint, ${BENCHMARK_NEXMARK_RISINGWAVE_Q13_SIDE_INPUT_ROW_COUNT}) = S.key
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q13-non-lookup-cond.drop.sql

@@ -0,0 +1,4 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q13_non_lookup_cond;
+DROP TABLE side_input;

ci/scripts/sql/nexmark/q13-non-lookup-cond.sql

@@ -0,0 +1,14 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE TABLE side_input(
+    key BIGINT PRIMARY KEY,
+    value VARCHAR
+);
+INSERT INTO side_input SELECT v, v::varchar FROM generate_series(0, ${BENCHMARK_NEXMARK_RISINGWAVE_Q13_SIDE_INPUT_ROW_COUNT} - 1) AS s(v);
+
+CREATE SINK nexmark_q13_non_lookup_cond AS
+SELECT B.auction, B.bidder, B.price, B.date_time, S.value
+FROM bid B join side_input FOR SYSTEM_TIME AS OF PROCTIME() S
+ON mod(B.auction, ${BENCHMARK_NEXMARK_RISINGWAVE_Q13_SIDE_INPUT_ROW_COUNT}) = S.key
+  AND S.key % 4 != 1
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q13.sql

@@ -1,11 +1,12 @@
 -- noinspection SqlNoDataSourceInspectionForFile
 -- noinspection SqlResolveForFile
 CREATE TABLE side_input(
-  key BIGINT PRIMARY KEY,
-  value VARCHAR
+    key BIGINT PRIMARY KEY,
+    value VARCHAR
 );
-INSERT INTO side_input SELECT i::bigint, i::varchar FROM(SELECT generate_series(0,9999,1) as i);
+INSERT INTO side_input SELECT v, v::varchar FROM generate_series(0, ${BENCHMARK_NEXMARK_RISINGWAVE_Q13_SIDE_INPUT_ROW_COUNT} - 1) AS s(v);
+
 CREATE SINK nexmark_q13 AS
 SELECT B.auction, B.bidder, B.price, B.date_time, S.value
-FROM bid B join side_input FOR SYSTEM_TIME AS OF PROCTIME() S on mod(B.auction, 10000) = S.key
-WITH ( connector = 'blackhole', type = 'append-only');
+FROM bid B join side_input FOR SYSTEM_TIME AS OF PROCTIME() S on mod(B.auction, ${BENCHMARK_NEXMARK_RISINGWAVE_Q13_SIDE_INPUT_ROW_COUNT}) = S.key
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q14-temporal-filter.drop.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q14_temporal_filter;

ci/scripts/sql/nexmark/q14-temporal-filter.sql

@@ -0,0 +1,26 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q14_temporal_filter AS
+SELECT auction,
+       bidder,
+       0.908 * price as price,
+       CASE
+           WHEN
+                       extract(hour from date_time) >= 8 AND
+                       extract(hour from date_time) <= 18
+               THEN 'dayTime'
+           WHEN
+                       extract(hour from date_time) <= 6 OR
+                       extract(hour from date_time) >= 20
+               THEN 'nightTime'
+           ELSE 'otherTime'
+           END       AS bidTimeType,
+       date_time
+       -- extra
+       -- TODO: count_char is an UDF, add it back when we support similar functionality.
+       -- https://github.com/nexmark/nexmark/blob/master/nexmark-flink/src/main/java/com/github/nexmark/flink/udf/CountChar.java
+       -- count_char(extra, 'c') AS c_counts
+FROM bid_filtered
+WHERE 0.908 * price > 1000000
+  AND 0.908 * price < 50000000
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q14.drop.sql

@@ -1,4 +1,4 @@
 -- noinspection SqlNoDataSourceInspectionForFile
 -- noinspection SqlResolveForFile
 DROP SINK nexmark_q14;
-DROP FUNCTION count_char;
+DROP FUNCTION count_char;