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 8d01380ff5b [FLINK-41034][table] Don't store rows whose join key can
never match in streaming joins
8d01380ff5b is described below
commit 8d01380ff5bf9892462ef9808b1b56fd9723f54d
Author: Rion Williams <[email protected]>
AuthorDate: Thu Oct 22 05:49:23 2026 -0500
[FLINK-41034][table] Don't store rows whose join key can never match in
streaming joins
Fixed in #29532.
---
.../join/JoinConditionWithNullFilters.java | 24 ++-
.../join/stream/StreamingJoinOperator.java | 13 ++
.../join/stream/StreamingSemiAntiJoinOperator.java | 15 +-
.../AsyncStateStreamingJoinOperator.java | 8 +
.../operators/join/stream/utils/JoinHelper.java | 1 +
.../join/JoinConditionWithNullFiltersTest.java | 86 ++++++++++
.../join/stream/StreamingJoinOperatorTest.java | 174 +++++++++++++++++++++
.../join/stream/StreamingJoinOperatorTestBase.java | 6 +-
.../stream/StreamingMiniBatchJoinOperatorTest.java | 23 ++-
.../stream/StreamingSemiAntiJoinOperatorTest.java | 38 +++++
.../flink/table/utils/HandwrittenSelectorUtil.java | 2 +-
11 files changed, 373 insertions(+), 17 deletions(-)
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/JoinConditionWithNullFilters.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/JoinConditionWithNullFilters.java
index 120dc9e5c97..8990e6e94de 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/JoinConditionWithNullFilters.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/JoinConditionWithNullFilters.java
@@ -21,7 +21,7 @@ package org.apache.flink.table.runtime.operators.join;
import org.apache.flink.api.common.functions.WrappingFunction;
import org.apache.flink.streaming.api.operators.KeyContext;
import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.data.binary.NullAwareGetters;
+import org.apache.flink.table.data.binary.BinaryRowData;
import org.apache.flink.table.runtime.generated.JoinCondition;
/** Utility to take null filters into consideration when apply join condition.
*/
@@ -52,15 +52,23 @@ public class JoinConditionWithNullFilters extends
WrappingFunction<JoinCondition
@Override
public boolean apply(RowData left, RowData right) {
- if (!nullSafe) { // is not null safe, return false if any null exists
- // key is always BinaryRowData
- NullAwareGetters joinKey = (NullAwareGetters)
keyContext.getCurrentKey();
- if (filterAllNulls ? joinKey.anyNull() :
joinKey.anyNull(nullFilterKeys)) {
- // find null present, return false directly
- return false;
- }
+ if (currentKeyCannotMatch()) {
+ return false;
}
// test condition
return wrappedFunction.apply(left, right);
}
+
+ /**
+ * 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 current key must be set.
+ */
+ public boolean currentKeyCannotMatch() {
+ if (nullSafe) {
+ return false;
+ }
+ // key is always BinaryRowData
+ return NullAwareJoinHelper.shouldFilter(
+ false, filterAllNulls, nullFilterKeys, (BinaryRowData)
keyContext.getCurrentKey());
+ }
}
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 83bf57f4f80..16b879ee9e3 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
@@ -179,6 +179,10 @@ public class StreamingJoinOperator extends
AbstractStreamingJoinOperator {
*
* <pre>
* if input record is accumulate
+ * | if the join key can never match
+ * | | if input side is outer, send +I[record+null]
+ * | | return without storing the record
+ * | endif
* | if record replaces a stored record (upsert) and the condition is
non-equi
* | | for each other that the replaced record matches but the record
does not
* | | | send -D[replaced+other]
@@ -260,6 +264,15 @@ public class StreamingJoinOperator extends
AbstractStreamingJoinOperator {
input.setRowKind(RowKind.INSERT); // erase RowKind for later state
updating
if (isAccumulateMsg) { // record is accumulate
+ if (joinCondition.currentKeyCannotMatch()) {
+ // do not store a record that can never match
+ if (inputIsOuter) {
+ // send +I[record+null]
+ outRow.setRowKind(RowKind.INSERT);
+ outputNullPadding(input, inputIsLeft);
+ }
+ return;
+ }
final RowData replacedRecord =
replacedRecord(input, inputSideStateView, inputIsLeft,
isSuppress);
boolean replacedRecordHadNoMatches = false;
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperator.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperator.java
index 2d7e15bbdb4..cc770771bb4 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperator.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperator.java
@@ -101,7 +101,8 @@ public class StreamingSemiAntiJoinOperator extends
AbstractStreamingJoinOperator
* if anti join, send input record
* if there are matched rows on the other side
* if semi join, send input record
- * if the input record is accumulate, state.add(record, matched size)
+ * if the input record is accumulate, state.add(record, matched size),
unless the join key can
+ * never match
* if the input record is retract, state.retract(record)
* </pre>
*/
@@ -123,7 +124,10 @@ public class StreamingSemiAntiJoinOperator extends
AbstractStreamingJoinOperator
if (RowDataUtil.isAccumulateMsg(input)) {
// erase RowKind for state updating
input.setRowKind(RowKind.INSERT);
- leftRecordStateView.addRecord(input,
Iterators.size(associatedRecords));
+ // do not store a record that can never match
+ if (!joinCondition.currentKeyCannotMatch()) {
+ leftRecordStateView.addRecord(input,
Iterators.size(associatedRecords));
+ }
} else { // input is retract
// erase RowKind for state updating
input.setRowKind(RowKind.INSERT);
@@ -142,7 +146,7 @@ public class StreamingSemiAntiJoinOperator extends
AbstractStreamingJoinOperator
*
* <pre>
* if input record is accumulate
- * | state.add(record)
+ * | state.add(record), unless the join key can never match
* | if there is no matched rows on the other side, skip
* | if there are matched rows on the other side
* | | if the matched num in the matched rows == 0
@@ -177,7 +181,10 @@ public class StreamingSemiAntiJoinOperator extends
AbstractStreamingJoinOperator
AbstractStreamingJoinOperator.iterator(
input, false, leftRecordStateView, joinCondition);
if (isAccumulateMsg) { // record is accumulate
- rightRecordStateView.addRecord(input);
+ // do not store a record that can never match
+ if (!joinCondition.currentKeyCannotMatch()) {
+ rightRecordStateView.addRecord(input);
+ }
while (associatedRecords.hasNext()) {
OuterRecord outerRecord = associatedRecords.next();
RowData other = outerRecord.record;
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/asyncprocessing/AsyncStateStreamingJoinOperator.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/asyncprocessing/AsyncStateStreamingJoinOperator.java
index 5c92d7a6ab2..8e127653d55 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/asyncprocessing/AsyncStateStreamingJoinOperator.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/asyncprocessing/AsyncStateStreamingJoinOperator.java
@@ -185,6 +185,10 @@ public class AsyncStateStreamingJoinOperator extends
AbstractAsyncStateStreaming
@Override
public void addRecord(JoinRecordAsyncStateView stateView, RowData
record) throws Exception {
+ // do not store a record that can never match
+ if (joinCondition.currentKeyCannotMatch()) {
+ return;
+ }
// no need to wait for the future
stateView.addRecord(record);
}
@@ -200,6 +204,10 @@ public class AsyncStateStreamingJoinOperator extends
AbstractAsyncStateStreaming
public void addRecordInOuterSide(
OuterJoinRecordAsyncStateView stateView, RowData record, int
numOfAssociations)
throws Exception {
+ // do not store a record that can never match
+ if (joinCondition.currentKeyCannotMatch()) {
+ return;
+ }
// no need to wait for the future
stateView.addRecord(record, numOfAssociations);
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/utils/JoinHelper.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/utils/JoinHelper.java
index 885a1ff25e9..50c3fa04a7e 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/utils/JoinHelper.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/utils/JoinHelper.java
@@ -68,6 +68,7 @@ public abstract class JoinHelper<STATE_VIEW, OUTER_STATE_VIEW
extends STATE_VIEW
*
* <pre>
* if input record is accumulate
+ * | skip every state.add below if the join key can never match
* | if input side is outer
* | | if there is no matched rows on the other side, send
+I[record+null], state.add(record, 0)
* | | if there are matched rows on the other side
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/JoinConditionWithNullFiltersTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/JoinConditionWithNullFiltersTest.java
new file mode 100644
index 00000000000..a08ff6d0216
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/JoinConditionWithNullFiltersTest.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.join;
+
+import org.apache.flink.streaming.api.operators.KeyContext;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.data.writer.BinaryRowWriter;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link JoinConditionWithNullFilters}. */
+class JoinConditionWithNullFiltersTest {
+
+ private final TestKeyContext keyContext = new TestKeyContext();
+
+ /** The first key field is compared with {@code =}, the second with IS NOT
DISTINCT FROM. */
+ @Test
+ void testMixedKeyCannotMatchOnlyWithNullInFilteredField() {
+ final JoinConditionWithNullFilters condition =
+ new JoinConditionWithNullFilters(null, new boolean[] {true,
false}, keyContext);
+
+ keyContext.setCurrentKey(key(null, 1));
+ assertThat(condition.currentKeyCannotMatch()).isTrue();
+ keyContext.setCurrentKey(key(1, null));
+ assertThat(condition.currentKeyCannotMatch()).isFalse();
+ keyContext.setCurrentKey(key(1, 1));
+ assertThat(condition.currentKeyCannotMatch()).isFalse();
+ }
+
+ @Test
+ void testNullSafeKeyCanAlwaysMatch() {
+ final JoinConditionWithNullFilters condition =
+ new JoinConditionWithNullFilters(null, new boolean[] {false,
false}, keyContext);
+
+ keyContext.setCurrentKey(key(null, null));
+ assertThat(condition.currentKeyCannotMatch()).isFalse();
+ }
+
+ private static BinaryRowData key(Integer first, Integer second) {
+ final BinaryRowData key = new BinaryRowData(2);
+ final BinaryRowWriter writer = new BinaryRowWriter(key);
+ final Integer[] fields = {first, second};
+ for (int i = 0; i < fields.length; i++) {
+ if (fields[i] == null) {
+ writer.setNullAt(i);
+ } else {
+ writer.writeInt(i, fields[i]);
+ }
+ }
+ writer.complete();
+ return key;
+ }
+
+ private static class TestKeyContext implements KeyContext {
+
+ private Object currentKey;
+
+ @Override
+ public void setCurrentKey(Object key) {
+ currentKey = key;
+ }
+
+ @Override
+ public Object getCurrentKey() {
+ return currentKey;
+ }
+ }
+}
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 b12f6a26dab..3a5b3083b69 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
@@ -18,9 +18,11 @@
package org.apache.flink.table.runtime.operators.join.stream;
+import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.KeyedTwoInputStreamOperatorTestHarness;
+import
org.apache.flink.streaming.util.asyncprocessing.AsyncKeyedTwoInputStreamOperatorTestHarness;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
@@ -48,6 +50,7 @@ import java.util.List;
import java.util.function.Function;
import java.util.stream.Stream;
+import static
org.apache.flink.table.runtime.util.StreamRecordUtils.binaryRecord;
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;
@@ -1123,6 +1126,177 @@ class StreamingJoinOperatorTest extends
StreamingJoinOperatorTestBase {
return enableAsyncState;
}
+ @TestTemplate
+ void testInnerJoinSkipsNullKeys() throws Exception {
+ useNullKeyJoinOperator(
+ JoinInputSideSpec.withoutUniqueKey(),
+ JoinInputSideSpec.withoutUniqueKey(),
+ false,
+ false,
+ true,
+ null);
+ testHarness.processElement1(insertRecord("Ord#1", null, ADDRESS));
+ testHarness.processElement2(insertRecord(null, "AIR"));
+ testHarness.processElement2(deleteRecord(null, "AIR"));
+ assertor.shouldEmitNothing(testHarness);
+ assertStoresOnlyNonNullKeys();
+ }
+
+ @TestTemplate
+ void testLeftOuterJoinSkipsNullKeys() throws Exception {
+ useNullKeyJoinOperator(leftHasUniqueKeySpec(), rightInputSpec, true,
false, true, null);
+ testHarness.processElement1(insertRecord("Ord#1", null, ADDRESS));
+ assertor.shouldEmit(
+ testHarness, rowOfKind(RowKind.INSERT, "Ord#1", null, ADDRESS,
null, null));
+ assertThat(numStateEntries()).isZero();
+ testHarness.processElement1(updateAfterRecord("Ord#1", null,
NEW_ADDRESS));
+ assertor.shouldEmit(
+ testHarness, rowOfKind(RowKind.INSERT, "Ord#1", null,
NEW_ADDRESS, null, null));
+ testHarness.processElement1(deleteRecord("Ord#1", null, NEW_ADDRESS));
+ assertor.shouldEmit(
+ testHarness, rowOfKind(RowKind.DELETE, "Ord#1", null,
NEW_ADDRESS, null, null));
+ assertStoresOnlyNonNullKeys();
+ }
+
+ @TestTemplate
+ void testRightOuterJoinSkipsNullKeys() throws Exception {
+ useNullKeyJoinOperator(leftInputSpec, rightInputSpec, false, true,
true, null);
+ testHarness.processElement2(insertRecord(null, "AIR"));
+ assertor.shouldEmit(testHarness, rowOfKind(RowKind.INSERT, null, null,
null, null, "AIR"));
+ testHarness.processElement1(insertRecord("Ord#1", null, ADDRESS));
+ testHarness.processElement2(deleteRecord(null, "AIR"));
+ assertor.shouldEmit(testHarness, rowOfKind(RowKind.DELETE, null, null,
null, null, "AIR"));
+ assertStoresOnlyNonNullKeys();
+ }
+
+ @TestTemplate
+ void testFullOuterJoinSkipsNullKeys() throws Exception {
+ useNullKeyJoinOperator(
+ JoinInputSideSpec.withoutUniqueKey(),
+ JoinInputSideSpec.withoutUniqueKey(),
+ true,
+ true,
+ true,
+ null);
+ testHarness.processElement1(insertRecord("Ord#1", null, ADDRESS));
+ testHarness.processElement2(insertRecord(null, "AIR"));
+ testHarness.processElement1(deleteRecord("Ord#1", null, ADDRESS));
+ assertor.shouldEmit(
+ testHarness,
+ rowOfKind(RowKind.INSERT, "Ord#1", null, ADDRESS, null, null),
+ rowOfKind(RowKind.INSERT, null, null, null, null, "AIR"),
+ rowOfKind(RowKind.DELETE, "Ord#1", null, ADDRESS, null, null));
+ assertStoresOnlyNonNullKeys();
+ }
+
+ @TestTemplate
+ void testNullSafeInnerJoinMatchesNullKeys() throws Exception {
+ useNullKeyJoinOperator(
+ JoinInputSideSpec.withoutUniqueKey(),
+ JoinInputSideSpec.withoutUniqueKey(),
+ false,
+ false,
+ false,
+ null);
+ testHarness.processElement1(insertRecord("Ord#1", null, ADDRESS));
+ testHarness.processElement2(insertRecord(null, "AIR"));
+ assertor.shouldEmit(
+ testHarness, rowOfKind(RowKind.INSERT, "Ord#1", null, ADDRESS,
null, "AIR"));
+ assertThat(numStateEntries()).isEqualTo(2);
+ }
+
+ /**
+ * A row with a null key that a null-safe join stored is still removed by
its retraction. Binary
+ * rows, because restored records are binary and compare unequal to
generic ones.
+ */
+ @TestTemplate
+ void testLeftOuterJoinRetractsRestoredNullKeys() throws Exception {
+ final JoinInputSideSpec spec = JoinInputSideSpec.withoutUniqueKey();
+ useNullKeyJoinOperator(spec, spec, true, false, false, null);
+ testHarness.processElement1(binaryRecord(RowKind.INSERT, "Ord#1",
null, ADDRESS));
+ assertor.shouldEmit(
+ testHarness, rowOfKind(RowKind.INSERT, "Ord#1", null, ADDRESS,
null, null));
+ assertThat(numStateEntries()).isOne();
+ final OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0L);
+
+ useNullKeyJoinOperator(spec, spec, true, false, true, snapshot);
+ assertThat(numStateEntries()).isOne();
+ testHarness.processElement1(binaryRecord(RowKind.DELETE, "Ord#1",
null, ADDRESS));
+ assertor.shouldEmit(
+ testHarness, rowOfKind(RowKind.DELETE, "Ord#1", null, ADDRESS,
null, null));
+ assertStoresOnlyNonNullKeys();
+ }
+
+ /** Asserts that nothing is stored, and that a row with a non-null key is.
*/
+ private void assertStoresOnlyNonNullKeys() throws Exception {
+ assertThat(numStateEntries()).isZero();
+ testHarness.processElement1(insertRecord("Ord#2", "LineOrd#2",
ADDRESS));
+ assertThat(numStateEntries()).isOne();
+ }
+
+ private int numStateEntries() throws Exception {
+ if (testHarness instanceof
AsyncKeyedTwoInputStreamOperatorTestHarness) {
+ ((AsyncKeyedTwoInputStreamOperatorTestHarness<?, ?, ?, ?>)
testHarness)
+ .drainStateRequests();
+ }
+ return testHarness.numKeyedStateEntries();
+ }
+
+ /** Uses a join of the given shape with the state API of the test,
restored if given a state. */
+ private void useNullKeyJoinOperator(
+ JoinInputSideSpec leftSpec,
+ JoinInputSideSpec rightSpec,
+ boolean leftIsOuter,
+ boolean rightIsOuter,
+ boolean filterNullKeys,
+ @Nullable OperatorSubtaskState restoredState)
+ throws Exception {
+ final boolean[] filterNulls = {filterNullKeys};
+ testHarness.close();
+ if (enableAsyncState) {
+ testHarness =
+ AsyncKeyedTwoInputStreamOperatorTestHarness.create(
+ new AsyncStateStreamingJoinOperator(
+ leftTypeInfo,
+ rightTypeInfo,
+ joinCondition,
+ leftSpec,
+ rightSpec,
+ leftIsOuter,
+ rightIsOuter,
+ filterNulls,
+ 0L,
+ 0L),
+ leftKeySelector,
+ rightKeySelector,
+ leftKeySelector.getProducedType());
+ } else {
+ testHarness =
+ new KeyedTwoInputStreamOperatorTestHarness<>(
+ new StreamingJoinOperator(
+ leftTypeInfo,
+ rightTypeInfo,
+ joinCondition,
+ leftSpec,
+ rightSpec,
+ leftIsOuter,
+ rightIsOuter,
+ null,
+ null,
+ false,
+ filterNulls,
+ 0L,
+ 0L),
+ leftKeySelector,
+ rightKeySelector,
+ leftKeySelector.getProducedType());
+ }
+ if (restoredState != null) {
+ testHarness.initializeState(restoredState);
+ }
+ testHarness.open();
+ }
+
private void useStreamingJoinOperator(
JoinInputSideSpec leftSpec,
JoinInputSideSpec rightSpec,
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperatorTestBase.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperatorTestBase.java
index 9002706e742..71f00ea758a 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperatorTestBase.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperatorTestBase.java
@@ -51,7 +51,7 @@ public abstract class StreamingJoinOperatorTestBase {
RowType.of(
new LogicalType[] {
new CharType(false, 20),
- new CharType(false, 20),
+ new CharType(true, 20),
VarCharType.STRING_TYPE
},
new String[] {"order_id", "line_order_id",
"shipping_address"}));
@@ -59,7 +59,7 @@ public abstract class StreamingJoinOperatorTestBase {
protected InternalTypeInfo<RowData> rightTypeInfo =
InternalTypeInfo.of(
RowType.of(
- new LogicalType[] {new CharType(false, 20), new
CharType(true, 10)},
+ new LogicalType[] {new CharType(true, 20), new
CharType(true, 10)},
new String[] {"line_order_id0",
"line_order_ship_mode"}));
protected RowDataKeySelector leftKeySelector =
@@ -77,7 +77,7 @@ public abstract class StreamingJoinOperatorTestBase {
JoinInputSideSpec.withUniqueKeyContainedByJoinKey(rightTypeInfo,
rightKeySelector);
protected final InternalTypeInfo<RowData> joinKeyTypeInfo =
- InternalTypeInfo.of(new CharType(false, 20));
+ InternalTypeInfo.of(new CharType(true, 20));
protected final String funcCode =
"public class ConditionFunction extends
org.apache.flink.api.common.functions.AbstractRichFunction "
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMiniBatchJoinOperatorTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMiniBatchJoinOperatorTest.java
index fde7a318c51..d2f38446958 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMiniBatchJoinOperatorTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMiniBatchJoinOperatorTest.java
@@ -48,6 +48,7 @@ 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.apache.flink.table.runtime.util.StreamRecordUtils.updateBeforeRecord;
+import static org.assertj.core.api.Assertions.assertThat;
/** Test for StreamingMiniBatchJoinOperatorTest. */
final class StreamingMiniBatchJoinOperatorTest extends
StreamingJoinOperatorTestBase {
@@ -55,6 +56,26 @@ final class StreamingMiniBatchJoinOperatorTest extends
StreamingJoinOperatorTest
private RowDataKeySelector leftUniqueKeySelector;
private RowDataKeySelector rightUniqueKeySelector;
+ @Tag("miniBatchSize=3")
+ @Test
+ void testLeftJoinSkipsNullKeys() throws Exception {
+ testHarness.processElement1(insertRecord("Ord#1", null, "3 Bellevue
Drive"));
+ testHarness.processElement2(insertRecord("Ord#X", null, "AIR"));
+ testHarness.processElement1(insertRecord("Ord#2", "LineOrd#2", "4
Bellevue Drive"));
+ assertor.shouldEmitAll(
+ testHarness,
+ rowOfKind(RowKind.INSERT, "Ord#1", null, "3 Bellevue Drive",
null, null, null),
+ rowOfKind(
+ RowKind.INSERT,
+ "Ord#2",
+ "LineOrd#2",
+ "4 Bellevue Drive",
+ null,
+ null,
+ null));
+ assertThat(testHarness.numKeyedStateEntries()).isOne();
+ }
+
@Tag("miniBatchSize=3")
@Test
void testLeftJoinWithLeftArriveFirst() throws Exception {
@@ -2091,7 +2112,7 @@ final class StreamingMiniBatchJoinOperatorTest extends
StreamingJoinOperatorTest
RowType.of(
new LogicalType[] {
new CharType(false, 20),
- new CharType(false, 20),
+ new CharType(true, 20),
new CharType(true, 10)
},
new String[] {
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperatorTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperatorTest.java
index af5201f023d..f9d82da1616 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperatorTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperatorTest.java
@@ -31,6 +31,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;
/** Test for {@link StreamingSemiAntiJoinOperator}. */
class StreamingSemiAntiJoinOperatorTest extends StreamingJoinOperatorTestBase {
@@ -289,6 +290,43 @@ class StreamingSemiAntiJoinOperatorTest extends
StreamingJoinOperatorTestBase {
assertor.shouldEmitNothing(testHarness);
}
+ @Test
+ void testLeftSemiJoinSkipsNullKeys() throws Exception {
+ assertSkipsNullKeys(false);
+ }
+
+ /** An anti join emits and retracts a left row with a null key without
storing it. */
+ @Test
+ void testLeftAntiJoinSkipsNullKeys() throws Exception {
+ assertSkipsNullKeys(true);
+ }
+
+ private void assertSkipsNullKeys(boolean isAntiJoin) throws Exception {
+ String address = "3 Bellevue Drive, Pottstown, PA 19464";
+ testHarness.processElement1(insertRecord("Ord#1", null, address));
+ if (isAntiJoin) {
+ assertor.shouldEmit(testHarness, rowOfKind(RowKind.INSERT,
"Ord#1", null, address));
+ } else {
+ assertor.shouldEmitNothing(testHarness);
+ }
+ testHarness.processElement2(insertRecord(null, "AIR"));
+ assertor.shouldEmitNothing(testHarness);
+ assertThat(testHarness.numKeyedStateEntries()).isZero();
+
+ testHarness.processElement1(deleteRecord("Ord#1", null, address));
+ if (isAntiJoin) {
+ assertor.shouldEmit(testHarness, rowOfKind(RowKind.DELETE,
"Ord#1", null, address));
+ } else {
+ assertor.shouldEmitNothing(testHarness);
+ }
+ testHarness.processElement2(deleteRecord(null, "AIR"));
+ assertor.shouldEmitNothing(testHarness);
+ assertThat(testHarness.numKeyedStateEntries()).isZero();
+
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ assertThat(testHarness.numKeyedStateEntries()).isOne();
+ }
+
private static final Predicate<String> ANTI_JOIN_CHECKER =
(testDisplayName) -> testDisplayName.contains("Anti");
}
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/utils/HandwrittenSelectorUtil.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/utils/HandwrittenSelectorUtil.java
index 73d0799c2e8..321992da0b7 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/utils/HandwrittenSelectorUtil.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/utils/HandwrittenSelectorUtil.java
@@ -72,7 +72,7 @@ public class HandwrittenSelectorUtil {
BinaryRowData ret = new BinaryRowData(keyFields.length);
BinaryRowWriter writer = new BinaryRowWriter(ret);
for (int i = 0; i < keyFields.length; i++) {
- if (value.isNullAt(i)) {
+ if (value.isNullAt(keyFields[i])) {
writer.setNullAt(i);
} else {
BinaryWriter.write(
