[ https://issues.apache.org/jira/browse/FLINK-10734?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17375207#comment-17375207 ]
Leonard Xu edited comment on FLINK-10734 at 7/6/21, 7:46 AM: ------------------------------------------------------------- [~twalthr] Although the bug still exits in legacy temporal join function, user can use new event-time versioned table syntax [1] which supports multiple field in primary key, thus I think we can close this one. [1] [https://ci.apache.org/projects/flink/flink-docs-master/docs/dev/table/concepts/versioned_tables/#versioned-table-sources] was (Author: leonard xu): [~twalthr] Although the bug still exits in legacy temporal join function, user can use new event-time versioned table syntax [1] to which support multiple field in primary key, thus I think we can close this one. [1] https://ci.apache.org/projects/flink/flink-docs-master/docs/dev/table/concepts/versioned_tables/#versioned-table-sources > Temporal joins on heavily filtered tables might fail in planning > ---------------------------------------------------------------- > > Key: FLINK-10734 > URL: https://issues.apache.org/jira/browse/FLINK-10734 > Project: Flink > Issue Type: Bug > Components: Table SQL / Planner > Affects Versions: 1.7.0 > Reporter: Piotr Nowojski > Priority: Minor > > Following query: > {code} > val sqlQuery = > """ > |SELECT > | o.amount * r.rate AS amount > |FROM > | Orders AS o, > | LATERAL TABLE (Rates(o.rowtime)) AS r > |WHERE r.currency = o.currency > |""".stripMargin > {code} > with {{Rates}} defined as follows: > {code} > tEnv.registerTable("EuroRatesHistory", > tEnv.scan("RatesHistory").filter('currency === "Euro")) > tEnv.registerFunction( > "Rates", > tEnv.scan("EuroRatesHistory").createTemporalTableFunction('rowtime, > 'currency)) > {code} > Will fail with: > {noformat} > org.apache.flink.table.api.ValidationException: Only single column join key > is supported. Found [] in [InnerJoin(where: > (__TEMPORAL_JOIN_CONDITION(rowtime, rowtime0, currency)), join: (amount, > rowtime, currency, rate, rowtime0))] > at > org.apache.flink.table.plan.nodes.datastream.DataStreamTemporalJoinToCoProcessTranslator$TemporalJoinConditionExtractor.validateRightPrimaryKey(DataStreamTemporalJoinToCoProcessTranslator.scala:215) > at > org.apache.flink.table.plan.nodes.datastream.DataStreamTemporalJoinToCoProcessTranslator$TemporalJoinConditionExtractor.visitCall(DataStreamTemporalJoinToCoProcessTranslator.scala:183) > at > org.apache.flink.table.plan.nodes.datastream.DataStreamTemporalJoinToCoProcessTranslator$TemporalJoinConditionExtractor.visitCall(DataStreamTemporalJoinToCoProcessTranslator.scala:152) > {noformat} > The problem is that filtering condition {{('currency === "Euro")}} interferes > with joining condition, simplifying it to nothing. Note how top > {{LogicalFilter(condition=[=($3, $1)])}} changes during optimising and > finally disappears: > {noformat} > LogicalProject(amount=[*($0, $4)]) > LogicalFilter(condition=[=($3, $1)]) > LogicalTemporalTableJoin(condition=[__TEMPORAL_JOIN_CONDITION($2, $5, > $3)], joinType=[inner]) > LogicalTableScan(table=[[_DataStreamTable_0]]) > LogicalFilter(condition=[=($0, _UTF-16LE'Euro')]) > LogicalTableScan(table=[[_DataStreamTable_1]]) > {noformat} > {noformat} > LogicalProject(amount=[*($0, $4)]) > LogicalFilter(condition=[=(_UTF-16LE'Euro', $1)]) > LogicalProject(amount=[$0], currency=[$1], rowtime=[$2], currency0=[$3], > rate=[$4], rowtime0=[CAST($5):TIMESTAMP(3) NOT NULL]) > LogicalTemporalTableJoin(condition=[__TEMPORAL_JOIN_CONDITION($2, $5, > $3)], joinType=[inner]) > LogicalTableScan(table=[[_DataStreamTable_0]]) > LogicalFilter(condition=[=($0, _UTF-16LE'Euro')]) > LogicalTableScan(table=[[_DataStreamTable_1]]) > {noformat} > {noformat} > FlinkLogicalCalc(expr#0..4=[{inputs}], expr#5=[*($t0, $t3)], amount=[$t5]) > FlinkLogicalTemporalTableJoin(condition=[__TEMPORAL_JOIN_CONDITION($1, $4, > $2)], joinType=[inner]) > FlinkLogicalCalc(expr#0..2=[{inputs}], expr#3=[_UTF-16LE'Euro'], > expr#4=[=($t3, $t1)], amount=[$t0], rowtime=[$t2], $condition=[$t4]) > FlinkLogicalNativeTableScan(table=[[_DataStreamTable_0]]) > FlinkLogicalCalc(expr#0..2=[{inputs}], expr#3=[_UTF-16LE'Euro'], > expr#4=[=($t0, $t3)], proj#0..2=[{exprs}], $condition=[$t4]) > FlinkLogicalNativeTableScan(table=[[_DataStreamTable_1]]) > {noformat} -- This message was sent by Atlassian Jira (v8.3.4#803005)