Flink SQL 1.14 : Match Recognize doesn't support consuming update and delete changes which is produced by node Join(joinType=[InnerJoin])

Question

I have tried to resolve this issue :

Could not execute SQL statement. Reason: org.apache.flink.table.api.TableException: Match Recognize doesn't support consuming update and delete changes which is produced by node Join(joinType=[InnerJoin], where=[(id = eventReference_id)], select=[type, id, isFired, eventMrid, createDateTime, eventReference_id], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[NoUniqueKey])

 CREATE TABLE Event (
>       isFired BOOLEAN
>       ,eventMrid STRING
>       ,createDateTime TIMESTAMP(3)
>      ,eventReference_id STRING
>      ,id STRING
>    ,WATERMARK FOR createDateTime AS createDateTime - INTERVAL '5' MINUTE
>      ) WITH (
>        'connector' = 'kafka'
>         ,'topic' = 'event'
>         ,'properties.bootstrap.servers' = 'localhost:9092'
>         ,'properties.group.id' = 'my-fourth-application'
>         ,'scan.startup.mode' = 'latest-offset'
>         ,'format' = 'json'
>       );

 CREATE TEMPORARY TABLE EventReference (
>       category STRING
>      ,commentt STRING
>    ,description STRING
>    ,type STRING
>    , id STRING PRIMARY KEY
>   )
>   WITH (
>     'connector' = 'postgres-cdc',
>     'hostname' = 'localhost',
>     'port' = '5432',
>     'database-name' = 'postgres',
>     'schema-name' = 'public',
>     'table-name' = 'EventReference',
>     'username' = 'postgres',
>     'password' = 'admin',
>     'decoding.plugin.name' = 'pgoutput'
>   );

 CREATE TEMPORARY VIEW event_with_eventReference AS
>     SELECT
>       e.isFired,
>       e.eventMrid,
>       e.createDateTime,
>       r.id AS eventReference_id,
>       r.type
>   FROM EventReference r
>   JOIN Event e ON r.id = e.eventReference_id
>   ;

But i find this issue when i used match_recognize :

 SELECT * from event_with_eventReference
>     MATCH_RECOGNIZE (
>      PARTITION BY eventMrid
>      ORDER BY createDateTime
>      MEASURES
>      FIRST(S.createDateTime) AS firstDateTime
>      ONE ROW PER MATCH
>      PATTERN (S)
>      DEFINE
>      S AS S.isFired = False
>     );
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.TableException: Match Recognize doesn't support consuming update and delete changes which is produced by node Join(joinType=[InnerJoin], where=[(id = eventReference_id)], select=[type, id, isFired, eventMrid, createDateTime, eventReference_id], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[NoUniqueKey])

Answer 1

Your temporary view event_with_eventReference results in a changelog stream where retractions and deletions can happen because of the join. You can find more information on https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/concepts/dynamic_tables/#table-to-stream-conversion

MATCH_RECOGNIZE doesn't support retractions and deletions, it expects an append-only stream.

I believe that if you use a lookup join when creating your view, you will end up with an append-only stream. See https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/sql/queries/joins/#lookup-join for details