[
https://issues.apache.org/jira/browse/FLINK-23751?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17410977#comment-17410977
]
Martijn Visser commented on FLINK-23751:
----------------------------------------
I've used Flink 1.14 RC0 and ran the following tests using the SQL client
1. Created a table backed by the [Flink
Faker|https://github.com/knaufk/flink-faker] connector using:
{code:sql}
CREATE TABLE orders (
bidtime TIMESTAMP(3),
price DOUBLE,
item STRING,
supplier STRING,
WATERMARK FOR bidtime AS bidtime - INTERVAL '5' SECONDS
) WITH (
'connector' = 'faker',
'fields.bidtime.expression' = '#{date.past ''30'',''SECONDS''}',
'fields.price.expression' = '#{Number.randomDouble ''2'',''1'',''150''}',
'fields.item.expression' = '#{Commerce.productName}',
'fields.supplier.expression' = '#{regexify
''(Alice|Bob|Carol|Alex|Joe|James|Jane|Jack)''}',
'rows-per-second' = '1'
);
{code}
2. Tested the Window Top-N after a Window TVF Tumble using:
{code:sql}
SELECT *
FROM (
SELECT *, ROW_NUMBER() OVER (PARTITION BY window_start, window_end
ORDER BY price DESC) as rownum
FROM (
SELECT window_start, window_end, supplier, SUM(price) as price,
COUNT(*) as cnt
FROM TABLE(
TUMBLE(TABLE orders, DESCRIPTOR(bidtime), INTERVAL '10'
SECONDS))
GROUP BY window_start, window_end, supplier
)
) WHERE rownum <= 3;
{code}
3. Tested the Window Top-N after a Window TVF Hop using:
{code:sql}
SELECT *
FROM (
SELECT *, ROW_NUMBER() OVER (PARTITION BY window_start, window_end
ORDER BY price DESC) as rownum
FROM (
SELECT window_start, window_end, supplier, SUM(price) as price,
COUNT(*) as cnt
FROM TABLE(
HOP(TABLE orders, DESCRIPTOR(bidtime), INTERVAL '10' SECONDS,
INTERVAL '20' SECONDS))
GROUP BY window_start, window_end, supplier
)
) WHERE rownum <= 3;
{code}
4. Tested the Window Top-N after a Window TVF Cumulate using:
{code:sql}
SELECT *
FROM (
SELECT *, ROW_NUMBER() OVER (PARTITION BY window_start, window_end
ORDER BY price DESC) as rownum
FROM (
SELECT window_start, window_end, supplier, SUM(price) as price,
COUNT(*) as cnt
FROM TABLE(
CUMULATE(TABLE orders, DESCRIPTOR(bidtime), INTERVAL '10'
SECONDS, INTERVAL '20' SECONDS))
GROUP BY window_start, window_end, supplier
)
) WHERE rownum <= 3;
{code}
Test results look good
> Testing Window Top-N after Windowing TVF
> ----------------------------------------
>
> Key: FLINK-23751
> URL: https://issues.apache.org/jira/browse/FLINK-23751
> Project: Flink
> Issue Type: Improvement
> Components: Tests
> Reporter: JING ZHANG
> Assignee: Martijn Visser
> Priority: Blocker
> Labels: release-testing
> Fix For: 1.14.0
>
>
> Currently, Flink not only supports Window Top-N which follows after [Window
> Aggregation|https://ci.apache.org/projects/flink/flink-docs-master/docs/dev/table/sql/queries/window-agg/].
> but also supports Window Top-N follows after [Windowing
> TVF.|https://ci.apache.org/projects/flink/flink-docs-master/docs/dev/table/sql/queries/window-tvf/]
> The following example shows how to calculate Top 3 items which have the
> highest price for every tumbling 10 minutes window.
> {code:java}
> SELECT * FROM Bid;
> +------------------+-------+------+-------------+
> | bidtime | price | item | supplier_id |
> +------------------+-------+------+-------------+
> | 2020-04-15 08:05 | 4.00 | A | supplier1 |
> | 2020-04-15 08:06 | 4.00 | C | supplier2 |
> | 2020-04-15 08:07 | 2.00 | G | supplier1 |
> | 2020-04-15 08:08 | 2.00 | B | supplier3 |
> | 2020-04-15 08:09 | 5.00 | D | supplier4 |
> | 2020-04-15 08:11 | 2.00 | B | supplier3 |
> | 2020-04-15 08:13 | 1.00 | E | supplier1 |
> | 2020-04-15 08:15 | 3.00 | H | supplier2 |
> | 2020-04-15 08:17 | 6.00 | F | supplier5 |
> +------------------+-------+------+-------------+
> Flink SQL> SELECT *
> FROM (
> SELECT *, ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY
> price DESC) as rownum
> FROM TABLE(
> TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES))
> ) WHERE rownum <= 3;
> +------------------+-------+------+-------------+------------------+------------------+--------+
> | bidtime | price | item | supplier_id | window_start | window_end | rownum |
> +------------------+-------+------+-------------+------------------+------------------+--------+
> | 2020-04-15 08:05 | 4.00 | A | supplier1 | 2020-04-15 08:00 | 2020-04-15
> 08:10 | 2 |
> | 2020-04-15 08:06 | 4.00 | C | supplier2 | 2020-04-15 08:00 | 2020-04-15
> 08:10 | 3 |
> | 2020-04-15 08:09 | 5.00 | D | supplier4 | 2020-04-15 08:00 | 2020-04-15
> 08:10 | 1 |
> | 2020-04-15 08:11 | 2.00 | B | supplier3 | 2020-04-15 08:10 | 2020-04-15
> 08:20 | 3 |
> | 2020-04-15 08:15 | 3.00 | H | supplier2 | 2020-04-15 08:10 | 2020-04-15
> 08:20 | 2 |
> | 2020-04-15 08:17 | 6.00 | F | supplier5 | 2020-04-15 08:10 | 2020-04-15
> 08:20 | 1 |
> +------------------+-------+------+-------------+------------------+------------------+--------+
> {code}
> Note: Currently, Flink only supports Window Top-N follows after Windowing TVF
> with Tumble Windows, Hop Windows and Cumulate Windows. Window Top-N follows
> after Windowing TVF with Session windows will be supported in the near future.
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
