rionmonster commented on code in PR #29570:
URL: https://github.com/apache/flink/pull/29570#discussion_r4405990083
##########
flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperatorTest.java:
##########
@@ -327,6 +335,355 @@ private void assertSkipsNullKeys(boolean isAntiJoin)
throws Exception {
assertThat(testHarness.numKeyedStateEntries()).isOne();
}
+ /** A right row replaced without UPDATE_BEFORE is counted once, so its
DELETE removes it. */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testRightUpsertIsCountedOnce(boolean isAntiJoin) throws Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightInputSpec, joinCondition,
false);
+ assertRightUpsertIsCountedOnce(isAntiJoin, "TRUCK");
+ }
+
+ /** The same holds if the unique key of the right side contains more than
the join key. */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testRightUpsertWithUniqueKeyIsCountedOnce(boolean isAntiJoin) throws
Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightHasUniqueKeySpec(),
joinCondition, false);
+ assertRightUpsertIsCountedOnce(isAntiJoin, "AIR");
+ }
+
+ private void assertRightUpsertIsCountedOnce(boolean isAntiJoin, String
newShipMode)
+ throws Exception {
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.getOutput().clear();
+
+ testHarness.processElement2(updateAfterRecord("LineOrd#1",
newShipMode));
+ assertor.shouldEmitNothing(testHarness);
+
+ testHarness.processElement2(deleteRecord("LineOrd#1", newShipMode));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS)));
+ }
+
+ /** A right row replaced without UPDATE_BEFORE that no longer matches
removes its match. */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testRightUpsertRemovesLostMatch(boolean isAntiJoin) throws Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightInputSpec,
knownValuesNonEquiCondition(), true);
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.getOutput().clear();
+
+ testHarness.processElement2(updateAfterRecord("LineOrd#1", UNKNOWN));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS)));
+
+ testHarness.processElement2(updateAfterRecord("LineOrd#1", "SHIP"));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.UPDATE_AFTER, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS)));
+ }
+
+ /** A DELETE of only the key of a right row is joined with the stored row.
*/
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testRightKeyOnlyDeleteRemovesMatch(boolean isAntiJoin) throws
Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightInputSpec,
knownValuesNonEquiCondition(), true);
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.getOutput().clear();
+
+ testHarness.processElement2(deleteRecord("LineOrd#1", null));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS)));
+ }
+
+ /**
+ * A left row replaced without UPDATE_BEFORE retracts the old version if
it was emitted but the
+ * new version is not.
+ */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testLeftUpsertRetractsOldVersion(boolean isAntiJoin) throws Exception
{
+ useSemiAntiJoinOperator(isAntiJoin, rightInputSpec,
knownValuesNonEquiCondition(), true);
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS),
+ null));
+
+ testHarness.processElement1(updateAfterRecord("Ord#1", "LineOrd#1",
UNKNOWN));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.UPDATE_AFTER, "Ord#1", "LineOrd#1",
UNKNOWN)));
+
+ testHarness.processElement1(updateAfterRecord("Ord#1", "LineOrd#1",
NEW_ADDRESS));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.UPDATE_AFTER, "Ord#1", "LineOrd#1",
NEW_ADDRESS),
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
UNKNOWN)));
+ }
+
+ /**
+ * A DELETE of only the key of a left row is joined with the stored row,
so it is emitted with
+ * all fields if the row was emitted, and not at all otherwise.
+ */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testLeftKeyOnlyDelete(boolean isAntiJoin) throws Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightInputSpec,
knownValuesNonEquiCondition(), true);
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.processElement1(insertRecord("Ord#2", "LineOrd#2",
ADDRESS));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.INSERT, "Ord#2", "LineOrd#2",
ADDRESS)));
+
+ testHarness.processElement1(deleteRecord("Ord#1", "LineOrd#1", null));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ null));
+
+ testHarness.processElement1(deleteRecord("Ord#2", "LineOrd#2", null));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ null,
+ rowOfKind(RowKind.DELETE, "Ord#2", "LineOrd#2",
ADDRESS)));
+ }
+
+ /** A DELETE of a right row that was never inserted does not remove a
match. */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testRightDeleteBeforeInsert(boolean isAntiJoin) throws Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightInputSpec, joinCondition,
false);
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.getOutput().clear();
+
+ testHarness.processElement2(deleteRecord("LineOrd#1", null));
+ assertor.shouldEmitNothing(testHarness);
+
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS)));
+
+ testHarness.processElement2(updateAfterRecord("LineOrd#1", "SHIP"));
+ assertor.shouldEmitNothing(testHarness);
+
+ testHarness.processElement2(deleteRecord("LineOrd#1", null));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS)));
+ }
+
+ /**
+ * A DELETE of a right row that was never inserted does not remove the
match of another right
+ * row with the same join key.
+ */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testRightDeleteOfRecordThatIsNotStored(boolean isAntiJoin) throws
Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightHasUniqueKeySpec(),
joinCondition, false);
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.getOutput().clear();
+
+ testHarness.processElement2(deleteRecord("LineOrd#1", "SHIP"));
+ assertor.shouldEmitNothing(testHarness);
+
+ testHarness.processElement2(deleteRecord("LineOrd#1", "AIR"));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS)));
+ }
+
+ /**
+ * A retraction of a right row that the left row did not count, because
the right row had
+ * expired when the left row was stored, does not make the count negative.
+ */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testRightRetractionOfExpiredRecord(boolean isAntiJoin) throws
Exception {
+ useSemiAntiJoinOperator(
+ isAntiJoin,
+ rightInputSpec,
+ joinCondition,
+ false,
+ RuntimeChangelogMode.serialize(ChangelogMode.all()),
+ 4000L,
+ 1000L);
+ testHarness.setStateTtlProcessingTime(1);
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.setStateTtlProcessingTime(2001);
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.getOutput().clear();
+
+ testHarness.processElement2(deleteRecord("LineOrd#1", "AIR"));
+ assertor.shouldEmitNothing(testHarness);
+
+ testHarness.processElement2(insertRecord("LineOrd#1", "SHIP"));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS)));
+ }
+
+ /**
+ * Whether the replaced left row was emitted is decided by its stored
number of matches, so it
+ * is retracted even if its matches expired.
+ */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testLeftUpsertRetractsOldVersionAfterMatchExpired(boolean isAntiJoin)
throws Exception {
+ useSemiAntiJoinOperator(
+ isAntiJoin,
+ rightInputSpec,
+ knownValuesNonEquiCondition(),
+ true,
+ UPSERT_WITH_KEY_ONLY_DELETES,
+ 4000L,
+ 1000L);
+ testHarness.setStateTtlProcessingTime(1);
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.getOutput().clear();
+
+ testHarness.setStateTtlProcessingTime(2001);
+ testHarness.processElement1(updateAfterRecord("Ord#1", "LineOrd#1",
UNKNOWN));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.UPDATE_AFTER, "Ord#1", "LineOrd#1",
UNKNOWN)));
+ }
+
+ /**
+ * A DELETE of only the key of a left row that is not stored matches
nothing, even if its null
+ * fields would match.
+ */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testLeftKeyOnlyDeleteOfRecordThatIsNotStored(boolean isAntiJoin)
throws Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightInputSpec,
nullTolerantNonEquiCondition(), true);
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.getOutput().clear();
+
+ testHarness.processElement1(deleteRecord("Ord#1", "LineOrd#1", null));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(isAntiJoin, null, rowOfKind(RowKind.DELETE, "Ord#1",
"LineOrd#1", null)));
+ }
+
+ private void useSemiAntiJoinOperator(
+ boolean isAntiJoin,
+ JoinInputSideSpec rightSpec,
+ GeneratedJoinCondition condition,
+ boolean hasNonEquiCondition)
+ throws Exception {
+ useSemiAntiJoinOperator(
+ isAntiJoin,
+ rightSpec,
+ condition,
+ hasNonEquiCondition,
+ UPSERT_WITH_KEY_ONLY_DELETES,
+ 0L,
+ 0L);
+ }
+
+ private void useSemiAntiJoinOperator(
+ boolean isAntiJoin,
+ JoinInputSideSpec rightSpec,
+ GeneratedJoinCondition condition,
+ boolean hasNonEquiCondition,
+ RuntimeChangelogMode rightChangelogMode,
+ long leftStateRetentionTime,
+ long rightStateRetentionTime)
+ throws Exception {
+ testHarness.close();
+ testHarness =
+ new KeyedTwoInputStreamOperatorTestHarness<>(
+ new StreamingSemiAntiJoinOperator(
+ isAntiJoin,
+ leftTypeInfo,
+ rightTypeInfo,
+ condition,
+ leftInputSpec,
Review Comment:
Added one specifically covered the left/unique case in
`testLeftUpsertsWithUniqueKeyWiderThanJoinKey`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
