[jira] [Commented] (FLINK-19059) Support to consume retractions for OVER WINDOW operator

["Martijn Visser (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-19059?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17842052#comment-17842052 ] 

Martijn Visser commented on FLINK-19059:
----------------------------------------

A relatively simple reproducer:
{code:sql}
CREATE OR REPLACE TABLE orders (
    order_number BIGINT,
    product_id   STRING,    
    price        DECIMAL(32,2),
    order_time   TIMESTAMP(3),
    WATERMARK for order_time AS order_time
) WITH (
  'connector' = 'datagen'
);
{code}
{code:sql}
CREATE OR REPLACE TABLE shipments (
    shipment_id BIGINT,
    order_number BIGINT,  
    product_id   STRING,
    shipment_time TIMESTAMP(3),
    WATERMARK FOR shipment_time AS shipment_time
) WITH (
    'connector' = 'datagen'
);
{code}
{code:sql}
WITH deduplicated AS (
	SELECT *, SUM(price)
	FROM (
	  SELECT *,
	    ROW_NUMBER() OVER (PARTITION BY order_number ORDER BY order_time ASC) AS row_num
	  FROM orders)
	WHERE row_num = 1
	GROUP BY order_number, product_id, price, order_time, row_num
)
SELECT order_number, order_time, price,
  SUM(price) OVER (
    PARTITION BY product_id
    ORDER BY order_time
    RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW
  ) AS one_hour_prod_amount_sum
FROM deduplicated;
{code}
Error:
{code:java}
org.apache.flink.table.api.TableException: StreamPhysicalOverAggregate doesn't support consuming update and delete changes which is produced by node GroupAggregate(groupBy=[order_number, product_id, price, order_time, w0$o0], select=[order_number, product_id, price, order_time, w0$o0]) {code}

> Support to consume retractions for OVER WINDOW operator
> -------------------------------------------------------
>
>                 Key: FLINK-19059
>                 URL: https://issues.apache.org/jira/browse/FLINK-19059
>             Project: Flink
>          Issue Type: New Feature
>          Components: Table SQL / Planner
>            Reporter: Jark Wu
>            Priority: Not a Priority
>              Labels: auto-deprioritized-major, auto-deprioritized-minor
>
> Currently, OVER WINDOW in streaming SQL only support to consume insert-only stream and doesn't support to consume retractions (i.e. contains UPDATE/DELETE messages). This results in OVER WINDOW can't be used to follow after unbounded group by or streaming join. 



--
This message was sent by Atlassian Jira
(v8.20.10#820010)