This is an automated email from the ASF dual-hosted git repository. gustavodemorais pushed a commit to branch release-2.2 in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3d20ecb1dfb9fdcb25a9d8140796fb8d2ceb506c Author: Gustavo de Morais <[email protected]> AuthorDate: Fri Oct 2 12:47:17 2026 +0200 [FLINK-40858][table] Non-equi joins fix: join key-only deletes with the stored record Generated-by: Claude Code (Claude Opus 5.5) (cherry picked from commit 5afe863af6262ba6600cebaa73578fa95511e89a) --- .../plan/nodes/exec/common/JoinTestPrograms.java | 49 ++++ .../plan/nodes/exec/stream/JoinRestoreTest.java | 3 +- .../join-inner-join-key-only-deletes-non-equi.json | 259 +++++++++++++++++++++ .../savepoint/_metadata | Bin 0 -> 11745 bytes .../join/stream/StreamingJoinOperator.java | 63 ++++- .../join/stream/StreamingJoinOperatorTest.java | 135 ++++++++++- .../join/stream/StreamingJoinOperatorTestBase.java | 12 +- 7 files changed, 510 insertions(+), 11 deletions(-) diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java index 4699e9ed29d..25e1c0342d6 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java @@ -44,6 +44,7 @@ public class JoinTestPrograms { public static final TableTestProgram SEMI_ANTI_JOIN_WITH_LITERAL_AGG; public static final TableTestProgram LEFT_JOIN_UPSERT_INPUT_NON_EQUI; public static final TableTestProgram LEFT_JOIN_UPSERT_INPUT_NON_EQUI_WITHOUT_CHANGELOG_MODES; + public static final TableTestProgram INNER_JOIN_KEY_ONLY_DELETES_NON_EQUI; static final SourceTestStep EMPLOYEE = SourceTestStep.newBuilder("EMPLOYEE") @@ -516,6 +517,54 @@ public class JoinTestPrograms { leftJoinWithUpsertInputAndNonEquiCondition( "join-left-join-upsert-input-non-equi-without-changelog-modes", "left join with an upsert input and a non-equi condition restored from a plan without input changelog modes"); + + // the full image of the deleted line is only in the restored state + INNER_JOIN_KEY_ONLY_DELETES_NON_EQUI = + TableTestProgram.of( + "join-inner-join-key-only-deletes-non-equi", + "inner join with an input that deletes by key and a non-equi condition") + .setupTableSource( + SourceTestStep.newBuilder("orders") + .addSchema( + "order_id INT PRIMARY KEY NOT ENFORCED", + "excluded_status STRING") + .producedBeforeRestore(Row.of(1, "cancelled")) + .build()) + .setupTableSource( + SourceTestStep.newBuilder("lines") + .addSchema( + "order_id INT", + "line_id STRING", + "status STRING", + "PRIMARY KEY (order_id, line_id) NOT ENFORCED") + .addOption("changelog-mode", "I,UA,D") + .addOption("source.produces-delete-by-key", "true") + .producedBeforeRestore( + Row.ofKind(RowKind.INSERT, 1, "l1", "new")) + .producedAfterRestore( + Row.ofKind(RowKind.DELETE, 1, "l1", null), + Row.ofKind(RowKind.INSERT, 1, "l2", "paid")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("MySink") + .addSchema( + "order_id INT", + "line_id STRING", + "status STRING", + "excluded_status STRING", + "PRIMARY KEY (order_id, line_id) NOT ENFORCED") + .addOption("changelog-mode", "I,UA,D") + .addOption("sink.supports-delete-by-key", "true") + .consumedBeforeRestore("+I[1, l1, new, cancelled]") + .consumedAfterRestore( + "-D[1, l1, new, cancelled]", + "+I[1, l2, paid, cancelled]") + .build()) + .runSql( + "INSERT INTO MySink SELECT l.order_id, l.line_id, l.status, o.excluded_status " + + "FROM orders o JOIN lines l " + + "ON o.order_id = l.order_id AND l.status <> o.excluded_status") + .build(); } /** diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/JoinRestoreTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/JoinRestoreTest.java index faa1403d929..44c8f3ff958 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/JoinRestoreTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/JoinRestoreTest.java @@ -51,6 +51,7 @@ public class JoinRestoreTest extends RestoreTestBase { JoinTestPrograms.JOIN_WITH_STATE_TTL_HINT, JoinTestPrograms.SEMI_ANTI_JOIN_WITH_LITERAL_AGG, JoinTestPrograms.LEFT_JOIN_UPSERT_INPUT_NON_EQUI, - JoinTestPrograms.LEFT_JOIN_UPSERT_INPUT_NON_EQUI_WITHOUT_CHANGELOG_MODES); + JoinTestPrograms.LEFT_JOIN_UPSERT_INPUT_NON_EQUI_WITHOUT_CHANGELOG_MODES, + JoinTestPrograms.INNER_JOIN_KEY_ONLY_DELETES_NON_EQUI); } } diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-inner-join-key-only-deletes-non-equi/plan/join-inner-join-key-only-deletes-non-equi.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-inner-join-key-only-deletes-non-equi/plan/join-inner-join-key-only-deletes-non-equi.json new file mode 100644 index 00000000000..44b33983af3 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-inner-join-key-only-deletes-non-equi/plan/join-inner-join-key-only-deletes-non-equi.json @@ -0,0 +1,259 @@ +{ + "flinkVersion" : "2.2", + "nodes" : [ { + "id" : 8, + "type" : "stream-exec-table-source-scan_1", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`orders`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "order_id", + "dataType" : "INT NOT NULL" + }, { + "name" : "excluded_status", + "dataType" : "VARCHAR(2147483647)" + } ], + "primaryKey" : { + "name" : "PK_order_id", + "type" : "PRIMARY_KEY", + "columns" : [ "order_id" ] + } + } + } + } + }, + "outputType" : "ROW<`order_id` INT NOT NULL, `excluded_status` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, orders]], fields=[order_id, excluded_status])" + }, { + "id" : 9, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`order_id` INT NOT NULL, `excluded_status` VARCHAR(2147483647)>", + "description" : "Exchange(distribution=[hash[order_id]])" + }, { + "id" : 10, + "type" : "stream-exec-table-source-scan_1", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`lines`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "order_id", + "dataType" : "INT NOT NULL" + }, { + "name" : "line_id", + "dataType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "status", + "dataType" : "VARCHAR(2147483647)" + } ], + "primaryKey" : { + "name" : "PK_order_id_line_id", + "type" : "PRIMARY_KEY", + "columns" : [ "order_id", "line_id" ] + } + } + } + } + }, + "outputType" : "ROW<`order_id` INT NOT NULL, `line_id` VARCHAR(2147483647) NOT NULL, `status` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, lines]], fields=[order_id, line_id, status])" + }, { + "id" : 11, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`order_id` INT NOT NULL, `line_id` VARCHAR(2147483647) NOT NULL, `status` VARCHAR(2147483647)>", + "description" : "Exchange(distribution=[hash[order_id]])" + }, { + "id" : 12, + "type" : "stream-exec-join_1", + "joinSpec" : { + "joinType" : "INNER", + "leftKeys" : [ 0 ], + "rightKeys" : [ 0 ], + "filterNulls" : [ true ], + "nonEquiCondition" : { + "kind" : "CALL", + "syntax" : "BINARY", + "internalName" : "$<>$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 4, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "BOOLEAN" + } + }, + "leftUpsertKeys" : [ [ 0 ] ], + "rightUpsertKeys" : [ [ 0, 1 ] ], + "leftInputChangelogMode" : [ "INSERT" ], + "rightInputChangelogMode" : [ "INSERT", "UPDATE_AFTER", "~DELETE" ], + "state" : [ { + "index" : 0, + "ttl" : "0 ms", + "name" : "leftState" + }, { + "index" : 1, + "ttl" : "0 ms", + "name" : "rightState" + } ], + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + }, { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`order_id` INT NOT NULL, `excluded_status` VARCHAR(2147483647), `order_id0` INT NOT NULL, `line_id` VARCHAR(2147483647) NOT NULL, `status` VARCHAR(2147483647)>", + "description" : "Join(joinType=[InnerJoin], where=[((order_id = order_id0) AND (status <> excluded_status))], select=[order_id, excluded_status, order_id0, line_id, status], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[HasUniqueKey])" + }, { + "id" : 13, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "INT NOT NULL" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "VARCHAR(2147483647) NOT NULL" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 4, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`order_id` INT NOT NULL, `line_id` VARCHAR(2147483647) NOT NULL, `status` VARCHAR(2147483647), `excluded_status` VARCHAR(2147483647)>", + "description" : "Calc(select=[order_id0 AS order_id, line_id, status, excluded_status])" + }, { + "id" : 14, + "type" : "stream-exec-sink_1", + "configuration" : { + "table.exec.sink.keyed-shuffle" : "AUTO", + "table.exec.sink.not-null-enforcer" : "ERROR", + "table.exec.sink.rowtime-inserter" : "ENABLED", + "table.exec.sink.type-length-enforcer" : "IGNORE", + "table.exec.sink.upsert-materialize" : "AUTO" + }, + "dynamicTableSink" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`MySink`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "order_id", + "dataType" : "INT NOT NULL" + }, { + "name" : "line_id", + "dataType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "status", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "excluded_status", + "dataType" : "VARCHAR(2147483647)" + } ], + "primaryKey" : { + "name" : "PK_order_id_line_id", + "type" : "PRIMARY_KEY", + "columns" : [ "order_id", "line_id" ] + } + } + } + } + }, + "inputChangelogMode" : [ "INSERT", "UPDATE_AFTER", "~DELETE" ], + "upsertMaterializeStrategy" : "MAP", + "inputUpsertKey" : [ 0, 1 ], + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`order_id` INT NOT NULL, `line_id` VARCHAR(2147483647) NOT NULL, `status` VARCHAR(2147483647), `excluded_status` VARCHAR(2147483647)>", + "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[order_id, line_id, status, excluded_status])" + } ], + "edges" : [ { + "source" : 8, + "target" : 9, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 10, + "target" : 11, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 9, + "target" : 12, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 11, + "target" : 12, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 12, + "target" : 13, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 13, + "target" : 14, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + } ] +} \ No newline at end of file diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-inner-join-key-only-deletes-non-equi/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-inner-join-key-only-deletes-non-equi/savepoint/_metadata new file mode 100644 index 00000000000..126603f8605 Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-inner-join-key-only-deletes-non-equi/savepoint/_metadata differ 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 076f119ae12..83bf57f4f80 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 @@ -55,6 +55,10 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { // record it replaces has to be looked up in state private final boolean leftIsUpsertOrUnknown; private final boolean rightIsUpsertOrUnknown; + // whether a DELETE of the input may contain only the unique key, so that the full record has + // to be looked up in state to evaluate a non-equi condition + private final boolean leftHasKeyOnlyDeletesOrUnknown; + private final boolean rightHasKeyOnlyDeletesOrUnknown; // whether the join condition has a non-equi part, e.g. A JOIN B ON A.k = B.k AND A.v > B.v protected final boolean hasNonEquiCondition; @@ -96,6 +100,8 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { this.rightInputChangelogMode = rightInputChangelogMode; this.leftIsUpsertOrUnknown = isUpsertOrUnknown(leftInputChangelogMode); this.rightIsUpsertOrUnknown = isUpsertOrUnknown(rightInputChangelogMode); + this.leftHasKeyOnlyDeletesOrUnknown = hasKeyOnlyDeletesOrUnknown(leftInputChangelogMode); + this.rightHasKeyOnlyDeletesOrUnknown = hasKeyOnlyDeletesOrUnknown(rightInputChangelogMode); this.hasNonEquiCondition = hasNonEquiCondition; } @@ -213,6 +219,9 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { * endif * * if input record is retract + * | if record is a DELETE that may contain only the unique key and the condition is non-equi + * | | record = the stored record with the same unique key, if there is one + * | endif * | state.retract(record) if there is no paired record * | if there is no matched rows on the other side * | | if input side is outer, send -D[record+null] @@ -343,18 +352,21 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { // skip when there is no matched rows on the other side } } else { // input record is retract + final RowData retractedRecord = + retractedRecord( + input, inputRowKind, inputSideStateView, inputIsLeft, isSuppress); // state.retract(record) if (!isSuppress) { - inputSideStateView.retractRecord(input); + inputSideStateView.retractRecord(retractedRecord); } Iterator<OuterRecord> associatedRecords = AbstractStreamingJoinOperator.iterator( - input, inputIsLeft, otherSideStateView, joinCondition); + retractedRecord, inputIsLeft, otherSideStateView, joinCondition); if (!associatedRecords.hasNext()) { // there is no matched rows on the other side if (inputIsOuter) { // input side is outer // send -D[record+null] outRow.setRowKind(RowKind.DELETE); - outputNullPadding(input, inputIsLeft); + outputNullPadding(retractedRecord, inputIsLeft); } // nothing to do when input side is not outer } else { // there are matched rows on the other side @@ -367,7 +379,7 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { outRow.setRowKind(inputRowKind); } OuterRecord outerRecord = associatedRecords.next(); - output(input, outerRecord.record, inputIsLeft); + output(retractedRecord, outerRecord.record, inputIsLeft); // if other side is outer if (otherIsOuter) { OuterJoinRecordStateView otherSideOuterStateView = @@ -410,6 +422,37 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { return stateView.getRecord(record); } + /** + * Returns the full record to retract. + * + * <p>A DELETE may contain only the unique key, e.g. from an upsert source, but a non-equi + * condition needs all fields. In that case, the stored record with the same unique key is + * returned, or the given retraction if nothing is stored. The retraction of a mini-batch pair + * is returned as it is. + */ + private RowData retractedRecord( + RowData retraction, + RowKind retractionKind, + JoinRecordStateView stateView, + boolean isLeft, + boolean isSuppress) + throws Exception { + if (isSuppress) { + return retraction; + } + final boolean inputHasKeyOnlyDeletesOrUnknown = + isLeft ? leftHasKeyOnlyDeletesOrUnknown : rightHasKeyOnlyDeletesOrUnknown; + final boolean mayBeKeyOnlyDelete = + retractionKind == RowKind.DELETE + && inputHasKeyOnlyDeletesOrUnknown + && hasUniqueKey(isLeft); + if (!mayBeKeyOnlyDelete || !hasNonEquiCondition) { + return retraction; + } + final RowData storedRecord = stateView.getRecord(retraction); + return storedRecord != null ? storedRecord : retraction; + } + /** * Retracts the joined rows that only the replaced record produced, i.e. with the rows of the * other side that it matched but the new record does not. Returns whether the replaced record @@ -488,6 +531,18 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { && !changelogMode.contains(RowKind.UPDATE_BEFORE); } + /** Returns whether a DELETE of the input may contain only the unique key. */ + private static boolean hasKeyOnlyDeletesOrUnknown( + @Nullable RuntimeChangelogMode inputChangelogMode) { + // compiled plans before Flink 2.4 do not contain the changelog mode, and handling + // their deletes as key-only is always correct + return inputChangelogMode == null || inputChangelogMode.deserialize().keyOnlyDeletes(); + } + + private boolean hasUniqueKey(boolean isLeft) { + return (isLeft ? leftInputSideSpec : rightInputSideSpec).hasUniqueKey(); + } + private boolean joinKeyContainsUniqueKey(boolean isLeft) { return (isLeft ? leftInputSideSpec : rightInputSideSpec).joinKeyContainsUniqueKey(); } 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 be4256640dd..b12f6a26dab 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperatorTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperatorTest.java @@ -19,12 +19,14 @@ package org.apache.flink.table.runtime.operators.join.stream; 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.table.data.RowData; import org.apache.flink.table.runtime.generated.GeneratedJoinCondition; import org.apache.flink.table.runtime.keyselector.RowDataKeySelector; import org.apache.flink.table.runtime.operators.join.stream.asyncprocessing.AsyncStateStreamingJoinOperator; import org.apache.flink.table.runtime.operators.join.stream.utils.JoinInputSideSpec; +import org.apache.flink.table.runtime.util.RuntimeChangelogMode; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.RowType; import org.apache.flink.table.utils.HandwrittenSelectorUtil; @@ -38,6 +40,8 @@ import org.junit.jupiter.api.TestInfo; import org.junit.jupiter.api.TestTemplate; import org.junit.jupiter.api.extension.ExtendWith; +import javax.annotation.Nullable; + import java.lang.reflect.Method; import java.util.Arrays; import java.util.List; @@ -987,6 +991,114 @@ class StreamingJoinOperatorTest extends StreamingJoinOperatorTestBase { rowOfKind(RowKind.INSERT, null, null, null, "LineOrd#1", "SHIP")); } + /** + * A DELETE of only the key is joined with the stored row, so it retracts the matches one by one + * and restores the null padding of the right row after the last one. + */ + @TestTemplate + void testRightOuterJoinKeyOnlyDeletesOfLeft() throws Exception { + assumeFalse(enableAsyncState); + useStreamingJoinOperator( + leftHasUniqueKeySpec(), + rightInputSpec, + false, + true, + knownValuesNonEquiCondition(), + true); + + testHarness.processElement2(insertRecord("LineOrd#1", "AIR")); + testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1", ADDRESS)); + testHarness.processElement1(insertRecord("Ord#2", "LineOrd#1", NEW_ADDRESS)); + testHarness.getOutput().clear(); + + testHarness.processElement1(deleteRecord("Ord#1", "LineOrd#1", null)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR")); + + testHarness.processElement1(deleteRecord("Ord#2", "LineOrd#1", null)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#2", "LineOrd#1", NEW_ADDRESS, "LineOrd#1", "AIR"), + rowOfKind(RowKind.INSERT, null, null, null, "LineOrd#1", "AIR")); + } + + /** + * Compiled plans before Flink 2.4 do not tell whether a DELETE contains only the key, so it is + * joined with the stored row. A row that is not stored, e.g. expired, is joined as it is. + */ + @TestTemplate + void testLeftOuterJoinKeyOnlyDeletesOfPreviousPlan() throws Exception { + assumeFalse(enableAsyncState); + useStreamingJoinOperator( + leftInputSpec, + rightInputSpec, + true, + false, + knownValuesNonEquiCondition(), + true, + null); + + 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, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR"), + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", ADDRESS, null, null)); + + testHarness.processElement1(deleteRecord("Ord#2", "LineOrd#2", null)); + assertor.shouldEmit( + testHarness, rowOfKind(RowKind.DELETE, "Ord#2", "LineOrd#2", null, null, null)); + } + + /** + * A DELETE is joined as it is if every stored row of the join key matches, if the input sends + * full deletes, or if the input has no unique key. + */ + @TestTemplate + void testLeftOuterJoinDeletesWithoutLookup() throws Exception { + assumeFalse(enableAsyncState); + useStreamingJoinOperator(leftInputSpec, rightInputSpec, true, false); + assertJoinedDeleteOfRight(deleteRecord("LineOrd#1", null), null); + + useStreamingJoinOperator( + leftInputSpec, + rightInputSpec, + true, + false, + knownValuesNonEquiCondition(), + true, + UPSERT_WITH_FULL_DELETES); + // differs from the stored row to show that it is not looked up + assertJoinedDeleteOfRight(deleteRecord("LineOrd#1", "SHIP"), "SHIP"); + + useStreamingJoinOperator( + leftInputSpec, + JoinInputSideSpec.withoutUniqueKey(), + true, + false, + knownValuesNonEquiCondition(), + true); + assertJoinedDeleteOfRight(deleteRecord("LineOrd#1", "AIR"), "AIR"); + } + + private void assertJoinedDeleteOfRight(StreamRecord<RowData> delete, String joinedShipMode) + throws Exception { + testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1", ADDRESS)); + testHarness.processElement2(insertRecord("LineOrd#1", "AIR")); + testHarness.getOutput().clear(); + + testHarness.processElement2(delete); + assertor.shouldEmit( + testHarness, + rowOfKind( + RowKind.DELETE, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", joinedShipMode), + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", ADDRESS, null, null)); + } + /** Views without a unique key never return a stored record, not even an equal one. */ @TestTemplate void testLeftOuterJoinGetRecord() throws Exception { @@ -1029,6 +1141,25 @@ class StreamingJoinOperatorTest extends StreamingJoinOperatorTestBase { GeneratedJoinCondition condition, boolean hasNonEquiCondition) throws Exception { + useStreamingJoinOperator( + leftSpec, + rightSpec, + leftIsOuter, + rightIsOuter, + condition, + hasNonEquiCondition, + UPSERT_WITH_KEY_ONLY_DELETES); + } + + private void useStreamingJoinOperator( + JoinInputSideSpec leftSpec, + JoinInputSideSpec rightSpec, + boolean leftIsOuter, + boolean rightIsOuter, + GeneratedJoinCondition condition, + boolean hasNonEquiCondition, + @Nullable RuntimeChangelogMode inputChangelogMode) + throws Exception { testHarness.close(); testHarness = new KeyedTwoInputStreamOperatorTestHarness<>( @@ -1040,8 +1171,8 @@ class StreamingJoinOperatorTest extends StreamingJoinOperatorTestBase { rightSpec, leftIsOuter, rightIsOuter, - UPSERT, - UPSERT, + inputChangelogMode, + inputChangelogMode, hasNonEquiCondition, new boolean[] {true}, 0L, 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 2af780faa0d..9002706e742 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 @@ -101,17 +101,21 @@ public abstract class StreamingJoinOperatorTestBase { protected static final String UNKNOWN = "unknown"; - protected static final RuntimeChangelogMode UPSERT = - RuntimeChangelogMode.serialize(ChangelogMode.upsert()); + protected static final RuntimeChangelogMode UPSERT_WITH_KEY_ONLY_DELETES = + RuntimeChangelogMode.serialize(ChangelogMode.upsert(true)); - /** Only rows whose last field is not {@link #UNKNOWN} join. */ + protected static final RuntimeChangelogMode UPSERT_WITH_FULL_DELETES = + RuntimeChangelogMode.serialize(ChangelogMode.upsert(false)); + + /** Only rows whose last field is neither null nor {@link #UNKNOWN} join. */ protected static GeneratedJoinCondition knownValuesNonEquiCondition() { final String code = "public class KnownValuesNonEquiCondition extends org.apache.flink.api.common.functions.AbstractRichFunction " + "implements org.apache.flink.table.runtime.generated.JoinCondition {\n" + " public KnownValuesNonEquiCondition(Object[] reference) {}\n" + " private boolean isKnown(org.apache.flink.table.data.RowData row) {\n" - + " return !row.getString(row.getArity() - 1).toString().equals(\"" + + " final int last = row.getArity() - 1;\n" + + " return !row.isNullAt(last) && !row.getString(last).toString().equals(\"" + UNKNOWN + "\");\n" + " }\n"
