This is an automated email from the ASF dual-hosted git repository. gustavodemorais pushed a commit to branch release-2.2 in repository https://gitbox.apache.org/repos/asf/flink.git
commit 798b4a76dcfc41f4c4a2dc4365c72513518f8670 Author: Gustavo de Morais <[email protected]> AuthorDate: Fri Sep 25 18:56:19 2026 +0200 [FLINK-40809][table] Outer regular joins fix: do not count an updated record as a new match Generated-by: Claude Code (Claude Opus 5.5) (cherry picked from commit 675f07c0abed3fcb492f7d682a69ab82cdfc0249) --- .../planner/runtime/stream/sql/JoinITCase.scala | 97 ++++++++++ .../join/stream/StreamingJoinOperator.java | 38 +++- .../join/stream/state/JoinRecordStateView.java | 6 + .../join/stream/state/JoinRecordStateViews.java | 16 ++ .../stream/state/OuterJoinRecordStateViews.java | 16 ++ .../join/stream/StreamingJoinOperatorTest.java | 212 ++++++++++++++++++++- 6 files changed, 369 insertions(+), 16 deletions(-) diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/JoinITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/JoinITCase.scala index fa5ef6a2752..3ecff0f01ae 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/JoinITCase.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/JoinITCase.scala @@ -17,12 +17,18 @@ */ package org.apache.flink.table.planner.runtime.stream.sql +import org.apache.flink.api.common.eventtime.WatermarkStrategy +import org.apache.flink.api.common.typeinfo.Types +import org.apache.flink.connector.datagen.source.{DataGeneratorSource, GeneratorFunction} import org.apache.flink.table.api._ import org.apache.flink.table.api.bridge.scala._ import org.apache.flink.table.api.config.ExecutionConfigOptions import org.apache.flink.table.api.config.OptimizerConfigOptions +import org.apache.flink.table.connector.ChangelogMode import org.apache.flink.table.planner.expressions.utils.FuncWithOpen import org.apache.flink.table.planner.factories.TestValuesTableFactory +import org.apache.flink.table.planner.factories.TestValuesTableFactory.changelogRow +import org.apache.flink.table.planner.runtime.stream.sql.JoinITCase.WaitForSinkThenEmit import org.apache.flink.table.planner.runtime.utils._ import org.apache.flink.table.planner.runtime.utils.BatchTestBase.row import org.apache.flink.table.planner.runtime.utils.StreamingWithMiniBatchTestBase.{MiniBatchMode, MiniBatchOff, MiniBatchOn} @@ -31,10 +37,12 @@ import org.apache.flink.testutils.junit.extensions.parameterized.{ParameterizedT import org.apache.flink.types.Row import org.assertj.core.api.Assertions.assertThat +import org.assertj.core.api.Assumptions.assumeThat import org.junit.jupiter.api.{BeforeEach, TestTemplate} import org.junit.jupiter.api.extension.ExtendWith import java.util +import java.util.UUID import scala.collection.{mutable, Seq} import scala.collection.JavaConversions._ @@ -760,6 +768,82 @@ class JoinITCase(miniBatch: MiniBatchMode, state: StateBackendMode, enableAsyncS assertThat(sink.getRetractResults.sorted).isEqualTo(expected.sorted) } + /** Only a line with a new unique key adds a match for the order, an updated line does not. */ + @TestTemplate + def testLeftJoinOnRowKey(): Unit = { + // mini-batch folds the +U/-D pair within a bundle and async state is not fixed yet + assumeThat(miniBatch == MiniBatchOff).isTrue + assumeThat(enableAsyncState).isFalse + env.setParallelism(1) + val sink = s"sink_${UUID.randomUUID().toString.replace('-', '_')}" + + val order = row(1, "EU") + val ordersId = TestValuesTableFactory.registerData(Seq(row(order))) + tEnv.executeSql(s""" + |CREATE TABLE orders ( + | order_key ROW<id INT, region STRING> PRIMARY KEY NOT ENFORCED + |) WITH ( + | 'connector' = 'values', + | 'data-id' = '$ordersId', + | 'changelog-mode' = 'I,UA,D' + |) + |""".stripMargin) + + val lines = Seq( + changelogRow("+I", order, "l1", "new"), + changelogRow("+I", order, "l2", "new"), + changelogRow("+U", order, "l1", "paid"), + changelogRow("-D", order, "l1", "paid"), + changelogRow("-D", order, "l2", "new") + ) + val lineType = Types.ROW_NAMED( + Array("order_key", "line_id", "status"), + Types.ROW_NAMED(Array("id", "region"), Types.INT, Types.STRING), + Types.STRING, + Types.STRING) + val lineSource = + new DataGeneratorSource(new WaitForSinkThenEmit(sink, lines), lines.size, lineType) + tEnv.createTemporaryView( + "lines", + tEnv.fromChangelogStream( + env.fromSource(lineSource, WatermarkStrategy.noWatermarks[Row](), "lines"), + Schema.newBuilder().primaryKey("order_key", "line_id").build(), + ChangelogMode.upsert(false) + ) + ) + + tEnv.executeSql(s""" + |CREATE TABLE $sink ( + | order_key ROW<id INT, region STRING>, + | line_order_key ROW<id INT, region STRING>, + | line_id STRING, + | status STRING + |) WITH ( + | 'connector' = 'values', + | 'sink-insert-only' = 'false', + | 'sink-changelog-mode-enforced' = 'I,UA,D' + |) + |""".stripMargin) + + val insert = + s"INSERT INTO $sink SELECT * FROM orders o LEFT JOIN lines l ON o.order_key = l.order_key" + assertThat(tEnv.explainSql(insert, ExplainDetail.CHANGELOG_MODE)) + .contains("rightInputSpec=[HasUniqueKey], changelogMode=[I,UA,D]") + tEnv.executeSql(insert).await() + + val expected = List( + "+I[+I[1, EU], null, null, null]", + "-D[+I[1, EU], null, null, null]", + "+I[+I[1, EU], +I[1, EU], l1, new]", + "+I[+I[1, EU], +I[1, EU], l2, new]", + "+I[+I[1, EU], +I[1, EU], l1, paid]", + "-D[+I[1, EU], +I[1, EU], l1, paid]", + "-D[+I[1, EU], +I[1, EU], l2, new]", + "+I[+I[1, EU], null, null, null]" + ) + assertThat(TestValuesTableFactory.getRawResultsAsStrings(sink).toList).isEqualTo(expected) + } + @TestTemplate def testRightJoinNonEqui(): Unit = { val query = "SELECT a1, b1 FROM A RIGHT JOIN B ON a1 = b1 AND a2 > b2" @@ -1698,4 +1782,17 @@ object JoinITCase { Array(MiniBatchOff, HEAP_BACKEND, Boolean.box(true)) ) } + + /** + * Emits the given rows once the sink received its first row, so that the other input goes first. + */ + class WaitForSinkThenEmit(sink: String, rows: Seq[Row]) + extends GeneratorFunction[java.lang.Long, Row] { + override def map(index: java.lang.Long): Row = { + while (TestValuesTableFactory.getRawResultsAsStrings(sink).isEmpty) { + Thread.sleep(10) + } + rows(index.toInt) + } + } } 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 dfc6bf25233..d3cd3e6905f 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 @@ -157,7 +157,7 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { * | | | 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) + * | | | | if matched num == 0 or record is an additional match, otherState.update(other, old + 1) * | | | endif * | | | send +I[record+other]s, state.add(record, other.size) * | | endif @@ -169,7 +169,7 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { * | | | 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) + * | | | | if matched num == 0 or record is an additional match, otherState.update(other, old + 1) * | | | | send +I[record+other]s * | | | else * | | | | send +I/+U[record+other]s (using input RowKind) @@ -215,6 +215,8 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { input.setRowKind(RowKind.INSERT); // erase RowKind for later state updating if (isAccumulateMsg) { // record is accumulate + final boolean isAdditionalMatch = + isAdditionalMatch(input, inputSideStateView, inputIsLeft); if (inputIsOuter) { // input side is outer Iterator<OuterRecord> associatedRecords = AbstractStreamingJoinOperator.iterator( @@ -239,9 +241,11 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { outputNullPadding(other, !inputIsLeft); } // ignore matched number > 0 // otherState.update(other, old + 1) - ((OuterJoinRecordStateView) otherSideStateView) - .updateNumOfAssociations( - other, outerRecord.numOfAssociations + 1); + if (outerRecord.numOfAssociations == 0 || isAdditionalMatch) { + ((OuterJoinRecordStateView) otherSideStateView) + .updateNumOfAssociations( + other, outerRecord.numOfAssociations + 1); + } } // send +I[record+other]s outRow.setRowKind(RowKind.INSERT); @@ -271,8 +275,10 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { outputNullPadding(outerRecord.record, !inputIsLeft); } // otherState.update(other, old + 1) - otherSideOuterStateView.updateNumOfAssociations( - outerRecord.record, outerRecord.numOfAssociations + 1); + if (outerRecord.numOfAssociations == 0 || isAdditionalMatch) { + otherSideOuterStateView.updateNumOfAssociations( + outerRecord.record, outerRecord.numOfAssociations + 1); + } // send +I[record+other]s outRow.setRowKind(RowKind.INSERT); output(input, outerRecord.record, inputIsLeft); @@ -332,6 +338,24 @@ 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. + * + * <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. + */ + private boolean isAdditionalMatch(RowData record, JoinRecordStateView stateView, boolean isLeft) + 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); + } + // ------------------------------------------------------------------------------------- private void output(RowData inputRow, RowData otherRow, boolean inputIsLeft) { diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateView.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateView.java index 2d349fd98b5..6f6b29ef7fc 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateView.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateView.java @@ -40,4 +40,10 @@ public interface JoinRecordStateView { /** Gets all the records under the current context (i.e. join key). */ Iterable<RowData> getRecords() throws Exception; + + /** + * Returns whether a record with the same unique key as the given record is stored under the + * current join key. + */ + boolean hasRecord(RowData record) throws Exception; } diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateViews.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateViews.java index 35c1770ce05..dd3477352a7 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateViews.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateViews.java @@ -99,6 +99,11 @@ public final class JoinRecordStateViews { recordState.clear(); } + @Override + public boolean hasRecord(RowData record) throws Exception { + return recordState.value() != null; + } + @Override public Iterable<RowData> getRecords() throws Exception { reusedList.clear(); @@ -146,6 +151,11 @@ public final class JoinRecordStateViews { recordState.remove(uniqueKey); } + @Override + public boolean hasRecord(RowData record) throws Exception { + return recordState.contains(uniqueKeySelector.getKey(record)); + } + @Override public Iterable<RowData> getRecords() throws Exception { return recordState.values(); @@ -193,6 +203,12 @@ public final class JoinRecordStateViews { // ignore cnt == null, which means state may be expired } + @Override + public boolean hasRecord(RowData record) { + // without a unique key, an equal record is a second record and not an update + return false; + } + @Override public Iterable<RowData> getRecords() throws Exception { return new IterableIterator<RowData>() { diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/OuterJoinRecordStateViews.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/OuterJoinRecordStateViews.java index a3f29439069..3edafd9bf2e 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/OuterJoinRecordStateViews.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/OuterJoinRecordStateViews.java @@ -118,6 +118,11 @@ public final class OuterJoinRecordStateViews { recordState.clear(); } + @Override + public boolean hasRecord(RowData record) throws Exception { + return recordState.value() != null; + } + @Override public Iterable<RowData> getRecords() throws Exception { Tuple2<RowData, Integer> tuple = recordState.value(); @@ -191,6 +196,11 @@ public final class OuterJoinRecordStateViews { recordState.remove(uniqueKey); } + @Override + public boolean hasRecord(RowData record) throws Exception { + return recordState.contains(uniqueKeySelector.getKey(record)); + } + @Override public Iterable<RowData> getRecords() throws Exception { return new RecordsIterable(getRecordsAndNumOfAssociations()); @@ -266,6 +276,12 @@ public final class OuterJoinRecordStateViews { } } + @Override + public boolean hasRecord(RowData record) { + // without a unique key, an equal record is a second record and not an update + return false; + } + @Override public Iterable<RowData> getRecords() throws Exception { return new RecordsIterable(getRecordsAndNumOfAssociations()); diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperatorTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperatorTest.java index 5a9e12feb13..53c0b49da50 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperatorTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperatorTest.java @@ -19,10 +19,14 @@ package org.apache.flink.table.runtime.operators.join.stream; import org.apache.flink.streaming.api.operators.TwoInputStreamOperator; +import org.apache.flink.streaming.util.KeyedTwoInputStreamOperatorTestHarness; import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.keyselector.RowDataKeySelector; import org.apache.flink.table.runtime.operators.join.stream.asyncprocessing.AsyncStateStreamingJoinOperator; +import org.apache.flink.table.runtime.operators.join.stream.utils.JoinInputSideSpec; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.utils.HandwrittenSelectorUtil; import org.apache.flink.testutils.junit.extensions.parameterized.Parameter; import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension; import org.apache.flink.testutils.junit.extensions.parameterized.Parameters; @@ -43,6 +47,7 @@ import static org.apache.flink.table.runtime.util.StreamRecordUtils.deleteRecord import static org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord; import static org.apache.flink.table.runtime.util.StreamRecordUtils.rowOfKind; import static org.apache.flink.table.runtime.util.StreamRecordUtils.updateAfterRecord; +import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assumptions.assumeFalse; /** Harness tests for {@link StreamingJoinOperator}. */ @@ -549,15 +554,34 @@ class StreamingJoinOperatorTest extends StreamingJoinOperatorTestBase { testHarness.setStateTtlProcessingTime(8001); testHarness.processElement2(deleteRecord("LineOrd#2", "SHIP")); - assertor.shouldEmit( - testHarness, - rowOfKind( - RowKind.DELETE, - "Ord#1", - "LineOrd#2", - "68 Manor Station Street, Honolulu, HI 96815", - "LineOrd#2", - "SHIP")); + if (increaseAssociationsOnReplace()) { + assertor.shouldEmit( + testHarness, + rowOfKind( + RowKind.DELETE, + "Ord#1", + "LineOrd#2", + "68 Manor Station Street, Honolulu, HI 96815", + "LineOrd#2", + "SHIP")); + } else { + assertor.shouldEmit( + testHarness, + rowOfKind( + RowKind.DELETE, + "Ord#1", + "LineOrd#2", + "68 Manor Station Street, Honolulu, HI 96815", + "LineOrd#2", + "SHIP"), + rowOfKind( + RowKind.INSERT, + "Ord#1", + "LineOrd#2", + "68 Manor Station Street, Honolulu, HI 96815", + null, + null)); + } } /** @@ -693,6 +717,176 @@ class StreamingJoinOperatorTest extends StreamingJoinOperatorTestBase { "AIR")); } + /** Deleting a left row replaced without UPDATE_BEFORE restores the right null padding. */ + @TestTemplate + void testRightOuterJoinRestoresNullPaddingAfterLeftUpsertAndDelete() throws Exception { + assumeFalse(enableAsyncState); + useStreamingJoinOperator(leftHasUniqueKeySpec(), rightInputSpec, false, true); + + testHarness.processElement2(insertRecord("LineOrd#1", "AIR")); + assertor.shouldEmit( + testHarness, rowOfKind(RowKind.INSERT, null, null, null, "LineOrd#1", "AIR")); + + testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1", ADDRESS)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, null, null, null, "LineOrd#1", "AIR"), + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR")); + + testHarness.processElement1(updateAfterRecord("Ord#1", "LineOrd#1", NEW_ADDRESS)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", NEW_ADDRESS, "LineOrd#1", "AIR")); + + testHarness.processElement1(deleteRecord("Ord#1", "LineOrd#1", NEW_ADDRESS)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", NEW_ADDRESS, "LineOrd#1", "AIR"), + rowOfKind(RowKind.INSERT, null, null, null, "LineOrd#1", "AIR")); + } + + /** A left row replaced without UPDATE_BEFORE is not counted twice by the right row. */ + @TestTemplate + void testFullOuterJoinRestoresNullPaddingAfterLeftUpsertAndDeletes() throws Exception { + assumeFalse(enableAsyncState); + useStreamingJoinOperator(leftHasUniqueKeySpec(), rightInputSpec, true, true); + + testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1", ADDRESS)); + assertor.shouldEmit( + testHarness, rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", ADDRESS, null, null)); + + testHarness.processElement2(insertRecord("LineOrd#1", "AIR")); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", ADDRESS, null, null), + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR")); + + testHarness.processElement1(insertRecord("Ord#2", "LineOrd#1", ADDRESS)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.INSERT, "Ord#2", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR")); + + testHarness.processElement1(updateAfterRecord("Ord#1", "LineOrd#1", NEW_ADDRESS)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", NEW_ADDRESS, "LineOrd#1", "AIR")); + + testHarness.processElement1(deleteRecord("Ord#1", "LineOrd#1", NEW_ADDRESS)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", NEW_ADDRESS, "LineOrd#1", "AIR")); + + testHarness.processElement1(deleteRecord("Ord#2", "LineOrd#1", ADDRESS)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#2", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR"), + rowOfKind(RowKind.INSERT, null, null, null, "LineOrd#1", "AIR")); + } + + /** Without a unique key, a repeated row is a second row and not a replacement. */ + @TestTemplate + void testLeftOuterJoinWithoutUniqueKeyKeepsDuplicateRows() throws Exception { + assumeFalse(enableAsyncState); + useStreamingJoinOperator(leftInputSpec, JoinInputSideSpec.withoutUniqueKey(), true, false); + + testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1", ADDRESS)); + assertor.shouldEmit( + testHarness, rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", ADDRESS, null, null)); + + testHarness.processElement2(insertRecord("LineOrd#1", "AIR")); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", ADDRESS, null, null), + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR")); + + testHarness.processElement2(insertRecord("LineOrd#1", "AIR")); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR")); + + testHarness.processElement2(deleteRecord("LineOrd#1", "AIR")); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR")); + } + + /** Views without a unique key never report a stored record, not even an equal one. */ + @TestTemplate + void testLeftOuterJoinHasRecord() throws Exception { + assumeFalse(enableAsyncState); + assertHasRecordAfterInsert(true); + + useStreamingJoinOperator( + JoinInputSideSpec.withoutUniqueKey(), + JoinInputSideSpec.withoutUniqueKey(), + true, + false); + assertHasRecordAfterInsert(false); + } + + // TODO FLINK-40681: remove once AsyncStateStreamingJoinOperator handles + // replacing records + /** + * {@link AsyncStateStreamingJoinOperator} counts a record that replaces another one as an + * additional association. + */ + private boolean increaseAssociationsOnReplace() { + return enableAsyncState; + } + + private void useStreamingJoinOperator( + JoinInputSideSpec leftSpec, + JoinInputSideSpec rightSpec, + boolean leftIsOuter, + boolean rightIsOuter) + throws Exception { + testHarness.close(); + testHarness = + new KeyedTwoInputStreamOperatorTestHarness<>( + new StreamingJoinOperator( + leftTypeInfo, + rightTypeInfo, + joinCondition, + leftSpec, + rightSpec, + leftIsOuter, + rightIsOuter, + new boolean[] {true}, + 0L, + 0L), + leftKeySelector, + rightKeySelector, + joinKeyTypeInfo); + testHarness.open(); + } + + private void assertHasRecordAfterInsert(boolean expected) throws Exception { + final StreamingJoinOperator operator = (StreamingJoinOperator) testHarness.getOperator(); + final RowData left = insertRecord("Ord#1", "LineOrd#1", ADDRESS).getValue(); + final RowData right = insertRecord("LineOrd#1", "AIR").getValue(); + + operator.setCurrentKey(rightKeySelector.getKey(right)); + assertThat(operator.leftRecordStateView.hasRecord(left)).isFalse(); + assertThat(operator.rightRecordStateView.hasRecord(right)).isFalse(); + + testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1", ADDRESS)); + testHarness.processElement2(insertRecord("LineOrd#1", "AIR")); + assertThat(operator.leftRecordStateView.hasRecord(left)).isEqualTo(expected); + assertThat(operator.rightRecordStateView.hasRecord(right)).isEqualTo(expected); + } + + private JoinInputSideSpec leftHasUniqueKeySpec() { + final RowDataKeySelector uniqueKeySelector = + HandwrittenSelectorUtil.getRowDataSelector( + new int[] {0, 1}, + leftTypeInfo.toRowType().getChildren().toArray(new LogicalType[0])); + return JoinInputSideSpec.withUniqueKey( + uniqueKeySelector.getProducedType(), uniqueKeySelector); + } + + private static final String ADDRESS = "3 Bellevue Drive, Pottstown, PA 19464"; + private static final String NEW_ADDRESS = "68 Manor Station Street, Honolulu, HI 96815"; + private static final Function<String, Boolean[]> JOIN_TYPE_EXTRACTOR = (testDisplayName) -> { if (testDisplayName.contains("InnerJoin")) {
