This is an automated email from the ASF dual-hosted git repository.
RocMarshal pushed a commit to branch release-2.4
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.4 by this push:
new 1fa0b04b8ac [FLINK-41067][table] Don't store rows whose join key can
never match in window and interval joins (#29646)
1fa0b04b8ac is described below
commit 1fa0b04b8ac46f3b02d82f45a3731cc7861a39d6
Author: Chan hae OH <[email protected]>
AuthorDate: Thu Oct 29 14:08:15 2026 +0900
[FLINK-41067][table] Don't store rows whose join key can never match in
window and interval joins (#29646)
* [FLINK-41067][table] Don't store rows whose join key can never match in
window and interval joins
Like the streaming joins before FLINK-41034, the window join and the
interval join stored rows whose join key has a NULL in a field that is
not null-safe, although such rows can never match. All of them share one
key, so they ended up in the state of a single subtask until the window
fired or the interval expired.
WindowJoinOperator, AsyncStateWindowJoinOperator and TimeIntervalJoin no
longer store these rows on an input whose unmatched rows are never
emitted. An input whose unmatched rows are emitted, such as the preserved
side of an outer join, still stores them, so they are emitted at the same
time as before.
Generated-by: Claude Code (Claude Opus 5.5)
* [FLINK-41067][table] Reuse NullAwareJoinHelper in IntervalJoinFunction
and test a partly null-safe key
IntervalJoinFunction#currentKeyCannotMatch repeated the body of
NullAwareJoinHelper#shouldFilter, so delegate to it like
JoinConditionWithNullFilters does.
Also cover a composite key that is only partly null-safe: a row with a
NULL in the field compared with = is not cached, while a row with a NULL
in the null-safe field is cached and matches.
Generated-by: Claude Code (Claude Opus 5.5)
* [FLINK-41067][table] Let NullAwareJoinHelper#shouldFilter take
NullAwareGetters
shouldFilter only needs anyNull, so it can take a NullAwareGetters.
IntervalJoinFunction then keeps its join key as NullAwareGetters, as
before, and the existing callers pass BinaryRowData unchanged.
Generated-by: Claude Code (Claude Opus 5.5)
---------
Co-authored-by: Rion Williams <[email protected]>
---
.../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) {
