You are viewing a plain text version of this content. The canonical link for it is here.

Posted to user-zh@flink.apache.org by Wayne <17...@163.com> on 2021/09/01 10:27:10 UTC

Temporal Joins 报 Currently the join key in Temporal Table Join can not be empty

我的flink 版本是  flink-1.12.2-bin-scala_2.12
我的sql 是
SELECT o.order_id, o.total, c.country, c.zip
FROM Orders AS o
  JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
    ON o.customer_id = c.id and o.customer_id is not null and  c.id is not null ;
或者
SELECT o.order_id, o.total, c.country, c.zip
FROM Orders AS o
  JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
    ON o.customer_id = c.id ;


都会报如下错误，麻烦帮我看看正确的写法是什么样的，灰常感谢


Exception in thread "main" org.apache.flink.table.api.ValidationException: Currently the join key in Temporal Table Join can not be empty.
at org.apache.flink.table.planner.plan.rules.logical.LogicalCorrelateToJoinFromGeneralTemporalTableRule.onMatch(LogicalCorrelateToJoinFromTemporalTableRule.scala:272)
at org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:333)
at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:542)
at org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:407)
at org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:243)
at org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
at org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:202)
at org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:189)
at org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
at org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$2(FlinkGroupProgram.scala:63)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1(FlinkGroupProgram.scala:60)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1$adapted(FlinkGroupProgram.scala:55)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
at scala.collection.immutable.Range.foreach(Range.scala:158)
at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.optimize(FlinkGroupProgram.scala:55)
at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
at org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
at org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:287)
at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:160)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1329)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:707)
at org.apache.flink.table.api.internal.TableImpl.execute(TableImpl.java:577)

回复: Re: Temporal Joins 报 Currently the join key in Temporal Table Join can not be empty

Posted by "wukong91@foxmail.com" <wu...@foxmail.com>.

Hi Wayne :
   请查考Flink Join 官方文档， 右侧构建侧 需要定义  PRIMARY KEY 同时 在join 条件的时候带上。
参考地址： https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/streaming/joins.html#%E5%9F%BA%E4%BA%8E%E4%BA%8B%E4%BB%B6%E6%97%B6%E9%97%B4%E7%9A%84%E6%97%B6%E6%80%81-join 

内容： 
注意: 基于事件时间的时态表 Join 是通过左右两侧的 watermark 触发，请确保为 join 两侧的表设置了合适的 watermark。
注意: 基于事件时间的时态表 Join 的 join key 必须包含时态表的主键，例如：表 product_changelog 的主键 P.product_id 必须包含在 join 条件 O.product_id = P.product_id 中。



wukong91@foxmail.com
 
发件人： Wayne
发送时间： 2021-09-09 23:29
收件人： user-zh
主题： Re:Re:Re: Re: Temporal Joins 报 Currently the join key in Temporal Table Join can not be empty
Hello 
 
 
打扰了我最近再次尝试，还是报这个错误
我的flink版本为 flink-1.12.2-bin-scala_2.12
使用sql client执行
我的sql 如下  
 
 
CREATE TABLE stub_trans (
`uuid` STRING,
`columnInfos` MAP<STRING NOT NULL,ROW<oldValue STRING NULL ,newValue STRING  ,name STRING NOT NULL ,isKeyColumn BOOLEAN NOT NULL,type STRING NOT NULL > NOT NULL>  NOT NULL,
procTime TIMESTAMP(3) METADATA FROM 'timestamp' , 
WATERMARK FOR procTime AS procTime   
) WITH (
    'connector' = 'kafka',
..............
    'format' = 'avro'
);
 
 
CREATE TABLE kapeng_test (
 
 
`event_id` STRING,
`genre_id` STRING,
`user_guid` STRING,
procTime TIMESTAMP(3) METADATA FROM 'timestamp'  ,  
WATERMARK FOR procTime AS procTime   
) WITH (
    'connector' = 'kafka',
........
    'format' = 'avro'
);
 
 
 
 
CREATE TABLE purchase_hist (
    rowkey STRING,
    d ROW < cost STRING, crdate STRING, currency STRING, eventid STRING, genreid STRING, quantity STRING >,
    PRIMARY KEY ( rowkey ) NOT ENFORCED 
) WITH ( 
    'connector' = 'hbase-1.4', 
.........);
 
 
 
 
INSERT INTO purchase_hist SELECT
rowkey,
ROW ( cost, crdate, currency, eventid, genreid, quantity ) 
FROM
( SELECT
    CONCAT_WS(
        '_',
        user_guid,
        CAST( CAST( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue, 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ) AS BIGINT ) AS STRING )) AS rowkey,
    columnInfos [ 'TICKET_COST' ].newValue AS cost,
    DATE_FORMAT( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue, 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ), 'yyyy-MM-dd' ) AS crdate,
    columnInfos [ 'EVENT_ID' ].newValue AS eventid,
    columnInfos [ 'CURRENCY_CODE' ].newValue AS currency,
    columnInfos [ 'QUANTITY' ].newValue AS quantity,
    genre_id AS genreid,
    user_guid AS userGuid 
    FROM
    stub_trans
   LEFT  JOIN kapeng_test FOR SYSTEM_TIME AS OF stub_trans.procTime ON stub_trans.columnInfos [ 'EVENT_ID' ].newValue = kapeng_test.event_id )m
 
 
 
 
报错如下
Exception in thread "main" org.apache.flink.table.api.ValidationException: Currently the join key in Temporal Table Join can not be empty.
at org.apache.flink.table.planner.plan.rules.logical.LogicalCorrelateToJoinFromGeneralTemporalTableRule.onMatch(LogicalCorrelateToJoinFromTemporalTableRule.scala:272)
at org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:333)
at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:542)
at org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:407)
at org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:243)
at org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
at org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:202)
at org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:189)
at org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
at org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$2(FlinkGroupProgram.scala:63)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1(FlinkGroupProgram.scala:60)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1$adapted(FlinkGroupProgram.scala:55)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
at scala.collection.immutable.Range.foreach(Range.scala:158)
at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.optimize(FlinkGroupProgram.scala:55)
at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
at org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
at org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:287)
at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:160)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1329)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:676)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeOperation(TableEnvironmentImpl.java:767)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:666)
 
 
 
 
 
 
如果我将
LEFT  JOIN kapeng_test FOR SYSTEM_TIME AS OF stub_trans.procTime ON stub_trans.columnInfos [ 'EVENT_ID' ].newValue = kapeng_test.event_id
中的 
stub_trans.columnInfos [ 'EVENT_ID' ].newValue  替换为 uuid  则这部分语句为
 
 
LEFT  JOIN kapeng_test FOR SYSTEM_TIME AS OF stub_trans.procTime ON stub_trans.uuid = kapeng_test.event_id
 
 
 
 
则报错如下:
 
 
Exception in thread "main" org.apache.flink.table.api.ValidationException: Temporal Table Join requires primary key in versioned table, but no primary key can be found. The physical plan is:
FlinkLogicalJoin(condition=[AND(=($0, $3), __INITIAL_TEMPORAL_JOIN_CONDITION($2, $6, __TEMPORAL_JOIN_LEFT_KEY($0), __TEMPORAL_JOIN_RIGHT_KEY($3)))], joinType=[left])
  FlinkLogicalCalc(select=[uuid, columnInfos, Reinterpret(CAST(timestamp)) AS procTime])
    FlinkLogicalTableSourceScan(table=[[default_catalog, default_database, stub_trans, watermark=[CAST($2):TIMESTAMP(3)]]], fields=[uuid, columnInfos, timestamp])
  FlinkLogicalSnapshot(period=[$cor0.procTime])
    FlinkLogicalCalc(select=[event_id, genre_id, user_guid, Reinterpret(CAST(timestamp)) AS procTime])
      FlinkLogicalTableSourceScan(table=[[default_catalog, default_database, kapeng_test, watermark=[CAST($3):TIMESTAMP(3)]]], fields=[event_id, genre_id, user_guid, timestamp])
 
 
 
 
