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..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,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, 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} -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("filterPushDown", filterPushdownString, filterPushdownString.nonEmpty) .itemIf( "joinCondition", joinConditionToString(resultFieldNames, preferExpressionFormat(pw), pw.getDetailLevel), @@ -209,6 +215,35 @@ abstract class CommonPhysicalLookupJoin( .itemIf("retry", retryOptions.getOrElse(""), retryOptions.isDefined) } + private def getTableFilterString(t: TableSourceTable): String = { + val filterOpt = t.abilitySpecs.collectFirst { case spec: FilterPushDownSpec => spec } + 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) + predicates + .map(FlinkRexUtil.getExpressionString(_, fieldNames)) + .reduceOption((l, r) => String.format("and(%s, %s)", l, r)) + .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 { 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..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 @@ -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/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