wuchong commented on a change in pull request #8302: [FLINK-12269][table-blink] Support Temporal Table Join in blink planner and runtime URL: https://github.com/apache/flink/pull/8302#discussion_r279395068
########## File path: flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/plan/batch/sql/join/TemporalTableJoinTest.xml ########## @@ -0,0 +1,335 @@ +<?xml version="1.0" ?> +<!-- +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with +this work for additional information regarding copyright ownership. +The ASF licenses this file to you under the Apache License, Version 2.0 +(the "License"); you may not use this file except in compliance with +the License. You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +--> +<Root> + <TestCase name="testAvoidAggregatePushDown"> + <Resource name="sql"> + <![CDATA[ +SELECT b, count(a), sum(c), sum(d) +FROM ( +SELECT T.* FROM ( +SELECT b, a, sum(c) c, sum(d) d, PROCTIME() as proctime +FROM T1 +GROUP BY a, b + ) AS T +JOIN temporalTest FOR SYSTEM_TIME AS OF T.proctime AS D +ON T.a = D.id +WHERE D.age > 10 + ) AS T +GROUP BY b + ]]> + </Resource> + <Resource name="planBefore"> + <![CDATA[ +LogicalAggregate(group=[{0}], EXPR$1=[COUNT($1)], EXPR$2=[SUM($2)], EXPR$3=[SUM($3)]) ++- LogicalProject(b=[$0], a=[$1], c=[$2], d=[$3]) + +- LogicalFilter(condition=[>($7, 10)]) + +- LogicalFilter(condition=[=($1, $5)]) + +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{4}]) + :- LogicalProject(b=[$1], a=[$0], c=[$2], d=[$3], proctime=[PROCTIME()]) + : +- LogicalAggregate(group=[{0, 1}], c=[SUM($2)], d=[SUM($3)]) + : +- LogicalTableScan(table=[[T1]]) + +- LogicalSnapshot(period=[$cor0.proctime]) + +- LogicalTableScan(table=[[temporalTest, source: [TestTemporalTable(id, name, age)]]]) +]]> + </Resource> + <Resource name="planAfter"> + <![CDATA[ +HashAggregate(isMerge=[true], groupBy=[b], select=[b, Final_COUNT(count$0) AS EXPR$1, Final_SUM(sum$1) AS EXPR$2, Final_SUM(sum$2) AS EXPR$3]) ++- Exchange(distribution=[hash[b]]) + +- LocalHashAggregate(groupBy=[b], select=[b, Partial_COUNT(a) AS count$0, Partial_SUM(c) AS sum$1, Partial_SUM(d) AS sum$2]) + +- Calc(select=[b, a, c, d]) + +- TemporalTableJoin(table=[TestTemporalTable(id, name, age)], joinType=[InnerJoin], async=[false], on=[a=id], where=[>(age, 10)], select=[b, a, c, d, id]) + +- Calc(select=[b, a, c, d]) + +- HashAggregate(isMerge=[true], groupBy=[a, b], select=[a, b, Final_SUM(sum$0) AS c, Final_SUM(sum$1) AS d]) + +- Exchange(distribution=[hash[a, b]]) + +- LocalHashAggregate(groupBy=[a, b], select=[a, b, Partial_SUM(c) AS sum$0, Partial_SUM(d) AS sum$1]) + +- BoundedStreamScan(table=[[T1]], fields=[a, b, c, d]) +]]> + </Resource> + </TestCase> + <TestCase name="testJoinTemporalTable"> + <Resource name="sql"> + <![CDATA[SELECT * FROM MyTable AS T JOIN temporalTest FOR SYSTEM_TIME AS OF T.proctime AS D ON T.a = D.id]]> + </Resource> + <Resource name="planBefore"> + <![CDATA[ +LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3], id=[$4], name=[$5], age=[$6]) ++- LogicalFilter(condition=[=($0, $4)]) + +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{3}]) + :- LogicalTableScan(table=[[MyTable]]) + +- LogicalSnapshot(period=[$cor0.proctime]) + +- LogicalTableScan(table=[[temporalTest, source: [TestTemporalTable(id, name, age)]]]) +]]> + </Resource> + <Resource name="planAfter"> + <![CDATA[ +Calc(select=[a, b, c, PROCTIME() AS proctime, id, name, age]) ++- TemporalTableJoin(table=[TestTemporalTable(id, name, age)], joinType=[InnerJoin], async=[false], on=[a=id], select=[a, b, c, id, name, age]) + +- BoundedStreamScan(table=[[T0]], fields=[a, b, c]) +]]> + </Resource> + </TestCase> + <TestCase name="testJoinTemporalTableWithFilterPushDown"> + <Resource name="sql"> + <