at org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule.org$apache$flink$table$planner$plan$rules$logical$TemporalJoinRewriteWithUniqueKeyRule$$validateRightPrimaryKey(TemporalJoinRewriteWithUniqueKeyRule.scala:127)
at org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule$$anon$1.visitCall(TemporalJoinRewriteWithUniqueKeyRule.scala:88)
at org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule$$anon$1.visitCall(TemporalJoinRewriteWithUniqueKeyRule.scala:70)
at org.apache.calcite.rex.RexCall.accept(RexCall.java:174)
at org.apache.calcite.rex.RexShuttle.visitList(RexShuttle.java:158)
at org.apache.calcite.rex.RexShuttle.visitCall(RexShuttle.java:110)
at org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule$$anon$1.visitCall(TemporalJoinRewriteWithUniqueKeyRule.scala:109)
at org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule$$anon$1.visitCall(TemporalJoinRewriteWithUniqueKeyRule.scala:70)
at org.apache.calcite.rex.RexCall.accept(RexCall.java:174)
at org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule.onMatch(TemporalJoinRewriteWithUniqueKeyRule.scala:70)
at org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:333)
at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:542)
at org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:407)
at org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:243)
at org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
at org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:202)
at org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:189)
at org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
at org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
at org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
at org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:287)
at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:160)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1329)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:676)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeOperation(TableEnvironmentImpl.java:767)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:666)
 
 
请问这个sql 我应该怎么写
 
 
 
 
 
 
 
――――――――――――――――――――――――――――――――――――分割线――――――――――――――――――――――――――――――――――――――――――――――――
 
 
在 2021-09-02 11:18:55，"Wayne" <17...@163.com> 写道：
>
>
>
>我是使用tableEnv.executeSql   执行那两个create 语句 然后用 tableEnv.sqlQuery  执行 select语句 ，然后报错， 没有再执行其他sql额 ， aaa.columnInfos [ 'EVENT_ID' ].newValue 就是取出map里面 EVENT_ID 的key的 newValue 值，从create 语句可以看出 这个是个string
>
> 另外   tableEnv.explainSql(xxx) 也不能执行那个select语句，报的一样的错误， 我先在本地复现一下试试吧
>
>
>
>
>
>
>
>
>在 2021-09-02 10:31:20，"Caizhi Weng" <ts...@gmail.com> 写道：
>>Hi！
>>
>>没有复现这个问题。后续是还有其他 SQL 吗？特别是对 aaa.columnInfos [ 'EVENT_ID' ].newValue 有
>>filter 的 SQL。
>>
>>可以尝试先在本地找一个能复现的 case，通过 tableEnv.explainSql() 就能快速检查能否产生 plan。如果找到了能复现的
>>SQL，可以在 jira 提一个 issue。
>>
>>Wayne <17...@163.com> 于2021年9月2日周四 上午10:43写道：
>>
>>> 表a
>>> CREATE TABLE aaa (
>>>
>>>
>>> `columnInfos` MAP<STRING NOT NULL,ROW<oldValue STRING NULL ,newValue
>>> STRING  ,name STRING NOT NULL ,isKeyColumn BOOLEAN NOT NULL,type STRING NOT
>>> NULL > NOT NULL>  NOT NULL,
>>> .......
>>> procTime AS PROCTIME()
>>> ) WITH (
>>>     'connector' = 'kafka' ,
>>>     .....
>>>     'format' = 'avro'
>>> )
>>>
>>>
>>>
>>>
>>> 表b
>>>
>>>
>>> CREATE TABLE bbb (
>>> `event_id` STRING,
>>> `genre_id` STRING,
>>> `user_guid` STRING,
>>> ......
>>> `uuid` STRING
>>> ) WITH (
>>>     'connector' = 'kafka' ,
>>>     .....
>>>     'format' = 'avro'
>>> )
>>>
>>>
>>>
>>>
>>>
>>>
>>> 我的sql
>>>
>>>
>>> SELECT
>>>     CONCAT_WS(
>>>         '_',
>>>         user_guid,
>>>         CAST( CAST( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue,
>>> 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ) AS BIGINT ) AS STRING )) AS rowkey,
>>>     columnInfos [ 'TICKET_COST' ].newValue AS cost,
>>>     DATE_FORMAT( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue,
>>> 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ), 'yyyy-MM-dd' ) AS crdate,
>>>     columnInfos [ 'EVENT_ID' ].newValue AS eventid,
>>>     columnInfos [ 'QUANTITY' ].newValue AS quantity,
>>>     genre_id AS genreId,
>>>     user_guid AS userGuid
>>>     FROM
>>>     aaa
>>>    LEFT  JOIN bbb FOR SYSTEM_TIME AS OF aaa.procTime ON aaa.columnInfos [
>>> 'EVENT_ID' ].newValue = bbb.event_id  and aaa.columnInfos [ 'EVENT_ID'
>>> ].newValue is not null and  bbb.event_id is not null
>>>
>>>
>>> 最后的  and aaa.columnInfos [ 'EVENT_ID' ].newValue is not null and
>>> bbb.event_id is not null   无论加不加，都是报这个错误
>>>
>>> 在 2021-09-02 09:13:11，"Caizhi Weng" <ts...@gmail.com> 写道：
>>> >Hi！
>>> >
>>> >Orders 和 Customers 就直接是 source 表吗？还是说 source 表到维表 join 之间有关于 customer_id 或
>>> >id 的 filter 条件？
>>> >
>>> >有一定可能是之前有关于 customer_id 或 id 的 filter 条件（比如 customer_id = 1），导致维表 join
>>> >这里的等值条件被优化成了 customer_id = 1 和 id = 1 并分别下推到维表 join 之前。
>>> >
>>> >Wayne <17...@163.com> 于2021年9月1日周三 下午6:27写道：
>>> >
>>> >> 我的flink 版本是  flink-1.12.2-bin-scala_2.12
>>> >> 我的sql 是
>>> >> SELECT o.order_id, o.total, c.country, c.zip
>>> >> FROM Orders AS o
>>> >>   JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
>>> >>     ON o.customer_id = c.id and o.customer_id is not null and  c.id is
>>> >> not null ;
>>> >> 或者
>>> >> SELECT o.order_id, o.total, c.country, c.zip
>>> >> FROM Orders AS o
>>> >>   JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
>>> >>     ON o.customer_id = c.id ;
>>> >>
>>> >> 都会报如下错误，麻烦帮我看看正确的写法是什么样的，灰常感谢
>>> >>
>>> >> Exception in thread "main"
>>> org.apache.flink.table.api.ValidationException:
>>> >> Currently the join key in Temporal Table Join can not be empty.
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.rules.logical.LogicalCorrelateToJoinFromGeneralTemporalTableRule.onMatch(LogicalCorrelateToJoinFromTemporalTableRule.scala:272)
>>> >> at
>>> >>
>>> org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:333)
>>> >> at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:542)
>>> >> at
>>> org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:407)
>>> >> at
>>> >>
>>> org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:243)
>>> >> at
>>> >>
>>> org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
>>> >> at
>>> >>
>>> org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:202)
>>> >> at
>>> org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:189)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$2(FlinkGroupProgram.scala:63)
>>> >> at
>>> >>
>>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
>>> >> at
>>> >>
>>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
>>> >> at scala.collection.Iterator.foreach(Iterator.scala:943)
>>> >> at scala.collection.Iterator.foreach$(Iterator.scala:943)
>>> >> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
>>> >> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
>>> >> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
>>> >> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
>>> >> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
>>> >> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
>>> >> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1(FlinkGroupProgram.scala:60)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1$adapted(FlinkGroupProgram.scala:55)
>>> >> at
>>> >>
>>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
>>> >> at
>>> >>
>>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
>>> >> at scala.collection.immutable.Range.foreach(Range.scala:158)
>>> >> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
>>> >> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
>>> >> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.optimize(FlinkGroupProgram.scala:55)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
>>> >> at
>>> >>
>>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
>>> >> at
>>> >>
>>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
>>> >> at scala.collection.Iterator.foreach(Iterator.scala:943)
>>> >> at scala.collection.Iterator.foreach$(Iterator.scala:943)
>>> >> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
>>> >> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
>>> >> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
>>> >> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
>>> >> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
>>> >> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
>>> >> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:287)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:160)
>>> >> at
>>> >>
>>> org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1329)
>>> >> at
>>> >>
>>> org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:707)
>>> >> at
>>> >>
>>> org.apache.flink.table.api.internal.TableImpl.execute(TableImpl.java:577)
>>> >>
>>> >>
>>> >>
>>> >>
>>> >>
>>> >>
>>> >>
>>> >>
>>> >>
>>> >>
>>> >>
>>>

Re: Re:Re: Re: Temporal Joins 报 Currently the join key in Temporal Table Join can not be empty

Posted by Caizhi Weng <ts...@gmail.com>.

Hi！

感谢持续反馈。这次确实在开发环境下复现了这个问题。

这是因为目前 event time temporal join 对 join key 的处理还不够完善，目前只能处理原原本本的输入列（比如直接用
uuid），暂时不能处理利用输入列做的运算（例如这里从 map 里取值）。我已经开了一个 issue 记录这个问题 [1]。一个绕过的方法是先在一个
view 里给 map 取值，比如这样：

