test: add create and drop sink test for nexmark

e2e_test/source/nexmark_endless_sinks/nexmark_endless_part1.slt

@@ -0,0 +1,22 @@
+include ../../nexmark/create_sources.slt.part
+
+include ../../streaming/nexmark/sinks/q0.slt.part
+include ../../streaming/nexmark/sinks/q1.slt.part
+include ../../streaming/nexmark/sinks/q2.slt.part
+include ../../streaming/nexmark/sinks/q3.slt.part
+
+sleep 20s
+
+statement ok
+drop sink nexmark_q0;
+
+statement ok
+drop sink nexmark_q1;
+
+statement ok
+drop sink nexmark_q2;
+
+statement ok
+drop sink nexmark_q3;
+
+include ../../nexmark/drop_sources.slt.part

e2e_test/source/nexmark_endless_sinks/nexmark_endless_part2.slt

@@ -0,0 +1,22 @@
+include ../../nexmark/create_sources.slt.part
+
+include ../../streaming/nexmark/sinks/q4.slt.part
+include ../../streaming/nexmark/sinks/q5.slt.part
+include ../../streaming/nexmark/sinks/q7.slt.part
+include ../../streaming/nexmark/sinks/q8.slt.part
+
+sleep 20s
+
+statement ok
+drop sink nexmark_q4;
+
+statement ok
+drop sink nexmark_q5;
+
+statement ok
+drop sink nexmark_q7;
+
+statement ok
+drop sink nexmark_q8;
+
+include ../../nexmark/drop_sources.slt.part

e2e_test/source/nexmark_endless_sinks/nexmark_endless_part3.slt

@@ -0,0 +1,22 @@
+include ../../nexmark/create_sources.slt.part
+
+include ../../streaming/nexmark/sinks/q9.slt.part
+include ../../streaming/nexmark/sinks/q10.slt.part
+include ../../streaming/nexmark/sinks/q14.slt.part
+include ../../streaming/nexmark/sinks/q15.slt.part
+
+sleep 20s
+
+statement ok
+drop sink nexmark_q9;
+
+statement ok
+drop sink nexmark_q10;
+
+statement ok
+drop sink nexmark_q14;
+
+statement ok
+drop sink nexmark_q15;
+
+include ../../nexmark/drop_sources.slt.part

e2e_test/source/nexmark_endless_sinks/nexmark_endless_part4.slt

@@ -0,0 +1,22 @@
+include ../../nexmark/create_sources.slt.part
+
+include ../../streaming/nexmark/sinks/q16.slt.part
+include ../../streaming/nexmark/sinks/q17.slt.part
+include ../../streaming/nexmark/sinks/q18.slt.part
+include ../../streaming/nexmark/sinks/q20.slt.part
+
+sleep 20s
+
+statement ok
+drop sink nexmark_q16;
+
+statement ok
+drop sink nexmark_q17;
+
+statement ok
+drop sink nexmark_q18;
+
+statement ok
+drop sink nexmark_q20;
+
+include ../../nexmark/drop_sources.slt.part

e2e_test/source/nexmark_endless_sinks/nexmark_endless_part5.slt

@@ -0,0 +1,22 @@
+include ../../nexmark/create_sources.slt.part
+
+include ../../streaming/nexmark/sinks/q21.slt.part
+include ../../streaming/nexmark/sinks/q22.slt.part
+include ../../streaming/nexmark/sinks/q101.slt.part
+include ../../streaming/nexmark/sinks/q102.slt.part
+
+sleep 20s
+
+statement ok
+drop sink nexmark_q21;
+
+statement ok
+drop sink nexmark_q22;
+
+statement ok
+drop sink nexmark_q101;
+
+statement ok
+drop sink nexmark_q102;
+
+include ../../nexmark/drop_sources.slt.part

e2e_test/source/nexmark_endless_sinks/nexmark_endless_part6.slt

