You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2026/09/30 08:26:05 UTC
(flink) 01/02: [FLINK-40809][table] Outer regular joins fix: do not count an updated record as a new match
This is an automated email from the ASF dual-hosted git repository.
gustavodemorais pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 675f07c0abed3fcb492f7d682a69ab82cdfc0249
Author: Gustavo de Morais <gd...@confluent.io>
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)
---
.../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 5042dd540ce..45bde458d4e 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._
@@ -809,6 +817,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"
@@ -1747,4 +1831,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")) {