CREATE TABLE A (
a MAP<STRING NOT NULL, INT>,
ts TIMESTAMP(3),
WATERMARK FOR ts AS ts
)；

CREATE TABLE B (
id INT,
ts TIMESTAMP(3),
WATERMARK FOR ts AS ts,
PRIMARY KEY (id) NOT ENFORCED
)；

CREATE VIEW myView AS SELECT A.a['ID'] AS id, ts FROM A；

SELECT * FROM myView AS a LEFT JOIN B FOR SYSTEM_TIME AS OF a.ts AS b ON
a.id = b.id；

后面那个 Temporal Table Join requires primary key in versioned table, but no
primary key can be found. 是说 event time temporal join 的维表必须定义 primary
key。这是预期行为，参见 [1] 中的 The event-time temporal join requires the primary key
contained in the equivalence condition of the temporal join condition.
Primary key 的定义方式见上述示例代码，也可以参见 [2]。

[1] https://issues.apache.org/jira/browse/FLINK-24239
[2]
https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/dev/table/sql/queries/joins/#event-time-temporal-join
[3]
https://ci.apache.org/projects/flink/flink-docs-master/zh/docs/dev/table/sql/create/#primary-key

Wayne <17...@163.com> 于2021年9月9日周四 下午11:29写道：

> Hello
>
>
> 打扰了我最近再次尝试，还是报这个错误
> 我的flink版本为 flink-1.12.2-bin-scala_2.12
> 使用sql client执行
> 我的sql 如下
>
>
> CREATE TABLE stub_trans (
> `uuid` STRING,
> `columnInfos` MAP<STRING NOT NULL,ROW<oldValue STRING NULL ,newValue
> STRING  ,name STRING NOT NULL ,isKeyColumn BOOLEAN NOT NULL,type STRING NOT
> NULL > NOT NULL>  NOT NULL,
> procTime TIMESTAMP(3) METADATA FROM 'timestamp' ,
> WATERMARK FOR procTime AS procTime
> ) WITH (
>     'connector' = 'kafka',
> ..............
>     'format' = 'avro'
> );
>
>
> CREATE TABLE kapeng_test (
>
>
> `event_id` STRING,
> `genre_id` STRING,
> `user_guid` STRING,
> procTime TIMESTAMP(3) METADATA FROM 'timestamp'  ,
> WATERMARK FOR procTime AS procTime
> ) WITH (
>     'connector' = 'kafka',
> ........
>     'format' = 'avro'
> );
>
>
>
>
> CREATE TABLE purchase_hist (
>     rowkey STRING,
>     d ROW < cost STRING, crdate STRING, currency STRING, eventid STRING,
> genreid STRING, quantity STRING >,
>     PRIMARY KEY ( rowkey ) NOT ENFORCED
> ) WITH (
>     'connector' = 'hbase-1.4',
> .........);
>
>
>
>
> INSERT INTO purchase_hist SELECT
> rowkey,
> ROW ( cost, crdate, currency, eventid, genreid, quantity )
> FROM
> ( SELECT
>     CONCAT_WS(
>         '_',
>         user_guid,
>         CAST( CAST( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue,
> 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ) AS BIGINT ) AS STRING )) AS rowkey,
>     columnInfos [ 'TICKET_COST' ].newValue AS cost,
>     DATE_FORMAT( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue,
> 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ), 'yyyy-MM-dd' ) AS crdate,
>     columnInfos [ 'EVENT_ID' ].newValue AS eventid,
>     columnInfos [ 'CURRENCY_CODE' ].newValue AS currency,
>     columnInfos [ 'QUANTITY' ].newValue AS quantity,
>     genre_id AS genreid,
>     user_guid AS userGuid
>     FROM
>     stub_trans
>    LEFT  JOIN kapeng_test FOR SYSTEM_TIME AS OF stub_trans.procTime ON
> stub_trans.columnInfos [ 'EVENT_ID' ].newValue = kapeng_test.event_id )m
>
>
>
>
> 报错如下
> Exception in thread "main" org.apache.flink.table.api.ValidationException:
> Currently the join key in Temporal Table Join can not be empty.
> at
> org.apache.flink.table.planner.plan.rules.logical.LogicalCorrelateToJoinFromGeneralTemporalTableRule.onMatch(LogicalCorrelateToJoinFromTemporalTableRule.scala:272)
> at
> org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:333)
> at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:542)
> at org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:407)
> at
> org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:243)
> at
> org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
> at
> org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:202)
> at org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:189)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$2(FlinkGroupProgram.scala:63)
> at
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
> at
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
> at scala.collection.Iterator.foreach(Iterator.scala:943)
> at scala.collection.Iterator.foreach$(Iterator.scala:943)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1(FlinkGroupProgram.scala:60)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1$adapted(FlinkGroupProgram.scala:55)
> at
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
> at
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
> at scala.collection.immutable.Range.foreach(Range.scala:158)
> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.optimize(FlinkGroupProgram.scala:55)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
> at
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
> at
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
> at scala.collection.Iterator.foreach(Iterator.scala:943)
> at scala.collection.Iterator.foreach$(Iterator.scala:943)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
> at
> org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
> at
> org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:287)
> at
> org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:160)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1329)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:676)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.executeOperation(TableEnvironmentImpl.java:767)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:666)
>
>
>
>
>
>
> 如果我将
> LEFT  JOIN kapeng_test FOR SYSTEM_TIME AS OF stub_trans.procTime ON
> stub_trans.columnInfos [ 'EVENT_ID' ].newValue = kapeng_test.event_id
> 中的
>  stub_trans.columnInfos [ 'EVENT_ID' ].newValue  替换为 uuid  则这部分语句为
>
>
> LEFT  JOIN kapeng_test FOR SYSTEM_TIME AS OF stub_trans.procTime ON
> stub_trans.uuid = kapeng_test.event_id
>
>
>
>
> 则报错如下:
>
>
> Exception in thread "main" org.apache.flink.table.api.ValidationException:
> Temporal Table Join requires primary key in versioned table, but no primary
> key can be found. The physical plan is:
> FlinkLogicalJoin(condition=[AND(=($0, $3),
> __INITIAL_TEMPORAL_JOIN_CONDITION($2, $6, __TEMPORAL_JOIN_LEFT_KEY($0),
> __TEMPORAL_JOIN_RIGHT_KEY($3)))], joinType=[left])
>   FlinkLogicalCalc(select=[uuid, columnInfos, Reinterpret(CAST(timestamp))
> AS procTime])
>     FlinkLogicalTableSourceScan(table=[[default_catalog, default_database,
> stub_trans, watermark=[CAST($2):TIMESTAMP(3)]]], fields=[uuid, columnInfos,
> timestamp])
>   FlinkLogicalSnapshot(period=[$cor0.procTime])
>     FlinkLogicalCalc(select=[event_id, genre_id, user_guid,
> Reinterpret(CAST(timestamp)) AS procTime])
>       FlinkLogicalTableSourceScan(table=[[default_catalog,
> default_database, kapeng_test, watermark=[CAST($3):TIMESTAMP(3)]]],
> fields=[event_id, genre_id, user_guid, timestamp])
>
>
>
>
> at
> org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule.org
> $apache$flink$table$planner$plan$rules$logical$TemporalJoinRewriteWithUniqueKeyRule$$validateRightPrimaryKey(TemporalJoinRewriteWithUniqueKeyRule.scala:127)
> at
> org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule$$anon$1.visitCall(TemporalJoinRewriteWithUniqueKeyRule.scala:88)
> at
> org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule$$anon$1.visitCall(TemporalJoinRewriteWithUniqueKeyRule.scala:70)
> at org.apache.calcite.rex.RexCall.accept(RexCall.java:174)
> at org.apache.calcite.rex.RexShuttle.visitList(RexShuttle.java:158)
> at org.apache.calcite.rex.RexShuttle.visitCall(RexShuttle.java:110)
> at
> org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule$$anon$1.visitCall(TemporalJoinRewriteWithUniqueKeyRule.scala:109)
> at
> org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule$$anon$1.visitCall(TemporalJoinRewriteWithUniqueKeyRule.scala:70)
> at org.apache.calcite.rex.RexCall.accept(RexCall.java:174)
> at
> org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule.onMatch(TemporalJoinRewriteWithUniqueKeyRule.scala:70)
> at
> org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:333)
> at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:542)
> at org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:407)
> at
> org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:243)
> at
> org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
> at
> org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:202)
> at org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:189)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
> at
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
> at
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
> at scala.collection.Iterator.foreach(Iterator.scala:943)
> at scala.collection.Iterator.foreach$(Iterator.scala:943)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
> at
> org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
> at
> org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:287)
> at
> org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:160)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1329)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:676)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.executeOperation(TableEnvironmentImpl.java:767)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:666)
>
>
> 请问这个sql 我应该怎么写
>
>
>
>
>
>
>
>
> ————————————————————————————————————分割线————————————————————————————————————————————————
>
>
> 在 2021-09-02 11:18:55，"Wayne" <17...@163.com> 写道：
> >
> >
> >
> >我是使用tableEnv.executeSql   执行那两个create 语句 然后用 tableEnv.sqlQuery  执行
> select语句 ，然后报错， 没有再执行其他sql额 ， aaa.columnInfos [ 'EVENT_ID' ].newValue
> 就是取出map里面 EVENT_ID 的key的 newValue 值，从create 语句可以看出 这个是个string
> >
> > 另外   tableEnv.explainSql(xxx) 也不能执行那个select语句，报的一样的错误， 我先在本地复现一下试试吧
> >
> >
> >
> >
> >
> >
> >
> >
> >在 2021-09-02 10:31:20，"Caizhi Weng" <ts...@gmail.com> 写道：
> >>Hi！
> >>
> >>没有复现这个问题。后续是还有其他 SQL 吗？特别是对 aaa.columnInfos [ 'EVENT_ID' ].newValue 有
> >>filter 的 SQL。
> >>
> >>可以尝试先在本地找一个能复现的 case，通过 tableEnv.explainSql() 就能快速检查能否产生 plan。如果找到了能复现的
> >>SQL，可以在 jira 提一个 issue。
> >>
> >>Wayne <17...@163.com> 于2021年9月2日周四 上午10:43写道：
> >>
> >>> 表a
> >>> CREATE TABLE aaa (
> >>>
> >>>
> >>> `columnInfos` MAP<STRING NOT NULL,ROW<oldValue STRING NULL ,newValue
> >>> STRING  ,name STRING NOT NULL ,isKeyColumn BOOLEAN NOT NULL,type
> STRING NOT
> >>> NULL > NOT NULL>  NOT NULL,
> >>> .......
> >>> procTime AS PROCTIME()
> >>> ) WITH (
> >>>     'connector' = 'kafka' ,
> >>>     .....
> >>>     'format' = 'avro'
> >>> )
> >>>
> >>>
> >>>
> >>>
> >>> 表b
> >>>
> >>>
> >>> CREATE TABLE bbb (
> >>> `event_id` STRING,
> >>> `genre_id` STRING,
> >>> `user_guid` STRING,
> >>> ......
> >>> `uuid` STRING
> >>> ) WITH (
> >>>     'connector' = 'kafka' ,
> >>>     .....
> >>>     'format' = 'avro'
> >>> )
> >>>
> >>>
> >>>
> >>>
> >>>
> >>>
> >>> 我的sql
> >>>
> >>>
> >>> SELECT
> >>>     CONCAT_WS(
> >>>         '_',
> >>>         user_guid,
> >>>         CAST( CAST( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS'
> ].newValue,
> >>> 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ) AS BIGINT ) AS STRING )) AS rowkey,
> >>>     columnInfos [ 'TICKET_COST' ].newValue AS cost,
> >>>     DATE_FORMAT( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue,
> >>> 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ), 'yyyy-MM-dd' ) AS crdate,
> >>>     columnInfos [ 'EVENT_ID' ].newValue AS eventid,
> >>>     columnInfos [ 'QUANTITY' ].newValue AS quantity,
> >>>     genre_id AS genreId,
> >>>     user_guid AS userGuid
> >>>     FROM
> >>>     aaa
> >>>    LEFT  JOIN bbb FOR SYSTEM_TIME AS OF aaa.procTime ON
> aaa.columnInfos [
> >>> 'EVENT_ID' ].newValue = bbb.event_id  and aaa.columnInfos [ 'EVENT_ID'
> >>> ].newValue is not null and  bbb.event_id is not null
> >>>
> >>>
> >>> 最后的  and aaa.columnInfos [ 'EVENT_ID' ].newValue is not null and
> >>> bbb.event_id is not null   无论加不加，都是报这个错误
> >>>
> >>> 在 2021-09-02 09:13:11，"Caizhi Weng" <ts...@gmail.com> 写道：
> >>> >Hi！
> >>> >
> >>> >Orders 和 Customers 就直接是 source 表吗？还是说 source 表到维表 join 之间有关于
> customer_id 或
> >>> >id 的 filter 条件？
> >>> >
> >>> >有一定可能是之前有关于 customer_id 或 id 的 filter 条件（比如 customer_id = 1），导致维表 join
> >>> >这里的等值条件被优化成了 customer_id = 1 和 id = 1 并分别下推到维表 join 之前。
> >>> >
> >>> >Wayne <17...@163.com> 于2021年9月1日周三 下午6:27写道：
> >>> >
> >>> >> 我的flink 版本是  flink-1.12.2-bin-scala_2.12
> >>> >> 我的sql 是
> >>> >> SELECT o.order_id, o.total, c.country, c.zip
> >>> >> FROM Orders AS o
> >>> >>   JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
> >>> >>     ON o.customer_id = c.id and o.customer_id is not null and  c.id
> is
> >>> >> not null ;
> >>> >> 或者
> >>> >> SELECT o.order_id, o.total, c.country, c.zip
> >>> >> FROM Orders AS o
> >>> >>   JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
> >>> >>     ON o.customer_id = c.id ;
> >>> >>
> >>> >> 都会报如下错误，麻烦帮我看看正确的写法是什么样的，灰常感谢
> >>> >>
> >>> >> Exception in thread "main"
> >>> org.apache.flink.table.api.ValidationException:
> >>> >> Currently the join key in Temporal Table Join can not be empty.
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.planner.plan.rules.logical.LogicalCorrelateToJoinFromGeneralTemporalTableRule.onMatch(LogicalCorrelateToJoinFromTemporalTableRule.scala:272)
> >>> >> at
> >>> >>
> >>>
> org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:333)
> >>> >> at
> org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:542)
> >>> >> at
> >>> org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:407)
> >>> >> at
> >>> >>
> >>>
> org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:243)
> >>> >> at
> >>> >>
> >>>
> org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
> >>> >> at
> >>> >>
> >>>
> org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:202)
> >>> >> at
> >>> org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:189)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$2(FlinkGroupProgram.scala:63)
> >>> >> at
> >>> >>
> >>>
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
> >>> >> at
> >>> >>
> >>>
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
> >>> >> at scala.collection.Iterator.foreach(Iterator.scala:943)
> >>> >> at scala.collection.Iterator.foreach$(Iterator.scala:943)
> >>> >> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
> >>> >> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
> >>> >> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
> >>> >> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
> >>> >> at
> scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
> >>> >> at
> scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
> >>> >> at
> scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1(FlinkGroupProgram.scala:60)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1$adapted(FlinkGroupProgram.scala:55)
> >>> >> at
> >>> >>
> >>>
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
> >>> >> at
> >>> >>
> >>>
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
> >>> >> at scala.collection.immutable.Range.foreach(Range.scala:158)
> >>> >> at
> scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
> >>> >> at
> scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
> >>> >> at
> scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.optimize(FlinkGroupProgram.scala:55)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
> >>> >> at
> >>> >>
> >>>
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
> >>> >> at
> >>> >>
> >>>
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
> >>> >> at scala.collection.Iterator.foreach(Iterator.scala:943)
> >>> >> at scala.collection.Iterator.foreach$(Iterator.scala:943)
> >>> >> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
> >>> >> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
> >>> >> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
> >>> >> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
> >>> >> at
> scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
> >>> >> at
> scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
> >>> >> at
> scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:287)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:160)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1329)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:707)
> >>> >> at
> >>> >>
> >>>
> org.apache.flink.table.api.internal.TableImpl.execute(TableImpl.java:577)
> >>> >>
> >>> >>
> >>> >>
> >>> >>
> >>> >>
> >>> >>
> >>> >>
> >>> >>
> >>> >>
> >>> >>
> >>> >>
> >>>
>

