This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 67594274366 [FLINK-41282][table-planner] Use each multi-join step's
own join columns for upsert keys
67594274366 is described below
commit 675942743661d688202c246dd64b66cdce2c757b
Author: Anton Borisov <[email protected]>
AuthorDate: Fri Nov 6 12:02:02 2026 +0000
[FLINK-41282][table-planner] Use each multi-join step's own join columns
for upsert keys
* [FLINK-41282][table-planner] Use each multi-join step's own join columns
for upsert keys
* Address review comments
* Update
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala
Co-authored-by: Sergey Nuyanzin <[email protected]>
* Reword getLeftSideJoinKeyIndices description
---------
Co-authored-by: Sergey Nuyanzin <[email protected]>
---
.../physical/stream/StreamPhysicalMultiJoin.java | 4 +
.../plan/metadata/FlinkRelMdUpsertKeys.scala | 4 +-
.../nodes/exec/stream/MultiJoinRestoreTest.java | 3 +-
.../nodes/exec/stream/MultiJoinTestPrograms.java | 73 +++
.../planner/plan/stream/sql/MultiJoinTest.java | 83 +++
.../planner/plan/stream/sql/MultiJoinTest.xml | 87 +++
...ree-way-join-on-fewer-columns-with-restore.json | 608 +++++++++++++++++++++
.../savepoint/_metadata | Bin 0 -> 57818 bytes
.../AttributeBasedJoinKeyExtractor.java | 13 +
.../join/stream/keyselector/JoinKeyExtractor.java | 10 +
10 files changed, 882 insertions(+), 3 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalMultiJoin.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalMultiJoin.java
index 4d7aa3a519f..a5735f368e9 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalMultiJoin.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalMultiJoin.java
@@ -247,6 +247,10 @@ public class StreamPhysicalMultiJoin extends
AbstractRelNode implements StreamPh
return keyExtractor.getJoinKeyIndices(inputId);
}
+ public int[] getLeftSideJoinKeyIndices(int inputId) {
+ return keyExtractor.getLeftSideJoinKeyIndices(inputId);
+ }
+
private @Nullable Set<ImmutableBitSet> getUniqueKeys(RelNode input) {
final FlinkRelMetadataQuery fmq =
FlinkRelMetadataQuery.reuseOrCreate(input.getCluster().getMetadataQuery());
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala
index ea94be9f61d..2532f0c256f 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala
@@ -322,13 +322,13 @@ class FlinkRelMdUpsertKeys private extends
MetadataHandler[UpsertKeys] {
var leftFieldCount = inputs.get(0).getRowType.getFieldCount
var leftUpsertKeys = fmq.getUpsertKeys(inputs.get(0))
- val leftJoinKeyIndices =
ImmutableBitSet.of(multiJoin.getJoinKeyIndices(0): _*)
-
// Process each subsequent input as right side of join
for (i <- 1 until inputs.size) {
val rightInput = inputs.get(i)
val rightUpsertKeys = fmq.getUpsertKeys(rightInput)
val joinType = joinTypes.get(i)
+ // Only the left columns that input i's own condition references, not
every previous join column.
+ val leftJoinKeyIndices =
ImmutableBitSet.of(multiJoin.getLeftSideJoinKeyIndices(i): _*)
val rightJoinKeyIndices =
ImmutableBitSet.of(multiJoin.getJoinKeyIndices(i): _*)
// We compute candidate unique keys that qualify as upsert keys if they
include
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinRestoreTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinRestoreTest.java
index fb81aab8c4a..e9f47684f88 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinRestoreTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinRestoreTest.java
@@ -43,6 +43,7 @@ public class MultiJoinRestoreTest extends RestoreTestBase {
MultiJoinTestPrograms.MULTI_JOIN_FOUR_WAY_COMPLEX_PRESERVES_UPSERT_KEY_WITH_RESTORE,
MultiJoinTestPrograms
.MULTI_JOIN_TWO_WAY_LEFT_JOIN_PRESERVES_UPSERT_KEY_WITH_RESTORE,
-
MultiJoinTestPrograms.MULTI_JOIN_THREE_WAY_JOIN_PRESERVES_UPSERT_KEY_WITH_RESTORE);
+
MultiJoinTestPrograms.MULTI_JOIN_THREE_WAY_JOIN_PRESERVES_UPSERT_KEY_WITH_RESTORE,
+
MultiJoinTestPrograms.MULTI_JOIN_THREE_WAY_JOIN_ON_FEWER_COLUMNS_WITH_RESTORE);
}
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinTestPrograms.java
index f5abff08279..b084c0c8524 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinTestPrograms.java
@@ -1391,6 +1391,79 @@ public class MultiJoinTestPrograms {
+ " ON o.user_id = p.user_id")
.build();
+ static final TableTestProgram
MULTI_JOIN_THREE_WAY_JOIN_ON_FEWER_COLUMNS_WITH_RESTORE =
+ TableTestProgram.of(
+ "three-way-join-on-fewer-columns-with-restore",
+ "three way join on fewer columns with restore")
+
.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(
+ SourceTestStep.newBuilder("UserRegions")
+ .addMode(ChangelogMode.upsert())
+ .addSchema(
+ "user_id STRING NOT NULL",
+ "region STRING NOT NULL",
+ "name STRING",
+ "CONSTRAINT `PRIMARY` PRIMARY KEY
(`user_id`, `region`) NOT ENFORCED")
+ .producedBeforeRestore(
+ Row.ofKind(RowKind.INSERT, "u1",
"US", "Ann"))
+ .producedAfterRestore(
+ Row.ofKind(RowKind.INSERT, "u1",
"EU", "Ann"),
+ Row.ofKind(RowKind.DELETE, "u1",
"EU", "Ann"))
+ .build())
+ .setupTableSource(
+ SourceTestStep.newBuilder("Orders")
+ .addMode(ChangelogMode.upsert())
+ .addSchema(
+ "order_id STRING NOT NULL",
+ "user_id STRING NOT NULL",
+ "product STRING",
+ "CONSTRAINT `PRIMARY` PRIMARY KEY
(`order_id`, `user_id`) NOT ENFORCED")
+ .producedBeforeRestore(
+ Row.ofKind(RowKind.INSERT, "o1",
"u1", "book"))
+ .producedAfterRestore(
+ Row.ofKind(RowKind.INSERT, "o2",
"u1", "pen"))
+ .build())
+ .setupTableSource(
+ SourceTestStep.newBuilder("RegionAddresses")
+ .addMode(ChangelogMode.upsert())
+ .addSchema(
+ "user_id STRING NOT NULL",
+ "region STRING NOT NULL",
+ "location STRING",
+ "CONSTRAINT `PRIMARY` PRIMARY KEY
(`user_id`, `region`) NOT ENFORCED")
+ .producedBeforeRestore(
+ Row.ofKind(RowKind.INSERT, "u1",
"EU", "Berlin"),
+ Row.ofKind(RowKind.INSERT, "u1",
"US", "NYC"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addMode(ChangelogMode.upsert())
+ .addSchema(
+ "`order_id` STRING NOT NULL",
+ "`user_id` STRING NOT NULL",
+ "name STRING",
+ "location STRING",
+ "CONSTRAINT `PRIMARY` PRIMARY KEY
(`order_id`, `user_id`) NOT ENFORCED")
+ // deleting the EU region must keep the
orders' rows for the US
+ .consumedBeforeRestore("+I[o1, u1, Ann,
NYC]")
+ .consumedAfterRestore("+I[o2, u1, Ann,
NYC]")
+ .testMaterializedData()
+ .build())
+ .runSql(
+ "INSERT INTO `sink`\n"
+ + "SELECT\n"
+ + " o.order_id,\n"
+ + " o.user_id,\n"
+ + " r.name,\n"
+ + " a.location\n"
+ + "FROM UserRegions r\n"
+ + "JOIN Orders o\n"
+ + " ON r.user_id = o.user_id\n"
+ + "JOIN RegionAddresses a\n"
+ + " ON r.user_id = a.user_id AND
r.region = a.region\n"
+ + "ON CONFLICT DO DEDUPLICATE")
+ .build();
+
public static final TableTestProgram
MULTI_JOIN_WITH_TIME_ATTRIBUTES_IN_CONDITIONS_MATERIALIZATION =
TableTestProgram.of(
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/MultiJoinTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/MultiJoinTest.java
index fc387fddeb0..6a77fc3fcaf 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/MultiJoinTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/MultiJoinTest.java
@@ -176,6 +176,24 @@ public class MultiJoinTest extends TableTestBase {
+ " user_id STRING PRIMARY KEY NOT ENFORCED,"
+ " location STRING"
+ ") WITH ('connector' = 'values',
'changelog-mode' = 'I,UA,D')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE UserRegionsPK ("
+ + " user_id STRING NOT NULL,"
+ + " region STRING NOT NULL,"
+ + " name STRING,"
+ + " CONSTRAINT `PRIMARY` PRIMARY KEY
(user_id, region) NOT ENFORCED"
+ + ") WITH ('connector' = 'values',
'changelog-mode' = 'I,UA,D')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE RegionAddressPK ("
+ + " user_id STRING NOT NULL,"
+ + " region STRING NOT NULL,"
+ + " location STRING,"
+ + " CONSTRAINT `PRIMARY` PRIMARY KEY
(user_id, region) NOT ENFORCED"
+ + ") WITH ('connector' = 'values',
'changelog-mode' = 'I,UA,D')");
}
@Test
@@ -865,6 +883,71 @@ public class MultiJoinTest extends TableTestBase {
+ " ON o.user_id = p.user_id");
}
+ @Test
+ void testPreservesUpsertKeyThreeWayJoinOnFewerLeftColumnsDoesNot() {
+ // UserRegionsPK is unique on its join columns (user_id, region), but
only user_id is joined
+ // to OrdersPK: an order matches the user's row of every region, so
the orders key is not
+ // unique in the output and the sink needs upsertMaterialize.
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE sink_orders ("
+ + " `order_id` STRING NOT NULL,"
+ + " `user_id` STRING NOT NULL,"
+ + " name STRING,"
+ + " location STRING,"
+ + " CONSTRAINT `PRIMARY` PRIMARY KEY
(`order_id`, `user_id`) NOT ENFORCED"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'sink-insert-only' = 'false'"
+ + ")");
+
+ util.verifyRelPlanInsert(
+ "\nINSERT INTO sink_orders\n"
+ + "SELECT\n"
+ + " o.order_id,\n"
+ + " o.user_id,\n"
+ + " r.name,\n"
+ + " a.location\n"
+ + "FROM UserRegionsPK r\n"
+ + "JOIN OrdersPK o\n"
+ + " ON r.user_id = o.user_id\n"
+ + "JOIN RegionAddressPK a\n"
+ + " ON r.user_id = a.user_id AND r.region =
a.region\n"
+ + "ON CONFLICT DO DEDUPLICATE");
+ }
+
+ @Test
+ void testPreservesUpsertKeyThreeWayJoinOnSecondInputColumns() {
+ // RegionAddressPK is joined on the key of UserRegionsPK, the second
input, which stays
+ // unique after joining UsersPK on user_id: an address matches at most
one row, so the
+ // address key is unique in the output and the sink needs no
upsertMaterialize.
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE sink_addresses ("
+ + " `user_id` STRING NOT NULL,"
+ + " `region` STRING NOT NULL,"
+ + " name STRING,"
+ + " location STRING,"
+ + " CONSTRAINT `PRIMARY` PRIMARY KEY
(`user_id`, `region`) NOT ENFORCED"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'sink-insert-only' = 'false'"
+ + ")");
+
+ util.verifyRelPlanInsert(
+ "\nINSERT INTO sink_addresses\n"
+ + "SELECT\n"
+ + " a.user_id,\n"
+ + " a.region,\n"
+ + " u.name,\n"
+ + " a.location\n"
+ + "FROM UsersPK u\n"
+ + "JOIN UserRegionsPK r\n"
+ + " ON u.user_id = r.user_id\n"
+ + "JOIN RegionAddressPK a\n"
+ + " ON r.user_id = a.user_id AND r.region =
a.region");
+ }
+
@Test
void testPreservesUpsertKeyFourWayComplex() {
util.tableEnv()
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/MultiJoinTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/MultiJoinTest.xml
index 7a0fd28351b..7f34caf615f 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/MultiJoinTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/MultiJoinTest.xml
@@ -1684,6 +1684,93 @@
Sink(table=[default_catalog.default_database.sink_three_way], fields=[user_id, o
: +- TableSourceScan(table=[[default_catalog, default_database,
OrdersPK, project=[order_id, user_id]]], fields=[order_id, user_id])
+- Exchange(distribution=[hash[user_id]])
+- TableSourceScan(table=[[default_catalog, default_database,
PaymentsPK, project=[payment_id, user_id]]], fields=[payment_id, user_id])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testPreservesUpsertKeyThreeWayJoinOnFewerLeftColumnsDoesNot">
+ <Resource name="sql">
+ <![CDATA[
+INSERT INTO sink_orders
+SELECT
+ o.order_id,
+ o.user_id,
+ r.name,
+ a.location
+FROM UserRegionsPK r
+JOIN OrdersPK o
+ ON r.user_id = o.user_id
+JOIN RegionAddressPK a
+ ON r.user_id = a.user_id AND r.region = a.region
+ON CONFLICT DO DEDUPLICATE]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalSink(table=[default_catalog.default_database.sink_orders],
fields=[order_id, user_id, name, location])
++- LogicalProject(order_id=[$3], user_id=[$4], name=[$2], location=[$8])
+ +- LogicalJoin(condition=[AND(=($0, $6), =($1, $7))], joinType=[inner])
+ :- LogicalJoin(condition=[=($0, $4)], joinType=[inner])
+ : :- LogicalTableScan(table=[[default_catalog, default_database,
UserRegionsPK]])
+ : +- LogicalTableScan(table=[[default_catalog, default_database,
OrdersPK]])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
RegionAddressPK]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Sink(table=[default_catalog.default_database.sink_orders], fields=[order_id,
user_id, name, location], upsertMaterialize=[true],
conflictStrategy=[DEDUPLICATE])
++- Calc(select=[order_id, user_id0 AS user_id, name, location])
+ +- MultiJoin(commonJoinKey=[user_id], joinTypes=[INNER, INNER],
inputUniqueKeys=[(user_id, region), (order_id, user_id), (user_id, region)],
joinConditions=[=(user_id, user_id0), AND(=(user_id, user_id1), =(region,
region0))],
select=[user_id,region,name,order_id,user_id0,user_id1,region0,location],
rowType=[RecordType(VARCHAR(2147483647) user_id, VARCHAR(2147483647) region,
VARCHAR(2147483647) name, VARCHAR(2147483647) order_id, VARCHAR(2147483647)
user_id0, VARCHAR(2147483647) user_ [...]
+ :- Exchange(distribution=[hash[user_id]])
+ : +- ChangelogNormalize(key=[user_id, region])
+ : +- Exchange(distribution=[hash[user_id, region]])
+ : +- TableSourceScan(table=[[default_catalog, default_database,
UserRegionsPK]], fields=[user_id, region, name])
+ :- Exchange(distribution=[hash[user_id]])
+ : +- ChangelogNormalize(key=[order_id, user_id])
+ : +- Exchange(distribution=[hash[order_id, user_id]])
+ : +- TableSourceScan(table=[[default_catalog, default_database,
OrdersPK, project=[order_id, user_id]]], fields=[order_id, user_id])
+ +- Exchange(distribution=[hash[user_id]])
+ +- ChangelogNormalize(key=[user_id, region])
+ +- Exchange(distribution=[hash[user_id, region]])
+ +- TableSourceScan(table=[[default_catalog, default_database,
RegionAddressPK]], fields=[user_id, region, location])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testPreservesUpsertKeyThreeWayJoinOnSecondInputColumns">
+ <Resource name="sql">
+ <![CDATA[
+INSERT INTO sink_addresses
+SELECT
+ a.user_id,
+ a.region,
+ u.name,
+ a.location
+FROM UsersPK u
+JOIN UserRegionsPK r
+ ON u.user_id = r.user_id
+JOIN RegionAddressPK a
+ ON r.user_id = a.user_id AND r.region = a.region]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalSink(table=[default_catalog.default_database.sink_addresses],
fields=[user_id, region, name, location])
++- LogicalProject(user_id=[$7], region=[$8], name=[$1], location=[$9])
+ +- LogicalJoin(condition=[AND(=($4, $7), =($5, $8))], joinType=[inner])
+ :- LogicalJoin(condition=[=($0, $4)], joinType=[inner])
+ : :- LogicalTableScan(table=[[default_catalog, default_database,
UsersPK]])
+ : +- LogicalTableScan(table=[[default_catalog, default_database,
UserRegionsPK]])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
RegionAddressPK]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Sink(table=[default_catalog.default_database.sink_addresses], fields=[user_id,
region, name, location])
++- Calc(select=[user_id1 AS user_id, region0 AS region, name, location])
+ +- MultiJoin(commonJoinKey=[user_id], joinTypes=[INNER, INNER],
inputUniqueKeys=[(user_id), (user_id, region), (user_id, region)],
joinConditions=[=(user_id, user_id0), AND(=(user_id0, user_id1), =(region,
region0))], select=[user_id,name,user_id0,region,user_id1,region0,location],
rowType=[RecordType(VARCHAR(2147483647) user_id, VARCHAR(2147483647) name,
VARCHAR(2147483647) user_id0, VARCHAR(2147483647) region, VARCHAR(2147483647)
user_id1, VARCHAR(2147483647) region0, VARCHAR(214748 [...]
+ :- Exchange(distribution=[hash[user_id]])
+ : +- TableSourceScan(table=[[default_catalog, default_database,
UsersPK, project=[user_id, name]]], fields=[user_id, name])
+ :- Exchange(distribution=[hash[user_id]])
+ : +- TableSourceScan(table=[[default_catalog, default_database,
UserRegionsPK, project=[user_id, region]]], fields=[user_id, region])
+ +- Exchange(distribution=[hash[user_id]])
+ +- TableSourceScan(table=[[default_catalog, default_database,
RegionAddressPK]], fields=[user_id, region, location])
]]>
</Resource>
</TestCase>
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/three-way-join-on-fewer-columns-with-restore/plan/three-way-join-on-fewer-columns-with-restore.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/three-way-join-on-fewer-columns-with-restore/plan/three-way-join-on-fewer-columns-with-restore.json
new file mode 100644
index 00000000000..742c17d1feb
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/three-way-join-on-fewer-columns-with-restore/plan/three-way-join-on-fewer-columns-with-restore.json
@@ -0,0 +1,608 @@
+{
+ "flinkVersion" : "2.5",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "stream-exec-table-source-scan_2",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`UserRegions`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "user_id",
+ "dataType" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "name" : "region",
+ "dataType" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "primaryKey" : {
+ "name" : "PRIMARY",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "user_id", "region" ]
+ }
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`user_id` VARCHAR(2147483647) NOT NULL, `region`
VARCHAR(2147483647) NOT NULL, `name` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, UserRegions]], fields=[user_id, region, name])"
+ }, {
+ "id" : 2,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0, 1 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`user_id` VARCHAR(2147483647) NOT NULL, `region`
VARCHAR(2147483647) NOT NULL, `name` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[user_id, region]])"
+ }, {
+ "id" : 3,
+ "type" : "stream-exec-changelog-normalize_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "false",
+ "table.exec.mini-batch.size" : "-1"
+ },
+ "uniqueKeys" : [ 0, 1 ],
+ "generateUpdateBefore" : true,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "changelogNormalizeState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`user_id` VARCHAR(2147483647) NOT NULL, `region`
VARCHAR(2147483647) NOT NULL, `name` VARCHAR(2147483647)>",
+ "description" : "ChangelogNormalize(key=[user_id, region])"
+ }, {
+ "id" : 4,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`user_id` VARCHAR(2147483647) NOT NULL, `region`
VARCHAR(2147483647) NOT NULL, `name` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[user_id]])"
+ }, {
+ "id" : 5,
+ "type" : "stream-exec-table-source-scan_2",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`Orders`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "order_id",
+ "dataType" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "name" : "user_id",
+ "dataType" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "name" : "product",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "primaryKey" : {
+ "name" : "PRIMARY",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "order_id", "user_id" ]
+ }
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 1 ] ],
+ "producedType" : "ROW<`order_id` VARCHAR(2147483647) NOT NULL,
`user_id` VARCHAR(2147483647) NOT NULL> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`order_id` VARCHAR(2147483647) NOT NULL, `user_id`
VARCHAR(2147483647) NOT NULL>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, Orders, project=[order_id, user_id]]], fields=[order_id,
user_id])"
+ }, {
+ "id" : 6,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0, 1 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` VARCHAR(2147483647) NOT NULL, `user_id`
VARCHAR(2147483647) NOT NULL>",
+ "description" : "Exchange(distribution=[hash[order_id, user_id]])"
+ }, {
+ "id" : 7,
+ "type" : "stream-exec-changelog-normalize_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "false",
+ "table.exec.mini-batch.size" : "-1"
+ },
+ "uniqueKeys" : [ 0, 1 ],
+ "generateUpdateBefore" : true,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "changelogNormalizeState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` VARCHAR(2147483647) NOT NULL, `user_id`
VARCHAR(2147483647) NOT NULL>",
+ "description" : "ChangelogNormalize(key=[order_id, user_id])"
+ }, {
+ "id" : 8,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 1 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` VARCHAR(2147483647) NOT NULL, `user_id`
VARCHAR(2147483647) NOT NULL>",
+ "description" : "Exchange(distribution=[hash[user_id]])"
+ }, {
+ "id" : 9,
+ "type" : "stream-exec-table-source-scan_2",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" :
"`default_catalog`.`default_database`.`RegionAddresses`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "user_id",
+ "dataType" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "name" : "region",
+ "dataType" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "name" : "location",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "primaryKey" : {
+ "name" : "PRIMARY",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "user_id", "region" ]
+ }
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`user_id` VARCHAR(2147483647) NOT NULL, `region`
VARCHAR(2147483647) NOT NULL, `location` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, RegionAddresses]], fields=[user_id, region, location])"
+ }, {
+ "id" : 10,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0, 1 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`user_id` VARCHAR(2147483647) NOT NULL, `region`
VARCHAR(2147483647) NOT NULL, `location` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[user_id, region]])"
+ }, {
+ "id" : 11,
+ "type" : "stream-exec-changelog-normalize_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "false",
+ "table.exec.mini-batch.size" : "-1"
+ },
+ "uniqueKeys" : [ 0, 1 ],
+ "generateUpdateBefore" : true,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "changelogNormalizeState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`user_id` VARCHAR(2147483647) NOT NULL, `region`
VARCHAR(2147483647) NOT NULL, `location` VARCHAR(2147483647)>",
+ "description" : "ChangelogNormalize(key=[user_id, region])"
+ }, {
+ "id" : 12,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`user_id` VARCHAR(2147483647) NOT NULL, `region`
VARCHAR(2147483647) NOT NULL, `location` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[user_id]])"
+ }, {
+ "id" : 13,
+ "type" : "stream-exec-multi-join_1",
+ "joinTypes" : [ "INNER", "INNER", "INNER" ],
+ "joinConditions" : [ {
+ "kind" : "LITERAL",
+ "value" : true,
+ "type" : "BOOLEAN NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$AND$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 5,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 6,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ } ],
+ "multiJoinCondition" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$AND$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 5,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 6,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ },
+ "joinAttributeMap" : {
+ "0" : [ {
+ "leftInputId" : -1,
+ "leftFieldIndex" : -1,
+ "rightInputId" : 0,
+ "rightFieldIndex" : 0
+ }, {
+ "leftInputId" : -1,
+ "leftFieldIndex" : -1,
+ "rightInputId" : 0,
+ "rightFieldIndex" : 0
+ }, {
+ "leftInputId" : -1,
+ "leftFieldIndex" : -1,
+ "rightInputId" : 0,
+ "rightFieldIndex" : 1
+ } ],
+ "1" : [ {
+ "leftInputId" : 0,
+ "leftFieldIndex" : 0,
+ "rightInputId" : 1,
+ "rightFieldIndex" : 1
+ } ],
+ "2" : [ {
+ "leftInputId" : 0,
+ "leftFieldIndex" : 0,
+ "rightInputId" : 2,
+ "rightFieldIndex" : 0
+ }, {
+ "leftInputId" : 0,
+ "leftFieldIndex" : 1,
+ "rightInputId" : 2,
+ "rightFieldIndex" : 1
+ } ]
+ },
+ "inputUniqueKeys" : [ [ [ 0, 1 ] ], [ [ 0, 1 ] ], [ [ 0, 1 ] ] ],
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "input-state-0"
+ }, {
+ "index" : 1,
+ "ttl" : "0 ms",
+ "name" : "input-state-1"
+ }, {
+ "index" : 2,
+ "ttl" : "0 ms",
+ "name" : "input-state-2"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ }, {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 1 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 1
+ }, {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 2
+ } ],
+ "outputType" : "ROW<`user_id` VARCHAR(2147483647) NOT NULL, `region`
VARCHAR(2147483647) NOT NULL, `name` VARCHAR(2147483647), `order_id`
VARCHAR(2147483647) NOT NULL, `user_id0` VARCHAR(2147483647) NOT NULL,
`user_id1` VARCHAR(2147483647) NOT NULL, `region0` VARCHAR(2147483647) NOT
NULL, `location` VARCHAR(2147483647)>",
+ "description" : "MultiJoin(commonJoinKey=[user_id], joinTypes=[INNER,
INNER], inputUniqueKeys=[(user_id, region), (order_id, user_id), (user_id,
region)], joinConditions=[(user_id = user_id0), ((user_id = user_id1) AND
(region = region0))],
select=[user_id,region,name,order_id,user_id0,user_id1,region0,location],
rowType=[RecordType(VARCHAR(2147483647) user_id, VARCHAR(2147483647) region,
VARCHAR(2147483647) name, VARCHAR(2147483647) order_id, VARCHAR(2147483647)
user_id0, VARCHAR(21 [...]
+ }, {
+ "id" : 14,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 7,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` VARCHAR(2147483647) NOT NULL, `user_id`
VARCHAR(2147483647) NOT NULL, `name` VARCHAR(2147483647), `location`
VARCHAR(2147483647)>",
+ "description" : "Calc(select=[order_id, user_id0 AS user_id, name,
location])"
+ }, {
+ "id" : 15,
+ "type" : "stream-exec-sink_2",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "order_id",
+ "dataType" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "name" : "user_id",
+ "dataType" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "location",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "primaryKey" : {
+ "name" : "PRIMARY",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "order_id", "user_id" ]
+ }
+ }
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_BEFORE", "UPDATE_AFTER",
"DELETE" ],
+ "requireUpsertMaterialize" : true,
+ "upsertMaterializeStrategy" : "ADAPTIVE",
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "sinkMaterializeState"
+ } ],
+ "conflictStrategy" : {
+ "behavior" : "DEDUPLICATE"
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` VARCHAR(2147483647) NOT NULL, `user_id`
VARCHAR(2147483647) NOT NULL, `name` VARCHAR(2147483647), `location`
VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.sink],
fields=[order_id, user_id, name, location], upsertMaterialize=[true],
conflictStrategy=[DEDUPLICATE])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 3,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 3,
+ "target" : 4,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 5,
+ "target" : 6,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 6,
+ "target" : 7,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 7,
+ "target" : 8,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 9,
+ "target" : 10,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 10,
+ "target" : 11,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 11,
+ "target" : 12,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 4,
+ "target" : 13,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 8,
+ "target" : 13,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 12,
+ "target" : 13,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 13,
+ "target" : 14,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 14,
+ "target" : 15,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/three-way-join-on-fewer-columns-with-restore/savepoint/_metadata
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/three-way-join-on-fewer-columns-with-restore/savepoint/_metadata
new file mode 100644
index 00000000000..6e708ab01a3
Binary files /dev/null and
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/three-way-join-on-fewer-columns-with-restore/savepoint/_metadata
differ
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/keyselector/AttributeBasedJoinKeyExtractor.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/keyselector/AttributeBasedJoinKeyExtractor.java
index 6de9e570adc..aeede3d7546 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/keyselector/AttributeBasedJoinKeyExtractor.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/keyselector/AttributeBasedJoinKeyExtractor.java
@@ -170,6 +170,15 @@ public class AttributeBasedJoinKeyExtractor implements
JoinKeyExtractor, Seriali
return
keyFieldIndices.stream().mapToInt(KeyExtractor::getFieldIndexInSourceRow).toArray();
}
+ @Override
+ public int[] getLeftSideJoinKeyIndices(final int inputId) {
+ final List<KeyExtractor> keyExtractors =
this.leftKeyExtractorsMap.get(inputId);
+ if (keyExtractors == null) {
+ return new int[0];
+ }
+ return
keyExtractors.stream().mapToInt(KeyExtractor::getAbsoluteFieldIndex).toArray();
+ }
+
@Override
public RowType getCommonJoinKeyType() {
// We return an empty RowType if no common join key is defined.
@@ -913,6 +922,10 @@ public class AttributeBasedJoinKeyExtractor implements
JoinKeyExtractor, Seriali
return fieldIndexInSourceRow;
}
+ public int getAbsoluteFieldIndex() {
+ return absoluteFieldIndex;
+ }
+
private void readObject(final java.io.ObjectInputStream in)
throws java.io.IOException, ClassNotFoundException {
in.defaultReadObject();
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/keyselector/JoinKeyExtractor.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/keyselector/JoinKeyExtractor.java
index 4fb96d28ed1..28b44081225 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/keyselector/JoinKeyExtractor.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/keyselector/JoinKeyExtractor.java
@@ -83,6 +83,16 @@ public interface JoinKeyExtractor extends Serializable {
*/
int[] getJoinKeyIndices(int inputId);
+ /**
+ * Gets the field indices in the joined row of the previous inputs that
make up the left-side
+ * join key for a given input.
+ *
+ * @param inputId The ID of the input stream.
+ * @return An array of integers representing the field indices in the
joined row that form the
+ * left-side join key.
+ */
+ int[] getLeftSideJoinKeyIndices(int inputId);
+
/**
* Extracts the common key from an input row. The common key consists of
attributes that are
* part of all equi-join conditions in the multi-join sequence.
