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
The following commit(s) were added to refs/heads/master by this push:
new eead57e7a07 [FLINK-41067][table] Don't store rows whose join key can
never match in window and interval joins
eead57e7a07 is described below
commit eead57e7a073c6d69e3fa48e0baebcd7f597fad5
Author: Rion Williams <[email protected]>
AuthorDate: Tue Oct 27 11:58:24 2026 -0500
[FLINK-41067][table] Don't store rows whose join key can never match in
window and interval joins
This fixes #29558.
---
.../operators/join/NullAwareJoinHelper.java | 4 +-
.../join/interval/IntervalJoinFunction.java | 15 +-
.../operators/join/interval/TimeIntervalJoin.java | 12 +-
.../operators/join/window/WindowJoinOperator.java | 2 +
.../AsyncStateWindowJoinOperator.java | 2 +
.../join/window/utils/WindowJoinHelper.java | 13 ++
.../join/interval/ProcTimeIntervalJoinTest.java | 42 ++++++
.../join/interval/RowTimeIntervalJoinTest.java | 156 +++++++++++++++++++++
.../interval/TimeIntervalStreamJoinTestBase.java | 6 +-
.../join/window/WindowJoinOperatorTest.java | 85 ++++++++++-
10 files changed, 324 insertions(+), 13 deletions(-)
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/NullAwareJoinHelper.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/NullAwareJoinHelper.java
index 9a1d87d3710..09f117e1a75 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/NullAwareJoinHelper.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/NullAwareJoinHelper.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.runtime.operators.join;
-import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.data.binary.NullAwareGetters;
import org.apache.commons.lang3.ArrayUtils;
@@ -42,7 +42,7 @@ public class NullAwareJoinHelper {
}
public static boolean shouldFilter(
- boolean nullSafe, boolean filterAllNulls, int[] nullFilterKeys,
BinaryRowData key) {
+ boolean nullSafe, boolean filterAllNulls, int[] nullFilterKeys,
NullAwareGetters key) {
return !nullSafe && (filterAllNulls ? key.anyNull() :
key.anyNull(nullFilterKeys));
}
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/IntervalJoinFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/IntervalJoinFunction.java
index f293feacbf2..2bdfc5ede43 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/IntervalJoinFunction.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/IntervalJoinFunction.java
@@ -71,11 +71,8 @@ public class IntervalJoinFunction extends
RichFlatJoinFunction<RowData, RowData,
@Override
public void join(RowData first, RowData second, Collector<RowData> out)
throws Exception {
- if (!nullSafe) { // is not null safe, return false if any null exists
- if (filterAllNulls ? joinKey.anyNull() :
joinKey.anyNull(nullFilterKeys)) {
- // find null present, return false directly
- return;
- }
+ if (currentKeyCannotMatch()) {
+ return;
}
if (joinCondition.apply(first, second)) {
out.collect(reusedJoinRowData.replace(first, second));
@@ -90,4 +87,12 @@ public class IntervalJoinFunction extends
RichFlatJoinFunction<RowData, RowData,
public void setJoinKey(RowData currentKey) {
this.joinKey = (NullAwareGetters) currentKey;
}
+
+ /**
+ * Returns whether no record of the current join key can ever match,
because the key has a null
+ * in a field that is not null-safe. The join key must be set.
+ */
+ public boolean currentKeyCannotMatch() {
+ return NullAwareJoinHelper.shouldFilter(nullSafe, filterAllNulls,
nullFilterKeys, joinKey);
+ }
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalJoin.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalJoin.java
index e699dfb7d9a..d5384013953 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalJoin.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalJoin.java
@@ -321,8 +321,10 @@ abstract class TimeIntervalJoin extends
KeyedCoProcessFunction<RowData, RowData,
// ordered keys.
}
}
- // Check if we need to cache the current row.
- if (rightOperatorTime < rightQualifiedUpperBound) {
+ // Check if we need to cache the current row. A row that can never
match is only cached to
+ // pad it with nulls once its window has closed.
+ if (rightOperatorTime < rightQualifiedUpperBound
+ && (joinType.isLeftOuter() ||
!joinFunction.currentKeyCannotMatch())) {
// Operator time of right stream has not exceeded the upper window
bound of the current
// row. Put it into the left cache, since later coming records
from the right stream are
// expected to be joined with it.
@@ -437,8 +439,10 @@ abstract class TimeIntervalJoin extends
KeyedCoProcessFunction<RowData, RowData,
// ordered keys.
}
}
- // Check if we need to cache the current row.
- if (leftOperatorTime < leftQualifiedUpperBound) {
+ // Check if we need to cache the current row. A row that can never
match is only cached to
+ // pad it with nulls once its window has closed.
+ if (leftOperatorTime < leftQualifiedUpperBound
+ && (joinType.isRightOuter() ||
!joinFunction.currentKeyCannotMatch())) {
// Operator time of left stream has not exceeded the upper window
bound of the current
// row. Put it into the right cache, since later coming records
from the left stream are
// expected to be joined with it.
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/window/WindowJoinOperator.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/window/WindowJoinOperator.java
index a087cfdada7..660969311b0 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/window/WindowJoinOperator.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/window/WindowJoinOperator.java
@@ -160,6 +160,7 @@ public class WindowJoinOperator extends
TableStreamOperator<RowData>
public void processElement1(StreamRecord<RowData> element) throws
Exception {
helper.processElement(
element,
+ true,
leftWindowEndIndex,
helper.getLeftLateRecordsDroppedRate(),
(windowEnd, rowData) -> leftWindowState.add(windowEnd,
rowData));
@@ -169,6 +170,7 @@ public class WindowJoinOperator extends
TableStreamOperator<RowData>
public void processElement2(StreamRecord<RowData> element) throws
Exception {
helper.processElement(
element,
+ false,
rightWindowEndIndex,
helper.getRightLateRecordsDroppedRate(),
(windowEnd, rowData) -> rightWindowState.add(windowEnd,
rowData));
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/window/asyncprocessing/AsyncStateWindowJoinOperator.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/window/asyncprocessing/AsyncStateWindowJoinOperator.java
index 50d2f1a0033..69e7ed50283 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/window/asyncprocessing/AsyncStateWindowJoinOperator.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/window/asyncprocessing/AsyncStateWindowJoinOperator.java
@@ -177,6 +177,7 @@ public class AsyncStateWindowJoinOperator extends
AsyncStateTableStreamOperator<
public void processElement1(StreamRecord<RowData> element) throws
Exception {
helper.processElement(
element,
+ true,
leftWindowEndIndex,
helper.getLeftLateRecordsDroppedRate(),
(windowEnd, rowData) -> leftWindowState.asyncAdd(windowEnd,
rowData));
@@ -186,6 +187,7 @@ public class AsyncStateWindowJoinOperator extends
AsyncStateTableStreamOperator<
public void processElement2(StreamRecord<RowData> element) throws
Exception {
helper.processElement(
element,
+ false,
rightWindowEndIndex,
helper.getRightLateRecordsDroppedRate(),
(windowEnd, rowData) -> rightWindowState.asyncAdd(windowEnd,
rowData));
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/window/utils/WindowJoinHelper.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/window/utils/WindowJoinHelper.java
index 27dc5c4dbb2..cc977af0ba0 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/window/utils/WindowJoinHelper.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/window/utils/WindowJoinHelper.java
@@ -75,6 +75,11 @@ public abstract class WindowJoinHelper {
private final WindowJoinProcessor windowJoinProcessor;
+ // Whether unmatched records of the left or right input are emitted when
the window fires. If
+ // not, a record of that input that can never match is not stored.
+ private final boolean leftUnmatchedIsEmitted;
+ private final boolean rightUnmatchedIsEmitted;
+
// ------------------------------------------------------------------------
// Metrics
// ------------------------------------------------------------------------
@@ -98,6 +103,8 @@ public abstract class WindowJoinHelper {
this.collector = collector;
this.windowJoinProcessor = getWindowJoinProcessor(joinType);
+ this.leftUnmatchedIsEmitted = joinType.isLeftOuter() || joinType ==
FlinkJoinType.ANTI;
+ this.rightUnmatchedIsEmitted = joinType.isRightOuter();
}
public void registerMetric(OperatorMetricGroup metrics) {
@@ -127,6 +134,7 @@ public abstract class WindowJoinHelper {
public void processElement(
StreamRecord<RowData> element,
+ boolean inputIsLeft,
int windowEndIndex,
Meter lateRecordsDroppedRate,
BiConsumerWithException<Long, RowData, Exception> accStateConsumer)
@@ -139,6 +147,11 @@ public abstract class WindowJoinHelper {
return;
}
if (RowDataUtil.isAccumulateMsg(inputRow)) {
+ final boolean unmatchedIsEmitted =
+ inputIsLeft ? leftUnmatchedIsEmitted :
rightUnmatchedIsEmitted;
+ if (!unmatchedIsEmitted && joinCondition.currentKeyCannotMatch()) {
+ return;
+ }
accStateConsumer.accept(windowEnd, inputRow);
} else {
// Window join could not handle retraction input stream
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/ProcTimeIntervalJoinTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/ProcTimeIntervalJoinTest.java
index 42a2e1dacbd..3b16ad4a371 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/ProcTimeIntervalJoinTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/ProcTimeIntervalJoinTest.java
@@ -248,6 +248,48 @@ class ProcTimeIntervalJoinTest extends
TimeIntervalStreamJoinTestBase {
testHarness.close();
}
+ @Test
+ void testDoesNotCacheRowsThatCanNeverMatch() throws Exception {
+ // Only rows of an outer input are cached, to pad them with nulls when
their window closes
+ assertNullKeyRows(FlinkJoinType.INNER, 0);
+ assertNullKeyRows(FlinkJoinType.LEFT, 2, insertRecord(null, "a", null,
null));
+ assertNullKeyRows(FlinkJoinType.RIGHT, 2, insertRecord(null, null,
null, "b"));
+ assertNullKeyRows(
+ FlinkJoinType.FULL,
+ 4,
+ insertRecord(null, "a", null, null),
+ insertRecord(null, null, null, "b"));
+ }
+
+ /**
+ * Processes a row with a null key on each input and checks the number of
state entries and the
+ * output once the windows of the rows have closed.
+ */
+ private void assertNullKeyRows(
+ FlinkJoinType joinType, int expectedStateEntries, Object...
expectedRows)
+ throws Exception {
+ ProcTimeIntervalJoin joinProcessFunc =
+ new ProcTimeIntervalJoin(
+ joinType, -5, 9, 0, rowType, rowType,
newJoinFunction(), -1L);
+ KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData,
RowData> testHarness =
+ createTestHarness(joinProcessFunc);
+ testHarness.open();
+
+ testHarness.setProcessingTime(1);
+ testHarness.processElement1(insertRecord(null, "a"));
+ testHarness.processElement2(insertRecord(null, "b"));
+ assertThat(testHarness.numKeyedStateEntries())
+ .as("state entries for %s", joinType)
+ .isEqualTo(expectedStateEntries);
+
+ testHarness.setProcessingTime(100);
+
+ assertor.assertOutputEqualsSorted(
+ "output wrong for " + joinType, List.of(expectedRows),
testHarness.getOutput());
+ assertThat(testHarness.numKeyedStateEntries()).isZero();
+ testHarness.close();
+ }
+
private KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData,
RowData>
createTestHarness(ProcTimeIntervalJoin intervalJoinFunc) throws
Exception {
KeyedCoProcessOperator<RowData, RowData, RowData, RowData> operator =
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoinTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoinTest.java
index 76efcfaad80..53e8bc93e87 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoinTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoinTest.java
@@ -32,6 +32,10 @@ import
org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
import
org.apache.flink.table.runtime.operators.join.KeyedCoProcessOperatorWithWatermarkDelay;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.runtime.util.RowDataHarnessAssertor;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.utils.HandwrittenSelectorUtil;
import org.junit.jupiter.api.Test;
@@ -491,6 +495,158 @@ class RowTimeIntervalJoinTest extends
TimeIntervalStreamJoinTestBase {
testHarness.close();
}
+ @Test
+ void testDoesNotCacheRowsThatCanNeverMatch() throws Exception {
+ // Only rows of an outer input are cached, to pad them with nulls when
their window closes
+ assertNullKeyRows(FlinkJoinType.INNER, 0);
+ assertNullKeyRows(FlinkJoinType.LEFT, 2, insertRecord(1L, null, null,
null));
+ assertNullKeyRows(FlinkJoinType.RIGHT, 2, insertRecord(null, null, 1L,
null));
+ assertNullKeyRows(
+ FlinkJoinType.FULL,
+ 4,
+ insertRecord(1L, null, null, null),
+ insertRecord(null, null, 1L, null));
+ }
+
+ /**
+ * Processes a row with a null key on each input and checks the number of
state entries and the
+ * output once the windows of the rows have closed.
+ */
+ private void assertNullKeyRows(
+ FlinkJoinType joinType, int expectedStateEntries, Object...
expectedRows)
+ throws Exception {
+ RowTimeIntervalJoin joinProcessFunc =
+ new RowTimeIntervalJoin(
+ joinType,
+ -5,
+ 9,
+ 0,
+ 7,
+ rowType,
+ rowType,
+ newJoinFunction(),
+ 0,
+ 0,
+ -1L,
+ false);
+ KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData,
RowData> testHarness =
+ createTestHarness(joinProcessFunc);
+ testHarness.open();
+
+ testHarness.processElement1(insertRecord(1L, null));
+ testHarness.processElement2(insertRecord(1L, null));
+ assertThat(testHarness.numKeyedStateEntries())
+ .as("state entries for %s", joinType)
+ .isEqualTo(expectedStateEntries);
+
+ testHarness.processWatermark1(new Watermark(100));
+ testHarness.processWatermark2(new Watermark(100));
+
+ List<Object> expectedOutput = new ArrayList<>(List.of(expectedRows));
+ expectedOutput.add(new Watermark(91));
+ assertor.assertOutputEqualsSorted(
+ "output wrong for " + joinType, expectedOutput,
testHarness.getOutput());
+ assertThat(testHarness.numKeyedStateEntries()).isZero();
+ testHarness.close();
+ }
+
+ @Test
+ void testCachesRowsWithNullSafeKeys() throws Exception {
+ // A null key can match if it is compared null-safely, as with IS NOT
DISTINCT FROM
+ RowTimeIntervalJoin joinProcessFunc =
+ new RowTimeIntervalJoin(
+ FlinkJoinType.INNER,
+ -5,
+ 9,
+ 0,
+ 7,
+ rowType,
+ rowType,
+ newJoinFunction(new boolean[] {false}),
+ 0,
+ 0,
+ -1L,
+ false);
+ KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData,
RowData> testHarness =
+ createTestHarness(joinProcessFunc);
+ testHarness.open();
+
+ testHarness.processElement1(insertRecord(1L, null));
+ testHarness.processElement2(insertRecord(1L, null));
+ assertThat(testHarness.numKeyedStateEntries()).isEqualTo(4);
+
+ testHarness.processWatermark1(new Watermark(100));
+ testHarness.processWatermark2(new Watermark(100));
+
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(insertRecord(1L, null, 1L, null));
+ expectedOutput.add(new Watermark(91));
+ assertor.assertOutputEqualsSorted("output wrong.", expectedOutput,
testHarness.getOutput());
+ testHarness.close();
+ }
+
+ @Test
+ void testCachesRowsWithNullsOnlyInNullSafeKeyFields() throws Exception {
+ // The key (a, b) compares a with = and b with IS NOT DISTINCT FROM
+ final InternalTypeInfo<RowData> compositeKeyRowType =
+ InternalTypeInfo.ofFields(
+ new BigIntType(), VarCharType.STRING_TYPE,
VarCharType.STRING_TYPE);
+ final RowDataKeySelector compositeKeySelector =
+ HandwrittenSelectorUtil.getRowDataSelector(
+ new int[] {1, 2},
compositeKeyRowType.toRowFieldTypes());
+ final RowTimeIntervalJoin joinProcessFunc =
+ new RowTimeIntervalJoin(
+ FlinkJoinType.INNER,
+ -5,
+ 9,
+ 0,
+ 7,
+ compositeKeyRowType,
+ compositeKeyRowType,
+ newJoinFunction(new boolean[] {true, false}),
+ 0,
+ 0,
+ -1L,
+ false);
+ final KeyedTwoInputStreamOperatorTestHarness<RowData, RowData,
RowData, RowData>
+ testHarness =
+ new KeyedTwoInputStreamOperatorTestHarness<>(
+ new KeyedCoProcessOperatorWithWatermarkDelay<>(
+ joinProcessFunc,
joinProcessFunc.getMaxOutputDelay()),
+ compositeKeySelector,
+ compositeKeySelector,
+ keyType);
+ testHarness.open();
+
+ testHarness.processElement1(insertRecord(1L, null, "x"));
+ testHarness.processElement2(insertRecord(1L, null, "x"));
+ assertThat(testHarness.numKeyedStateEntries()).isZero();
+
+ testHarness.processElement1(insertRecord(1L, "a", null));
+ testHarness.processElement2(insertRecord(1L, "a", null));
+ assertThat(testHarness.numKeyedStateEntries()).isEqualTo(4);
+
+ testHarness.processWatermark1(new Watermark(100));
+ testHarness.processWatermark2(new Watermark(100));
+
+ final RowDataHarnessAssertor compositeKeyAssertor =
+ new RowDataHarnessAssertor(
+ new LogicalType[] {
+ new BigIntType(),
+ VarCharType.STRING_TYPE,
+ VarCharType.STRING_TYPE,
+ new BigIntType(),
+ VarCharType.STRING_TYPE,
+ VarCharType.STRING_TYPE
+ });
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(insertRecord(1L, "a", null, 1L, "a", null));
+ expectedOutput.add(new Watermark(91));
+ compositeKeyAssertor.assertOutputEqualsSorted(
+ "output wrong.", expectedOutput, testHarness.getOutput());
+ testHarness.close();
+ }
+
@Test
public void testInterruptibleTimers() throws Exception {
final boolean[] isMailProcessed = new boolean[1]; // e.g. checkpoint
mail
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalStreamJoinTestBase.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalStreamJoinTestBase.java
index 1a1ea06847e..6d94fafba0c 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalStreamJoinTestBase.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalStreamJoinTestBase.java
@@ -56,9 +56,13 @@ abstract class TimeIntervalStreamJoinTestBase {
// instance cannot be opened twice. Restore tests that open a second
operator must build a fresh
// function for the restored harness.
protected IntervalJoinFunction newJoinFunction() {
+ return newJoinFunction(new boolean[] {true});
+ }
+
+ protected IntervalJoinFunction newJoinFunction(boolean[] filterNulls) {
return new IntervalJoinFunction(
new GeneratedJoinCondition("TestIntervalJoinCondition",
funcCode, new Object[0]),
outputRowType,
- new boolean[] {true});
+ filterNulls);
}
}
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/window/WindowJoinOperatorTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/window/WindowJoinOperatorTest.java
index b1fe275087c..8f5fd251e8c 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/window/WindowJoinOperatorTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/window/WindowJoinOperatorTest.java
@@ -504,8 +504,91 @@ class WindowJoinOperatorTest {
testHarness.close();
}
+ @TestTemplate
+ void testDoesNotStoreRecordsThatCanNeverMatch() throws Exception {
+ final long time = toUtcTimestampMills(3L, shiftTimeZone);
+ // Only records of an input whose unmatched records are emitted are
stored
+ assertNullKeyRecords(FlinkJoinType.INNER, 0);
+ assertNullKeyRecords(FlinkJoinType.SEMI, 0);
+ assertNullKeyRecords(FlinkJoinType.ANTI, 1, insertRecord(time, null));
+ assertNullKeyRecords(FlinkJoinType.LEFT, 1, insertRecord(time, null,
null, null));
+ assertNullKeyRecords(FlinkJoinType.RIGHT, 1, insertRecord(null, null,
time, null));
+ assertNullKeyRecords(
+ FlinkJoinType.FULL,
+ 2,
+ insertRecord(time, null, null, null),
+ insertRecord(null, null, time, null));
+ }
+
+ /**
+ * Processes a record with a null key on each input and checks the number
of stored records and
+ * the output when the window fires.
+ */
+ private void assertNullKeyRecords(
+ FlinkJoinType joinType, int expectedStateEntries, Object...
expectedRecords)
+ throws Exception {
+ KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData,
RowData> testHarness =
+ createTestHarness(joinType);
+ testHarness.open();
+ testHarness.processWatermark1(new Watermark(1));
+ testHarness.processWatermark2(new Watermark(1));
+
+ final long time = toUtcTimestampMills(3L, shiftTimeZone);
+ testHarness.processElement1(insertRecord(time, null));
+ testHarness.processElement2(insertRecord(time, null));
+ assertThat(testHarness.numKeyedStateEntries())
+ .as("stored records for %s", joinType)
+ .isEqualTo(expectedStateEntries);
+
+ testHarness.processWatermark1(new Watermark(10));
+ testHarness.processWatermark2(new Watermark(10));
+
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(new Watermark(1));
+ expectedOutput.addAll(Arrays.asList(expectedRecords));
+ expectedOutput.add(new Watermark(10));
+ RowDataHarnessAssertor assertor =
+ joinType == FlinkJoinType.SEMI || joinType ==
FlinkJoinType.ANTI
+ ? SEMI_ANTI_JOIN_ASSERTER
+ : ASSERTER;
+ assertor.assertOutputEqualsSorted(
+ "output wrong for " + joinType, expectedOutput,
testHarness.getOutput());
+ assertThat(testHarness.numKeyedStateEntries()).isZero();
+ testHarness.close();
+ }
+
+ @TestTemplate
+ void testStoresRecordsWithNullSafeKeys() throws Exception {
+ // A null key can match if it is compared null-safely, as with IS NOT
DISTINCT FROM
+ final long time = toUtcTimestampMills(3L, shiftTimeZone);
+ KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData,
RowData> testHarness =
+ createTestHarness(FlinkJoinType.INNER, new boolean[] {false});
+ testHarness.open();
+ testHarness.processWatermark1(new Watermark(1));
+ testHarness.processWatermark2(new Watermark(1));
+
+ testHarness.processElement1(insertRecord(time, null));
+ testHarness.processElement2(insertRecord(time, null));
+ assertThat(testHarness.numKeyedStateEntries()).isEqualTo(2);
+
+ testHarness.processWatermark1(new Watermark(10));
+ testHarness.processWatermark2(new Watermark(10));
+
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(new Watermark(1));
+ expectedOutput.add(insertRecord(time, null, time, null));
+ expectedOutput.add(new Watermark(10));
+ ASSERTER.assertOutputEqualsSorted("output wrong.", expectedOutput,
testHarness.getOutput());
+ testHarness.close();
+ }
+
private KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData,
RowData>
createTestHarness(FlinkJoinType joinType) throws Exception {
+ return createTestHarness(joinType, new boolean[] {true});
+ }
+
+ private KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData,
RowData>
+ createTestHarness(FlinkJoinType joinType, boolean[]
filterNullKeys) throws Exception {
String funcCode =
"public class TestWindowJoinCondition extends
org.apache.flink.api.common.functions.AbstractRichFunction "
+ "implements
org.apache.flink.table.runtime.generated.JoinCondition {\n"
@@ -533,7 +616,7 @@ class WindowJoinOperatorTest {
.generatedJoinCondition(joinFunction)
.leftWindowEndIndex(0)
.rightWindowEndIndex(0)
- .filterNullKeys(new boolean[] {true})
+ .filterNullKeys(filterNullKeys)
.joinType(joinType)
.withShiftTimezone(shiftTimeZone);
if (enableAsyncState) {