Re:Re:Re: Re: Temporal Joins 报 Currently the join key in Temporal Table Join can not be empty

Posted by Wayne <17...@163.com>.

Hello 


打扰了我最近再次尝试，还是报这个错误
我的flink版本为 flink-1.12.2-bin-scala_2.12
使用sql client执行
我的sql 如下  


CREATE TABLE stub_trans (
`uuid` STRING,
`columnInfos` MAP<STRING NOT NULL,ROW<oldValue STRING NULL ,newValue STRING  ,name STRING NOT NULL ,isKeyColumn BOOLEAN NOT NULL,type STRING NOT NULL > NOT NULL>  NOT NULL,
procTime TIMESTAMP(3) METADATA FROM 'timestamp' , 
WATERMARK FOR procTime AS procTime   
) WITH (
    'connector' = 'kafka',
..............
    'format' = 'avro'
);


CREATE TABLE kapeng_test (


`event_id` STRING,
`genre_id` STRING,
`user_guid` STRING,
procTime TIMESTAMP(3) METADATA FROM 'timestamp'  ,  
WATERMARK FOR procTime AS procTime   
) WITH (
    'connector' = 'kafka',
........
    'format' = 'avro'
);




CREATE TABLE purchase_hist (
    rowkey STRING,
    d ROW < cost STRING, crdate STRING, currency STRING, eventid STRING, genreid STRING, quantity STRING >,
    PRIMARY KEY ( rowkey ) NOT ENFORCED 
) WITH ( 
    'connector' = 'hbase-1.4', 
.........);