@@ -0,0 +1,22 @@
+include ../../nexmark/create_sources.slt.part
+
+include ../../streaming/nexmark/sinks/q103.slt.part
+include ../../streaming/nexmark/sinks/q104.slt.part
+include ../../streaming/nexmark/sinks/q105.slt.part
+include ../../streaming/nexmark/sinks/q106.slt.part
+
+sleep 20s
+
+statement ok
+drop sink nexmark_q103;
+
+statement ok
+drop sink nexmark_q104;
+
+statement ok
+drop sink nexmark_q105;
+
+statement ok
+drop sink nexmark_q106;
+
+include ../../nexmark/drop_sources.slt.part

e2e_test/streaming/nexmark/sinks/q0.slt.part

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

e2e_test/streaming/nexmark/sinks/q1.slt.part

@@ -0,0 +1,10 @@
+statement ok
+CREATE SINK nexmark_q1
+AS
+SELECT
+    auction,
+    bidder,
+    0.908 * price as price,
+    date_time
+FROM bid
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

e2e_test/streaming/nexmark/sinks/q10.slt.part

@@ -0,0 +1,4 @@
+statement ok
+CREATE SINK nexmark_q10 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
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

e2e_test/streaming/nexmark/sinks/q101.slt.part

@@ -0,0 +1,16 @@
+statement ok
+CREATE SINK nexmark_q101
+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 b1
+    GROUP BY b1.auction
+) b ON a.id = b.auction
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

e2e_test/streaming/nexmark/sinks/q102.slt.part

@@ -0,0 +1,14 @@
+statement ok
+CREATE SINK nexmark_q102
+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
+HAVING COUNT(b.auction) >= (
+    SELECT COUNT(*) / COUNT(DISTINCT auction) FROM bid
+)
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

e2e_test/streaming/nexmark/sinks/q103.slt.part

@@ -0,0 +1,13 @@
+statement ok
+CREATE SINK nexmark_q103
+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 b
+    GROUP BY b.auction
+    HAVING COUNT(*) >= 20
+)
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

e2e_test/streaming/nexmark/sinks/q104.slt.part

@@ -0,0 +1,13 @@
+statement ok
+CREATE SINK nexmark_q104
+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 b
+    GROUP BY b.auction
+    HAVING COUNT(*) < 20
+)
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

e2e_test/streaming/nexmark/sinks/q105.slt.part

@@ -0,0 +1,13 @@
+statement ok
+CREATE SINK nexmark_q105
+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
+ORDER BY bid_count DESC
+LIMIT 1000
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

e2e_test/streaming/nexmark/sinks/q106.slt.part

@@ -0,0 +1,20 @@
+statement ok
+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');

e2e_test/streaming/nexmark/sinks/q14.slt.part

@@ -0,0 +1,25 @@
+statement ok
+CREATE SINK nexmark_q14 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
+WHERE 0.908 * price > 1000000 AND 0.908 * price < 50000000
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

e2e_test/streaming/nexmark/sinks/q15.slt.part

@@ -0,0 +1,19 @@
+statement ok
+CREATE SINK nexmark_q15 AS
+SELECT
+     to_char(date_time, 'YYYY-MM-DD') as "day",
+     count(*) AS total_bids,
+     count(*) filter (where price < 10000) AS rank1_bids,
+     count(*) filter (where price >= 10000 and price < 1000000) AS rank2_bids,
+     count(*) filter (where price >= 1000000) AS rank3_bids,
+     count(distinct bidder) AS total_bidders,
+     count(distinct bidder) filter (where price < 10000) AS rank1_bidders,
+     count(distinct bidder) filter (where price >= 10000 and price < 1000000) AS rank2_bidders,
+     count(distinct bidder) filter (where price >= 1000000) AS rank3_bidders,
+     count(distinct auction) AS total_auctions,
+     count(distinct auction) filter (where price < 10000) AS rank1_auctions,
+     count(distinct auction) filter (where price >= 10000 and price < 1000000) AS rank2_auctions,
+     count(distinct auction) filter (where price >= 1000000) AS rank3_auctions
+FROM bid
+GROUP BY to_char(date_time, 'YYYY-MM-DD')
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

e2e_test/streaming/nexmark/sinks/q16.slt.part

