From eca1bf6da5f8aa736595308c5bfcf05299a876e1 Mon Sep 17 00:00:00 2001 From: morazow Date: Sat, 19 Apr 2025 11:42:14 +0200 Subject: [PATCH 1/2] [FLINK-36808][table-planner] Fix LookupJoin bug when used with filterable table source --- .../common/CommonPhysicalLookupJoin.scala | 24 ++++- .../plan/stream/sql/join/LookupJoinTest.xml | 49 ++++++++++ .../plan/stream/sql/join/LookupJoinTest.scala | 33 ++++++- .../stream/sql/UnionLookupJoinITCase.scala | 96 +++++++++++++++++++ 4 files changed, 198 insertions(+), 4 deletions(-) create mode 100644 flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnionLookupJoinITCase.scala diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala index cd5def1c00ae71..b1a347df29d3b3 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala @@ -21,16 +21,17 @@ import org.apache.flink.table.api.{TableConfig, TableException} import org.apache.flink.table.catalog.{ObjectIdentifier, UniqueConstraint} import org.apache.flink.table.connector.ChangelogMode import org.apache.flink.table.planner.calcite.FlinkTypeFactory +import org.apache.flink.table.planner.plan.abilities.source.FilterPushDownSpec import org.apache.flink.table.planner.plan.nodes.FlinkRelNode import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalRel import org.apache.flink.table.planner.plan.schema.{IntermediateRelTable, LegacyTableSourceTable, TableSourceTable} -import org.apache.flink.table.planner.plan.utils.{ChangelogPlanUtils, ExpressionFormat, InputRefVisitor, JoinTypeUtil, LookupJoinUtil, RelExplainUtil, TemporalJoinUtil} +import org.apache.flink.table.planner.plan.utils.{ChangelogPlanUtils, ExpressionFormat, FlinkRexUtil, InputRefVisitor, JoinTypeUtil, LookupJoinUtil, RelExplainUtil, TemporalJoinUtil} import org.apache.flink.table.planner.plan.utils.ExpressionFormat.ExpressionFormat import org.apache.flink.table.planner.plan.utils.LookupJoinUtil._ import org.apache.flink.table.planner.plan.utils.PythonUtil.containsPythonCall import org.apache.flink.table.planner.plan.utils.RelExplainUtil.preferExpressionFormat +import org.apache.flink.table.planner.utils.JavaScalaConversionUtil import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig -import org.apache.flink.table.runtime.types.PlannerTypeUtils import org.apache.calcite.plan.{RelOptCluster, RelOptTable, RelTraitSet} import org.apache.calcite.plan.hep.HepRelVertex @@ -190,6 +191,10 @@ abstract class CommonPhysicalLookupJoin( case t: TableSourceTable => t.contextResolvedTable.getIdentifier case t: LegacyTableSourceTable[_] => t.tableIdentifier } + val filterPushdownString: String = temporalTable match { + case t: TableSourceTable => getTableFilterString(t) + case _: LegacyTableSourceTable[_] => "" + } super .explainTerms(pw) @@ -197,6 +202,7 @@ abstract class CommonPhysicalLookupJoin( .item("joinType", JoinTypeUtil.getFlinkJoinType(joinType)) .item("lookup", lookupKeys) .itemIf("where", whereString, whereString.nonEmpty) + .itemIf("filter", filterPushdownString, filterPushdownString.nonEmpty) .itemIf( "joinCondition", joinConditionToString(resultFieldNames, preferExpressionFormat(pw), pw.getDetailLevel), @@ -209,6 +215,20 @@ abstract class CommonPhysicalLookupJoin( .itemIf("retry", retryOptions.getOrElse(""), retryOptions.isDefined) } + private def getTableFilterString(t: TableSourceTable): String = { + val filterOpt = t.abilitySpecs.collectFirst { case spec: FilterPushDownSpec => spec } + filterOpt match { + case Some(filter) if !filter.getPredicates.isEmpty => + val fieldNames = JavaScalaConversionUtil.toScala(t.getRowType.getFieldNames) + val predicates = JavaScalaConversionUtil.toScala(filter.getPredicates) + predicates + .map(FlinkRexUtil.getExpressionString(_, fieldNames)) + .reduceOption((l, r) => String.format("and(%s, %s)", l, r)) + .getOrElse("") + case _ => "" + } + } + private def getInputChangelogMode(rel: RelNode): ChangelogMode = rel match { case streamPhysicalRel: StreamPhysicalRel => ChangelogPlanUtils.getChangelogMode(streamPhysicalRel).getOrElse(ChangelogMode.insertOnly()) diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml index 8686130a866b65..79b2cafb940983 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml @@ -1607,4 +1607,53 @@ Calc(select=[a, b, c, PROCTIME_MATERIALIZE(proctime) AS proctime, rowtime, id, n ]]> + + + + + + + + + + + diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala index 44e08fe6d113d5..4751563f1f56d8 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala @@ -427,7 +427,6 @@ class LookupJoinTest extends TableTestBase with Serializable { @Test def testJoinTemporalTableWithFunctionAndConstantCondition(): Unit = { - val sql = """ |SELECT * FROM MyTable AS T @@ -440,7 +439,6 @@ class LookupJoinTest extends TableTestBase with Serializable { @Test def testJoinTemporalTableWithMultiFunctionAndConstantCondition(): Unit = { - val sql = """ |SELECT * FROM MyTable AS T @@ -573,6 +571,37 @@ class LookupJoinTest extends TableTestBase with Serializable { util.verifyExecPlan(sql) } + @Test + def testJoinFilterableTemporalTableWithUnion(): Unit = { + util.addTable(""" + |CREATE TABLE LookupTableWithFilterableFields ( + | `id` INT, + | `status` STRING, + | PRIMARY KEY(id) NOT ENFORCED + |) WITH ( + | 'connector' = 'values', + | 'filterable-fields' = 'id;status' + |) + |""".stripMargin) + + val sql = + """ + |SELECT s.a, s.b, s.proctime, d.status + |FROM MyTable AS `s` + |INNER JOIN LookupTableWithFilterableFields FOR SYSTEM_TIME AS OF `s`.proctime AS `d` + |ON `s`.a = `d`.`id` + |WHERE `d`.`status` = 'OK' + |UNION ALL + |SELECT s.a, s.b, s.proctime, d.status + |FROM MyTable AS `s` + |INNER JOIN LookupTableWithFilterableFields FOR SYSTEM_TIME AS OF `s`.proctime AS `d` + |ON `s`.a = `d`.`id` + |WHERE `d`.`status` = 'KO' + |""".stripMargin + + util.verifyExecPlan(sql) + } + @Test def testAggAndAllConstantLookupKeyWithTryResolveMode(): Unit = { // expect lookup join using single parallelism due to all constant lookup key diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnionLookupJoinITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnionLookupJoinITCase.scala new file mode 100644 index 00000000000000..8886b05e7ac0f8 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnionLookupJoinITCase.scala @@ -0,0 +1,96 @@ +/* + * 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. + */ +package org.apache.flink.table.planner.runtime.stream.sql + +import org.apache.flink.table.api.bridge.scala.tableConversions +import org.apache.flink.table.planner.factories.TestValuesTableFactory +import org.apache.flink.table.planner.runtime.utils.{StreamingTestBase, TestingAppendSink} +import org.apache.flink.table.planner.runtime.utils.BatchTestBase.row + +import org.assertj.core.api.Assertions.assertThat +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.CsvSource + +class UnionLookupJoinITCase extends StreamingTestBase { + + @BeforeEach + def beforeEach(): Unit = { + super.before() + val streamTableId = TestValuesTableFactory.registerData(Seq(row(1L, "Alice"), row(2L, "Bob"))) + val streamTableDDL = + s""" + | CREATE TABLE `stream` ( + | `id` BIGINT, + | `name` STRING, + | `txn_time` AS PROCTIME(), + | PRIMARY KEY (`id`) NOT ENFORCED + | ) WITH ( + | 'connector' = 'values', + | 'data-id' = '$streamTableId' + | ) + |""".stripMargin + tEnv.executeSql(streamTableDDL) + + val dimTableId = TestValuesTableFactory.registerData(Seq(row(1L, "OK"), row(2L, "OK"))) + val dimTableDDL = + s""" + | CREATE TABLE `dim` ( + | `id` BIGINT, + | `status` STRING, + | PRIMARY KEY (`id`) NOT ENFORCED + | ) WITH ( + | 'connector' = 'values', + | 'filterable-fields' = 'id;status', + | 'data-id' = '$dimTableId' + | ) + |""".stripMargin + tEnv.executeSql(dimTableDDL) + } + + @ParameterizedTest + @CsvSource(Array[String]("OK, NOT_EXISTS", "NOT_EXISTS, OK")) + def testUnionTemporalJoinWithFilterPushdownSource( + firstFilterValue: String, + secondFilterValue: String): Unit = { + val query = getUnionQuery(firstFilterValue, secondFilterValue); + val expectedData = Seq("1,Alice,OK", "2,Bob,OK") + + val result = tEnv.sqlQuery(query).toDataStream + val sink = new TestingAppendSink() + result.addSink(sink) + + env.execute() + assertThat(sink.getAppendResults.sorted).isEqualTo(expectedData.sorted) + } + + private def getUnionQuery(firstFilterValue: String, secondFilterValue: String): String = { + s""" + | SELECT s.id, s.name, d.status + | FROM `stream` AS `s` INNER JOIN `dim` FOR SYSTEM_TIME AS OF `s`.`txn_time` AS `d` + | ON `s`.`id` = `d`.`id` + | WHERE `d`.`status` = '$firstFilterValue' + | UNION ALL + | SELECT s.id, s.name, d.status + | FROM `stream` AS `s` INNER JOIN `dim` FOR SYSTEM_TIME AS OF `s`.`txn_time` AS `d` + | ON `s`.`id` = `d`.`id` + | WHERE `d`.`status` = '$secondFilterValue' + |""".stripMargin + } + +} From 11a51dc1523064dc07f93567d791c6af8afd7741 Mon Sep 17 00:00:00 2001 From: morazow Date: Sun, 31 Aug 2025 11:00:23 +0200 Subject: [PATCH 2/2] [FLINK-36808][table-planner] Address PR reviews suggestions --- .../common/CommonPhysicalLookupJoin.scala | 21 +++- .../plan/stream/sql/join/LookupJoinTest.xml | 4 +- .../runtime/stream/sql/LookupJoinITCase.scala | 79 ++++++++++++++- .../stream/sql/UnionLookupJoinITCase.scala | 96 ------------------- 4 files changed, 97 insertions(+), 103 deletions(-) delete mode 100644 flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnionLookupJoinITCase.scala diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala index b1a347df29d3b3..c9b5db6c8065f5 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala @@ -21,7 +21,7 @@ import org.apache.flink.table.api.{TableConfig, TableException} import org.apache.flink.table.catalog.{ObjectIdentifier, UniqueConstraint} import org.apache.flink.table.connector.ChangelogMode import org.apache.flink.table.planner.calcite.FlinkTypeFactory -import org.apache.flink.table.planner.plan.abilities.source.FilterPushDownSpec +import org.apache.flink.table.planner.plan.abilities.source.{FilterPushDownSpec, PartitionPushDownSpec} import org.apache.flink.table.planner.plan.nodes.FlinkRelNode import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalRel import org.apache.flink.table.planner.plan.schema.{IntermediateRelTable, LegacyTableSourceTable, TableSourceTable} @@ -202,7 +202,7 @@ abstract class CommonPhysicalLookupJoin( .item("joinType", JoinTypeUtil.getFlinkJoinType(joinType)) .item("lookup", lookupKeys) .itemIf("where", whereString, whereString.nonEmpty) - .itemIf("filter", filterPushdownString, filterPushdownString.nonEmpty) + .itemIf("filterPushDown", filterPushdownString, filterPushdownString.nonEmpty) .itemIf( "joinCondition", joinConditionToString(resultFieldNames, preferExpressionFormat(pw), pw.getDetailLevel), @@ -217,7 +217,9 @@ abstract class CommonPhysicalLookupJoin( private def getTableFilterString(t: TableSourceTable): String = { val filterOpt = t.abilitySpecs.collectFirst { case spec: FilterPushDownSpec => spec } - filterOpt match { + val partitionOpt = t.abilitySpecs.collectFirst { case spec: PartitionPushDownSpec => spec } + + val filterString = filterOpt match { case Some(filter) if !filter.getPredicates.isEmpty => val fieldNames = JavaScalaConversionUtil.toScala(t.getRowType.getFieldNames) val predicates = JavaScalaConversionUtil.toScala(filter.getPredicates) @@ -227,6 +229,19 @@ abstract class CommonPhysicalLookupJoin( .getOrElse("") case _ => "" } + + val partitionString = partitionOpt match { + case Some(partition) if !partition.getPartitions.isEmpty => + s"partitions=${partition.getDigests(null)}" + case _ => "" + } + + (filterString, partitionString) match { + case ("", "") => "" + case (f, "") => f + case ("", p) => p + case (f, p) => s"$f, $p" + } } private def getInputChangelogMode(rel: RelNode): ChangelogMode = rel match { diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml index 79b2cafb940983..20c1df1dfb9abf 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml @@ -1647,11 +1647,11 @@ LogicalUnion(all=[true]) Calc(select=[a, b, PROCTIME_MATERIALIZE(proctime) AS proctime, status]) +- Union(all=[true], union=[a, b, proctime, status]) :- Calc(select=[a, b, proctime, CAST('OK' AS VARCHAR(2147483647)) AS status]) - : +- LookupJoin(table=[default_catalog.default_database.LookupTableWithFilterableFields], joinType=[InnerJoin], lookup=[id=a], filter=[=(status, _UTF-16LE'OK':VARCHAR(2147483647) CHARACTER SET "UTF-16LE")], select=[a, b, proctime, id]) + : +- LookupJoin(table=[default_catalog.default_database.LookupTableWithFilterableFields], joinType=[InnerJoin], lookup=[id=a], filterPushDown=[=(status, _UTF-16LE'OK':VARCHAR(2147483647) CHARACTER SET "UTF-16LE")], select=[a, b, proctime, id]) : +- Calc(select=[a, b, proctime])(reuse_id=[1]) : +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime]) +- Calc(select=[a, b, proctime, CAST('KO' AS VARCHAR(2147483647)) AS status]) - +- LookupJoin(table=[default_catalog.default_database.LookupTableWithFilterableFields], joinType=[InnerJoin], lookup=[id=a], filter=[=(status, _UTF-16LE'KO':VARCHAR(2147483647) CHARACTER SET "UTF-16LE")], select=[a, b, proctime, id]) + +- LookupJoin(table=[default_catalog.default_database.LookupTableWithFilterableFields], joinType=[InnerJoin], lookup=[id=a], filterPushDown=[=(status, _UTF-16LE'KO':VARCHAR(2147483647) CHARACTER SET "UTF-16LE")], select=[a, b, proctime, id]) +- Reused(reference_id=[1]) ]]> diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/LookupJoinITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/LookupJoinITCase.scala index 07f1fb69a33a08..05a4cb74eb6042 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/LookupJoinITCase.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/LookupJoinITCase.scala @@ -39,6 +39,8 @@ import org.assertj.core.api.Assumptions.assumeThat import org.assertj.core.api.IterableAssert.assertThatIterable import org.junit.jupiter.api.{AfterEach, BeforeEach, TestTemplate} import org.junit.jupiter.api.extension.ExtendWith +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.CsvSource import java.time.LocalDateTime import java.util.{Collection => JCollection} @@ -104,6 +106,9 @@ class LookupJoinITCase(cacheType: LookupCacheType) extends StreamingTestBase { // lookup will start from the 3rd time, first lookup will always get null result createLookupTable("user_table_with_lookup_threshold3", userData, 3) createLookupTableWithComputedColumn("userTableWithComputedColumn", userData) + // Union test tables + createUnionScanTable("union_scan_table", List(rowOf(1L, "Alice"), rowOf(2L, "Bob"))) + createUnionLookupTable("union_dim_table", List(rowOf(1L, "OK"), rowOf(2L, "OK"))) } @AfterEach @@ -225,6 +230,76 @@ class LookupJoinITCase(cacheType: LookupCacheType) extends StreamingTestBase { |""".stripMargin) } + private def createUnionScanTable(tableName: String, data: List[Row]): Unit = { + val dataId = TestValuesTableFactory.registerData(data) + tEnv.executeSql(s""" + |CREATE TABLE $tableName ( + | `id` BIGINT, + | `name` STRING, + | `txn_time` AS PROCTIME(), + | PRIMARY KEY (`id`) NOT ENFORCED + |) WITH ( + | 'connector' = 'values', + | 'data-id' = '$dataId' + |) + |""".stripMargin) + } + + private def createUnionLookupTable(tableName: String, data: List[Row]): Unit = { + val dataId = TestValuesTableFactory.registerData(data) + tEnv.executeSql(s""" + |CREATE TABLE $tableName ( + | `id` BIGINT, + | `status` STRING, + | PRIMARY KEY (`id`) NOT ENFORCED + |) WITH ( + | 'connector' = 'values', + | 'filterable-fields' = 'id;status', + | 'data-id' = '$dataId' + |) + |""".stripMargin) + } + + @TestTemplate + def testUnionTemporalJoinWithFilterPushdownSourceOK(): Unit = { + // First filter parameter exists, the second one does not exist + val query = getUnionQuery("OK", "NOT_EXISTS"); + val expectedData = Seq("1,Alice,OK", "2,Bob,OK") + + val sink = new TestingAppendSink + tEnv.sqlQuery(query).toDataStream.addSink(sink) + env.execute() + + assertThat(sink.getAppendResults.sorted).isEqualTo(expectedData.sorted) + } + + @TestTemplate + def testUnionTemporalJoinWithFilterPushdownSourceKO(): Unit = { + // First filter parameter does not exist, the second one exists + val query = getUnionQuery("NOT_EXISTS", "OK"); + val expectedData = Seq("1,Alice,OK", "2,Bob,OK") + + val sink = new TestingAppendSink + tEnv.sqlQuery(query).toDataStream.addSink(sink) + env.execute() + + assertThat(sink.getAppendResults.sorted).isEqualTo(expectedData.sorted) + } + + private def getUnionQuery(firstFilterValue: String, secondFilterValue: String): String = { + s""" + | SELECT s.id, s.name, d.status + | FROM union_scan_table AS `s` INNER JOIN union_dim_table FOR SYSTEM_TIME AS OF `s`.`txn_time` AS `d` + | ON `s`.`id` = `d`.`id` + | WHERE `d`.`status` = '$firstFilterValue' + | UNION ALL + | SELECT s.id, s.name, d.status + | FROM union_scan_table AS `s` INNER JOIN union_dim_table FOR SYSTEM_TIME AS OF `s`.`txn_time` AS `d` + | ON `s`.`id` = `d`.`id` + | WHERE `d`.`status` = '$secondFilterValue' + |""".stripMargin + } + @TestTemplate def testJoinTemporalTable(): Unit = { val sql = "SELECT T.id, T.len, T.content, D.name FROM src AS T JOIN user_table " + @@ -664,8 +739,8 @@ class LookupJoinITCase(cacheType: LookupCacheType) extends StreamingTestBase { tEnv.executeSql(sourceDdl) val sql = """ - |SELECT T.id, D.name, D.age FROM T - |LEFT JOIN user_table FOR SYSTEM_TIME AS OF T.proc AS D + |SELECT T.id, D.name, D.age FROM T + |LEFT JOIN user_table FOR SYSTEM_TIME AS OF T.proc AS D |ON T.id = D.id |""".stripMargin val sink = new TestingAppendSink diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnionLookupJoinITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnionLookupJoinITCase.scala deleted file mode 100644 index 8886b05e7ac0f8..00000000000000 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnionLookupJoinITCase.scala +++ /dev/null @@ -1,96 +0,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. - */ -package org.apache.flink.table.planner.runtime.stream.sql - -import org.apache.flink.table.api.bridge.scala.tableConversions -import org.apache.flink.table.planner.factories.TestValuesTableFactory -import org.apache.flink.table.planner.runtime.utils.{StreamingTestBase, TestingAppendSink} -import org.apache.flink.table.planner.runtime.utils.BatchTestBase.row - -import org.assertj.core.api.Assertions.assertThat -import org.junit.jupiter.api.BeforeEach -import org.junit.jupiter.params.ParameterizedTest -import org.junit.jupiter.params.provider.CsvSource - -class UnionLookupJoinITCase extends StreamingTestBase { - - @BeforeEach - def beforeEach(): Unit = { - super.before() - val streamTableId = TestValuesTableFactory.registerData(Seq(row(1L, "Alice"), row(2L, "Bob"))) - val streamTableDDL = - s""" - | CREATE TABLE `stream` ( - | `id` BIGINT, - | `name` STRING, - | `txn_time` AS PROCTIME(), - | PRIMARY KEY (`id`) NOT ENFORCED - | ) WITH ( - | 'connector' = 'values', - | 'data-id' = '$streamTableId' - | ) - |""".stripMargin - tEnv.executeSql(streamTableDDL) - - val dimTableId = TestValuesTableFactory.registerData(Seq(row(1L, "OK"), row(2L, "OK"))) - val dimTableDDL = - s""" - | CREATE TABLE `dim` ( - | `id` BIGINT, - | `status` STRING, - | PRIMARY KEY (`id`) NOT ENFORCED - | ) WITH ( - | 'connector' = 'values', - | 'filterable-fields' = 'id;status', - | 'data-id' = '$dimTableId' - | ) - |""".stripMargin - tEnv.executeSql(dimTableDDL) - } - - @ParameterizedTest - @CsvSource(Array[String]("OK, NOT_EXISTS", "NOT_EXISTS, OK")) - def testUnionTemporalJoinWithFilterPushdownSource( - firstFilterValue: String, - secondFilterValue: String): Unit = { - val query = getUnionQuery(firstFilterValue, secondFilterValue); - val expectedData = Seq("1,Alice,OK", "2,Bob,OK") - - val result = tEnv.sqlQuery(query).toDataStream - val sink = new TestingAppendSink() - result.addSink(sink) - - env.execute() - assertThat(sink.getAppendResults.sorted).isEqualTo(expectedData.sorted) - } - - private def getUnionQuery(firstFilterValue: String, secondFilterValue: String): String = { - s""" - | SELECT s.id, s.name, d.status - | FROM `stream` AS `s` INNER JOIN `dim` FOR SYSTEM_TIME AS OF `s`.`txn_time` AS `d` - | ON `s`.`id` = `d`.`id` - | WHERE `d`.`status` = '$firstFilterValue' - | UNION ALL - | SELECT s.id, s.name, d.status - | FROM `stream` AS `s` INNER JOIN `dim` FOR SYSTEM_TIME AS OF `s`.`txn_time` AS `d` - | ON `s`.`id` = `d`.`id` - | WHERE `d`.`status` = '$secondFilterValue' - |""".stripMargin - } - -}