INSERT INTO purchase_hist SELECT
rowkey,
ROW ( cost, crdate, currency, eventid, genreid, quantity ) 
FROM
( SELECT
    CONCAT_WS(
        '_',
        user_guid,
        CAST( CAST( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue, 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ) AS BIGINT ) AS STRING )) AS rowkey,
    columnInfos [ 'TICKET_COST' ].newValue AS cost,
    DATE_FORMAT( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue, 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ), 'yyyy-MM-dd' ) AS crdate,
    columnInfos [ 'EVENT_ID' ].newValue AS eventid,
    columnInfos [ 'CURRENCY_CODE' ].newValue AS currency,
    columnInfos [ 'QUANTITY' ].newValue AS quantity,
    genre_id AS genreid,
    user_guid AS userGuid 
    FROM
    stub_trans
   LEFT  JOIN kapeng_test FOR SYSTEM_TIME AS OF stub_trans.procTime ON stub_trans.columnInfos [ 'EVENT_ID' ].newValue = kapeng_test.event_id )m




报错如下
Exception in thread "main" org.apache.flink.table.api.ValidationException: Currently the join key in Temporal Table Join can not be empty.
at org.apache.flink.table.planner.plan.rules.logical.LogicalCorrelateToJoinFromGeneralTemporalTableRule.onMatch(LogicalCorrelateToJoinFromTemporalTableRule.scala:272)
at org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:333)
at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:542)
at org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:407)
at org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:243)
at org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
at org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:202)
at org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:189)
at org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
at org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$2(FlinkGroupProgram.scala:63)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1(FlinkGroupProgram.scala:60)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1$adapted(FlinkGroupProgram.scala:55)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
at scala.collection.immutable.Range.foreach(Range.scala:158)
at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.optimize(FlinkGroupProgram.scala:55)
at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
at org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
at org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:287)
at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:160)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1329)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:676)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeOperation(TableEnvironmentImpl.java:767)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:666)






如果我将
LEFT  JOIN kapeng_test FOR SYSTEM_TIME AS OF stub_trans.procTime ON stub_trans.columnInfos [ 'EVENT_ID' ].newValue = kapeng_test.event_id
中的 
 stub_trans.columnInfos [ 'EVENT_ID' ].newValue  替换为 uuid  则这部分语句为


LEFT  JOIN kapeng_test FOR SYSTEM_TIME AS OF stub_trans.procTime ON stub_trans.uuid = kapeng_test.event_id




则报错如下:


Exception in thread "main" org.apache.flink.table.api.ValidationException: Temporal Table Join requires primary key in versioned table, but no primary key can be found. The physical plan is:
FlinkLogicalJoin(condition=[AND(=($0, $3), __INITIAL_TEMPORAL_JOIN_CONDITION($2, $6, __TEMPORAL_JOIN_LEFT_KEY($0), __TEMPORAL_JOIN_RIGHT_KEY($3)))], joinType=[left])
  FlinkLogicalCalc(select=[uuid, columnInfos, Reinterpret(CAST(timestamp)) AS procTime])
    FlinkLogicalTableSourceScan(table=[[default_catalog, default_database, stub_trans, watermark=[CAST($2):TIMESTAMP(3)]]], fields=[uuid, columnInfos, timestamp])
  FlinkLogicalSnapshot(period=[$cor0.procTime])
    FlinkLogicalCalc(select=[event_id, genre_id, user_guid, Reinterpret(CAST(timestamp)) AS procTime])
      FlinkLogicalTableSourceScan(table=[[default_catalog, default_database, kapeng_test, watermark=[CAST($3):TIMESTAMP(3)]]], fields=[event_id, genre_id, user_guid, timestamp])




at org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule.org$apache$flink$table$planner$plan$rules$logical$TemporalJoinRewriteWithUniqueKeyRule$$validateRightPrimaryKey(TemporalJoinRewriteWithUniqueKeyRule.scala:127)
at org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule$$anon$1.visitCall(TemporalJoinRewriteWithUniqueKeyRule.scala:88)
at org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule$$anon$1.visitCall(TemporalJoinRewriteWithUniqueKeyRule.scala:70)
at org.apache.calcite.rex.RexCall.accept(RexCall.java:174)
at org.apache.calcite.rex.RexShuttle.visitList(RexShuttle.java:158)
at org.apache.calcite.rex.RexShuttle.visitCall(RexShuttle.java:110)
at org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule$$anon$1.visitCall(TemporalJoinRewriteWithUniqueKeyRule.scala:109)
at org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule$$anon$1.visitCall(TemporalJoinRewriteWithUniqueKeyRule.scala:70)
at org.apache.calcite.rex.RexCall.accept(RexCall.java:174)
at org.apache.flink.table.planner.plan.rules.logical.TemporalJoinRewriteWithUniqueKeyRule.onMatch(TemporalJoinRewriteWithUniqueKeyRule.scala:70)
at org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:333)
at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:542)
at org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:407)
at org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:243)
at org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
at org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:202)
at org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:189)
at org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
at org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
at org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
at org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:287)
at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:160)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1329)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:676)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeOperation(TableEnvironmentImpl.java:767)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:666)


请问这个sql 我应该怎么写







————————————————————————————————————分割线————————————————————————————————————————————————


