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 5a4f05bd645 [FLINK-40842][table] Handle upsert inputs in the streaming
semi/anti join
5a4f05bd645 is described below
commit 5a4f05bd645caf1eca56f44d723b6e787ee32822
Author: Rion Williams <[email protected]>
AuthorDate: Fri Oct 30 09:29:14 2026 -0500
[FLINK-40842][table] Handle upsert inputs in the streaming semi/anti join
This fixes #29570.
---
.../plan/nodes/exec/stream/StreamExecJoin.java | 13 +-
.../plan/nodes/exec/common/JoinTestPrograms.java | 62 ++++
.../plan/nodes/exec/stream/JoinRestoreTest.java | 4 +-
...n-semi-join-upsert-inputs-non-equi-pre-2-4.json | 211 ++++++++++++
.../savepoint/_metadata | Bin 0 -> 9670 bytes
.../join-semi-join-upsert-inputs-non-equi.json | 213 ++++++++++++
.../savepoint/_metadata | Bin 0 -> 9670 bytes
.../join/stream/AbstractStreamingJoinOperator.java | 108 ++++++
.../join/stream/StreamingJoinOperator.java | 143 ++------
.../join/stream/StreamingSemiAntiJoinOperator.java | 199 +++++++----
.../stream/state/OuterJoinRecordStateView.java | 9 +
.../stream/state/OuterJoinRecordStateViews.java | 17 +
.../join/stream/StreamingJoinOperatorTest.java | 16 -
.../join/stream/StreamingJoinOperatorTestBase.java | 24 ++
.../stream/StreamingSemiAntiJoinOperatorTest.java | 362 ++++++++++++++++++++-
15 files changed, 1157 insertions(+), 224 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java
index 684bdc1e10b..aa3175b1080 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java
@@ -224,6 +224,11 @@ public class StreamExecJoin extends ExecNodeBase<RowData>
TwoInputStreamOperator<RowData, RowData, RowData> operator;
FlinkJoinType joinType = joinSpec.getJoinType();
final boolean isMiniBatchEnabled =
MinibatchUtil.isMiniBatchEnabled(config);
+ final RuntimeChangelogMode leftRuntimeChangelogMode =
+ toRuntimeChangelogMode(leftInputChangelogMode);
+ final RuntimeChangelogMode rightRuntimeChangelogMode =
+ toRuntimeChangelogMode(rightInputChangelogMode);
+ final boolean hasNonEquiCondition =
joinSpec.getNonEquiCondition().isPresent();
if (joinType == FlinkJoinType.ANTI || joinType == FlinkJoinType.SEMI) {
operator =
new StreamingSemiAntiJoinOperator(
@@ -233,6 +238,9 @@ public class StreamExecJoin extends ExecNodeBase<RowData>
generatedCondition,
leftInputSpec,
rightInputSpec,
+ leftRuntimeChangelogMode,
+ rightRuntimeChangelogMode,
+ hasNonEquiCondition,
joinSpec.getFilterNulls(),
leftStateRetentionTime,
rightStateRetentionTime);
@@ -240,11 +248,6 @@ public class StreamExecJoin extends ExecNodeBase<RowData>
boolean leftIsOuter = joinType == FlinkJoinType.LEFT || joinType
== FlinkJoinType.FULL;
boolean rightIsOuter =
joinType == FlinkJoinType.RIGHT || joinType ==
FlinkJoinType.FULL;
- final RuntimeChangelogMode leftRuntimeChangelogMode =
- toRuntimeChangelogMode(leftInputChangelogMode);
- final RuntimeChangelogMode rightRuntimeChangelogMode =
- toRuntimeChangelogMode(rightInputChangelogMode);
- final boolean hasNonEquiCondition =
joinSpec.getNonEquiCondition().isPresent();
if (isMiniBatchEnabled) {
operator =
MiniBatchStreamingJoinOperator.newMiniBatchStreamJoinOperator(
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 150a2b31b60..7cb27378d12 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
@@ -45,6 +45,8 @@ public class JoinTestPrograms {
public static final TableTestProgram LEFT_JOIN_UPSERT_INPUT_NON_EQUI;
public static final TableTestProgram
LEFT_JOIN_UPSERT_INPUT_NON_EQUI_PRE_2_4;
public static final TableTestProgram INNER_JOIN_KEY_ONLY_DELETES_NON_EQUI;
+ public static final TableTestProgram SEMI_JOIN_UPSERT_INPUTS_NON_EQUI;
+ public static final TableTestProgram
SEMI_JOIN_UPSERT_INPUTS_NON_EQUI_PRE_2_4;
static final SourceTestStep EMPLOYEE =
SourceTestStep.newBuilder("EMPLOYEE")
@@ -565,6 +567,66 @@ public class JoinTestPrograms {
+ "FROM orders o JOIN lines l "
+ "ON o.order_id = l.order_id AND
l.status <> o.excluded_status")
.build();
+
+ SEMI_JOIN_UPSERT_INPUTS_NON_EQUI =
+ semiJoinWithUpsertInputsAndNonEquiCondition(
+ "join-semi-join-upsert-inputs-non-equi",
+ "semi join with upsert inputs that delete by key and a
non-equi condition");
+
+ // its plan and savepoint are compiled with Flink 2.3, whose plans
have no input changelog
+ // modes
+ SEMI_JOIN_UPSERT_INPUTS_NON_EQUI_PRE_2_4 =
+ semiJoinWithUpsertInputsAndNonEquiCondition(
+ "join-semi-join-upsert-inputs-non-equi-pre-2-4",
+ "semi join with upsert inputs that delete by key and a
non-equi condition restored from a Flink 2.3 plan");
+ }
+
+ /**
+ * The line of order 1 is updated without UPDATE_BEFORE and deleted by
key, the replaced version
+ * of the line is only in the restored state; order 2 stops matching its
line.
+ */
+ private static TableTestProgram
semiJoinWithUpsertInputsAndNonEquiCondition(
+ String id, String description) {
+ return TableTestProgram.of(id, description)
+ .setupTableSource(
+ SourceTestStep.newBuilder("orders")
+ .addSchema(
+ "order_id INT PRIMARY KEY NOT
ENFORCED",
+ "excluded_status STRING")
+ .addOption("changelog-mode", "I,UA,D")
+ .addOption("source.produces-delete-by-key",
"true")
+ .producedBeforeRestore(
+ Row.ofKind(RowKind.INSERT, 1,
"cancelled"),
+ Row.ofKind(RowKind.INSERT, 2,
"cancelled"))
+
.producedAfterRestore(Row.ofKind(RowKind.UPDATE_AFTER, 2, "paid"))
+ .build())
+ .setupTableSource(
+ SourceTestStep.newBuilder("lines")
+ .addSchema("order_id INT PRIMARY KEY NOT
ENFORCED", "status STRING")
+ .addOption("changelog-mode", "I,UA,D")
+ .addOption("source.produces-delete-by-key",
"true")
+ .producedBeforeRestore(
+ Row.ofKind(RowKind.INSERT, 1, "new"),
+ Row.ofKind(RowKind.INSERT, 2, "paid"))
+ .producedAfterRestore(
+ Row.ofKind(RowKind.UPDATE_AFTER, 1,
"paid"),
+ Row.ofKind(RowKind.DELETE, 1, null))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("MySink")
+ .addSchema(
+ "order_id INT PRIMARY KEY NOT
ENFORCED",
+ "excluded_status STRING")
+ .addOption("changelog-mode", "I,UA,D")
+ .addOption("sink.supports-delete-by-key",
"true")
+ .consumedBeforeRestore("+I[1, cancelled]",
"+I[2, cancelled]")
+ .consumedAfterRestore("-D[2, cancelled]",
"-D[1, cancelled]")
+ .build())
+ .runSql(
+ "INSERT INTO MySink SELECT * FROM orders o WHERE
EXISTS ("
+ + "SELECT 1 FROM lines l WHERE l.order_id =
o.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 813d85e45c3..517c5b8c8a8 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
@@ -52,6 +52,8 @@ public class JoinRestoreTest extends RestoreTestBase {
JoinTestPrograms.SEMI_ANTI_JOIN_WITH_LITERAL_AGG,
JoinTestPrograms.LEFT_JOIN_UPSERT_INPUT_NON_EQUI,
JoinTestPrograms.LEFT_JOIN_UPSERT_INPUT_NON_EQUI_PRE_2_4,
- JoinTestPrograms.INNER_JOIN_KEY_ONLY_DELETES_NON_EQUI);
+ JoinTestPrograms.INNER_JOIN_KEY_ONLY_DELETES_NON_EQUI,
+ JoinTestPrograms.SEMI_JOIN_UPSERT_INPUTS_NON_EQUI,
+ JoinTestPrograms.SEMI_JOIN_UPSERT_INPUTS_NON_EQUI_PRE_2_4);
}
}
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-semi-join-upsert-inputs-non-equi-pre-2-4/plan/join-semi-join-upsert-inputs-non-equi-pre-2-4.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-semi-join-upsert-inputs-non-equi-pre-2-4/plan/join-semi-join-upsert-inputs-non-equi-pre-2-4.json
new file mode 100644
index 00000000000..e1dcdd79b3a
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-semi-join-upsert-inputs-non-equi-pre-2-4/plan/join-semi-join-upsert-inputs-non-equi-pre-2-4.json
@@ -0,0 +1,211 @@
+{
+ "flinkVersion" : "2.3",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "stream-exec-table-source-scan_2",
+ "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" : 2,
+ "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" : 3,
+ "type" : "stream-exec-table-source-scan_2",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`lines`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "order_id",
+ "dataType" : "INT NOT NULL"
+ }, {
+ "name" : "status",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "primaryKey" : {
+ "name" : "PK_order_id",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "order_id" ]
+ }
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`order_id` INT NOT NULL, `status`
VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, lines]], fields=[order_id, status])"
+ }, {
+ "id" : 4,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` INT NOT NULL, `status`
VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[order_id]])"
+ }, {
+ "id" : 5,
+ "type" : "stream-exec-join_1",
+ "joinSpec" : {
+ "joinType" : "SEMI",
+ "leftKeys" : [ 0 ],
+ "rightKeys" : [ 0 ],
+ "filterNulls" : [ true ],
+ "nonEquiCondition" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$<>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "BOOLEAN"
+ }
+ },
+ "leftUpsertKeys" : [ [ 0 ] ],
+ "rightUpsertKeys" : [ [ 0 ] ],
+ "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)>",
+ "description" : "Join(joinType=[LeftSemiJoin], where=[((order_id0 =
order_id) AND (status <> excluded_status))], select=[order_id,
excluded_status], leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])"
+ }, {
+ "id" : 6,
+ "type" : "stream-exec-sink_2",
+ "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" : "excluded_status",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "primaryKey" : {
+ "name" : "PK_order_id",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "order_id" ]
+ }
+ }
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_AFTER", "~DELETE" ],
+ "upsertMaterializeStrategy" : "ADAPTIVE",
+ "inputUpsertKey" : [ 0 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` INT NOT NULL, `excluded_status`
VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.MySink],
fields=[order_id, excluded_status])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 3,
+ "target" : 4,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 5,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 4,
+ "target" : 5,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 5,
+ "target" : 6,
+ "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-semi-join-upsert-inputs-non-equi-pre-2-4/savepoint/_metadata
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-semi-join-upsert-inputs-non-equi-pre-2-4/savepoint/_metadata
new file mode 100644
index 00000000000..cdd3061de08
Binary files /dev/null and
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-semi-join-upsert-inputs-non-equi-pre-2-4/savepoint/_metadata
differ
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-semi-join-upsert-inputs-non-equi/plan/join-semi-join-upsert-inputs-non-equi.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-semi-join-upsert-inputs-non-equi/plan/join-semi-join-upsert-inputs-non-equi.json
new file mode 100644
index 00000000000..24e34cee0ab
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-semi-join-upsert-inputs-non-equi/plan/join-semi-join-upsert-inputs-non-equi.json
@@ -0,0 +1,213 @@
+{
+ "flinkVersion" : "2.4",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "stream-exec-table-source-scan_2",
+ "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" : 2,
+ "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" : 3,
+ "type" : "stream-exec-table-source-scan_2",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`lines`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "order_id",
+ "dataType" : "INT NOT NULL"
+ }, {
+ "name" : "status",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "primaryKey" : {
+ "name" : "PK_order_id",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "order_id" ]
+ }
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`order_id` INT NOT NULL, `status`
VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, lines]], fields=[order_id, status])"
+ }, {
+ "id" : 4,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` INT NOT NULL, `status`
VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[order_id]])"
+ }, {
+ "id" : 5,
+ "type" : "stream-exec-join_1",
+ "joinSpec" : {
+ "joinType" : "SEMI",
+ "leftKeys" : [ 0 ],
+ "rightKeys" : [ 0 ],
+ "filterNulls" : [ true ],
+ "nonEquiCondition" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$<>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "BOOLEAN"
+ }
+ },
+ "leftUpsertKeys" : [ [ 0 ] ],
+ "rightUpsertKeys" : [ [ 0 ] ],
+ "leftInputChangelogMode" : [ "INSERT", "UPDATE_AFTER", "~DELETE" ],
+ "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)>",
+ "description" : "Join(joinType=[LeftSemiJoin], where=[((order_id0 =
order_id) AND (status <> excluded_status))], select=[order_id,
excluded_status], leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])"
+ }, {
+ "id" : 6,
+ "type" : "stream-exec-sink_2",
+ "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" : "excluded_status",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "primaryKey" : {
+ "name" : "PK_order_id",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "order_id" ]
+ }
+ }
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_AFTER", "~DELETE" ],
+ "upsertMaterializeStrategy" : "VALUE",
+ "inputUpsertKey" : [ 0 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`order_id` INT NOT NULL, `excluded_status`
VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.MySink],
fields=[order_id, excluded_status])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 3,
+ "target" : 4,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 5,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 4,
+ "target" : 5,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 5,
+ "target" : 6,
+ "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-semi-join-upsert-inputs-non-equi/savepoint/_metadata
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-semi-join-upsert-inputs-non-equi/savepoint/_metadata
new file mode 100644
index 00000000000..ed6c84374d1
Binary files /dev/null and
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-semi-join-upsert-inputs-non-equi/savepoint/_metadata
differ
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/AbstractStreamingJoinOperator.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/AbstractStreamingJoinOperator.java
index 507d06cbe73..9594365830c 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/AbstractStreamingJoinOperator.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/AbstractStreamingJoinOperator.java
@@ -22,6 +22,7 @@ import
org.apache.flink.api.common.functions.DefaultOpenContext;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.TimestampedCollector;
import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
import org.apache.flink.table.runtime.generated.JoinCondition;
@@ -30,6 +31,10 @@ import
org.apache.flink.table.runtime.operators.join.stream.state.JoinRecordStat
import
org.apache.flink.table.runtime.operators.join.stream.state.OuterJoinRecordStateView;
import
org.apache.flink.table.runtime.operators.join.stream.utils.JoinInputSideSpec;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.runtime.util.RuntimeChangelogMode;
+import org.apache.flink.types.RowKind;
+
+import javax.annotation.Nullable;
import java.util.Iterator;
import java.util.function.Function;
@@ -55,6 +60,17 @@ public abstract class AbstractStreamingJoinOperator extends
AbstractStreamOperat
protected final JoinInputSideSpec leftInputSideSpec;
protected final JoinInputSideSpec rightInputSideSpec;
+ // whether the input may update a record without retracting it first
(upsert), so that the
+ // 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;
+
private final boolean[] filterNullKeys;
protected final long leftStateRetentionTime;
@@ -69,6 +85,9 @@ public abstract class AbstractStreamingJoinOperator extends
AbstractStreamOperat
GeneratedJoinCondition generatedJoinCondition,
JoinInputSideSpec leftInputSideSpec,
JoinInputSideSpec rightInputSideSpec,
+ @Nullable RuntimeChangelogMode leftInputChangelogMode,
+ @Nullable RuntimeChangelogMode rightInputChangelogMode,
+ boolean hasNonEquiCondition,
boolean[] filterNullKeys,
long leftStateRetentionTime,
long rightStateRetentionTime) {
@@ -77,6 +96,11 @@ public abstract class AbstractStreamingJoinOperator extends
AbstractStreamOperat
this.generatedJoinCondition = generatedJoinCondition;
this.leftInputSideSpec = leftInputSideSpec;
this.rightInputSideSpec = rightInputSideSpec;
+ this.leftIsUpsertOrUnknown = isUpsertOrUnknown(leftInputChangelogMode);
+ this.rightIsUpsertOrUnknown =
isUpsertOrUnknown(rightInputChangelogMode);
+ this.leftHasKeyOnlyDeletesOrUnknown =
hasKeyOnlyDeletesOrUnknown(leftInputChangelogMode);
+ this.rightHasKeyOnlyDeletesOrUnknown =
hasKeyOnlyDeletesOrUnknown(rightInputChangelogMode);
+ this.hasNonEquiCondition = hasNonEquiCondition;
this.leftStateRetentionTime = leftStateRetentionTime;
this.rightStateRetentionTime = rightStateRetentionTime;
this.filterNullKeys = filterNullKeys;
@@ -102,6 +126,90 @@ public abstract class AbstractStreamingJoinOperator
extends AbstractStreamOperat
}
}
+ /**
+ * Returns the stored record that the given record replaces, or null if
there is none. An upsert
+ * input updates a record by sending only the new version, without
retracting the old one.
+ *
+ * <p>The lookup only happens if it can change the result.
+ */
+ protected @Nullable RowData replacedRecord(
+ RowData record, JoinRecordStateView stateView, boolean isLeft,
boolean otherIsOuter)
+ throws Exception {
+ return needsReplacedRecord(isLeft, otherIsOuter) ?
stateView.getRecord(record) : null;
+ }
+
+ /**
+ * Returns whether looking up the record that an input record replaces can
change the result.
+ */
+ protected boolean needsReplacedRecord(boolean isLeft, boolean
otherIsOuter) {
+ final boolean inputIsUpsertOrUnknown =
+ isLeft ? leftIsUpsertOrUnknown : rightIsUpsertOrUnknown;
+ // with at most one record per join key, the count of the other side
is right without it
+ final boolean mayCountTwice = otherIsOuter &&
!joinKeyContainsUniqueKey(isLeft);
+ final boolean mayMatchDifferently = hasNonEquiCondition;
+ return inputIsUpsertOrUnknown && (mayCountTwice ||
mayMatchDifferently);
+ }
+
+ /**
+ * 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.
+ */
+ protected RowData retractedRecord(
+ RowData retraction,
+ RowKind retractionKind,
+ JoinRecordStateView stateView,
+ boolean isLeft)
+ throws Exception {
+ 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;
+ }
+
+ protected boolean matches(RowData inputRecord, RowData otherRecord,
boolean inputIsLeft) {
+ return inputIsLeft
+ ? joinCondition.apply(inputRecord, otherRecord)
+ : joinCondition.apply(otherRecord, inputRecord);
+ }
+
+ protected boolean joinKeyContainsUniqueKey(boolean isLeft) {
+ return (isLeft ? leftInputSideSpec :
rightInputSideSpec).joinKeyContainsUniqueKey();
+ }
+
+ private boolean hasUniqueKey(boolean isLeft) {
+ return (isLeft ? leftInputSideSpec :
rightInputSideSpec).hasUniqueKey();
+ }
+
+ /** Returns whether the input may update a record without retracting it
first. */
+ private static boolean isUpsertOrUnknown(@Nullable RuntimeChangelogMode
inputChangelogMode) {
+ // compiled plans before Flink 2.4 do not contain the changelog mode,
and handling
+ // their inputs as upsert is always correct
+ if (inputChangelogMode == null) {
+ return true;
+ }
+ final ChangelogMode changelogMode = inputChangelogMode.deserialize();
+ return changelogMode.contains(RowKind.UPDATE_AFTER)
+ && !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();
+ }
+
/** Wrap the passed iterator with filter and mapper. */
private static <I, O> Iterator<O> collect(
final Iterator<I> delegate, Function<I, Boolean> matcher,
Function<I, O> mapper) {
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 16b879ee9e3..00c4f9395e2 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
@@ -19,7 +19,6 @@
package org.apache.flink.table.runtime.operators.join.stream;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.util.RowDataUtil;
@@ -47,20 +46,6 @@ public class StreamingJoinOperator extends
AbstractStreamingJoinOperator {
protected final boolean leftIsOuter;
// whether right side is outer side, e.g. right is outer but left is not
when RIGHT OUTER JOIN
protected final boolean rightIsOuter;
- // changelog mode of the left input, null for compiled plans before Flink
2.4
- @Nullable protected final RuntimeChangelogMode leftInputChangelogMode;
- // changelog mode of the right input, null for compiled plans before Flink
2.4
- @Nullable protected final RuntimeChangelogMode rightInputChangelogMode;
- // whether the input may update a record without retracting it first
(upsert), so that the
- // 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;
private transient JoinedRowData outRow;
private transient RowData leftNullRow;
@@ -91,18 +76,14 @@ public class StreamingJoinOperator extends
AbstractStreamingJoinOperator {
generatedJoinCondition,
leftInputSideSpec,
rightInputSideSpec,
+ leftInputChangelogMode,
+ rightInputChangelogMode,
+ hasNonEquiCondition,
filterNullKeys,
leftStateRetentionTime,
rightStateRetentionTime);
this.leftIsOuter = leftIsOuter;
this.rightIsOuter = rightIsOuter;
- this.leftInputChangelogMode = leftInputChangelogMode;
- this.rightInputChangelogMode = rightInputChangelogMode;
- this.leftIsUpsertOrUnknown = isUpsertOrUnknown(leftInputChangelogMode);
- this.rightIsUpsertOrUnknown =
isUpsertOrUnknown(rightInputChangelogMode);
- this.leftHasKeyOnlyDeletesOrUnknown =
hasKeyOnlyDeletesOrUnknown(leftInputChangelogMode);
- this.rightHasKeyOnlyDeletesOrUnknown =
hasKeyOnlyDeletesOrUnknown(rightInputChangelogMode);
- this.hasNonEquiCondition = hasNonEquiCondition;
}
@Override
@@ -273,8 +254,11 @@ public class StreamingJoinOperator extends
AbstractStreamingJoinOperator {
}
return;
}
+ // in a mini-batch pair, the retraction already handled the old
version
final RowData replacedRecord =
- replacedRecord(input, inputSideStateView, inputIsLeft,
isSuppress);
+ isSuppress
+ ? null
+ : replacedRecord(input, inputSideStateView,
inputIsLeft, otherIsOuter);
boolean replacedRecordHadNoMatches = false;
if (replacedRecord != null && hasNonEquiCondition) {
replacedRecordHadNoMatches =
@@ -365,9 +349,11 @@ public class StreamingJoinOperator extends
AbstractStreamingJoinOperator {
// skip when there is no matched rows on the other side
}
} else { // input record is retract
+ // the retraction of a mini-batch pair is used as it is
final RowData retractedRecord =
- retractedRecord(
- input, inputRowKind, inputSideStateView,
inputIsLeft, isSuppress);
+ isSuppress
+ ? input
+ : retractedRecord(input, inputRowKind,
inputSideStateView, inputIsLeft);
// state.retract(record)
if (!isSuppress) {
inputSideStateView.retractRecord(retractedRecord);
@@ -414,56 +400,19 @@ public class StreamingJoinOperator extends
AbstractStreamingJoinOperator {
}
/**
- * Returns the stored record that the given record replaces, or null if
there is none. An upsert
- * input updates a record by sending only the new version, without
retracting the old one.
- *
- * <p>The lookup only happens if it can change the result. In a mini-batch
pair, the retraction
- * already handled the old version.
- */
- private @Nullable RowData replacedRecord(
- RowData record, JoinRecordStateView stateView, boolean isLeft,
boolean isSuppress)
- throws Exception {
- final boolean inputIsUpsertOrUnknown =
- isLeft ? leftIsUpsertOrUnknown : rightIsUpsertOrUnknown;
- final boolean otherIsOuter = isLeft ? rightIsOuter : leftIsOuter;
- // with at most one record per join key, the count of the other side
is right without it
- final boolean mayCountTwice = otherIsOuter &&
!joinKeyContainsUniqueKey(isLeft);
- final boolean mayMatchDifferently = hasNonEquiCondition;
- if (isSuppress || !inputIsUpsertOrUnknown || !(mayCountTwice ||
mayMatchDifferently)) {
- return null;
- }
- 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.
+ * Returns whether the other record gains a match, so that its number of
associations increases.
+ * It does not if the replaced record matched it already.
*/
- private RowData retractedRecord(
- RowData retraction,
- RowKind retractionKind,
- JoinRecordStateView stateView,
- boolean isLeft,
- boolean isSuppress)
- throws Exception {
- if (isSuppress) {
- return retraction;
+ private boolean isNewMatch(
+ OuterRecord other, @Nullable RowData replacedRecord, boolean
inputIsLeft) {
+ if (other.numOfAssociations == 0) {
+ return true;
}
- final boolean inputHasKeyOnlyDeletesOrUnknown =
- isLeft ? leftHasKeyOnlyDeletesOrUnknown :
rightHasKeyOnlyDeletesOrUnknown;
- final boolean mayBeKeyOnlyDelete =
- retractionKind == RowKind.DELETE
- && inputHasKeyOnlyDeletesOrUnknown
- && hasUniqueKey(isLeft);
- if (!mayBeKeyOnlyDelete || !hasNonEquiCondition) {
- return retraction;
+ // with at most one record per join key, the existing match is the
replaced record
+ if (joinKeyContainsUniqueKey(inputIsLeft)) {
+ return false;
}
- final RowData storedRecord = stateView.getRecord(retraction);
- return storedRecord != null ? storedRecord : retraction;
+ return replacedRecord == null || !matches(replacedRecord,
other.record, inputIsLeft);
}
/**
@@ -506,22 +455,6 @@ public class StreamingJoinOperator extends
AbstractStreamingJoinOperator {
return replacedRecordHadMatches;
}
- /**
- * Returns whether the other record gains a match, so that its number of
associations increases.
- * It does not if the replaced record matched it already.
- */
- private boolean isNewMatch(
- OuterRecord other, @Nullable RowData replacedRecord, boolean
inputIsLeft) {
- if (other.numOfAssociations == 0) {
- return true;
- }
- // with at most one record per join key, the existing match is the
replaced record
- if (joinKeyContainsUniqueKey(inputIsLeft)) {
- return false;
- }
- return replacedRecord == null || !matches(replacedRecord,
other.record, inputIsLeft);
- }
-
/**
* Returns whether the paired record of a -U/+U pair cancels out the null
padding change of the
* other record, which is the case if both records of the pair match it.
@@ -532,40 +465,6 @@ public class StreamingJoinOperator extends
AbstractStreamingJoinOperator {
&& (!hasNonEquiCondition || matches(pairedRecord, other,
inputIsLeft));
}
- /** Returns whether the input may update a record without retracting it
first. */
- private static boolean isUpsertOrUnknown(@Nullable RuntimeChangelogMode
inputChangelogMode) {
- // compiled plans before Flink 2.4 do not contain the changelog mode,
and handling
- // their inputs as upsert is always correct
- if (inputChangelogMode == null) {
- return true;
- }
- final ChangelogMode changelogMode = inputChangelogMode.deserialize();
- return changelogMode.contains(RowKind.UPDATE_AFTER)
- && !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();
- }
-
- private boolean matches(RowData inputRecord, RowData otherRecord, boolean
inputIsLeft) {
- return inputIsLeft
- ? joinCondition.apply(inputRecord, otherRecord)
- : joinCondition.apply(otherRecord, inputRecord);
- }
-
//
-------------------------------------------------------------------------------------
private void output(RowData inputRow, RowData otherRow, boolean
inputIsLeft) {
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 cc770771bb4..8d3c6482598 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
@@ -18,6 +18,7 @@
package org.apache.flink.table.runtime.operators.join.stream;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.util.RowDataUtil;
@@ -28,10 +29,13 @@ import
org.apache.flink.table.runtime.operators.join.stream.state.OuterJoinRecor
import
org.apache.flink.table.runtime.operators.join.stream.state.OuterJoinRecordStateViews;
import
org.apache.flink.table.runtime.operators.join.stream.utils.JoinInputSideSpec;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.runtime.util.RuntimeChangelogMode;
import org.apache.flink.types.RowKind;
import org.apache.flink.shaded.guava33.com.google.common.collect.Iterators;
+import javax.annotation.Nullable;
+
import java.util.Iterator;
/** Streaming unbounded Join operator which supports SEMI/ANTI JOIN. */
@@ -54,6 +58,9 @@ public class StreamingSemiAntiJoinOperator extends
AbstractStreamingJoinOperator
GeneratedJoinCondition generatedJoinCondition,
JoinInputSideSpec leftInputSideSpec,
JoinInputSideSpec rightInputSideSpec,
+ @Nullable RuntimeChangelogMode leftInputChangelogMode,
+ @Nullable RuntimeChangelogMode rightInputChangelogMode,
+ boolean hasNonEquiCondition,
boolean[] filterNullKeys,
long leftStateRetentionTime,
long rightStateRetentionTIme) {
@@ -63,6 +70,9 @@ public class StreamingSemiAntiJoinOperator extends
AbstractStreamingJoinOperator
generatedJoinCondition,
leftInputSideSpec,
rightInputSideSpec,
+ leftInputChangelogMode,
+ rightInputChangelogMode,
+ hasNonEquiCondition,
filterNullKeys,
leftStateRetentionTime,
rightStateRetentionTIme);
@@ -97,6 +107,10 @@ public class StreamingSemiAntiJoinOperator extends
AbstractStreamingJoinOperator
* <p>Following is the pseudo code to describe the core logic of this
method.
*
* <pre>
+ * 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 any
+ * if record replaces a stored record (upsert) and the condition is
non-equi
+ * if the replaced record is sent (by its matched num) but the record is
not, send -D[replaced]
* if there is no matched rows on the other side
* if anti join, send input record
* if there are matched rows on the other side
@@ -109,29 +123,40 @@ public class StreamingSemiAntiJoinOperator extends
AbstractStreamingJoinOperator
@Override
public void processElement1(StreamRecord<RowData> element) throws
Exception {
RowData input = element.getValue();
- Iterator<OuterRecord> associatedRecords =
- AbstractStreamingJoinOperator.iterator(
- input, true, rightRecordStateView, joinCondition);
- if (!associatedRecords.hasNext()) {
- if (isAntiJoin) {
- collector.collect(input);
- }
- } else { // there are matched rows on the other side
- if (!isAntiJoin) {
- collector.collect(input);
+ boolean isAccumulateMsg = RowDataUtil.isAccumulateMsg(input);
+ RowKind inputRowKind = input.getRowKind();
+ input.setRowKind(RowKind.INSERT); // erase RowKind for later state
updating
+
+ if (isAccumulateMsg) {
+ final int numOfAssociations =
+ Iterators.size(
+ AbstractStreamingJoinOperator.iterator(
+ input, true, rightRecordStateView,
joinCondition));
+ if (isSent(numOfAssociations)) {
+ collect(input, inputRowKind);
+ } else if (needsReplacedRecord(true, false)) {
+ final Tuple2<RowData, Integer> replaced =
+
leftRecordStateView.getRecordAndNumOfAssociations(input);
+ if (replaced != null && isSent(replaced.f1)) {
+ // send -D[replaced]
+ collect(replaced.f0, RowKind.DELETE);
+ }
}
- }
- if (RowDataUtil.isAccumulateMsg(input)) {
- // erase RowKind for state updating
- input.setRowKind(RowKind.INSERT);
// do not store a record that can never match
if (!joinCondition.currentKeyCannotMatch()) {
- leftRecordStateView.addRecord(input,
Iterators.size(associatedRecords));
+ leftRecordStateView.addRecord(input, numOfAssociations);
+ }
+ } else {
+ final RowData retractedRecord =
+ retractedRecord(input, inputRowKind, leftRecordStateView,
true);
+ final boolean hasMatches =
+ AbstractStreamingJoinOperator.iterator(
+ retractedRecord, true,
rightRecordStateView, joinCondition)
+ .hasNext();
+ if (isAntiJoin != hasMatches) {
+ collect(retractedRecord, inputRowKind);
}
- } else { // input is retract
- // erase RowKind for state updating
- input.setRowKind(RowKind.INSERT);
- leftRecordStateView.retractRecord(input);
+ leftRecordStateView.retractRecord(retractedRecord);
}
}
@@ -147,27 +172,34 @@ public class StreamingSemiAntiJoinOperator extends
AbstractStreamingJoinOperator
* <pre>
* if input record is accumulate
* | 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
- * | | if anti join, send -D[other]s
- * | | if semi join, send +I/+U[other]s (using input RowKind)
- * | | if the matched num in the matched rows > 0, skip
- * | | otherState.update(other, old+1)
- * | endif
+ * | for each other on the other side
+ * | | if record matches other
+ * | | | if the matched num of other <= 0
+ * | | | if anti join, send -D[other]
+ * | | | if semi join, send +I/+U[other] (using input RowKind)
+ * | | | otherState.update(other, 1)
+ * | | | else if the record replaces a stored record (upsert) that did not
match other
+ * | | | otherState.update(other, old+1)
+ * | | else if the record replaces a stored record (upsert) that matched
other
+ * | | | remove the match of other
+ * | endfor
* endif
* if input record is retract
- * | state.retract(record)
- * | 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, this should never
happen!
- * | | if the matched num in the matched rows == 1
- * | | if semi join, send -D/-U[other] (using input RowKind)
- * | | if anti join, send +I[other]
- * | | if the matched num in the matched rows > 1, skip
- * | | otherState.update(other, old-1)
+ * | 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 any
* | endif
+ * | state.retract(record)
+ * | for each other on the other side that record matches
+ * | | remove the match of other
+ * | endfor
* endif
+ *
+ * remove the match of other:
+ * if the matched num of other <= 0, skip, as the record was not counted
+ * if the matched num of other == 1
+ * if semi join, send -D/-U[other] (using input RowKind)
+ * if anti join, send +I[other]
+ * otherState.update(other, old-1)
* </pre>
*/
@Override
@@ -177,52 +209,75 @@ public class StreamingSemiAntiJoinOperator extends
AbstractStreamingJoinOperator
RowKind inputRowKind = input.getRowKind();
input.setRowKind(RowKind.INSERT); // erase RowKind for later state
updating
- Iterator<OuterRecord> associatedRecords =
- AbstractStreamingJoinOperator.iterator(
- input, false, leftRecordStateView, joinCondition);
if (isAccumulateMsg) { // record is accumulate
+ final RowData replacedRecord = replacedRecord(input,
rightRecordStateView, false, true);
// 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;
- if (outerRecord.numOfAssociations == 0) {
- if (isAntiJoin) {
- // send -D[other]
- other.setRowKind(RowKind.DELETE);
- } else {
- // send +I/+U[other] (using input RowKind)
- other.setRowKind(inputRowKind);
+ // a single pass for the matches that the record gains and the
replaced record loses
+ for (Tuple2<RowData, Integer> left :
+ leftRecordStateView.getRecordsAndNumOfAssociations()) {
+ final RowData other = left.f0;
+ final int numOfAssociations = left.f1;
+ final boolean replacedRecordMatches =
+ replacedRecord != null && matches(replacedRecord,
other, false);
+ if (matches(input, other, false)) {
+ if (numOfAssociations <= 0) {
+ // send -D[other] if anti join, +I/+U[other] (using
input RowKind) if semi
+ collect(other, isAntiJoin ? RowKind.DELETE :
inputRowKind);
+ leftRecordStateView.updateNumOfAssociations(other, 1);
+ } else if (!replacedRecordMatches &&
!joinKeyContainsUniqueKey(false)) {
+ // with at most one record per join key, the existing
match is the
+ // replaced record
+ leftRecordStateView.updateNumOfAssociations(other,
numOfAssociations + 1);
}
- collector.collect(other);
- // set header back to INSERT, because we will update the
other row to state
- other.setRowKind(RowKind.INSERT);
- } // ignore when number > 0
- leftRecordStateView.updateNumOfAssociations(
- other, outerRecord.numOfAssociations + 1);
+ } else if (replacedRecordMatches) {
+ removeMatch(other, numOfAssociations, RowKind.DELETE);
+ }
}
} else { // retract input
- rightRecordStateView.retractRecord(input);
+ final RowData retractedRecord =
+ retractedRecord(input, inputRowKind, rightRecordStateView,
false);
+ rightRecordStateView.retractRecord(retractedRecord);
+ Iterator<OuterRecord> associatedRecords =
+ AbstractStreamingJoinOperator.iterator(
+ retractedRecord, false, leftRecordStateView,
joinCondition);
while (associatedRecords.hasNext()) {
- OuterRecord outerRecord = associatedRecords.next();
- RowData other = outerRecord.record;
- if (outerRecord.numOfAssociations == 1) {
- if (!isAntiJoin) {
- // send -D/-U[other] (using input RowKind)
- other.setRowKind(inputRowKind);
- } else {
- // send +I[other]
- other.setRowKind(RowKind.INSERT);
- }
- collector.collect(other);
- // set RowKind back, because we will update the other row
to state
- other.setRowKind(RowKind.INSERT);
- } // ignore when number > 0
- leftRecordStateView.updateNumOfAssociations(
- other, outerRecord.numOfAssociations - 1);
+ final OuterRecord outerRecord = associatedRecords.next();
+ removeMatch(outerRecord.record, outerRecord.numOfAssociations,
inputRowKind);
}
}
}
+
+ /**
+ * Removes a match of the left record. If it was the last one, a semi join
retracts the record
+ * with the given RowKind and an anti join sends it.
+ */
+ private void removeMatch(RowData other, int numOfAssociations, RowKind
retractionKind)
+ throws Exception {
+ if (numOfAssociations <= 0) {
+ // the retracted record was not counted, e.g. it had expired when
the left record was
+ // stored
+ return;
+ }
+ if (numOfAssociations == 1) {
+ // send -D/-U[other] if semi join, +I[other] if anti join
+ collect(other, isAntiJoin ? RowKind.INSERT : retractionKind);
+ } // ignore when number > 1
+ leftRecordStateView.updateNumOfAssociations(other, numOfAssociations -
1);
+ }
+
+ /** Returns whether a left record with the given number of matches is
sent. */
+ private boolean isSent(int numOfAssociations) {
+ return isAntiJoin != (numOfAssociations > 0);
+ }
+
+ /** Sends the left record with the given RowKind. */
+ private void collect(RowData left, RowKind rowKind) {
+ left.setRowKind(rowKind);
+ collector.collect(left);
+ // set RowKind back, because the record may be the one in state
+ left.setRowKind(RowKind.INSERT);
+ }
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/OuterJoinRecordStateView.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/OuterJoinRecordStateView.java
index e9171e2cc19..e2547cf8021 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/OuterJoinRecordStateView.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/OuterJoinRecordStateView.java
@@ -21,6 +21,8 @@ package
org.apache.flink.table.runtime.operators.join.stream.state;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.data.RowData;
+import javax.annotation.Nullable;
+
/**
* A {@link OuterJoinRecordStateView} is an extension to {@link
JoinRecordStateView}. The {@link
* OuterJoinRecordStateView} is used to store records for the outer input side
of the Join, e.g. the
@@ -54,4 +56,11 @@ public interface OuterJoinRecordStateView extends
JoinRecordStateView {
* Gets all the records and number of associations under the current
context (i.e. join key).
*/
Iterable<Tuple2<RowData, Integer>> getRecordsAndNumOfAssociations() throws
Exception;
+
+ /**
+ * Returns the record with the same unique key as the given record and its
number of
+ * associations, or null if there is none. Always null if the input side
has no unique key.
+ */
+ @Nullable
+ Tuple2<RowData, Integer> getRecordAndNumOfAssociations(RowData record)
throws Exception;
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/OuterJoinRecordStateViews.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/OuterJoinRecordStateViews.java
index c077933d57b..a792c687eac 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/OuterJoinRecordStateViews.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/OuterJoinRecordStateViews.java
@@ -124,6 +124,12 @@ public final class OuterJoinRecordStateViews {
return tuple == null ? null : tuple.f0;
}
+ @Override
+ public Tuple2<RowData, Integer> getRecordAndNumOfAssociations(RowData
record)
+ throws Exception {
+ return recordState.value();
+ }
+
@Override
public Iterable<RowData> getRecords() throws Exception {
Tuple2<RowData, Integer> tuple = recordState.value();
@@ -204,6 +210,12 @@ public final class OuterJoinRecordStateViews {
return tuple == null ? null : tuple.f0;
}
+ @Override
+ public Tuple2<RowData, Integer> getRecordAndNumOfAssociations(RowData
record)
+ throws Exception {
+ return recordState.get(uniqueKeySelector.getKey(record));
+ }
+
@Override
public Iterable<RowData> getRecords() throws Exception {
return new RecordsIterable(getRecordsAndNumOfAssociations());
@@ -285,6 +297,11 @@ public final class OuterJoinRecordStateViews {
return null;
}
+ @Override
+ public Tuple2<RowData, Integer> getRecordAndNumOfAssociations(RowData
record) {
+ return null;
+ }
+
@Override
public Iterable<RowData> getRecords() throws Exception {
return new RecordsIterable(getRecordsAndNumOfAssociations());
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 3a5b3083b69..c5b9b64ee7e 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
@@ -25,13 +25,11 @@ 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;
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;
import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
import
org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
@@ -1374,20 +1372,6 @@ class StreamingJoinOperatorTest extends
StreamingJoinOperatorTestBase {
.isEqualTo(hasUniqueKey ? right : null);
}
- /**
- * The unique key (order_id, line_order_id) contains the join key
(line_order_id), which the
- * planner requires for inputs without UPDATE_BEFORE, so a replacing
record has the same join
- * key.
- */
- private JoinInputSideSpec leftHasUniqueKeySpec() {
- final RowDataKeySelector uniqueKeySelector =
- HandwrittenSelectorUtil.getRowDataSelector(
- new int[] {0, 1},
- leftTypeInfo.toRowType().getChildren().toArray(new
LogicalType[0]));
- return JoinInputSideSpec.withUniqueKey(
- uniqueKeySelector.getProducedType(), uniqueKeySelector);
- }
-
private static final String ADDRESS = "3 Bellevue Drive, Pottstown, PA
19464";
private static final String NEW_ADDRESS = "68 Manor Station Street,
Honolulu, HI 96815";
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 71f00ea758a..88c73101cd4 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
@@ -127,6 +127,30 @@ public abstract class StreamingJoinOperatorTestBase {
return new GeneratedJoinCondition("KnownValuesNonEquiCondition", code,
new Object[0]);
}
+ /**
+ * The unique key (order_id, line_order_id) contains the join key
(line_order_id), which the
+ * planner requires for inputs without UPDATE_BEFORE, so a replacing
record has the same join
+ * key.
+ */
+ protected JoinInputSideSpec leftHasUniqueKeySpec() {
+ final RowDataKeySelector uniqueKeySelector =
+ HandwrittenSelectorUtil.getRowDataSelector(
+ new int[] {0, 1},
+ leftTypeInfo.toRowType().getChildren().toArray(new
LogicalType[0]));
+ return JoinInputSideSpec.withUniqueKey(
+ uniqueKeySelector.getProducedType(), uniqueKeySelector);
+ }
+
+ /** The unique key (line_order_id, line_order_ship_mode) contains the join
key. */
+ protected JoinInputSideSpec rightHasUniqueKeySpec() {
+ final RowDataKeySelector uniqueKeySelector =
+ HandwrittenSelectorUtil.getRowDataSelector(
+ new int[] {0, 1},
+ rightTypeInfo.toRowType().getChildren().toArray(new
LogicalType[0]));
+ return JoinInputSideSpec.withUniqueKey(
+ uniqueKeySelector.getProducedType(), uniqueKeySelector);
+ }
+
protected RowDataHarnessAssertor assertor;
protected KeyedTwoInputStreamOperatorTestHarness<RowData, RowData,
RowData, RowData>
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 f9d82da1616..36a2a1b2360 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
@@ -18,12 +18,22 @@
package org.apache.flink.table.runtime.operators.join.stream;
+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.operators.join.stream.state.OuterJoinRecordStateViews;
+import
org.apache.flink.table.runtime.operators.join.stream.utils.JoinInputSideSpec;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.types.RowKind;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import javax.annotation.Nullable;
import java.util.function.Predicate;
@@ -45,6 +55,9 @@ class StreamingSemiAntiJoinOperatorTest extends
StreamingJoinOperatorTestBase {
joinCondition,
leftInputSpec,
rightInputSpec,
+ null,
+ null,
+ false,
new boolean[] {true},
ttl[0],
ttl[1]);
@@ -88,10 +101,6 @@ class StreamingSemiAntiJoinOperatorTest extends
StreamingJoinOperatorTestBase {
assertor.shouldEmitNothing(testHarness);
testHarness.processElement2(deleteRecord("LineOrd#2", "TRUCK"));
- assertor.shouldEmitNothing(testHarness);
-
- // numOfAssociations is reduced to 1, retract the record
- testHarness.processElement2(deleteRecord("LineOrd#2", "AIR"));
assertor.shouldEmit(
testHarness,
rowOfKind(
@@ -141,10 +150,6 @@ class StreamingSemiAntiJoinOperatorTest extends
StreamingJoinOperatorTestBase {
assertor.shouldEmitNothing(testHarness);
testHarness.processElement2(deleteRecord("LineOrd#2", "TRUCK"));
- assertor.shouldEmitNothing(testHarness);
-
- // numOfAssociations is reduced to 1, retract the record
- testHarness.processElement2(deleteRecord("LineOrd#2", "AIR"));
assertor.shouldEmit(
testHarness,
rowOfKind(
@@ -327,6 +332,347 @@ class StreamingSemiAntiJoinOperatorTest extends
StreamingJoinOperatorTestBase {
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 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, leftInputSpec, rightInputSpec, joinCondition,
false, 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,
+ leftInputSpec,
+ rightInputSpec,
+ knownValuesNonEquiCondition(),
+ true,
+ 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)));
+ }
+
+ /**
+ * Left rows whose unique key is wider than the join key are updated and
deleted by their stored
+ * version.
+ */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testLeftUpsertsWithUniqueKeyWiderThanJoinKey(boolean isAntiJoin)
throws Exception {
+ useSemiAntiJoinOperator(
+ isAntiJoin,
+ leftHasUniqueKeySpec(),
+ rightInputSpec,
+ knownValuesNonEquiCondition(),
+ true,
+ 0L,
+ 0L);
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.processElement1(insertRecord("Ord#2", "LineOrd#1",
ADDRESS));
+ testHarness.getOutput().clear();
+
+ 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(deleteRecord("Ord#2", "LineOrd#1", null));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#2", "LineOrd#1",
ADDRESS),
+ null));
+
+ testHarness.processElement1(deleteRecord("Ord#1", "LineOrd#1", null));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ null,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
UNKNOWN)));
+ }
+
+ /** A negative number of matches, as state written before Flink 2.4 may
contain, recovers. */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testRightInsertMatchesLeftRecordWithNegativeCount(boolean isAntiJoin)
throws Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightInputSpec, joinCondition,
false);
+ final RowData order = rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS);
+ testHarness.processElement1(new StreamRecord<>(order));
+ setNumOfAssociations(order, -1);
+ testHarness.getOutput().clear();
+
+ 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(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)));
+ }
+
+ private void useSemiAntiJoinOperator(
+ boolean isAntiJoin,
+ JoinInputSideSpec rightSpec,
+ GeneratedJoinCondition condition,
+ boolean hasNonEquiCondition)
+ throws Exception {
+ useSemiAntiJoinOperator(
+ isAntiJoin, leftInputSpec, rightSpec, condition,
hasNonEquiCondition, 0L, 0L);
+ }
+
+ private void useSemiAntiJoinOperator(
+ boolean isAntiJoin,
+ JoinInputSideSpec leftSpec,
+ JoinInputSideSpec rightSpec,
+ GeneratedJoinCondition condition,
+ boolean hasNonEquiCondition,
+ long leftStateRetentionTime,
+ long rightStateRetentionTime)
+ throws Exception {
+ testHarness.close();
+ testHarness =
+ new KeyedTwoInputStreamOperatorTestHarness<>(
+ new StreamingSemiAntiJoinOperator(
+ isAntiJoin,
+ leftTypeInfo,
+ rightTypeInfo,
+ condition,
+ leftSpec,
+ rightSpec,
+ UPSERT_WITH_KEY_ONLY_DELETES,
+ UPSERT_WITH_KEY_ONLY_DELETES,
+ hasNonEquiCondition,
+ new boolean[] {true},
+ leftStateRetentionTime,
+ rightStateRetentionTime),
+ leftKeySelector,
+ rightKeySelector,
+ joinKeyTypeInfo);
+ testHarness.open();
+ }
+
+ /** Overwrites the stored number of matches of the left row, e.g. as old
state may contain. */
+ private void setNumOfAssociations(RowData record, int numOfAssociations)
throws Exception {
+ final StreamingSemiAntiJoinOperator operator =
+ (StreamingSemiAntiJoinOperator) testHarness.getOperator();
+ operator.setCurrentKey(leftKeySelector.getKey(record));
+ OuterJoinRecordStateViews.create(
+ operator.getRuntimeContext(),
+ AbstractStreamingJoinOperator.LEFT_RECORDS_STATE_NAME,
+ leftInputSpec,
+ leftTypeInfo,
+ 0L)
+ .updateNumOfAssociations(record, numOfAssociations);
+ }
+
+ /** Returns the row that the semi or the anti join emits, if any. */
+ private static RowData[] emitted(
+ boolean isAntiJoin, @Nullable RowData semiJoinRow, @Nullable
RowData antiJoinRow) {
+ final RowData row = isAntiJoin ? antiJoinRow : semiJoinRow;
+ return row == null ? new RowData[0] : new RowData[] {row};
+ }
+
+ private static final String ADDRESS = "3 Bellevue Drive, Pottstown, PA
19464";
+ private static final String NEW_ADDRESS = "68 Manor Station Street,
Honolulu, HI 96815";
+
private static final Predicate<String> ANTI_JOIN_CHECKER =
(testDisplayName) -> testDisplayName.contains("Anti");
}
