This is an automated email from the ASF dual-hosted git repository. gustavodemorais pushed a commit to branch release-2.3 in repository https://gitbox.apache.org/repos/asf/flink.git
commit b0c6f5c40ae72bd72f2cebe6da6ec19fbe1a6512 Author: Gustavo de Morais <[email protected]> AuthorDate: Mon Sep 28 12:38:31 2026 +0200 [FLINK-40831][table] MiniBatch joins fix: count an updated record again after a suppressed retraction Generated-by: Claude Code (Claude Opus 5.5) (cherry picked from commit 076753358ce080a8fac6e98a2f50d4881ac050f1) --- .../join/stream/StreamingJoinOperator.java | 17 +++++++---- .../stream/StreamingMiniBatchJoinOperatorTest.java | 33 ++++++++++++++++++++++ 2 files changed, 44 insertions(+), 6 deletions(-) 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 d3cd3e6905f..af5b1457f5f 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 @@ -216,7 +216,7 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { if (isAccumulateMsg) { // record is accumulate final boolean isAdditionalMatch = - isAdditionalMatch(input, inputSideStateView, inputIsLeft); + isAdditionalMatch(input, inputSideStateView, inputIsLeft, isSuppress); if (inputIsOuter) { // input side is outer Iterator<OuterRecord> associatedRecords = AbstractStreamingJoinOperator.iterator( @@ -339,21 +339,26 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { } /** - * Returns whether the record adds a new match for other-side records, rather than replacing a - * stored record with the same unique key. + * Returns whether the record adds a new match for other-side records, which indicates that we + * have to increase the number of associations for a specific key match. * * <p>Matches are only counted when the other side is outer, so the result is false otherwise * and no lookup happens. If the join key contains the unique key, there is at most one record * per join key, so a match is never additional and no lookup is needed either. + * + * <p>A suppressed retraction in mini-batch mode removes the matches of a record but keeps it in + * state, so the suppressed accumulate message that follows is always an additional match. */ - private boolean isAdditionalMatch(RowData record, JoinRecordStateView stateView, boolean isLeft) + private boolean isAdditionalMatch( + RowData record, JoinRecordStateView stateView, boolean isLeft, boolean isSuppress) throws Exception { final boolean otherIsOuter = isLeft ? rightIsOuter : leftIsOuter; final JoinInputSideSpec inputSideSpec = isLeft ? leftInputSideSpec : rightInputSideSpec; // TODO FLINK-40841: assumes the replaced record had the same matches, not true for non-equi return otherIsOuter - && !inputSideSpec.joinKeyContainsUniqueKey() - && !stateView.hasRecord(record); + && (isSuppress + || (!inputSideSpec.joinKeyContainsUniqueKey() + && !stateView.hasRecord(record))); } // ------------------------------------------------------------------------------------- diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMiniBatchJoinOperatorTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMiniBatchJoinOperatorTest.java index a42e258a340..98abbe1764a 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMiniBatchJoinOperatorTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMiniBatchJoinOperatorTest.java @@ -1252,6 +1252,39 @@ final class StreamingMiniBatchJoinOperatorTest extends StreamingJoinOperatorTest testLeftJoinWithUpdateRecordsMultipleCases(); } + /** + * A suppressed -U/+U pair on one of two matches of the left row keeps both matches, so deleting + * the other one does not null pad the left row. + */ + @Tag("miniBatchSize=10") + @Test + void testLeftJoinHasUniqueKeyUpdatesOneOfTwoMatchesWithinBundle() throws Exception { + final String address = "3 Bellevue Drive, Pottstown, PA 19464"; + testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1", address)); + testHarness.processElement2(insertRecord("Ord#1", "LineOrd#1", "AIR")); + testHarness.processElement2(insertRecord("Ord#2", "LineOrd#1", "SHIP")); + testHarness.prepareSnapshotPreBarrier(1L); + testHarness.getOutput().clear(); + + testHarness.processElement2(updateBeforeRecord("Ord#1", "LineOrd#1", "AIR")); + testHarness.processElement2(updateAfterRecord("Ord#1", "LineOrd#1", "TRUCK")); + testHarness.prepareSnapshotPreBarrier(2L); + testHarness.getOutput().clear(); + + testHarness.processElement2(deleteRecord("Ord#2", "LineOrd#1", "SHIP")); + testHarness.prepareSnapshotPreBarrier(3L); + assertor.shouldEmit( + testHarness, + rowOfKind( + RowKind.DELETE, + "Ord#1", + "LineOrd#1", + address, + "Ord#2", + "LineOrd#1", + "SHIP")); + } + @Tag("miniBatchSize=4") @Test void testLeftJoinJoinKeyContainsUniqueKeyWithUpdateMultipleCases() throws Exception {