在 2021-09-02 11:18:55，"Wayne" <17...@163.com> 写道：
>
>
>
>我是使用tableEnv.executeSql   执行那两个create 语句 然后用 tableEnv.sqlQuery  执行 select语句 ，然后报错， 没有再执行其他sql额 ， aaa.columnInfos [ 'EVENT_ID' ].newValue 就是取出map里面 EVENT_ID 的key的 newValue 值，从create 语句可以看出 这个是个string
>
> 另外   tableEnv.explainSql(xxx) 也不能执行那个select语句，报的一样的错误， 我先在本地复现一下试试吧
>
>
>
>
>
>
>
>
>在 2021-09-02 10:31:20，"Caizhi Weng" <ts...@gmail.com> 写道：
>>Hi！
>>
>>没有复现这个问题。后续是还有其他 SQL 吗？特别是对 aaa.columnInfos [ 'EVENT_ID' ].newValue 有
>>filter 的 SQL。
>>
>>可以尝试先在本地找一个能复现的 case，通过 tableEnv.explainSql() 就能快速检查能否产生 plan。如果找到了能复现的
>>SQL，可以在 jira 提一个 issue。
>>
>>Wayne <17...@163.com> 于2021年9月2日周四 上午10:43写道：
>>
>>> 表a
>>> CREATE TABLE aaa (
>>>
>>>
>>> `columnInfos` MAP<STRING NOT NULL,ROW<oldValue STRING NULL ,newValue
>>> STRING  ,name STRING NOT NULL ,isKeyColumn BOOLEAN NOT NULL,type STRING NOT
>>> NULL > NOT NULL>  NOT NULL,
>>> .......
>>> procTime AS PROCTIME()
>>> ) WITH (
>>>     'connector' = 'kafka' ,
>>>     .....
>>>     'format' = 'avro'
>>> )
>>>
>>>
>>>
>>>
>>> 表b
>>>
>>>
>>> CREATE TABLE bbb (
>>> `event_id` STRING,
>>> `genre_id` STRING,
>>> `user_guid` STRING,
>>> ......
>>> `uuid` STRING
>>> ) WITH (
>>>     'connector' = 'kafka' ,
>>>     .....
>>>     'format' = 'avro'
>>> )
>>>
>>>
>>>
>>>
>>>
>>>
>>> 我的sql
>>>
>>>
>>> SELECT
>>>     CONCAT_WS(
>>>         '_',
>>>         user_guid,
>>>         CAST( CAST( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue,
>>> 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ) AS BIGINT ) AS STRING )) AS rowkey,
>>>     columnInfos [ 'TICKET_COST' ].newValue AS cost,
>>>     DATE_FORMAT( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue,
>>> 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ), 'yyyy-MM-dd' ) AS crdate,
>>>     columnInfos [ 'EVENT_ID' ].newValue AS eventid,
>>>     columnInfos [ 'QUANTITY' ].newValue AS quantity,
>>>     genre_id AS genreId,
>>>     user_guid AS userGuid
>>>     FROM
>>>     aaa
>>>    LEFT  JOIN bbb FOR SYSTEM_TIME AS OF aaa.procTime ON aaa.columnInfos [
>>> 'EVENT_ID' ].newValue = bbb.event_id  and aaa.columnInfos [ 'EVENT_ID'
>>> ].newValue is not null and  bbb.event_id is not null
>>>
>>>
>>> 最后的  and aaa.columnInfos [ 'EVENT_ID' ].newValue is not null and
>>> bbb.event_id is not null   无论加不加，都是报这个错误
>>>
>>> 在 2021-09-02 09:13:11，"Caizhi Weng" <ts...@gmail.com> 写道：
>>> >Hi！
>>> >
>>> >Orders 和 Customers 就直接是 source 表吗？还是说 source 表到维表 join 之间有关于 customer_id 或
>>> >id 的 filter 条件？
>>> >
>>> >有一定可能是之前有关于 customer_id 或 id 的 filter 条件（比如 customer_id = 1），导致维表 join
>>> >这里的等值条件被优化成了 customer_id = 1 和 id = 1 并分别下推到维表 join 之前。
>>> >
>>> >Wayne <17...@163.com> 于2021年9月1日周三 下午6:27写道：
>>> >
>>> >> 我的flink 版本是  flink-1.12.2-bin-scala_2.12
>>> >> 我的sql 是
>>> >> SELECT o.order_id, o.total, c.country, c.zip
>>> >> FROM Orders AS o
>>> >>   JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
>>> >>     ON o.customer_id = c.id and o.customer_id is not null and  c.id is
>>> >> not null ;
>>> >> 或者
>>> >> SELECT o.order_id, o.total, c.country, c.zip
>>> >> FROM Orders AS o
>>> >>   JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
>>> >>     ON o.customer_id = c.id ;
>>> >>
>>> >> 都会报如下错误，麻烦帮我看看正确的写法是什么样的，灰常感谢
>>> >>
>>> >> Exception in thread "main"
>>> org.apache.flink.table.api.ValidationException:
>>> >> Currently the join key in Temporal Table Join can not be empty.
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.rules.logical.LogicalCorrelateToJoinFromGeneralTemporalTableRule.onMatch(LogicalCorrelateToJoinFromTemporalTableRule.scala:272)
>>> >> at
>>> >>
>>> org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:333)
>>> >> at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:542)
>>> >> at
>>> org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:407)
>>> >> at
>>> >>
>>> org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:243)
>>> >> at
>>> >>
>>> org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
>>> >> at
>>> >>
>>> org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:202)
>>> >> at
>>> org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:189)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$2(FlinkGroupProgram.scala:63)
>>> >> at
>>> >>
>>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
>>> >> at
>>> >>
>>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
>>> >> at scala.collection.Iterator.foreach(Iterator.scala:943)
>>> >> at scala.collection.Iterator.foreach$(Iterator.scala:943)
>>> >> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
>>> >> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
>>> >> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
>>> >> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
>>> >> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
>>> >> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
>>> >> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1(FlinkGroupProgram.scala:60)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1$adapted(FlinkGroupProgram.scala:55)
>>> >> at
>>> >>
>>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
>>> >> at
>>> >>
>>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
>>> >> at scala.collection.immutable.Range.foreach(Range.scala:158)
>>> >> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
>>> >> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
>>> >> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.optimize(FlinkGroupProgram.scala:55)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
>>> >> at
>>> >>
>>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
>>> >> at
>>> >>
>>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
>>> >> at scala.collection.Iterator.foreach(Iterator.scala:943)
>>> >> at scala.collection.Iterator.foreach$(Iterator.scala:943)
>>> >> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
>>> >> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
>>> >> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
>>> >> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
>>> >> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
>>> >> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
>>> >> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:287)
>>> >> at
>>> >>
>>> org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:160)
>>> >> at
>>> >>
>>> org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1329)
>>> >> at
>>> >>
>>> org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:707)
>>> >> at
>>> >>
>>> org.apache.flink.table.api.internal.TableImpl.execute(TableImpl.java:577)
>>> >>
>>> >>
>>> >>
>>> >>
>>> >>
>>> >>
>>> >>
>>> >>
>>> >>
>>> >>
>>> >>
>>>

Re:Re: Re: Temporal Joins 报 Currently the join key in Temporal Table Join can not be empty

Posted by Wayne <17...@163.com>.



我是使用tableEnv.executeSql   执行那两个create 语句 然后用 tableEnv.sqlQuery  执行 select语句 ，然后报错， 没有再执行其他sql额 ， aaa.columnInfos [ 'EVENT_ID' ].newValue 就是取出map里面 EVENT_ID 的key的 newValue 值，从create 语句可以看出 这个是个string

 另外   tableEnv.explainSql(xxx) 也不能执行那个select语句，报的一样的错误， 我先在本地复现一下试试吧








