Flink SQL1.14: Match Recognize不支持使用由节点联接(joinType=[InnerJoin])产生的更新和删除更改

Question

我曾试图解决这个问题：

无法执行SQL语句。原因: org.apache.flink.table.api.TableException: Match Recognize不支持由节点连接(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
>   ;

但是当我使用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-to-stream-conversion的更多信息。

MATCH_RECOGNIZE不支持还原和删除，它需要一个仅附加的流。

我相信，如果您在创建视图时使用查找联接，您将得到一个仅附加的流。详细信息请参见https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/sql/queries/joins/#lookup-join