@@ -0,0 +1,21 @@
+statement ok
+CREATE SINK nexmark_q16 AS
+SELECT
+    channel,
+    to_char(date_time, 'YYYY-MM-DD') as "day",
+    max(to_char(date_time, 'HH:mm')) as "minute",
+    count(*) AS total_bids,
+    count(*) filter (where price < 10000) AS rank1_bids,
+    count(*) filter (where price >= 10000 and price < 1000000) AS rank2_bids,
+    count(*) filter (where price >= 1000000) AS rank3_bids,
+    count(distinct bidder) AS total_bidders,
+    count(distinct bidder) filter (where price < 10000) AS rank1_bidders,
+    count(distinct bidder) filter (where price >= 10000 and price < 1000000) AS rank2_bidders,
+    count(distinct bidder) filter (where price >= 1000000) AS rank3_bidders,
+    count(distinct auction) AS total_auctions,
+    count(distinct auction) filter (where price < 10000) AS rank1_auctions,
+    count(distinct auction) filter (where price >= 10000 and price < 1000000) AS rank2_auctions,
+    count(distinct auction) filter (where price >= 1000000) AS rank3_auctions
+FROM bid
+GROUP BY channel, to_char(date_time, 'YYYY-MM-DD')
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

e2e_test/streaming/nexmark/sinks/q17.slt.part

@@ -0,0 +1,16 @@
+statement ok
+CREATE SINK nexmark_q17 AS
+SELECT
+    auction,
+    to_char(date_time, 'YYYY-MM-DD') AS day,
+    count(*) AS total_bids,
+    count(*) filter (where price < 10000) AS rank1_bids,
+    count(*) filter (where price >= 10000 and price < 1000000) AS rank2_bids,
+    count(*) filter (where price >= 1000000) AS rank3_bids,
+    min(price) AS min_price,
+    max(price) AS max_price,
+    avg(price) AS avg_price,
+    sum(price) AS sum_price
+FROM bid
+GROUP BY auction, to_char(date_time, 'YYYY-MM-DD')
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

e2e_test/streaming/nexmark/sinks/q18.slt.part

@@ -0,0 +1,14 @@
+statement ok
+CREATE SINK nexmark_q18 AS
+SELECT auction, bidder, price, channel, url, date_time, extra
+FROM (
+    SELECT *, 
+        ROW_NUMBER() OVER (
+            PARTITION BY bidder, auction 
+            ORDER BY date_time DESC, extra
+            -- extra is addtionally added here to make the result deterministic
+        ) AS rank_number
+    FROM bid
+)
+WHERE rank_number <= 1
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

e2e_test/streaming/nexmark/sinks/q2.slt.part

@@ -0,0 +1,6 @@
+statement ok
+CREATE SINK nexmark_q2
+AS
+SELECT auction, price FROM bid 
+WHERE auction = 1007 OR auction = 1020 OR auction = 2001 OR auction = 2019 OR auction = 2087
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

e2e_test/streaming/nexmark/sinks/q20.slt.part

@@ -0,0 +1,9 @@
+statement ok
+CREATE SINK nexmark_q20 AS
+SELECT
+    auction, bidder, price, channel, url, B.date_time as bid_date_time, B.extra as bid_extra,
+    item_name, description, initial_bid, reserve, A.date_time as auction_date_time, expires, seller, category, A.extra as auction_extra
+FROM
+    bid AS B INNER JOIN auction AS A on B.auction = A.id
+WHERE A.category = 10
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

e2e_test/streaming/nexmark/sinks/q21.slt.part

@@ -0,0 +1,17 @@
+statement ok
+CREATE SINK nexmark_q21 AS
+SELECT
+    auction, bidder, price, channel,
+    CASE
+        WHEN LOWER(channel) = 'apple' THEN '0'
+        WHEN LOWER(channel) = 'google' THEN '1'
+        WHEN LOWER(channel) = 'facebook' THEN '2'
+        WHEN LOWER(channel) = 'baidu' THEN '3'
+    ELSE (regexp_match(url, '(&|^)channel_id=([^&]*)'))[2]
+    END
+    AS channel_id 
+FROM 
+    bid
+WHERE 
+    (regexp_match(url, '(&|^)channel_id=([^&]*)'))[2] is not null or LOWER(channel) in ('apple', 'google', 'facebook', 'baidu')
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');