Skip to content

[FLINK-36808][table-planner] Fix LookupJoin bug when used with filterable table source #26514

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
morazow wants to merge 2 commits into from
Closed

[FLINK-36808][table-planner] Fix LookupJoin bug when used with filterable table source #26514

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
eca1bf6
[FLINK-36808][table-planner] Fix LookupJoin bug when used with filter…
morazow Apr 19, 2025
11a51dc
[FLINK-36808][table-planner] Address PR reviews suggestions
morazow Aug 31, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 37 additions & 2 deletions .../org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -190,13 +191,18 @@ 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)
.item("table", tableIdentifier.asSummaryString())

@xuyangzhong xuyangzhong Dec 23, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What about reusing this item table with:

val tableDesc: String = temporalTable match {
  case t: TableSourceTable => t.getQualifiedName.asScala.mkString(", ")
  case t: LegacyTableSourceTable[_] => t.tableIdentifier.asSummaryString()
}

super
  .explainTerms(pw)
  .item("table", tableDesc)
  ...

Although the plans for some tests may change, this will not actually affect compatibility, and it can be aligned with the digest in CommonPhysicalTableSourceScan.

.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),
Expand All @@ -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 }

@xuyangzhong xuyangzhong May 9, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm wondering should PartitionPushDownSpec also need to be added into this part.

@morazow morazow Sep 1, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added 👍

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())
Expand Down
49 changes: 49 additions & 0 deletions ...src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1607,4 +1607,53 @@ Calc(select=[a, b, c, PROCTIME_MATERIALIZE(proctime) AS proctime, rowtime, id, n
]]>
</Resource>
</TestCase>
<TestCase name="testJoinFilterableTemporalTableWithUnion">
<Resource name="sql">
<![CDATA[
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'
]]>
</Resource>
<Resource name="ast">
<![CDATA[
LogicalUnion(all=[true])
:- LogicalProject(a=[$0], b=[$1], proctime=[$3], status=[$6])
: +- LogicalFilter(condition=[=($6, _UTF-16LE'OK')])
: +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{0, 3}])
: :- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
: +- LogicalFilter(condition=[=($cor0.a, $0)])
: +- LogicalSnapshot(period=[$cor0.proctime])
: +- LogicalTableScan(table=[[default_catalog, default_database, LookupTableWithFilterableFields]])
+- LogicalProject(a=[$0], b=[$1], proctime=[$3], status=[$6])
+- LogicalFilter(condition=[=($6, _UTF-16LE'KO')])
+- LogicalCorrelate(correlation=[$cor1], joinType=[inner], requiredColumns=[{0, 3}])
:- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+- LogicalFilter(condition=[=($cor1.a, $0)])
+- LogicalSnapshot(period=[$cor1.proctime])
+- LogicalTableScan(table=[[default_catalog, default_database, LookupTableWithFilterableFields]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
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], 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], filterPushDown=[=(status, _UTF-16LE'KO':VARCHAR(2147483647) CHARACTER SET "UTF-16LE")], select=[a, b, proctime, id])
+- Reused(reference_id=[1])
]]>
</Resource>
</TestCase>
</Root>
33 changes: 31 additions & 2 deletions ...r/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -427,7 +427,6 @@ class LookupJoinTest extends TableTestBase with Serializable {

@Test
def testJoinTemporalTableWithFunctionAndConstantCondition(): Unit = {

val sql =
"""
|SELECT * FROM MyTable AS T
Expand All @@ -440,7 +439,6 @@ class LookupJoinTest extends TableTestBase with Serializable {

@Test
def testJoinTemporalTableWithMultiFunctionAndConstantCondition(): Unit = {

val sql =
"""
|SELECT * FROM MyTable AS T
Expand Down Expand Up @@ -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
Expand Down
79 changes: 77 additions & 2 deletions ...r/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/LookupJoinITCase.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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 " +
Expand Down Expand Up @@ -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
Expand Down