Flink SQL 1.14：匹配识别不支持使用由节点 Join(joinType=[InnerJoin]) 生成的更新和删除更改

Question

我已经尝试解决这个问题：

无法执行SQL语句。原因：org.apache.flink.table.api.TableException：Match Recognition不支持使用由节点Join(joinType=[InnerJoin]，其中=[(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
>   ;

但当我使用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

您的临时视图event_with_eventReference会产生一个变更日志流，其中可能会因为联接而发生撤回和删除。您可以在上找到更多信息https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/concepts/dynamic_tables/#table-流转换

MATCH_RECOGNIZE不支持撤回和删除，它需要一个仅追加的流。

我相信，如果在创建视图时使用查找联接，最终会得到一个仅附加流。看见https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/sql/queries/joins/#lookup-加入了解详细信息