在 2021-09-02 10:31:20，"Caizhi Weng" <ts...@gmail.com> 写道：
>Hi！
>
>没有复现这个问题。后续是还有其他 SQL 吗？特别是对 aaa.columnInfos [ 'EVENT_ID' ].newValue 有
>filter 的 SQL。
>
>可以尝试先在本地找一个能复现的 case，通过 tableEnv.explainSql() 就能快速检查能否产生 plan。如果找到了能复现的
>SQL，可以在 jira 提一个 issue。
>
>Wayne <17...@163.com> 于2021年9月2日周四 上午10:43写道：
>
>> 表a
>> CREATE TABLE aaa (
>>
>>
>> `columnInfos` MAP<STRING NOT NULL,ROW<oldValue STRING NULL ,newValue
>> STRING  ,name STRING NOT NULL ,isKeyColumn BOOLEAN NOT NULL,type STRING NOT
>> NULL > NOT NULL>  NOT NULL,
>> .......
>> procTime AS PROCTIME()
>> ) WITH (
>>     'connector' = 'kafka' ,
>>     .....
>>     'format' = 'avro'
>> )
>>
>>
>>
>>
>> 表b
>>
>>
>> CREATE TABLE bbb (
>> `event_id` STRING,
>> `genre_id` STRING,
>> `user_guid` STRING,
>> ......
>> `uuid` STRING
>> ) WITH (
>>     'connector' = 'kafka' ,
>>     .....
>>     'format' = 'avro'
>> )
>>
>>
>>
>>
>>
>>
>> 我的sql
>>
>>
>> SELECT
>>     CONCAT_WS(
>>         '_',
>>         user_guid,
>>         CAST( CAST( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue,
>> 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ) AS BIGINT ) AS STRING )) AS rowkey,
>>     columnInfos [ 'TICKET_COST' ].newValue AS cost,
>>     DATE_FORMAT( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue,
>> 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ), 'yyyy-MM-dd' ) AS crdate,
>>     columnInfos [ 'EVENT_ID' ].newValue AS eventid,
>>     columnInfos [ 'QUANTITY' ].newValue AS quantity,
>>     genre_id AS genreId,
>>     user_guid AS userGuid
>>     FROM
>>     aaa
>>    LEFT  JOIN bbb FOR SYSTEM_TIME AS OF aaa.procTime ON aaa.columnInfos [
>> 'EVENT_ID' ].newValue = bbb.event_id  and aaa.columnInfos [ 'EVENT_ID'
>> ].newValue is not null and  bbb.event_id is not null
>>
>>
>> 最后的  and aaa.columnInfos [ 'EVENT_ID' ].newValue is not null and
>> bbb.event_id is not null   无论加不加，都是报这个错误
>>
>> 在 2021-09-02 09:13:11，"Caizhi Weng" <ts...@gmail.com> 写道：
>> >Hi！
>> >
>> >Orders 和 Customers 就直接是 source 表吗？还是说 source 表到维表 join 之间有关于 customer_id 或
>> >id 的 filter 条件？
>> >
>> >有一定可能是之前有关于 customer_id 或 id 的 filter 条件（比如 customer_id = 1），导致维表 join
>> >这里的等值条件被优化成了 customer_id = 1 和 id = 1 并分别下推到维表 join 之前。
>> >
>> >Wayne <17...@163.com> 于2021年9月1日周三 下午6:27写道：
>> >
>> >> 我的flink 版本是  flink-1.12.2-bin-scala_2.12
>> >> 我的sql 是
>> >> SELECT o.order_id, o.total, c.country, c.zip
>> >> FROM Orders AS o
>> >>   JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
>> >>     ON o.customer_id = c.id and o.customer_id is not null and  c.id is
>> >> not null ;
>> >> 或者
>> >> SELECT o.order_id, o.total, c.country, c.zip
>> >> FROM Orders AS o
>> >>   JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
>> >>     ON o.customer_id = c.id ;
>> >>
>> >> 都会报如下错误，麻烦帮我看看正确的写法是什么样的，灰常感谢
>> >>
>> >> Exception in thread "main"
>> org.apache.flink.table.api.ValidationException:
>> >> Currently the join key in Temporal Table Join can not be empty.
>> >> at
>> >>
>> org.apache.flink.table.planner.plan.rules.logical.LogicalCorrelateToJoinFromGeneralTemporalTableRule.onMatch(LogicalCorrelateToJoinFromTemporalTableRule.scala:272)
>> >> at
>> >>
>> org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:333)
>> >> at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:542)
>> >> at
>> org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:407)
>> >> at
>> >>
>> org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:243)
>> >> at
>> >>
>> org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
>> >> at
>> >>
>> org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:202)
>> >> at
>> org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:189)
>> >> at
>> >>
>> org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
>> >> at
>> >>
>> org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
>> >> at
>> >>
>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$2(FlinkGroupProgram.scala:63)
>> >> at
>> >>
>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
>> >> at
>> >>
>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
>> >> at scala.collection.Iterator.foreach(Iterator.scala:943)
>> >> at scala.collection.Iterator.foreach$(Iterator.scala:943)
>> >> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
>> >> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
>> >> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
>> >> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
>> >> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
>> >> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
>> >> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
>> >> at
>> >>
>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1(FlinkGroupProgram.scala:60)
>> >> at
>> >>
>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1$adapted(FlinkGroupProgram.scala:55)
>> >> at
>> >>
>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
>> >> at
>> >>
>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
>> >> at scala.collection.immutable.Range.foreach(Range.scala:158)
>> >> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
>> >> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
>> >> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
>> >> at
>> >>
>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.optimize(FlinkGroupProgram.scala:55)
>> >> at
>> >>
>> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
>> >> at
>> >>
>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
>> >> at
>> >>
>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
>> >> at scala.collection.Iterator.foreach(Iterator.scala:943)
>> >> at scala.collection.Iterator.foreach$(Iterator.scala:943)
>> >> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
>> >> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
>> >> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
>> >> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
>> >> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
>> >> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
>> >> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
>> >> at
>> >>
>> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
>> >> at
>> >>
>> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
>> >> at
>> >>
>> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
>> >> at
>> >>
>> org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
>> >> at
>> >>
>> org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:287)
>> >> at
>> >>
>> org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:160)
>> >> at
>> >>
>> org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1329)
>> >> at
>> >>
>> org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:707)
>> >> at
>> >>
>> org.apache.flink.table.api.internal.TableImpl.execute(TableImpl.java:577)
>> >>
>> >>
>> >>
>> >>
>> >>
>> >>
>> >>
>> >>
>> >>
>> >>
>> >>
>>

Re: Re: Temporal Joins 报 Currently the join key in Temporal Table Join can not be empty

Posted by Caizhi Weng <ts...@gmail.com>.

Hi！

没有复现这个问题。后续是还有其他 SQL 吗？特别是对 aaa.columnInfos [ 'EVENT_ID' ].newValue 有
filter 的 SQL。

可以尝试先在本地找一个能复现的 case，通过 tableEnv.explainSql() 就能快速检查能否产生 plan。如果找到了能复现的
SQL，可以在 jira 提一个 issue。

Wayne <17...@163.com> 于2021年9月2日周四 上午10:43写道：

> 表a
> CREATE TABLE aaa (
>
>
> `columnInfos` MAP<STRING NOT NULL,ROW<oldValue STRING NULL ,newValue
> STRING  ,name STRING NOT NULL ,isKeyColumn BOOLEAN NOT NULL,type STRING NOT
> NULL > NOT NULL>  NOT NULL,
> .......
> procTime AS PROCTIME()
> ) WITH (
>     'connector' = 'kafka' ,
>     .....
>     'format' = 'avro'
> )
>
>
>
>
> 表b
>
>
> CREATE TABLE bbb (
> `event_id` STRING,
> `genre_id` STRING,
> `user_guid` STRING,
> ......
> `uuid` STRING
> ) WITH (
>     'connector' = 'kafka' ,
>     .....
>     'format' = 'avro'
> )
>
>
>
>
>
>
> 我的sql
>
>
> SELECT
>     CONCAT_WS(
>         '_',
>         user_guid,
>         CAST( CAST( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue,
> 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ) AS BIGINT ) AS STRING )) AS rowkey,
>     columnInfos [ 'TICKET_COST' ].newValue AS cost,
>     DATE_FORMAT( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue,
> 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ), 'yyyy-MM-dd' ) AS crdate,
>     columnInfos [ 'EVENT_ID' ].newValue AS eventid,
>     columnInfos [ 'QUANTITY' ].newValue AS quantity,
>     genre_id AS genreId,
>     user_guid AS userGuid
>     FROM
>     aaa
>    LEFT  JOIN bbb FOR SYSTEM_TIME AS OF aaa.procTime ON aaa.columnInfos [
> 'EVENT_ID' ].newValue = bbb.event_id  and aaa.columnInfos [ 'EVENT_ID'
> ].newValue is not null and  bbb.event_id is not null
>
>
> 最后的  and aaa.columnInfos [ 'EVENT_ID' ].newValue is not null and
> bbb.event_id is not null   无论加不加，都是报这个错误
>
> 在 2021-09-02 09:13:11，"Caizhi Weng" <ts...@gmail.com> 写道：
> >Hi！
> >
> >Orders 和 Customers 就直接是 source 表吗？还是说 source 表到维表 join 之间有关于 customer_id 或
> >id 的 filter 条件？
> >
> >有一定可能是之前有关于 customer_id 或 id 的 filter 条件（比如 customer_id = 1），导致维表 join
> >这里的等值条件被优化成了 customer_id = 1 和 id = 1 并分别下推到维表 join 之前。
> >
> >Wayne <17...@163.com> 于2021年9月1日周三 下午6:27写道：
> >
> >> 我的flink 版本是  flink-1.12.2-bin-scala_2.12
> >> 我的sql 是
> >> SELECT o.order_id, o.total, c.country, c.zip
> >> FROM Orders AS o
> >>   JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
> >>     ON o.customer_id = c.id and o.customer_id is not null and  c.id is
> >> not null ;
> >> 或者
> >> SELECT o.order_id, o.total, c.country, c.zip
> >> FROM Orders AS o
> >>   JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
> >>     ON o.customer_id = c.id ;
> >>
> >> 都会报如下错误，麻烦帮我看看正确的写法是什么样的，灰常感谢
> >>
> >> Exception in thread "main"
> org.apache.flink.table.api.ValidationException:
> >> Currently the join key in Temporal Table Join can not be empty.
> >> at
> >>
> org.apache.flink.table.planner.plan.rules.logical.LogicalCorrelateToJoinFromGeneralTemporalTableRule.onMatch(LogicalCorrelateToJoinFromTemporalTableRule.scala:272)
> >> at
> >>
> org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:333)
> >> at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:542)
> >> at
> org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:407)
> >> at
> >>
> org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:243)
> >> at
> >>
> org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
> >> at
> >>
> org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:202)
> >> at
> org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:189)
> >> at
> >>
> org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
> >> at
> >>
> org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
> >> at
> >>
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$2(FlinkGroupProgram.scala:63)
> >> at
> >>
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
> >> at
> >>
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
> >> at scala.collection.Iterator.foreach(Iterator.scala:943)
> >> at scala.collection.Iterator.foreach$(Iterator.scala:943)
> >> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
> >> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
> >> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
> >> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
> >> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
> >> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
> >> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> >> at
> >>
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1(FlinkGroupProgram.scala:60)
> >> at
> >>
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1$adapted(FlinkGroupProgram.scala:55)
> >> at
> >>
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
> >> at
> >>
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
> >> at scala.collection.immutable.Range.foreach(Range.scala:158)
> >> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
> >> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
> >> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> >> at
> >>
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.optimize(FlinkGroupProgram.scala:55)
> >> at
> >>
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
> >> at
> >>
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
> >> at
> >>
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
> >> at scala.collection.Iterator.foreach(Iterator.scala:943)
> >> at scala.collection.Iterator.foreach$(Iterator.scala:943)
> >> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
> >> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
> >> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
> >> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
> >> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
> >> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
> >> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> >> at
> >>
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
> >> at
> >>
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
> >> at
> >>
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
> >> at
> >>
> org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
> >> at
> >>
> org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:287)
> >> at
> >>
> org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:160)
> >> at
> >>
> org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1329)
> >> at
> >>
> org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:707)
> >> at
> >>
> org.apache.flink.table.api.internal.TableImpl.execute(TableImpl.java:577)
> >>
> >>
> >>
> >>
> >>
> >>
> >>
> >>
> >>
> >>
> >>
>

