Hey folks, We are looking into performing deduplication + over aggregation (flink v1.15)? We tried the below but fell into error:

StreamPhysicalOverAggregate doesn't support consuming update and delete changes which is produced by node Deduplicate(keep=[LastRow], key=[home, room, sn], order=[ROWTIME])

Any thoughts? Thanks.

WITH t1 AS (
    /* DEDUPLICATION */
    SELECT 
        home, 
        room, 
        sn, 
        temp,
        event_ts, 
        arrival_ts
    FROM ( 
        SELECT 
            *, 
            ROW_NUMBER() OVER (PARTITION BY home, room, sn ORDER BY event_ts DESC) AS row_num
        FROM telemetry
        WHERE 
            event_ts >= CURRENT_WATERMARK(event_ts)
            AND ABS(TIMESTAMPDIFF(SECOND, event_ts, arrival_ts)) <= 60
    )
    WHERE row_num = 1
)

/* OVER AGGREGATION */
SELECT 
    home, 
    room, 
    sn, 
    event_ts, 
    arrival_ts, 
    AVG(temp) OVER ( 
        PARTITION BY home, room
        ORDER BY event_ts
        RANGE BETWEEN INTERVAL '1' MINUTE PRECEDING AND CURRENT ROW
    ) as avg_temp
FROM t1