Source表定义的proctime，在view中select后，不能用于Temporal table join.
sql如下：
```sql
CREATE TABLE SourceA (
id STRING,
procTime AS PROCTIME()
) WITH (
'connector' = 'kafka-0.11'
);
CREATE TABLE DimTable (
id STRING,
dim1 STRING,
primary key(id) NOT ENFORCED
) WITH (
'connector' = 'jdbc'
);
CREATE VIEW TestView AS
SELECT
id,
procTime
FROM RouteStaffPostQueue;
INSERT INTO Sink
SELECT
id,
dim1,
sum(post_count)
FROM TestView t1 JOIN DimTable FOR SYSTEM_TIME AS OF t1.procTime d1 ON d1.id =
t1.id
GROUP BY id, dim1;
// 换成如下的sql 没有问题。
INSERT INTO Sink
SELECT
id,
dim1,
sum(post_count)
FROM (SELECT * FROM SourceA) t1 JOIN DimTable FOR SYSTEM_TIME AS OF t1.procTime
d1 ON d1.id = t1.id
GROUP BY id, dim1;
```
如果把view的语句替换成TestView则没有问题，报错信息如下：
```
Loading class `com.mysql.jdbc.Driver'. This is deprecated. The new driver class
is `com.mysql.cj.jdbc.Driver'. The driver is automatically registered via the
SPI and manual loading of the driver class is generally unnecessary.
2020-11-02 19:21:17.363 [main] INFO
org.apache.flink.table.catalog.CatalogManager - Set the current default
catalog as [mysql] and the current default database as [default].
Exception in thread "main" org.apache.flink.table.api.TableException: Cannot
generate a valid execution plan for the given query:
FlinkLogicalSink(table=[mysql.default.dis_pair_day], fields=[sta_date,
dis_code, post_count])
+- FlinkLogicalCalc(select=[CAST(sta_date) AS sta_date, dis_code, CAST(EXPR$2)
AS post_count])
+- FlinkLogicalAggregate(group=[{0, 1}], EXPR$2=[SUM($2)])
+- FlinkLogicalCalc(select=[sta_date, dis_code, 1 AS post_count])
+- FlinkLogicalJoin(condition=[=($2, $1)], joinType=[inner])
:- FlinkLogicalCalc(select=[REPLACE(SUBSTR(postTime, 1, 10),
_UTF-16LE'-', _UTF-16LE'') AS sta_date, assetCode AS edCode])
: +- FlinkLogicalTableSourceScan(table=[[mysql, default,
RouteStaffPostQueue]], fields=[postTime, serviceChargeType, assetCode,
couponId, realCouponPay, serviceChargeAmount, busyStatus])
+- FlinkLogicalSnapshot(period=[$cor0.procTime])
+- FlinkLogicalCalc(select=[ed_code, dis_code])
+- FlinkLogicalTableSourceScan(table=[[mysql, default,
BasicEdInfo]], fields=[ed_code, dis_code, throw_area_type, is_free])
Temporal table join currently only supports 'FOR SYSTEM_TIME AS OF' left
table's proctime field, doesn't support 'PROCTIME()'
Please check the documentation for the set of currently supported SQL features.
at
org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:78)
at
org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram$$anonfun$optimize$1.apply(FlinkChainedProgram.scala:62)
at
org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram$$anonfun$optimize$1.apply(FlinkChainedProgram.scala:58)
at
scala.collection.TraversableOnce$$anonfun$foldLeft$1.apply(TraversableOnce.scala:157)
at
scala.collection.TraversableOnce$$anonfun$foldLeft$1.apply(TraversableOnce.scala:157)
at scala.collection.Iterator$class.foreach(Iterator.scala:891)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at scala.collection.TraversableOnce$class.foldLeft(TraversableOnce.scala:157)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
at
org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:57)
at
org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:164)
at
org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:80)
at
org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
at
org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:279)
at
org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:164)
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1264)
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:700)
at
org.apache.flink.table.api.internal.StatementSetImpl.execute(StatementSetImpl.java:97)
at com.fcbox.streaming.sql.submit.StreamingJob.run(StreamingJob.java:145)
at com.fcbox.streaming.sql.submit.StreamingJob.main(StreamingJob.java:77)
Caused by: org.apache.flink.table.api.TableException: Temporal table join
currently only supports 'FOR SYSTEM_TIME AS OF' left table's proctime field,
doesn't support 'PROCTIME()'
at
org.apache.flink.table.planner.plan.rules.physical.common.CommonLookupJoinRule$class.matches(CommonLookupJoinRule.scala:68)
at
org.apache.flink.table.planner.plan.rules.physical.common.BaseSnapshotOnCalcTableScanRule.matches(CommonLookupJoinRule.scala:150)
at
org.apache.flink.table.planner.plan.rules.physical.common.BaseSnapshotOnCalcTableScanRule.matches(CommonLookupJoinRule.scala:164)
at
org.apache.calcite.plan.volcano.VolcanoRuleCall.matchRecurse(VolcanoRuleCall.java:272)
at
org.apache.calcite.plan.volcano.VolcanoRuleCall.matchRecurse(VolcanoRuleCall.java:386)
at
org.apache.calcite.plan.volcano.VolcanoRuleCall.matchRecurse(VolcanoRuleCall.java:386)
at
org.apache.calcite.plan.volcano.VolcanoRuleCall.matchRecurse(VolcanoRuleCall.java:386)
at
org.apache.calcite.plan.volcano.VolcanoRuleCall.matchRecurse(VolcanoRuleCall.java:386)
at
org.apache.calcite.plan.volcano.VolcanoRuleCall.match(VolcanoRuleCall.java:256)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.fireRules(VolcanoPlanner.java:1476)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.registerImpl(VolcanoPlanner.java:1742)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.register(VolcanoPlanner.java:851)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:866)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:91)
at org.apache.calcite.rel.AbstractRelNode.onRegister(AbstractRelNode.java:321)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.registerImpl(VolcanoPlanner.java:1611)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.register(VolcanoPlanner.java:851)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:866)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:91)
at org.apache.calcite.rel.AbstractRelNode.onRegister(AbstractRelNode.java:321)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.registerImpl(VolcanoPlanner.java:1611)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.register(VolcanoPlanner.java:851)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:866)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:91)
at org.apache.calcite.rel.AbstractRelNode.onRegister(AbstractRelNode.java:321)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.registerImpl(VolcanoPlanner.java:1611)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.register(VolcanoPlanner.java:851)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:866)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:91)
at org.apache.calcite.rel.AbstractRelNode.onRegister(AbstractRelNode.java:321)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.registerImpl(VolcanoPlanner.java:1611)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.register(VolcanoPlanner.java:851)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:866)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.changeTraits(VolcanoPlanner.java:534)
at org.apache.calcite.tools.Programs$RuleSetProgram.run(Programs.java:324)
at
org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:64)
... 21 more
```