Re:Re: Temporal Joins 报 Currently the join key in Temporal Table Join can not be empty

Posted by Wayne <17...@163.com>.

表a
CREATE TABLE aaa (


`columnInfos` MAP<STRING NOT NULL,ROW<oldValue STRING NULL ,newValue STRING  ,name STRING NOT NULL ,isKeyColumn BOOLEAN NOT NULL,type STRING NOT NULL > NOT NULL>  NOT NULL,
.......
procTime AS PROCTIME()   
) WITH (
    'connector' = 'kafka' ,
    .....
    'format' = 'avro'
)




表b


CREATE TABLE bbb (
`event_id` STRING,
`genre_id` STRING,
`user_guid` STRING,
......
`uuid` STRING
) WITH (
    'connector' = 'kafka' ,
    .....
    'format' = 'avro'
)






我的sql


SELECT
    CONCAT_WS(
        '_',
        user_guid,
        CAST( CAST( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue, 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ) AS BIGINT ) AS STRING )) AS rowkey,
    columnInfos [ 'TICKET_COST' ].newValue AS cost,
    DATE_FORMAT( TO_TIMESTAMP ( columnInfos [ 'CREATED_TS' ].newValue, 'yyyy-MM-dd:HH:mm:ss.SSSSSSSSS' ), 'yyyy-MM-dd' ) AS crdate,
    columnInfos [ 'EVENT_ID' ].newValue AS eventid,
    columnInfos [ 'QUANTITY' ].newValue AS quantity,
    genre_id AS genreId,
    user_guid AS userGuid 
    FROM
    aaa
   LEFT  JOIN bbb FOR SYSTEM_TIME AS OF aaa.procTime ON aaa.columnInfos [ 'EVENT_ID' ].newValue = bbb.event_id  and aaa.columnInfos [ 'EVENT_ID' ].newValue is not null and  bbb.event_id is not null 


最后的  and aaa.columnInfos [ 'EVENT_ID' ].newValue is not null and  bbb.event_id is not null   无论加不加，都是报这个错误

在 2021-09-02 09:13:11，"Caizhi Weng" <ts...@gmail.com> 写道：
>Hi！
>
>Orders 和 Customers 就直接是 source 表吗？还是说 source 表到维表 join 之间有关于 customer_id 或
>id 的 filter 条件？
>
>有一定可能是之前有关于 customer_id 或 id 的 filter 条件（比如 customer_id = 1），导致维表 join
>这里的等值条件被优化成了 customer_id = 1 和 id = 1 并分别下推到维表 join 之前。
>
>Wayne <17...@163.com> 于2021年9月1日周三 下午6:27写道：
>
>> 我的flink 版本是  flink-1.12.2-bin-scala_2.12
>> 我的sql 是
>> SELECT o.order_id, o.total, c.country, c.zip
>> FROM Orders AS o
>>   JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
>>     ON o.customer_id = c.id and o.customer_id is not null and  c.id is
>> not null ;
>> 或者
>> SELECT o.order_id, o.total, c.country, c.zip
>> FROM Orders AS o
>>   JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
>>     ON o.customer_id = c.id ;
>>
>> 都会报如下错误，麻烦帮我看看正确的写法是什么样的，灰常感谢
>>
>> Exception in thread "main" org.apache.flink.table.api.ValidationException:
>> Currently the join key in Temporal Table Join can not be empty.
>> at
>> org.apache.flink.table.planner.plan.rules.logical.LogicalCorrelateToJoinFromGeneralTemporalTableRule.onMatch(LogicalCorrelateToJoinFromTemporalTableRule.scala:272)
>> at
>> org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:333)
>> at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:542)
>> at org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:407)
>> at
>> org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:243)
>> at
>> org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
>> at
>> org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:202)
>> at org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:189)
>> at
>> org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
>> at
>> org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
>> at
>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$2(FlinkGroupProgram.scala:63)
>> at
>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
>> at
>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
>> at scala.collection.Iterator.foreach(Iterator.scala:943)
>> at scala.collection.Iterator.foreach$(Iterator.scala:943)
>> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
>> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
>> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
>> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
>> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
>> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
>> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
>> at
>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1(FlinkGroupProgram.scala:60)
>> at
>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1$adapted(FlinkGroupProgram.scala:55)
>> at
>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
>> at
>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
>> at scala.collection.immutable.Range.foreach(Range.scala:158)
>> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
>> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
>> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
>> at
>> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.optimize(FlinkGroupProgram.scala:55)
>> at
>> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
>> at
>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
>> at
>> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
>> at scala.collection.Iterator.foreach(Iterator.scala:943)
>> at scala.collection.Iterator.foreach$(Iterator.scala:943)
>> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
>> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
>> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
>> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
>> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
>> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
>> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
>> at
>> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
>> at
>> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
>> at
>> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
>> at
>> org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
>> at
>> org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:287)
>> at
>> org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:160)
>> at
>> org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1329)
>> at
>> org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:707)
>> at
>> org.apache.flink.table.api.internal.TableImpl.execute(TableImpl.java:577)
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>

Re: Temporal Joins 报 Currently the join key in Temporal Table Join can not be empty

Posted by Caizhi Weng <ts...@gmail.com>.

Hi！

Orders 和 Customers 就直接是 source 表吗？还是说 source 表到维表 join 之间有关于 customer_id 或
id 的 filter 条件？

有一定可能是之前有关于 customer_id 或 id 的 filter 条件（比如 customer_id = 1），导致维表 join
这里的等值条件被优化成了 customer_id = 1 和 id = 1 并分别下推到维表 join 之前。

Wayne <17...@163.com> 于2021年9月1日周三 下午6:27写道：

> 我的flink 版本是  flink-1.12.2-bin-scala_2.12
> 我的sql 是
> SELECT o.order_id, o.total, c.country, c.zip
> FROM Orders AS o
>   JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
>     ON o.customer_id = c.id and o.customer_id is not null and  c.id is
> not null ;
> 或者
> SELECT o.order_id, o.total, c.country, c.zip
> FROM Orders AS o
>   JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
>     ON o.customer_id = c.id ;
>
> 都会报如下错误，麻烦帮我看看正确的写法是什么样的，灰常感谢
>
> Exception in thread "main" org.apache.flink.table.api.ValidationException:
> Currently the join key in Temporal Table Join can not be empty.
> at
> org.apache.flink.table.planner.plan.rules.logical.LogicalCorrelateToJoinFromGeneralTemporalTableRule.onMatch(LogicalCorrelateToJoinFromTemporalTableRule.scala:272)
> at
> org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:333)
> at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:542)
> at org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:407)
> at
> org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:243)
> at
> org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
> at
> org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:202)
> at org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:189)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$2(FlinkGroupProgram.scala:63)
> at
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
> at
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
> at scala.collection.Iterator.foreach(Iterator.scala:943)
> at scala.collection.Iterator.foreach$(Iterator.scala:943)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1(FlinkGroupProgram.scala:60)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1$adapted(FlinkGroupProgram.scala:55)
> at
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
> at
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
> at scala.collection.immutable.Range.foreach(Range.scala:158)
> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.optimize(FlinkGroupProgram.scala:55)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
> at
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196)
> at
> scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194)
> at scala.collection.Iterator.foreach(Iterator.scala:943)
> at scala.collection.Iterator.foreach$(Iterator.scala:943)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199)
> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192)
> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
> at
> org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:83)
> at
> org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
> at
> org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:287)
> at
> org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:160)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1329)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:707)
> at
> org.apache.flink.table.api.internal.TableImpl.execute(TableImpl.java:577)
>
>
>
>
>
>
>
>
>
>
>