You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2025/08/11 09:08:30 UTC
(flink) branch master updated: [FLINK-38219][table-runtime] Fix row kind in StreamingMultiJoinOperator
This is an automated email from the ASF dual-hosted git repository.
twalthr 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 4a98ded5524 [FLINK-38219][table-runtime] Fix row kind in StreamingMultiJoinOperator
4a98ded5524 is described below
commit 4a98ded552473202085947a6f7d2f4cd73e17bfe
Author: Gustavo de Morais <gd...@confluent.io>
AuthorDate: Mon Aug 11 10:08:23 2025 +0100
[FLINK-38219][table-runtime] Fix row kind in StreamingMultiJoinOperator
This closes #26886.
---
.../nodes/exec/stream/MultiJoinSemanticTests.java | 3 +-
.../nodes/exec/stream/MultiJoinTestPrograms.java | 64 ++++++++++++++++++++++
.../join/stream/StreamingMultiJoinOperator.java | 9 ++-
.../join/stream/state/MultiJoinStateViews.java | 21 ++-----
4 files changed, 78 insertions(+), 19 deletions(-)
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinSemanticTests.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinSemanticTests.java
index 137e37fb56d..4b026c83223 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinSemanticTests.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinSemanticTests.java
@@ -37,6 +37,7 @@ public class MultiJoinSemanticTests extends SemanticTestBase {
MultiJoinTestPrograms.MULTI_JOIN_WITH_TIME_ATTRIBUTES_MATERIALIZATION,
MultiJoinTestPrograms.MULTI_JOIN_THREE_WAY_INNER_JOIN_NO_JOIN_KEY,
MultiJoinTestPrograms.MULTI_JOIN_FOUR_WAY_NO_COMMON_JOIN_KEY,
- MultiJoinTestPrograms.MULTI_JOIN_MIXED_CHANGELOG_MODES);
+ MultiJoinTestPrograms.MULTI_JOIN_MIXED_CHANGELOG_MODES,
+ MultiJoinTestPrograms.MULTI_JOIN_THREE_WAY_LEFT_OUTER_JOIN_WITH_CTE);
}
}
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 461cc2b2c8c..32c32728c34 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
@@ -967,4 +967,68 @@ public class MultiJoinTestPrograms {
+ "LEFT JOIN RetractTable r ON a.id = r.ref_id "
+ "LEFT JOIN UpsertTable u ON a.id = u.key_id")
.build();
+
+ public static final TableTestProgram MULTI_JOIN_THREE_WAY_LEFT_OUTER_JOIN_WITH_CTE =
+ TableTestProgram.of(
+ "left-outer-join-with-cte",
+ "CTE with three-way left outer join and aggregation")
+ .setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(USERS_SOURCE)
+ .setupTableSource(
+ SourceTestStep.newBuilder("Orders")
+ .addSchema(
+ "user_id STRING",
+ "order_id STRING PRIMARY KEY NOT ENFORCED",
+ "product STRING")
+ .addOption("changelog-mode", "I, UA,D")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, "2", "order2", "Product B"),
+ Row.ofKind(
+ RowKind.UPDATE_AFTER,
+ "2",
+ "order2",
+ "Product B"),
+ Row.ofKind(
+ RowKind.UPDATE_AFTER,
+ "2",
+ "order2",
+ "Product C"),
+ Row.ofKind(
+ RowKind.UPDATE_AFTER,
+ "2",
+ "order2",
+ "Product C"))
+ .build())
+ .setupTableSource(PAYMENTS_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("user_id STRING", "name STRING", "cnt BIGINT")
+ .testMaterializedData()
+ .consumedValues(Row.of("1", "Gus", 2), Row.of("2", "Bob", 1))
+ .build())
+ .runSql(
+ "INSERT INTO sink WITH "
+ + " order_details AS ( "
+ + " SELECT o.user_id "
+ + " FROM Orders o "
+ + " ), "
+ + " user_elements AS ( "
+ + " SELECT "
+ + " u.id AS user_id "
+ + " FROM ( "
+ + " SELECT '2' AS id, '2' AS order_user_id "
+ + " UNION ALL "
+ + " SELECT '1' AS id, '2' AS order_user_id "
+ + " UNION ALL "
+ + " SELECT '5' AS id, '5' AS order_user_id "
+ + " ) u "
+ + " LEFT JOIN order_details od "
+ + " ON od.user_id = u.order_user_id "
+ + " ) "
+ + "SELECT ue.user_id, us.name, COUNT(*) AS cnt "
+ + "FROM user_elements ue "
+ + "INNER JOIN Users us ON ue.user_id = us.user_id "
+ + "LEFT JOIN Payments p ON ue.user_id = p.user_id "
+ + "GROUP BY ue.user_id, us.name")
+ .build();
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMultiJoinOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMultiJoinOperator.java
index 062f6852c7d..3764b886d00 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMultiJoinOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMultiJoinOperator.java
@@ -777,12 +777,15 @@ public class StreamingMultiJoinOperator extends AbstractStreamOperatorV2<RowData
}
private void addRecordToState(RowData input, int inputId) throws Exception {
+ final boolean isUpsert = isUpsert(input);
RowData joinKey = keyExtractor.getJoinKey(input, inputId);
- if (isRetraction(input)) {
- stateHandlers.get(inputId).retractRecord(joinKey, input);
- } else {
+ // Always use insert so we store and retract records correctly from state
+ input.setRowKind(RowKind.INSERT);
+ if (isUpsert) {
stateHandlers.get(inputId).addRecord(joinKey, input);
+ } else {
+ stateHandlers.get(inputId).retractRecord(joinKey, input);
}
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/MultiJoinStateViews.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/MultiJoinStateViews.java
index a0a59922c51..6d3caba2ba3 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/MultiJoinStateViews.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/MultiJoinStateViews.java
@@ -32,7 +32,6 @@ import org.apache.flink.table.runtime.operators.join.stream.utils.JoinInputSideS
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
import org.apache.flink.table.types.logical.RowType;
-import org.apache.flink.types.RowKind;
import org.apache.flink.util.IterableIterator;
import javax.annotation.Nonnull;
@@ -238,15 +237,15 @@ public final class MultiJoinStateViews {
@Override
public void addRecord(@Nullable RowData joinKey, RowData record) throws Exception {
- RowData uniqueKey = uniqueKeySelector.getKey(record);
- RowData stateKey = getStateKey(joinKey, uniqueKey);
+ final RowData uniqueKey = uniqueKeySelector.getKey(record);
+ final RowData stateKey = getStateKey(joinKey, uniqueKey);
recordState.put(stateKey, record);
}
@Override
public void retractRecord(@Nullable RowData joinKey, RowData record) throws Exception {
- RowData uniqueKey = uniqueKeySelector.getKey(record);
- RowData stateKey = getStateKey(joinKey, uniqueKey);
+ final RowData uniqueKey = uniqueKeySelector.getKey(record);
+ final RowData stateKey = getStateKey(joinKey, uniqueKey);
recordState.remove(stateKey);
}
@@ -364,25 +363,18 @@ public final class MultiJoinStateViews {
@Override
public void addRecord(@Nullable RowData joinKey, RowData record) throws Exception {
- // Normalize RowKind for consistent state representation
- RowKind originalKind = record.getRowKind();
- record.setRowKind(RowKind.INSERT); // Normalize for key creation
- RowData stateKey = getStateKey(joinKey, record);
+ final RowData stateKey = getStateKey(joinKey, record);
Integer currentCount = recordState.get(stateKey);
if (currentCount == null) {
currentCount = 0;
}
recordState.put(stateKey, currentCount + 1);
- record.setRowKind(originalKind); // Restore original RowKind
}
@Override
public void retractRecord(@Nullable RowData joinKey, RowData record) throws Exception {
- // Normalize RowKind for consistent state representation and lookup
- RowKind originalKind = record.getRowKind();
- record.setRowKind(RowKind.INSERT); // Normalize for key lookup
- RowData stateKey = getStateKey(joinKey, record);
+ final RowData stateKey = getStateKey(joinKey, record);
Integer currentCount = recordState.get(stateKey);
if (currentCount != null) {
@@ -392,7 +384,6 @@ public final class MultiJoinStateViews {
recordState.remove(stateKey);
}
}
- record.setRowKind(originalKind); // Restore original RowKind
}
@Override