[
https://issues.apache.org/jira/browse/FLINK-23751?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17411031#comment-17411031
]
Martijn Visser commented on FLINK-23751:
----------------------------------------
[~twalthr] Didn't read it correctly, thanks for re-opening.
I've adjusted my test to the following:
{code:sql}
CREATE TABLE Bid (
bidtime TIMESTAMP(3),
price DOUBLE,
item STRING,
supplier_id 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_id.expression' = '#{regexify
''(Alice|Bob|Carol|Alex|Joe|James|Jane|Jack)''}',
'rows-per-second' = '100'
);
{code}
However, when using the example from the description
{code: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;
{code}
The following error would be returned:
{code:java}
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.TableException: Found more than one rowtime field:
[bidtime, window_time] in the query when insert into
'default_catalog.default_database.Unregistered_Collect_Sink_8'.
Please select the rowtime field that should be used as event-time timestamp for
the DataStream by casting all other fields to TIMESTAMP.
{code}
When adjusting the SQL statement to this, it does work:
{code:sql}
SELECT *
FROM (
SELECT bidtime, price, item, supplier_id, window_start, window_end,
ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY price DESC)
as rownum
FROM TABLE(
TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' SECONDS))
) WHERE rownum <= 3;
{code}
Results:
| bidtime | price | item | supplier\_id |
window\_start | window\_end | rownum |
| 2021-09-07 08:31:07.000 | 146.44 | Awesome Paper Keyboard | Joe |
2021-09-07 08:31:00.000 | 2021-09-07 08:31:10.000 | 3 |
| 2021-09-07 08:31:16.000 | 146.59 | Small Iron Gloves | Jack |
2021-09-07 08:31:10.000 | 2021-09-07 08:31:20.000 | 1 |
| 2021-09-07 08:31:11.000 | 146.45 | Sleek Paper Chair | Carol |
2021-09-07 08:31:10.000 | 2021-09-07 08:31:20.000 | 2 |
| 2021-09-07 08:31:12.000 | 145.55 | Rustic Plastic Bottle | Carol |
2021-09-07 08:31:10.000 | 2021-09-07 08:31:20.000 | 3 |
| 2021-09-07 08:31:24.000 | 146.99 | Incredible Aluminum Chair | Carol |
2021-09-07 08:31:20.000 | 2021-09-07 08:31:30.000 | 1 |
| 2021-09-07 08:31:25.000 | 146.55 | Aerodynamic Granite Plate | Bob |
2021-09-07 08:31:20.000 | 2021-09-07 08:31:30.000 | 2 |
| 2021-09-07 08:31:22.000 | 146.43 | Durable Marble Watch | Bob |
2021-09-07 08:31:20.000 | 2021-09-07 08:31:30.000 | 3 |
| 2021-09-07 08:31:37.000 | 147.45 | Fantastic Concrete Hat | Bob |
2021-09-07 08:31:30.000 | 2021-09-07 08:31:40.000 | 1 |
| 2021-09-07 08:31:30.000 | 146.95 | Rustic Copper Clock | Alex |
2021-09-07 08:31:30.000 | 2021-09-07 08:31:40.000 | 2 |
| 2021-09-07 08:31:35.000 | 146.69 | Synergistic Cotton Knife | Carol |
2021-09-07 08:31:30.000 | 2021-09-07 08:31:40.000 | 3 |
> 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)
