You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2025/09/18 09:43:48 UTC
(flink) branch master updated (80a95ef5a64 -> d08abb7b169)
This is an automated email from the ASF dual-hosted git repository.
roman pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 80a95ef5a64 [FLINK-37295][doc] Add state ttl migration compatibility to docs
new dfdba3dd18e [FLINK-37955] Don't load all records into main memory to avoid OOM
new d08abb7b169 [FLINK-37955] Add test for duplicated results from StreamingJoinOperator
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../plan/nodes/exec/common/JoinTestPrograms.java | 61 +++++
...nkSemanticTests.java => JoinSemanticTests.java} | 10 +-
.../join/stream/AbstractStreamingJoinOperator.java | 87 +++++++
.../join/stream/StreamingJoinOperator.java | 251 +++++++++++++++++----
.../join/stream/StreamingSemiAntiJoinOperator.java | 92 ++++----
5 files changed, 403 insertions(+), 98 deletions(-)
copy flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/{SinkSemanticTests.java => JoinSemanticTests.java} (81%)
(flink) 01/02: [FLINK-37955] Don't load all records into main memory to avoid OOM
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit dfdba3dd18e56c0b4f288c9a350245f982b27d2f
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Wed Jun 4 23:11:33 2025 +0200
[FLINK-37955] Don't load all records into main memory to avoid OOM
---
.../join/stream/AbstractStreamingJoinOperator.java | 87 +++++++
.../join/stream/StreamingJoinOperator.java | 251 +++++++++++++++++----
.../join/stream/StreamingSemiAntiJoinOperator.java | 92 ++++----
3 files changed, 338 insertions(+), 92 deletions(-)
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/AbstractStreamingJoinOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/AbstractStreamingJoinOperator.java
index c41c6615446..507d06cbe73 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/AbstractStreamingJoinOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/AbstractStreamingJoinOperator.java
@@ -26,9 +26,16 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
import org.apache.flink.table.runtime.generated.JoinCondition;
import org.apache.flink.table.runtime.operators.join.JoinConditionWithNullFilters;
+import org.apache.flink.table.runtime.operators.join.stream.state.JoinRecordStateView;
+import org.apache.flink.table.runtime.operators.join.stream.state.OuterJoinRecordStateView;
import org.apache.flink.table.runtime.operators.join.stream.utils.JoinInputSideSpec;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import java.util.Iterator;
+import java.util.function.Function;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
/**
* Abstract implementation for streaming unbounded Join operator which defines some member fields
* can be shared between different implementations.
@@ -94,4 +101,84 @@ public abstract class AbstractStreamingJoinOperator extends AbstractStreamOperat
joinCondition.close();
}
}
+
+ /** Wrap the passed iterator with filter and mapper. */
+ private static <I, O> Iterator<O> collect(
+ final Iterator<I> delegate, Function<I, Boolean> matcher, Function<I, O> mapper) {
+ return new Iterator<O>() {
+ private I next;
+
+ @Override
+ public boolean hasNext() {
+ advance();
+ return next != null;
+ }
+
+ @Override
+ public O next() {
+ checkState(hasNext());
+ I tmp = next;
+ next = null;
+ return mapper.apply(tmp);
+ }
+
+ private void advance() {
+ while (next == null && delegate.hasNext()) {
+ I record = delegate.next();
+ if (matcher.apply(record)) {
+ next = record;
+ }
+ }
+ }
+ };
+ }
+
+ /** Creates an {@link Iterator} over the records associated to the input row. */
+ protected static Iterator<OuterRecord> iterator(
+ RowData input,
+ boolean inputIsLeft,
+ JoinRecordStateView otherSideStateView,
+ JoinCondition condition)
+ throws Exception {
+ if (otherSideStateView instanceof OuterJoinRecordStateView) {
+ return collect(
+ ((OuterJoinRecordStateView) otherSideStateView)
+ .getRecordsAndNumOfAssociations()
+ .iterator(),
+ record ->
+ inputIsLeft
+ ? condition.apply(input, record.f0)
+ : condition.apply(record.f0, input),
+ record -> new OuterRecord(record.f0, record.f1));
+ } else {
+ return collect(
+ otherSideStateView.getRecords().iterator(),
+ record ->
+ inputIsLeft
+ ? condition.apply(input, record)
+ : condition.apply(record, input),
+ // use -1 as the default number of associations
+ record -> new OuterRecord(record, -1));
+ }
+ }
+
+ /**
+ * An {@link OuterRecord} is a composite of record and {@code numOfAssociations}. The {@code
+ * numOfAssociations} represents the number of associated records in the other side. It is used
+ * when the record is from outer side (e.g. left side in LEFT OUTER JOIN). When the {@code
+ * numOfAssociations} is ZERO, we need to send a null padding row. This is useful to avoid
+ * recompute the associated numbers every time.
+ *
+ * <p>When the record is from inner side (e.g. right side in LEFT OUTER JOIN), the {@code
+ * numOfAssociations} will always be {@code -1}.
+ */
+ protected static final class OuterRecord {
+ public final RowData record;
+ public final int numOfAssociations;
+
+ private OuterRecord(RowData record, int numOfAssociations) {
+ this.record = record;
+ this.numOfAssociations = numOfAssociations;
+ }
+ }
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperator.java
index 5a80c7ff0e0..dfc6bf25233 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperator.java
@@ -21,18 +21,19 @@ package org.apache.flink.table.runtime.operators.join.stream;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.util.RowDataUtil;
import org.apache.flink.table.data.utils.JoinedRowData;
import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
import org.apache.flink.table.runtime.operators.join.stream.state.JoinRecordStateView;
import org.apache.flink.table.runtime.operators.join.stream.state.JoinRecordStateViews;
import org.apache.flink.table.runtime.operators.join.stream.state.OuterJoinRecordStateView;
import org.apache.flink.table.runtime.operators.join.stream.state.OuterJoinRecordStateViews;
-import org.apache.flink.table.runtime.operators.join.stream.utils.AssociatedRecords;
-import org.apache.flink.table.runtime.operators.join.stream.utils.JoinHelper;
import org.apache.flink.table.runtime.operators.join.stream.utils.JoinInputSideSpec;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.types.RowKind;
+import java.util.Iterator;
+
/** Streaming unbounded Join operator which supports INNER/LEFT/RIGHT/FULL JOIN. */
public class StreamingJoinOperator extends AbstractStreamingJoinOperator {
@@ -52,8 +53,6 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator {
// right join state
protected transient JoinRecordStateView rightRecordStateView;
- private transient SyncStateJoinHelper joinHelper;
-
public StreamingJoinOperator(
InternalTypeInfo<RowData> leftType,
InternalTypeInfo<RowData> rightType,
@@ -122,7 +121,6 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator {
rightType,
rightStateRetentionTime);
}
- this.joinHelper = new SyncStateJoinHelper();
}
@Override
@@ -135,6 +133,74 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator {
processElement(element.getValue(), rightRecordStateView, leftRecordStateView, false, false);
}
+ /**
+ * Process an input element and output incremental joined records, retraction messages will be
+ * sent in some scenarios.
+ *
+ * <p>Following is the pseudo code to describe the core logic of this method. The logic of this
+ * method is too complex, so we provide the pseudo code to help understand the logic. We should
+ * keep sync the following pseudo code with the real logic of the method.
+ *
+ * <p>Note: "+I" represents "INSERT", "-D" represents "DELETE", "+U" represents "UPDATE_AFTER",
+ * "-U" represents "UPDATE_BEFORE". We forward input RowKind if it is inner join, otherwise, we
+ * always send insert and delete for simplification. We can optimize this to send -U & +U
+ * instead of D & I in the future (see FLINK-17337). They are equivalent in this join case. It
+ * may need some refactoring if we want to send -U & +U, so we still keep -D & +I for now for
+ * simplification. See {@code
+ * FlinkChangelogModeInferenceProgram.SatisfyModifyKindSetTraitVisitor}.
+ *
+ * <pre>
+ * if input record is accumulate
+ * | if input side is outer
+ * | | if there is no matched rows on the other side, send +I[record+null], state.add(record, 0)
+ * | | if there are matched rows on the other side
+ * | | | if other side is outer
+ * | | | | if the matched num in the matched rows == 0, send -D[null+other]
+ * | | | | if the matched num in the matched rows > 0, skip
+ * | | | | otherState.update(other, old + 1)
+ * | | | endif
+ * | | | send +I[record+other]s, state.add(record, other.size)
+ * | | endif
+ * | endif
+ * | if input side not outer
+ * | | state.add(record)
+ * | | if there is no matched rows on the other side, skip
+ * | | if there are matched rows on the other side
+ * | | | if other side is outer
+ * | | | | if the matched num in the matched rows == 0, send -D[null+other]
+ * | | | | if the matched num in the matched rows > 0, skip
+ * | | | | otherState.update(other, old + 1)
+ * | | | | send +I[record+other]s
+ * | | | else
+ * | | | | send +I/+U[record+other]s (using input RowKind)
+ * | | | endif
+ * | | endif
+ * | endif
+ * endif
+ *
+ * if input record is retract
+ * | state.retract(record)
+ * | if there is no matched rows on the other side
+ * | | if input side is outer, send -D[record+null]
+ * | endif
+ * | if there are matched rows on the other side, send -D[record+other]s if outer, send -D/-U[record+other]s if inner.
+ * | | if other side is outer
+ * | | | if the matched num in the matched rows == 0, this should never happen!
+ * | | | if the matched num in the matched rows == 1, send +I[null+other]
+ * | | | if the matched num in the matched rows > 1, skip
+ * | | | otherState.update(other, old - 1)
+ * | | endif
+ * | endif
+ * endif
+ * </pre>
+ *
+ * @param input the input element
+ * @param inputSideStateView state of input side
+ * @param otherSideStateView state of other side
+ * @param inputIsLeft whether input side is left side
+ * @param isSuppress whether suppress the output of redundant messages when the other side is
+ * outer join. This only applies to the case of mini-batch.
+ */
protected void processElement(
RowData input,
JoinRecordStateView inputSideStateView,
@@ -142,52 +208,147 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator {
boolean inputIsLeft,
boolean isSuppress)
throws Exception {
- RowKind originalRowKind = input.getRowKind();
- // erase RowKind for later state updating
- input.setRowKind(RowKind.INSERT);
-
- AssociatedRecords associatedRecords =
- AssociatedRecords.fromSyncStateView(
- input, inputIsLeft, otherSideStateView, joinCondition);
- // set back RowKind
- input.setRowKind(originalRowKind);
- joinHelper.processJoin(
- input,
- inputSideStateView,
- otherSideStateView,
- inputIsLeft,
- associatedRecords,
- isSuppress);
- }
-
- private class SyncStateJoinHelper
- extends JoinHelper<JoinRecordStateView, OuterJoinRecordStateView> {
- public SyncStateJoinHelper() {
- super(leftIsOuter, rightIsOuter, outRow, leftNullRow, rightNullRow, collector);
- }
+ boolean inputIsOuter = inputIsLeft ? leftIsOuter : rightIsOuter;
+ boolean otherIsOuter = inputIsLeft ? rightIsOuter : leftIsOuter;
+ boolean isAccumulateMsg = RowDataUtil.isAccumulateMsg(input);
+ RowKind inputRowKind = input.getRowKind();
+ input.setRowKind(RowKind.INSERT); // erase RowKind for later state updating
- @Override
- public void addRecord(JoinRecordStateView stateView, RowData record) throws Exception {
- stateView.addRecord(record);
+ if (isAccumulateMsg) { // record is accumulate
+ if (inputIsOuter) { // input side is outer
+ Iterator<OuterRecord> associatedRecords =
+ AbstractStreamingJoinOperator.iterator(
+ input, inputIsLeft, otherSideStateView, joinCondition);
+ if (!associatedRecords.hasNext()) { // there is no matched rows on the other side
+ // send +I[record+null]
+ outRow.setRowKind(RowKind.INSERT);
+ outputNullPadding(input, inputIsLeft);
+ // state.add(record, 0)
+ ((OuterJoinRecordStateView) inputSideStateView).addRecord(input, 0);
+ return;
+ } else { // there are matched rows on the other side
+ int numAssociations = 0;
+ while (associatedRecords.hasNext()) {
+ OuterRecord outerRecord = associatedRecords.next();
+ RowData other = outerRecord.record;
+ if (otherIsOuter) { // other side is outer
+ // if the matched num in the matched rows == 0
+ if (outerRecord.numOfAssociations == 0 && !isSuppress) {
+ // send -D[null+other]
+ outRow.setRowKind(RowKind.DELETE);
+ outputNullPadding(other, !inputIsLeft);
+ } // ignore matched number > 0
+ // otherState.update(other, old + 1)
+ ((OuterJoinRecordStateView) otherSideStateView)
+ .updateNumOfAssociations(
+ other, outerRecord.numOfAssociations + 1);
+ }
+ // send +I[record+other]s
+ outRow.setRowKind(RowKind.INSERT);
+ output(input, other, inputIsLeft);
+ numAssociations++;
+ }
+ // state.add(record, other.size)
+ ((OuterJoinRecordStateView) inputSideStateView)
+ .addRecord(input, numAssociations);
+ }
+ } else { // input side not outer
+ // state.add(record)
+ inputSideStateView.addRecord(input);
+ Iterator<OuterRecord> associatedRecords =
+ AbstractStreamingJoinOperator.iterator(
+ input, inputIsLeft, otherSideStateView, joinCondition);
+ if (associatedRecords.hasNext()) {
+ if (otherIsOuter) { // if other side is outer
+ OuterJoinRecordStateView otherSideOuterStateView =
+ (OuterJoinRecordStateView) otherSideStateView;
+ while (associatedRecords.hasNext()) {
+ OuterRecord outerRecord = associatedRecords.next();
+ if (outerRecord.numOfAssociations == 0
+ && !isSuppress) { // if the matched num in the matched rows == 0
+ // send -D[null+other]
+ outRow.setRowKind(RowKind.DELETE);
+ outputNullPadding(outerRecord.record, !inputIsLeft);
+ }
+ // otherState.update(other, old + 1)
+ otherSideOuterStateView.updateNumOfAssociations(
+ outerRecord.record, outerRecord.numOfAssociations + 1);
+ // send +I[record+other]s
+ outRow.setRowKind(RowKind.INSERT);
+ output(input, outerRecord.record, inputIsLeft);
+ }
+ } else {
+ // send +I/+U[record+other]s (using input RowKind)
+ outRow.setRowKind(inputRowKind);
+ while (associatedRecords.hasNext()) {
+ OuterRecord other = associatedRecords.next();
+ output(input, other.record, inputIsLeft);
+ }
+ }
+ }
+ // skip when there is no matched rows on the other side
+ }
+ } else { // input record is retract
+ // state.retract(record)
+ if (!isSuppress) {
+ inputSideStateView.retractRecord(input);
+ }
+ Iterator<OuterRecord> associatedRecords =
+ AbstractStreamingJoinOperator.iterator(
+ input, inputIsLeft, otherSideStateView, joinCondition);
+ if (!associatedRecords.hasNext()) { // there is no matched rows on the other side
+ if (inputIsOuter) { // input side is outer
+ // send -D[record+null]
+ outRow.setRowKind(RowKind.DELETE);
+ outputNullPadding(input, inputIsLeft);
+ }
+ // nothing to do when input side is not outer
+ } else { // there are matched rows on the other side
+ while (associatedRecords.hasNext()) {
+ if (inputIsOuter) {
+ // send -D[record+other]s
+ outRow.setRowKind(RowKind.DELETE);
+ } else {
+ // send -D/-U[record+other]s (using input RowKind)
+ outRow.setRowKind(inputRowKind);
+ }
+ OuterRecord outerRecord = associatedRecords.next();
+ output(input, outerRecord.record, inputIsLeft);
+ // if other side is outer
+ if (otherIsOuter) {
+ OuterJoinRecordStateView otherSideOuterStateView =
+ (OuterJoinRecordStateView) otherSideStateView;
+ if (outerRecord.numOfAssociations == 1 && !isSuppress) {
+ // send +I[null+other]
+ outRow.setRowKind(RowKind.INSERT);
+ outputNullPadding(outerRecord.record, !inputIsLeft);
+ } // nothing else to do when number of associations > 1
+ // otherState.update(other, old - 1)
+ otherSideOuterStateView.updateNumOfAssociations(
+ outerRecord.record, outerRecord.numOfAssociations - 1);
+ }
+ }
+ }
}
+ }
- @Override
- public void retractRecord(JoinRecordStateView stateView, RowData record) throws Exception {
- stateView.retractRecord(record);
- }
+ // -------------------------------------------------------------------------------------
- @Override
- public void addRecordInOuterSide(
- OuterJoinRecordStateView stateView, RowData record, int numOfAssociations)
- throws Exception {
- stateView.addRecord(record, numOfAssociations);
+ private void output(RowData inputRow, RowData otherRow, boolean inputIsLeft) {
+ if (inputIsLeft) {
+ outRow.replace(inputRow, otherRow);
+ } else {
+ outRow.replace(otherRow, inputRow);
}
+ collector.collect(outRow);
+ }
- @Override
- public void updateNumOfAssociationsInOuterSide(
- OuterJoinRecordStateView stateView, RowData record, int numOfAssociations)
- throws Exception {
- stateView.updateNumOfAssociations(record, numOfAssociations);
+ private void outputNullPadding(RowData row, boolean isLeft) {
+ if (isLeft) {
+ outRow.replace(row, rightNullRow);
+ } else {
+ outRow.replace(leftNullRow, row);
}
+ collector.collect(outRow);
}
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperator.java
index fa018a8ab60..2d7e15bbdb4 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperator.java
@@ -26,12 +26,14 @@ import org.apache.flink.table.runtime.operators.join.stream.state.JoinRecordStat
import org.apache.flink.table.runtime.operators.join.stream.state.JoinRecordStateViews;
import org.apache.flink.table.runtime.operators.join.stream.state.OuterJoinRecordStateView;
import org.apache.flink.table.runtime.operators.join.stream.state.OuterJoinRecordStateViews;
-import org.apache.flink.table.runtime.operators.join.stream.utils.AssociatedRecords;
import org.apache.flink.table.runtime.operators.join.stream.utils.JoinInputSideSpec;
-import org.apache.flink.table.runtime.operators.join.stream.utils.OuterRecord;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.types.RowKind;
+import org.apache.flink.shaded.guava33.com.google.common.collect.Iterators;
+
+import java.util.Iterator;
+
/** Streaming unbounded Join operator which supports SEMI/ANTI JOIN. */
public class StreamingSemiAntiJoinOperator extends AbstractStreamingJoinOperator {
@@ -106,10 +108,10 @@ public class StreamingSemiAntiJoinOperator extends AbstractStreamingJoinOperator
@Override
public void processElement1(StreamRecord<RowData> element) throws Exception {
RowData input = element.getValue();
- AssociatedRecords associatedRecords =
- AssociatedRecords.fromSyncStateView(
+ Iterator<OuterRecord> associatedRecords =
+ AbstractStreamingJoinOperator.iterator(
input, true, rightRecordStateView, joinCondition);
- if (associatedRecords.isEmpty()) {
+ if (!associatedRecords.hasNext()) {
if (isAntiJoin) {
collector.collect(input);
}
@@ -121,7 +123,7 @@ public class StreamingSemiAntiJoinOperator extends AbstractStreamingJoinOperator
if (RowDataUtil.isAccumulateMsg(input)) {
// erase RowKind for state updating
input.setRowKind(RowKind.INSERT);
- leftRecordStateView.addRecord(input, associatedRecords.size());
+ leftRecordStateView.addRecord(input, Iterators.size(associatedRecords));
} else { // input is retract
// erase RowKind for state updating
input.setRowKind(RowKind.INSERT);
@@ -171,53 +173,49 @@ public class StreamingSemiAntiJoinOperator extends AbstractStreamingJoinOperator
RowKind inputRowKind = input.getRowKind();
input.setRowKind(RowKind.INSERT); // erase RowKind for later state updating
- AssociatedRecords associatedRecords =
- AssociatedRecords.fromSyncStateView(
+ Iterator<OuterRecord> associatedRecords =
+ AbstractStreamingJoinOperator.iterator(
input, false, leftRecordStateView, joinCondition);
if (isAccumulateMsg) { // record is accumulate
rightRecordStateView.addRecord(input);
- if (!associatedRecords.isEmpty()) {
- // there are matched rows on the other side
- for (OuterRecord outerRecord : associatedRecords.getOuterRecords()) {
- RowData other = outerRecord.record;
- if (outerRecord.numOfAssociations == 0) {
- if (isAntiJoin) {
- // send -D[other]
- other.setRowKind(RowKind.DELETE);
- } else {
- // send +I/+U[other] (using input RowKind)
- other.setRowKind(inputRowKind);
- }
- collector.collect(other);
- // set header back to INSERT, because we will update the other row to state
- other.setRowKind(RowKind.INSERT);
- } // ignore when number > 0
- leftRecordStateView.updateNumOfAssociations(
- other, outerRecord.numOfAssociations + 1);
- }
- } // ignore when associated number == 0
+ while (associatedRecords.hasNext()) {
+ OuterRecord outerRecord = associatedRecords.next();
+ RowData other = outerRecord.record;
+ if (outerRecord.numOfAssociations == 0) {
+ if (isAntiJoin) {
+ // send -D[other]
+ other.setRowKind(RowKind.DELETE);
+ } else {
+ // send +I/+U[other] (using input RowKind)
+ other.setRowKind(inputRowKind);
+ }
+ collector.collect(other);
+ // set header back to INSERT, because we will update the other row to state
+ other.setRowKind(RowKind.INSERT);
+ } // ignore when number > 0
+ leftRecordStateView.updateNumOfAssociations(
+ other, outerRecord.numOfAssociations + 1);
+ }
} else { // retract input
rightRecordStateView.retractRecord(input);
- if (!associatedRecords.isEmpty()) {
- // there are matched rows on the other side
- for (OuterRecord outerRecord : associatedRecords.getOuterRecords()) {
- RowData other = outerRecord.record;
- if (outerRecord.numOfAssociations == 1) {
- if (!isAntiJoin) {
- // send -D/-U[other] (using input RowKind)
- other.setRowKind(inputRowKind);
- } else {
- // send +I[other]
- other.setRowKind(RowKind.INSERT);
- }
- collector.collect(other);
- // set RowKind back, because we will update the other row to state
+ while (associatedRecords.hasNext()) {
+ OuterRecord outerRecord = associatedRecords.next();
+ RowData other = outerRecord.record;
+ if (outerRecord.numOfAssociations == 1) {
+ if (!isAntiJoin) {
+ // send -D/-U[other] (using input RowKind)
+ other.setRowKind(inputRowKind);
+ } else {
+ // send +I[other]
other.setRowKind(RowKind.INSERT);
- } // ignore when number > 0
- leftRecordStateView.updateNumOfAssociations(
- other, outerRecord.numOfAssociations - 1);
- }
- } // ignore when associated number == 0
+ }
+ collector.collect(other);
+ // set RowKind back, because we will update the other row to state
+ other.setRowKind(RowKind.INSERT);
+ } // ignore when number > 0
+ leftRecordStateView.updateNumOfAssociations(
+ other, outerRecord.numOfAssociations - 1);
+ }
}
}
}
(flink) 02/02: [FLINK-37955] Add test for duplicated results from StreamingJoinOperator
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d08abb7b1697a3ba78ee0bc5d31f841eb0c9d386
Author: Gustavo de Morais <gd...@confluent.io>
AuthorDate: Fri Aug 8 17:43:16 2025 +0200
[FLINK-37955] Add test for duplicated results from StreamingJoinOperator
---
.../plan/nodes/exec/common/JoinTestPrograms.java | 61 ++++++++++++++++++++++
.../plan/nodes/exec/stream/JoinSemanticTests.java | 33 ++++++++++++
2 files changed, 94 insertions(+)
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java
index a6d3582964c..6f41ef4ff1e 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java
@@ -18,10 +18,14 @@
package org.apache.flink.table.planner.plan.nodes.exec.common;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecJoin;
import org.apache.flink.table.test.program.SinkTestStep;
import org.apache.flink.table.test.program.SourceTestStep;
import org.apache.flink.table.test.program.TableTestProgram;
import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+
+import java.util.stream.IntStream;
/** {@link TableTestProgram} definitions for testing {@link StreamExecJoin}. */
public class JoinTestPrograms {
@@ -41,6 +45,63 @@ public class JoinTestPrograms {
public static final TableTestProgram JOIN_WITH_STATE_TTL_HINT;
public static final TableTestProgram SEMI_ANTI_JOIN_WITH_LITERAL_AGG;
+ public static final TableTestProgram OUTER_JOIN_CHANGELOG_TEST =
+ TableTestProgram.of("join-duplicate-emission-bug", "bug with CTE and left join")
+ .setupTableSource(
+ SourceTestStep.newBuilder("upsert_table_with_duplicates")
+ .addSchema(
+ "`execution_plan_id` VARCHAR(2147483647) NOT NULL",
+ "`workflow_id` VARCHAR(2147483647) NOT NULL",
+ "`event_section_id` VARCHAR(2147483647) NOT NULL",
+ "CONSTRAINT `PRIMARY` PRIMARY KEY (`execution_plan_id`, `event_section_id`) NOT ENFORCED")
+ .addOption("changelog-mode", "I, UA,D")
+ .producedValues(
+ IntStream.range(0, 13)
+ .mapToObj(
+ i ->
+ Row.ofKind(
+ RowKind.UPDATE_AFTER,
+ "section_id_1",
+ "section_id_2",
+ "section_id_3"))
+ .toArray(Row[]::new))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("event_element_id STRING", "cnt BIGINT")
+ .testMaterializedData()
+ .consumedValues(Row.of("pk-1", 1), Row.of("pk-2", 1))
+ .build())
+ .runSql(
+ "INSERT INTO sink WITH\n"
+ + " section_detail as (\n"
+ + " SELECT s.event_section_id\n"
+ + " \n"
+ + " FROM upsert_table_with_duplicates s\n"
+ + " ),\n"
+ + "\n"
+ + " event_element as (\n"
+ + " SELECT\n"
+ + " ed.id as event_element_id\n"
+ + " FROM (\n"
+ + " SELECT\n"
+ + " 'pk-2' id,\n"
+ + " 'section_id_3' section_id\n"
+ + " UNION ALL\n"
+ + " SELECT\n"
+ + " 'pk-1' id,\n"
+ + " 'section_id_3' section_id\n"
+ + " ) ed \n"
+ + " LEFT JOIN\n"
+ + " section_detail as s\n"
+ + " ON s.event_section_id = ed.section_id\n"
+ + " )\n"
+ + "\n"
+ + "SELECT event_element_id, COUNT(*) cnt\n"
+ + "FROM event_element\n"
+ + "GROUP BY event_element_id")
+ .build();
+
static final SourceTestStep EMPLOYEE =
SourceTestStep.newBuilder("EMPLOYEE")
.addSchema("deptno int", "salary bigint", "name varchar")
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/JoinSemanticTests.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/JoinSemanticTests.java
new file mode 100644
index 00000000000..b7970e3aa3f
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/JoinSemanticTests.java
@@ -0,0 +1,33 @@
+/*
+ * 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.plan.nodes.exec.stream;
+
+import org.apache.flink.table.planner.plan.nodes.exec.common.JoinTestPrograms;
+import org.apache.flink.table.planner.plan.nodes.exec.testutils.SemanticTestBase;
+import org.apache.flink.table.test.program.TableTestProgram;
+
+import java.util.List;
+
+/** Semantic tests for {@link StreamExecJoin}. */
+public class JoinSemanticTests extends SemanticTestBase {
+ @Override
+ public List<TableTestProgram> programs() {
+ return List.of(JoinTestPrograms.OUTER_JOIN_CHANGELOG_TEST);
+ }
+}