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 9c11b0bf01472f0f52fabc4024ab3b1af316a075 Author: Gustavo de Morais <[email protected]> AuthorDate: Fri Oct 2 12:47:10 2026 +0200 [FLINK-40841][table] Non-equi joins fix: retract the matches that a replaced record loses Generated-by: Claude Code (Claude Opus 5.5) (cherry picked from commit 0c619470086cbc9e09fdd6fff642086d3c743aa7) --- .../plan/nodes/exec/stream/StreamExecJoin.java | 43 ++++ .../stream/StreamExecProcessTableFunction.java | 2 +- .../nodes/physical/stream/StreamPhysicalJoin.scala | 4 +- .../plan/nodes/exec/common/JoinTestPrograms.java | 67 ++++++ .../plan/nodes/exec/stream/JoinRestoreTest.java | 4 +- ...ert-input-non-equi-without-changelog-modes.json | 252 ++++++++++++++++++++ .../savepoint/_metadata | Bin 0 -> 15561 bytes .../plan/join-left-join-upsert-input-non-equi.json | 254 +++++++++++++++++++++ .../savepoint/_metadata | Bin 0 -> 15561 bytes .../planner/runtime/stream/sql/JoinITCase.scala | 88 +++++-- .../stream/MiniBatchStreamingJoinOperator.java | 45 +++- .../join/stream/StreamingJoinOperator.java | 203 +++++++++++++--- .../join/stream/state/JoinRecordStateView.java | 9 +- .../join/stream/state/JoinRecordStateViews.java | 12 +- .../stream/state/OuterJoinRecordStateViews.java | 15 +- .../process/AbstractProcessTableOperator.java | 1 + .../operators/process/ProcessRowTableOperator.java | 1 + .../operators/process/ProcessSetTableOperator.java | 1 + .../process/ProcessTableOperatorFactory.java | 1 + .../operators/process/RuntimeTableSemantics.java | 1 + .../process => util}/RuntimeChangelogMode.java | 2 +- .../join/stream/StreamingJoinOperatorTest.java | 219 +++++++++++++++++- .../join/stream/StreamingJoinOperatorTestBase.java | 26 +++ .../stream/StreamingMiniBatchJoinOperatorTest.java | 88 ++++++- 24 files changed, 1246 insertions(+), 92 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 29ef49fca2a..684bdc1e10b 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 @@ -25,6 +25,7 @@ import org.apache.flink.configuration.ReadableConfig; import org.apache.flink.streaming.api.operators.TwoInputStreamOperator; import org.apache.flink.streaming.api.transformations.TwoInputTransformation; import org.apache.flink.table.api.config.ExecutionConfigOptions; +import org.apache.flink.table.connector.ChangelogMode; import org.apache.flink.table.data.RowData; import org.apache.flink.table.planner.delegation.PlannerBase; import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge; @@ -49,6 +50,7 @@ import org.apache.flink.table.runtime.operators.join.stream.StreamingSemiAntiJoi 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.typeutils.InternalTypeInfo; +import org.apache.flink.table.runtime.util.RuntimeChangelogMode; import org.apache.flink.table.types.logical.RowType; import org.apache.flink.shaded.guava33.com.google.common.collect.Lists; @@ -88,6 +90,8 @@ public class StreamExecJoin extends ExecNodeBase<RowData> public static final String FIELD_NAME_JOIN_SPEC = "joinSpec"; public static final String FIELD_NAME_LEFT_UPSERT_KEYS = "leftUpsertKeys"; public static final String FIELD_NAME_RIGHT_UPSERT_KEYS = "rightUpsertKeys"; + public static final String FIELD_NAME_LEFT_INPUT_CHANGELOG_MODE = "leftInputChangelogMode"; + public static final String FIELD_NAME_RIGHT_INPUT_CHANGELOG_MODE = "rightInputChangelogMode"; @JsonProperty(FIELD_NAME_JOIN_SPEC) private final JoinSpec joinSpec; @@ -100,6 +104,16 @@ public class StreamExecJoin extends ExecNodeBase<RowData> @JsonInclude(JsonInclude.Include.NON_DEFAULT) private final List<int[]> rightUpsertKeys; + @Nullable + @JsonProperty(FIELD_NAME_LEFT_INPUT_CHANGELOG_MODE) + @JsonInclude(JsonInclude.Include.NON_NULL) + private final ChangelogMode leftInputChangelogMode; + + @Nullable + @JsonProperty(FIELD_NAME_RIGHT_INPUT_CHANGELOG_MODE) + @JsonInclude(JsonInclude.Include.NON_NULL) + private final ChangelogMode rightInputChangelogMode; + @Nullable @JsonProperty(FIELD_NAME_STATE) @JsonInclude(JsonInclude.Include.NON_NULL) @@ -110,6 +124,8 @@ public class StreamExecJoin extends ExecNodeBase<RowData> JoinSpec joinSpec, List<int[]> leftUpsertKeys, List<int[]> rightUpsertKeys, + ChangelogMode leftInputChangelogMode, + ChangelogMode rightInputChangelogMode, InputProperty leftInputProperty, InputProperty rightInputProperty, Map<Integer, Long> stateTtlFromHint, @@ -122,6 +138,8 @@ public class StreamExecJoin extends ExecNodeBase<RowData> joinSpec, leftUpsertKeys, rightUpsertKeys, + leftInputChangelogMode, + rightInputChangelogMode, StateMetadata.getMultiInputOperatorDefaultMeta( stateTtlFromHint, tableConfig, LEFT_STATE_NAME, RIGHT_STATE_NAME), Lists.newArrayList(leftInputProperty, rightInputProperty), @@ -137,6 +155,10 @@ public class StreamExecJoin extends ExecNodeBase<RowData> @JsonProperty(FIELD_NAME_JOIN_SPEC) JoinSpec joinSpec, @JsonProperty(FIELD_NAME_LEFT_UPSERT_KEYS) List<int[]> leftUpsertKeys, @JsonProperty(FIELD_NAME_RIGHT_UPSERT_KEYS) List<int[]> rightUpsertKeys, + @Nullable @JsonProperty(FIELD_NAME_LEFT_INPUT_CHANGELOG_MODE) + ChangelogMode leftInputChangelogMode, + @Nullable @JsonProperty(FIELD_NAME_RIGHT_INPUT_CHANGELOG_MODE) + ChangelogMode rightInputChangelogMode, @Nullable @JsonProperty(FIELD_NAME_STATE) List<StateMetadata> stateMetadataList, @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty> inputProperties, @JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType, @@ -146,6 +168,8 @@ public class StreamExecJoin extends ExecNodeBase<RowData> this.joinSpec = checkNotNull(joinSpec); this.leftUpsertKeys = leftUpsertKeys; this.rightUpsertKeys = rightUpsertKeys; + this.leftInputChangelogMode = leftInputChangelogMode; + this.rightInputChangelogMode = rightInputChangelogMode; this.stateMetadataList = stateMetadataList; } @@ -216,6 +240,11 @@ 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( @@ -227,6 +256,9 @@ public class StreamExecJoin extends ExecNodeBase<RowData> rightInputSpec, leftIsOuter, rightIsOuter, + leftRuntimeChangelogMode, + rightRuntimeChangelogMode, + hasNonEquiCondition, joinSpec.getFilterNulls(), leftStateRetentionTime, rightStateRetentionTime, @@ -255,6 +287,9 @@ public class StreamExecJoin extends ExecNodeBase<RowData> rightInputSpec, leftIsOuter, rightIsOuter, + leftRuntimeChangelogMode, + rightRuntimeChangelogMode, + hasNonEquiCondition, joinSpec.getFilterNulls(), leftStateRetentionTime, rightStateRetentionTime); @@ -284,4 +319,12 @@ public class StreamExecJoin extends ExecNodeBase<RowData> transform.setStateKeyType(leftSelect.getProducedType()); return transform; } + + /** Compiled plans before Flink 2.4 do not contain the changelog modes of the inputs. */ + private static @Nullable RuntimeChangelogMode toRuntimeChangelogMode( + @Nullable ChangelogMode inputChangelogMode) { + return inputChangelogMode == null + ? null + : RuntimeChangelogMode.serialize(inputChangelogMode); + } } diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecProcessTableFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecProcessTableFunction.java index d04ee8757c8..46adc829067 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecProcessTableFunction.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecProcessTableFunction.java @@ -50,10 +50,10 @@ import org.apache.flink.table.runtime.generated.GeneratedProcessTableRunner; import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser; import org.apache.flink.table.runtime.keyselector.RowDataKeySelector; import org.apache.flink.table.runtime.operators.process.ProcessTableOperatorFactory; -import org.apache.flink.table.runtime.operators.process.RuntimeChangelogMode; import org.apache.flink.table.runtime.operators.process.RuntimeStateInfo; import org.apache.flink.table.runtime.operators.process.RuntimeTableSemantics; import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.runtime.util.RuntimeChangelogMode; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.inference.StaticArgument; import org.apache.flink.table.types.inference.StaticArgumentTrait; diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalJoin.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalJoin.scala index 77774cd02a8..a10401fd08a 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalJoin.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalJoin.scala @@ -23,7 +23,7 @@ import org.apache.flink.table.planner.hint.StateTtlHint import org.apache.flink.table.planner.plan.nodes.exec.{ExecNode, InputProperty} import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecJoin import org.apache.flink.table.planner.plan.nodes.physical.common.CommonPhysicalJoin -import org.apache.flink.table.planner.plan.utils.{JoinUtil, MinibatchUtil} +import org.apache.flink.table.planner.plan.utils.{ChangelogPlanUtils, JoinUtil, MinibatchUtil} import org.apache.flink.table.planner.utils.ShortcutUtils.{unwrapClassLoader, unwrapTableConfig} import org.apache.flink.table.runtime.typeutils.InternalTypeInfo @@ -126,6 +126,8 @@ class StreamPhysicalJoin( joinSpec, getUpsertKeys(left, joinSpec.getLeftKeys), getUpsertKeys(right, joinSpec.getRightKeys), + ChangelogPlanUtils.getChangelogMode(left.asInstanceOf[StreamPhysicalRel]).get, + ChangelogPlanUtils.getChangelogMode(right.asInstanceOf[StreamPhysicalRel]).get, InputProperty.DEFAULT, InputProperty.DEFAULT, StateTtlHint.getStateTtlFromHintOnBiRel(getHints), 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 02d3e92b711..4699e9ed29d 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 @@ -23,6 +23,7 @@ import org.apache.flink.table.test.program.SinkTestStep; import org.apache.flink.table.test.program.SourceTestStep; import org.apache.flink.table.test.program.TableTestProgram; import org.apache.flink.types.Row; +import org.apache.flink.types.RowKind; /** {@link TableTestProgram} definitions for testing {@link StreamExecJoin}. */ public class JoinTestPrograms { @@ -41,6 +42,8 @@ public class JoinTestPrograms { public static final TableTestProgram ANTI_JOIN; public static final TableTestProgram JOIN_WITH_STATE_TTL_HINT; 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; static final SourceTestStep EMPLOYEE = SourceTestStep.newBuilder("EMPLOYEE") @@ -501,5 +504,69 @@ public class JoinTestPrograms { + " WHEN a NOT IN (SELECT CAST(j AS INTEGER) FROM source_t3) THEN 2 ELSE 3 END)" + " NOT IN (SELECT d FROM source_t2 WHERE source_t1.c = source_t2.f)") .build(); + + LEFT_JOIN_UPSERT_INPUT_NON_EQUI = + leftJoinWithUpsertInputAndNonEquiCondition( + "join-left-join-upsert-input-non-equi", + "left join with an upsert input and a non-equi condition"); + + // its plan and savepoint are compiled before this fix, whose plans have no input changelog + // modes + LEFT_JOIN_UPSERT_INPUT_NON_EQUI_WITHOUT_CHANGELOG_MODES = + 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"); + } + + /** + * An updated line that no longer satisfies the non-equi condition after the restore retracts + * its match, before a further update matches again. + */ + private static TableTestProgram leftJoinWithUpsertInputAndNonEquiCondition( + String id, String description) { + return TableTestProgram.of(id, description) + .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") + .producedBeforeRestore( + Row.ofKind(RowKind.INSERT, 1, "l1", "cancelled")) + .producedAfterRestore( + Row.ofKind(RowKind.UPDATE_AFTER, 1, "l1", "new"), + Row.ofKind(RowKind.UPDATE_AFTER, 1, "l1", "cancelled"), + Row.ofKind(RowKind.UPDATE_AFTER, 1, "l1", "paid")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("MySink") + .addSchema( + "order_id INT", + "line_order_id INT", + "line_id STRING", + "status STRING") + .addOption("sink-changelog-mode-enforced", "I,UA,D") + .consumedBeforeRestore("+I[1, null, null, null]") + .consumedAfterRestore( + "-D[1, null, null, null]", + "+I[1, 1, l1, new]", + "-D[1, 1, l1, new]", + "+I[1, null, null, null]", + "-D[1, null, null, null]", + "+I[1, 1, l1, paid]") + .build()) + .runSql( + "INSERT INTO MySink SELECT o.order_id, l.* FROM orders o " + + "LEFT 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 4ff137572e1..faa1403d929 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 @@ -49,6 +49,8 @@ public class JoinRestoreTest extends RestoreTestBase { JoinTestPrograms.SEMI_JOIN, JoinTestPrograms.ANTI_JOIN, JoinTestPrograms.JOIN_WITH_STATE_TTL_HINT, - JoinTestPrograms.SEMI_ANTI_JOIN_WITH_LITERAL_AGG); + JoinTestPrograms.SEMI_ANTI_JOIN_WITH_LITERAL_AGG, + JoinTestPrograms.LEFT_JOIN_UPSERT_INPUT_NON_EQUI, + JoinTestPrograms.LEFT_JOIN_UPSERT_INPUT_NON_EQUI_WITHOUT_CHANGELOG_MODES); } } diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-left-join-upsert-input-non-equi-without-changelog-modes/plan/join-left-join-upsert-input-non-equi-without-changelog-modes.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-left-join-upsert-input-non-equi-without-changelog-modes/plan/join-left-join-upsert-input-non-equi-without-changelog-modes.json new file mode 100644 index 00000000000..83c22ddf621 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-left-join-upsert-input-non-equi-without-changelog-modes/plan/join-left-join-upsert-input-non-equi-without-changelog-modes.json @@ -0,0 +1,252 @@ +{ + "flinkVersion" : "2.2", + "nodes" : [ { + "id" : 1, + "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" : 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_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" : 4, + "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" : 5, + "type" : "stream-exec-join_1", + "joinSpec" : { + "joinType" : "LEFT", + "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 ] ], + "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, `line_id` VARCHAR(2147483647), `status` VARCHAR(2147483647)>", + "description" : "Join(joinType=[LeftOuterJoin], where=[((order_id = order_id0) AND (status <> excluded_status))], select=[order_id, excluded_status, order_id0, line_id, status], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[HasUniqueKey])" + }, { + "id" : 6, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "INT NOT NULL" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "INT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 4, + "type" : "VARCHAR(2147483647)" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`order_id` INT NOT NULL, `order_id0` INT, `line_id` VARCHAR(2147483647), `status` VARCHAR(2147483647)>", + "description" : "Calc(select=[order_id, order_id0, line_id, status])" + }, { + "id" : 7, + "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" + }, { + "name" : "line_order_id", + "dataType" : "INT" + }, { + "name" : "line_id", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "status", + "dataType" : "VARCHAR(2147483647)" + } ] + } + } + } + }, + "inputChangelogMode" : [ "INSERT", "UPDATE_AFTER", "DELETE" ], + "upsertMaterializeStrategy" : "MAP", + "inputUpsertKey" : [ 0, 1, 2 ], + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`order_id` INT NOT NULL, `order_id0` INT, `line_id` VARCHAR(2147483647), `status` VARCHAR(2147483647)>", + "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[order_id, order_id0, line_id, 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" + }, { + "source" : 6, + "target" : 7, + "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-left-join-upsert-input-non-equi-without-changelog-modes/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-left-join-upsert-input-non-equi-without-changelog-modes/savepoint/_metadata new file mode 100644 index 00000000000..4432b998be1 Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-left-join-upsert-input-non-equi-without-changelog-modes/savepoint/_metadata differ diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-left-join-upsert-input-non-equi/plan/join-left-join-upsert-input-non-equi.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-left-join-upsert-input-non-equi/plan/join-left-join-upsert-input-non-equi.json new file mode 100644 index 00000000000..92f39414875 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-left-join-upsert-input-non-equi/plan/join-left-join-upsert-input-non-equi.json @@ -0,0 +1,254 @@ +{ + "flinkVersion" : "2.2", + "nodes" : [ { + "id" : 1, + "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" : 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_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" : 4, + "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" : 5, + "type" : "stream-exec-join_1", + "joinSpec" : { + "joinType" : "LEFT", + "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, `line_id` VARCHAR(2147483647), `status` VARCHAR(2147483647)>", + "description" : "Join(joinType=[LeftOuterJoin], where=[((order_id = order_id0) AND (status <> excluded_status))], select=[order_id, excluded_status, order_id0, line_id, status], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[HasUniqueKey])" + }, { + "id" : 6, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "INT NOT NULL" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "INT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 4, + "type" : "VARCHAR(2147483647)" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`order_id` INT NOT NULL, `order_id0` INT, `line_id` VARCHAR(2147483647), `status` VARCHAR(2147483647)>", + "description" : "Calc(select=[order_id, order_id0, line_id, status])" + }, { + "id" : 7, + "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" + }, { + "name" : "line_order_id", + "dataType" : "INT" + }, { + "name" : "line_id", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "status", + "dataType" : "VARCHAR(2147483647)" + } ] + } + } + } + }, + "inputChangelogMode" : [ "INSERT", "UPDATE_AFTER", "DELETE" ], + "upsertMaterializeStrategy" : "MAP", + "inputUpsertKey" : [ 0, 1, 2 ], + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`order_id` INT NOT NULL, `order_id0` INT, `line_id` VARCHAR(2147483647), `status` VARCHAR(2147483647)>", + "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[order_id, order_id0, line_id, 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" + }, { + "source" : 6, + "target" : 7, + "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-left-join-upsert-input-non-equi/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-left-join-upsert-input-non-equi/savepoint/_metadata new file mode 100644 index 00000000000..d644f4c5a1a Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-left-join-upsert-input-non-equi/savepoint/_metadata differ diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/JoinITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/JoinITCase.scala index 3ecff0f01ae..01cbb3942f7 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/JoinITCase.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/JoinITCase.scala @@ -774,14 +774,74 @@ class JoinITCase(miniBatch: MiniBatchMode, state: StateBackendMode, enableAsyncS // mini-batch folds the +U/-D pair within a bundle and async state is not fixed yet assumeThat(miniBatch == MiniBatchOff).isTrue assumeThat(enableAsyncState).isFalse + + val lines = Seq( + changelogRow("+I", order, "l1", "new"), + changelogRow("+I", order, "l2", "new"), + changelogRow("+U", order, "l1", "paid"), + changelogRow("-D", order, "l1", "paid"), + changelogRow("-D", order, "l2", "new") + ) + + val expected = List( + "+I[+I[1, EU], null, null, null]", + "-D[+I[1, EU], null, null, null]", + "+I[+I[1, EU], +I[1, EU], l1, new]", + "+I[+I[1, EU], +I[1, EU], l2, new]", + "+I[+I[1, EU], +I[1, EU], l1, paid]", + "-D[+I[1, EU], +I[1, EU], l1, paid]", + "-D[+I[1, EU], +I[1, EU], l2, new]", + "+I[+I[1, EU], null, null, null]" + ) + val results = getRawResultsOfLeftJoinOnRowKey("o.order_key = l.order_key", lines) + assertThat(results).isEqualTo(expected) + } + + /** An updated line that no longer satisfies the non-equi condition retracts its match. */ + @TestTemplate + def testLeftJoinOnRowKeyWithNonEquiCondition(): Unit = { + // mini-batch folds the +U/+U pair within a bundle and async state is not fixed yet + assumeThat(miniBatch == MiniBatchOff).isTrue + assumeThat(enableAsyncState).isFalse + + val lines = Seq( + changelogRow("+I", order, "l1", "new"), + changelogRow("+U", order, "l1", "cancelled"), + changelogRow("+U", order, "l1", "paid") + ) + + val expected = List( + "+I[+I[1, EU], null, null, null]", + "-D[+I[1, EU], null, null, null]", + "+I[+I[1, EU], +I[1, EU], l1, new]", + "-D[+I[1, EU], +I[1, EU], l1, new]", + "+I[+I[1, EU], null, null, null]", + "-D[+I[1, EU], null, null, null]", + "+I[+I[1, EU], +I[1, EU], l1, paid]" + ) + val results = getRawResultsOfLeftJoinOnRowKey( + "o.order_key = l.order_key AND l.status <> o.excluded_status", + lines) + assertThat(results).isEqualTo(expected) + } + + private val order = row(1, "EU") + + /** + * Left joins an upsert table of orders with the given upsert changes of their lines, which are + * emitted once the sink received the first row, and returns the raw results of the sink. + */ + private def getRawResultsOfLeftJoinOnRowKey( + joinCondition: String, + lines: Seq[Row]): List[String] = { env.setParallelism(1) val sink = s"sink_${UUID.randomUUID().toString.replace('-', '_')}" - val order = row(1, "EU") - val ordersId = TestValuesTableFactory.registerData(Seq(row(order))) + val ordersId = TestValuesTableFactory.registerData(Seq(row(order, "cancelled"))) tEnv.executeSql(s""" |CREATE TABLE orders ( - | order_key ROW<id INT, region STRING> PRIMARY KEY NOT ENFORCED + | order_key ROW<id INT, region STRING> PRIMARY KEY NOT ENFORCED, + | excluded_status STRING |) WITH ( | 'connector' = 'values', | 'data-id' = '$ordersId', @@ -789,13 +849,6 @@ class JoinITCase(miniBatch: MiniBatchMode, state: StateBackendMode, enableAsyncS |) |""".stripMargin) - val lines = Seq( - changelogRow("+I", order, "l1", "new"), - changelogRow("+I", order, "l2", "new"), - changelogRow("+U", order, "l1", "paid"), - changelogRow("-D", order, "l1", "paid"), - changelogRow("-D", order, "l2", "new") - ) val lineType = Types.ROW_NAMED( Array("order_key", "line_id", "status"), Types.ROW_NAMED(Array("id", "region"), Types.INT, Types.STRING), @@ -826,22 +879,11 @@ class JoinITCase(miniBatch: MiniBatchMode, state: StateBackendMode, enableAsyncS |""".stripMargin) val insert = - s"INSERT INTO $sink SELECT * FROM orders o LEFT JOIN lines l ON o.order_key = l.order_key" + s"INSERT INTO $sink SELECT o.order_key, l.* FROM orders o LEFT JOIN lines l ON $joinCondition" assertThat(tEnv.explainSql(insert, ExplainDetail.CHANGELOG_MODE)) .contains("rightInputSpec=[HasUniqueKey], changelogMode=[I,UA,D]") tEnv.executeSql(insert).await() - - val expected = List( - "+I[+I[1, EU], null, null, null]", - "-D[+I[1, EU], null, null, null]", - "+I[+I[1, EU], +I[1, EU], l1, new]", - "+I[+I[1, EU], +I[1, EU], l2, new]", - "+I[+I[1, EU], +I[1, EU], l1, paid]", - "-D[+I[1, EU], +I[1, EU], l1, paid]", - "-D[+I[1, EU], +I[1, EU], l2, new]", - "+I[+I[1, EU], null, null, null]" - ) - assertThat(TestValuesTableFactory.getRawResultsAsStrings(sink).toList).isEqualTo(expected) + TestValuesTableFactory.getRawResultsAsStrings(sink).toList } @TestTemplate diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/MiniBatchStreamingJoinOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/MiniBatchStreamingJoinOperator.java index a31cd7da576..fa0224cecc3 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/MiniBatchStreamingJoinOperator.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/MiniBatchStreamingJoinOperator.java @@ -35,6 +35,9 @@ import org.apache.flink.table.runtime.operators.join.stream.state.JoinRecordStat import org.apache.flink.table.runtime.operators.join.stream.utils.JoinInputSideSpec; import org.apache.flink.table.runtime.operators.metrics.SimpleGauge; import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.runtime.util.RuntimeChangelogMode; + +import javax.annotation.Nullable; import java.io.Serializable; import java.util.Iterator; @@ -65,6 +68,9 @@ public abstract class MiniBatchStreamingJoinOperator extends StreamingJoinOperat parameter.rightInputSideSpec, parameter.leftIsOuter, parameter.rightIsOuter, + parameter.leftInputChangelogMode, + parameter.rightInputChangelogMode, + parameter.hasNonEquiCondition, parameter.filterNullKeys, parameter.leftStateRetentionTime, parameter.rightStateRetentionTime); @@ -199,10 +205,14 @@ public abstract class MiniBatchStreamingJoinOperator extends StreamingJoinOperat pre = next; } processElement( - current, inputSideStateView, otherSideStateView, inputIsLeft, isSuppress); + current, + inputSideStateView, + otherSideStateView, + inputIsLeft, + isSuppress ? next : null); if (isSuppress) { processElement( - next, inputSideStateView, otherSideStateView, inputIsLeft, isSuppress); + next, inputSideStateView, otherSideStateView, inputIsLeft, current); } } else { // 1. current is accumulateMsg 2. current is retractMsg and no next row @@ -211,11 +221,19 @@ public abstract class MiniBatchStreamingJoinOperator extends StreamingJoinOperat isSuppress = true; } processElement( - pre, inputSideStateView, otherSideStateView, inputIsLeft, isSuppress); - pre = null; + pre, + inputSideStateView, + otherSideStateView, + inputIsLeft, + isSuppress ? current : null); } processElement( - current, inputSideStateView, otherSideStateView, inputIsLeft, isSuppress); + current, + inputSideStateView, + otherSideStateView, + inputIsLeft, + isSuppress ? pre : null); + pre = null; } } } @@ -238,7 +256,7 @@ public abstract class MiniBatchStreamingJoinOperator extends StreamingJoinOperat setCurrentKey(entry.getKey()); for (RowData rowData : entry.getValue()) { processElement( - rowData, inputSideStateView, otherSideStateView, inputIsLeft, false); + rowData, inputSideStateView, otherSideStateView, inputIsLeft, null); } } } else if (inputBuffer instanceof JoinKeyContainsUniqueKeyBundle) { @@ -272,6 +290,9 @@ public abstract class MiniBatchStreamingJoinOperator extends StreamingJoinOperat JoinInputSideSpec rightInputSideSpec, boolean leftIsOuter, boolean rightIsOuter, + @Nullable RuntimeChangelogMode leftInputChangelogMode, + @Nullable RuntimeChangelogMode rightInputChangelogMode, + boolean hasNonEquiCondition, boolean[] filterNullKeys, long leftStateRetentionTime, long rightStateRetentionTime, @@ -285,6 +306,9 @@ public abstract class MiniBatchStreamingJoinOperator extends StreamingJoinOperat rightInputSideSpec, leftIsOuter, rightIsOuter, + leftInputChangelogMode, + rightInputChangelogMode, + hasNonEquiCondition, filterNullKeys, leftStateRetentionTime, rightStateRetentionTime, @@ -311,6 +335,9 @@ public abstract class MiniBatchStreamingJoinOperator extends StreamingJoinOperat JoinInputSideSpec rightInputSideSpec; boolean leftIsOuter; boolean rightIsOuter; + @Nullable RuntimeChangelogMode leftInputChangelogMode; + @Nullable RuntimeChangelogMode rightInputChangelogMode; + boolean hasNonEquiCondition; boolean[] filterNullKeys; long leftStateRetentionTime; long rightStateRetentionTime; @@ -325,6 +352,9 @@ public abstract class MiniBatchStreamingJoinOperator extends StreamingJoinOperat JoinInputSideSpec rightInputSideSpec, boolean leftIsOuter, boolean rightIsOuter, + @Nullable RuntimeChangelogMode leftInputChangelogMode, + @Nullable RuntimeChangelogMode rightInputChangelogMode, + boolean hasNonEquiCondition, boolean[] filterNullKeys, long leftStateRetentionTime, long rightStateRetentionTime, @@ -336,6 +366,9 @@ public abstract class MiniBatchStreamingJoinOperator extends StreamingJoinOperat this.rightInputSideSpec = rightInputSideSpec; this.leftIsOuter = leftIsOuter; this.rightIsOuter = rightIsOuter; + this.leftInputChangelogMode = leftInputChangelogMode; + this.rightInputChangelogMode = rightInputChangelogMode; + this.hasNonEquiCondition = hasNonEquiCondition; this.filterNullKeys = filterNullKeys; this.leftStateRetentionTime = leftStateRetentionTime; this.rightStateRetentionTime = rightStateRetentionTime; 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 af5b1457f5f..076f119ae12 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,6 +19,7 @@ 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; @@ -30,8 +31,11 @@ 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 javax.annotation.Nullable; + import java.util.Iterator; /** Streaming unbounded Join operator which supports INNER/LEFT/RIGHT/FULL JOIN. */ @@ -43,6 +47,16 @@ 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 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; @@ -61,6 +75,9 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { JoinInputSideSpec rightInputSideSpec, boolean leftIsOuter, boolean rightIsOuter, + @Nullable RuntimeChangelogMode leftInputChangelogMode, + @Nullable RuntimeChangelogMode rightInputChangelogMode, + boolean hasNonEquiCondition, boolean[] filterNullKeys, long leftStateRetentionTime, long rightStateRetentionTime) { @@ -75,6 +92,11 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { rightStateRetentionTime); this.leftIsOuter = leftIsOuter; this.rightIsOuter = rightIsOuter; + this.leftInputChangelogMode = leftInputChangelogMode; + this.rightInputChangelogMode = rightInputChangelogMode; + this.leftIsUpsertOrUnknown = isUpsertOrUnknown(leftInputChangelogMode); + this.rightIsUpsertOrUnknown = isUpsertOrUnknown(rightInputChangelogMode); + this.hasNonEquiCondition = hasNonEquiCondition; } @Override @@ -125,12 +147,12 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { @Override public void processElement1(StreamRecord<RowData> element) throws Exception { - processElement(element.getValue(), leftRecordStateView, rightRecordStateView, true, false); + processElement(element.getValue(), leftRecordStateView, rightRecordStateView, true, null); } @Override public void processElement2(StreamRecord<RowData> element) throws Exception { - processElement(element.getValue(), rightRecordStateView, leftRecordStateView, false, false); + processElement(element.getValue(), rightRecordStateView, leftRecordStateView, false, null); } /** @@ -151,13 +173,24 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { * * <pre> * if input record is accumulate + * | if record replaces a stored record (upsert) and the condition is non-equi + * | | for each other that the replaced record matches but the record does not + * | | | send -D[replaced+other] + * | | | if other side is outer + * | | | | if the matched num in the matched rows == 1, send +I[null+other] + * | | | | otherState.update(other, old - 1) + * | | | endif + * | | endfor + * | endif * | if input side is outer * | | if there is no matched rows on the other side, send +I[record+null], state.add(record, 0) * | | if there are matched rows on the other side + * | | | if the replaced record was null padded, send -D[replaced+null] * | | | if other side is outer - * | | | | if the matched num in the matched rows == 0, send -D[null+other] + * | | | | if the matched num in the matched rows == 0, send -D[null+other] unless the paired + * | | | | record matches other * | | | | if the matched num in the matched rows > 0, skip - * | | | | if matched num == 0 or record is an additional match, otherState.update(other, old + 1) + * | | | | if the replaced record did not match other, otherState.update(other, old + 1) * | | | endif * | | | send +I[record+other]s, state.add(record, other.size) * | | endif @@ -167,9 +200,10 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { * | | if there is no matched rows on the other side, skip * | | if there are matched rows on the other side * | | | if other side is outer - * | | | | if the matched num in the matched rows == 0, send -D[null+other] + * | | | | if the matched num in the matched rows == 0, send -D[null+other] unless the + * | | | | paired record matches other * | | | | if the matched num in the matched rows > 0, skip - * | | | | if matched num == 0 or record is an additional match, otherState.update(other, old + 1) + * | | | | if the replaced record did not match other, otherState.update(other, old + 1) * | | | | send +I[record+other]s * | | | else * | | | | send +I/+U[record+other]s (using input RowKind) @@ -179,14 +213,15 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { * endif * * if input record is retract - * | state.retract(record) + * | 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] * | endif * | if there are matched rows on the other side, send -D[record+other]s if outer, send -D/-U[record+other]s if inner. * | | if other side is outer * | | | if the matched num in the matched rows == 0, this should never happen! - * | | | if the matched num in the matched rows == 1, send +I[null+other] + * | | | if the matched num in the matched rows == 1, send +I[null+other] unless the paired + * | | | record matches other * | | | if the matched num in the matched rows > 1, skip * | | | otherState.update(other, old - 1) * | | endif @@ -198,16 +233,17 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { * @param inputSideStateView state of input side * @param otherSideStateView state of other side * @param inputIsLeft whether input side is left side - * @param isSuppress whether suppress the output of redundant messages when the other side is - * outer join. This only applies to the case of mini-batch. + * @param pairedRecord the other record of a -U/+U pair in mini-batch mode, or null. Null + * padding changes that the pair cancels out are suppressed. */ protected void processElement( RowData input, JoinRecordStateView inputSideStateView, JoinRecordStateView otherSideStateView, boolean inputIsLeft, - boolean isSuppress) + @Nullable RowData pairedRecord) throws Exception { + final boolean isSuppress = pairedRecord != null; boolean inputIsOuter = inputIsLeft ? leftIsOuter : rightIsOuter; boolean otherIsOuter = inputIsLeft ? rightIsOuter : leftIsOuter; boolean isAccumulateMsg = RowDataUtil.isAccumulateMsg(input); @@ -215,8 +251,13 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { input.setRowKind(RowKind.INSERT); // erase RowKind for later state updating if (isAccumulateMsg) { // record is accumulate - final boolean isAdditionalMatch = - isAdditionalMatch(input, inputSideStateView, inputIsLeft, isSuppress); + final RowData replacedRecord = + replacedRecord(input, inputSideStateView, inputIsLeft, isSuppress); + boolean replacedRecordHadNoMatches = false; + if (replacedRecord != null && hasNonEquiCondition) { + replacedRecordHadNoMatches = + !retractLostMatches(input, replacedRecord, otherSideStateView, inputIsLeft); + } if (inputIsOuter) { // input side is outer Iterator<OuterRecord> associatedRecords = AbstractStreamingJoinOperator.iterator( @@ -229,19 +270,25 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { ((OuterJoinRecordStateView) inputSideStateView).addRecord(input, 0); return; } else { // there are matched rows on the other side + if (replacedRecordHadNoMatches) { + // the replaced record was null padded, send -D[replaced+null] + outRow.setRowKind(RowKind.DELETE); + outputNullPadding(replacedRecord, inputIsLeft); + } int numAssociations = 0; while (associatedRecords.hasNext()) { OuterRecord outerRecord = associatedRecords.next(); RowData other = outerRecord.record; if (otherIsOuter) { // other side is outer // if the matched num in the matched rows == 0 - if (outerRecord.numOfAssociations == 0 && !isSuppress) { + if (outerRecord.numOfAssociations == 0 + && !pairCancelsNullPadding(pairedRecord, other, inputIsLeft)) { // send -D[null+other] outRow.setRowKind(RowKind.DELETE); outputNullPadding(other, !inputIsLeft); } // ignore matched number > 0 // otherState.update(other, old + 1) - if (outerRecord.numOfAssociations == 0 || isAdditionalMatch) { + if (isNewMatch(outerRecord, replacedRecord, inputIsLeft)) { ((OuterJoinRecordStateView) otherSideStateView) .updateNumOfAssociations( other, outerRecord.numOfAssociations + 1); @@ -269,13 +316,14 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { while (associatedRecords.hasNext()) { OuterRecord outerRecord = associatedRecords.next(); if (outerRecord.numOfAssociations == 0 - && !isSuppress) { // if the matched num in the matched rows == 0 + && !pairCancelsNullPadding( + pairedRecord, outerRecord.record, inputIsLeft)) { // send -D[null+other] outRow.setRowKind(RowKind.DELETE); outputNullPadding(outerRecord.record, !inputIsLeft); } // otherState.update(other, old + 1) - if (outerRecord.numOfAssociations == 0 || isAdditionalMatch) { + if (isNewMatch(outerRecord, replacedRecord, inputIsLeft)) { otherSideOuterStateView.updateNumOfAssociations( outerRecord.record, outerRecord.numOfAssociations + 1); } @@ -324,7 +372,9 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { if (otherIsOuter) { OuterJoinRecordStateView otherSideOuterStateView = (OuterJoinRecordStateView) otherSideStateView; - if (outerRecord.numOfAssociations == 1 && !isSuppress) { + if (outerRecord.numOfAssociations == 1 + && !pairCancelsNullPadding( + pairedRecord, outerRecord.record, inputIsLeft)) { // send +I[null+other] outRow.setRowKind(RowKind.INSERT); outputNullPadding(outerRecord.record, !inputIsLeft); @@ -339,26 +389,113 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator { } /** - * Returns whether the record adds a new match for other-side records, which indicates that we - * have to increase the number of associations for a specific key match. + * 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>Matches are only counted when the other side is outer, so the result is false otherwise - * and no lookup happens. If the join key contains the unique key, there is at most one record - * per join key, so a match is never additional and no lookup is needed either. - * - * <p>A suppressed retraction in mini-batch mode removes the matches of a record but keeps it in - * state, so the suppressed accumulate message that follows is always an additional match. + * <p>The lookup only happens if it can change the result. In a mini-batch pair, the retraction + * already handled the old version. */ - private boolean isAdditionalMatch( + 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; - final JoinInputSideSpec inputSideSpec = isLeft ? leftInputSideSpec : rightInputSideSpec; - // TODO FLINK-40841: assumes the replaced record had the same matches, not true for non-equi - return otherIsOuter - && (isSuppress - || (!inputSideSpec.joinKeyContainsUniqueKey() - && !stateView.hasRecord(record))); + // 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); + } + + /** + * 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 + * matched any row. + */ + private boolean retractLostMatches( + RowData record, + RowData replacedRecord, + JoinRecordStateView otherSideStateView, + boolean inputIsLeft) + throws Exception { + final boolean otherIsOuter = inputIsLeft ? rightIsOuter : leftIsOuter; + final Iterator<OuterRecord> matchesOfReplacedRecord = + AbstractStreamingJoinOperator.iterator( + replacedRecord, inputIsLeft, otherSideStateView, joinCondition); + final boolean replacedRecordHadMatches = matchesOfReplacedRecord.hasNext(); + while (matchesOfReplacedRecord.hasNext()) { + final OuterRecord other = matchesOfReplacedRecord.next(); + final boolean bothOldAndNewVersionsMatch = matches(record, other.record, inputIsLeft); + if (bothOldAndNewVersionsMatch) { + continue; + } + // send -D[replaced+other] + outRow.setRowKind(RowKind.DELETE); + output(replacedRecord, other.record, inputIsLeft); + if (otherIsOuter) { + final boolean wasLastMatch = other.numOfAssociations == 1; + if (wasLastMatch) { + // send +I[null+other] + outRow.setRowKind(RowKind.INSERT); + outputNullPadding(other.record, !inputIsLeft); + } + // otherState.update(other, old - 1) + ((OuterJoinRecordStateView) otherSideStateView) + .updateNumOfAssociations(other.record, other.numOfAssociations - 1); + } + } + 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. + */ + private boolean pairCancelsNullPadding( + @Nullable RowData pairedRecord, RowData other, boolean inputIsLeft) { + return pairedRecord != null + && (!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); + } + + 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); } // ------------------------------------------------------------------------------------- diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateView.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateView.java index 6f6b29ef7fc..cc2b63c5712 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateView.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateView.java @@ -21,6 +21,8 @@ package org.apache.flink.table.runtime.operators.join.stream.state; import org.apache.flink.table.data.RowData; import org.apache.flink.table.runtime.operators.join.stream.utils.JoinInputSideSpec; +import javax.annotation.Nullable; + /** * A {@link JoinRecordStateView} is a view to the join state. It encapsulates the join state and * provides some APIs facing the input records. The join state is used to store input records. The @@ -42,8 +44,9 @@ public interface JoinRecordStateView { Iterable<RowData> getRecords() throws Exception; /** - * Returns whether a record with the same unique key as the given record is stored under the - * current join key. + * Returns the record with the same unique key as the given record that is stored under the + * current join key, or null if there is none. Always null if the input side has no unique key. */ - boolean hasRecord(RowData record) throws Exception; + @Nullable + RowData getRecord(RowData record) throws Exception; } diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateViews.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateViews.java index dd3477352a7..582273e13e8 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateViews.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/JoinRecordStateViews.java @@ -100,8 +100,8 @@ public final class JoinRecordStateViews { } @Override - public boolean hasRecord(RowData record) throws Exception { - return recordState.value() != null; + public RowData getRecord(RowData record) throws Exception { + return recordState.value(); } @Override @@ -152,8 +152,8 @@ public final class JoinRecordStateViews { } @Override - public boolean hasRecord(RowData record) throws Exception { - return recordState.contains(uniqueKeySelector.getKey(record)); + public RowData getRecord(RowData record) throws Exception { + return recordState.get(uniqueKeySelector.getKey(record)); } @Override @@ -204,9 +204,9 @@ public final class JoinRecordStateViews { } @Override - public boolean hasRecord(RowData record) { + public RowData getRecord(RowData record) { // without a unique key, an equal record is a second record and not an update - return false; + return null; } @Override 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 3edafd9bf2e..c077933d57b 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 @@ -119,8 +119,9 @@ public final class OuterJoinRecordStateViews { } @Override - public boolean hasRecord(RowData record) throws Exception { - return recordState.value() != null; + public RowData getRecord(RowData record) throws Exception { + final Tuple2<RowData, Integer> tuple = recordState.value(); + return tuple == null ? null : tuple.f0; } @Override @@ -197,8 +198,10 @@ public final class OuterJoinRecordStateViews { } @Override - public boolean hasRecord(RowData record) throws Exception { - return recordState.contains(uniqueKeySelector.getKey(record)); + public RowData getRecord(RowData record) throws Exception { + final Tuple2<RowData, Integer> tuple = + recordState.get(uniqueKeySelector.getKey(record)); + return tuple == null ? null : tuple.f0; } @Override @@ -277,9 +280,9 @@ public final class OuterJoinRecordStateViews { } @Override - public boolean hasRecord(RowData record) { + public RowData getRecord(RowData record) { // without a unique key, an equal record is a second record and not an update - return false; + return null; } @Override diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java index 5432a826cc2..e9f0e005640 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java @@ -58,6 +58,7 @@ import org.apache.flink.table.runtime.operators.process.TimeConverter.LongTimeCo import org.apache.flink.table.runtime.typeutils.ExternalSerializer; import org.apache.flink.table.runtime.typeutils.InternalSerializers; import org.apache.flink.table.runtime.typeutils.StringDataSerializer; +import org.apache.flink.table.runtime.util.RuntimeChangelogMode; import org.apache.flink.table.runtime.util.StateConfigUtil; import org.apache.flink.table.types.CollectionDataType; import org.apache.flink.table.types.DataType; diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessRowTableOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessRowTableOperator.java index 075f589e926..acb91cc9e74 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessRowTableOperator.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessRowTableOperator.java @@ -30,6 +30,7 @@ import org.apache.flink.table.functions.TableSemantics; import org.apache.flink.table.runtime.generated.HashFunction; import org.apache.flink.table.runtime.generated.ProcessTableRunner; import org.apache.flink.table.runtime.generated.RecordEqualiser; +import org.apache.flink.table.runtime.util.RuntimeChangelogMode; import javax.annotation.Nullable; diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperator.java index 948464d5f3f..7b303c7a400 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperator.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperator.java @@ -29,6 +29,7 @@ import org.apache.flink.table.functions.TableSemantics; import org.apache.flink.table.runtime.generated.HashFunction; import org.apache.flink.table.runtime.generated.ProcessTableRunner; import org.apache.flink.table.runtime.generated.RecordEqualiser; +import org.apache.flink.table.runtime.util.RuntimeChangelogMode; import java.util.List; import java.util.stream.Collectors; diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessTableOperatorFactory.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessTableOperatorFactory.java index d0eed417f11..cb3fb77a1d7 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessTableOperatorFactory.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessTableOperatorFactory.java @@ -28,6 +28,7 @@ import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser; import org.apache.flink.table.runtime.generated.HashFunction; import org.apache.flink.table.runtime.generated.ProcessTableRunner; import org.apache.flink.table.runtime.generated.RecordEqualiser; +import org.apache.flink.table.runtime.util.RuntimeChangelogMode; import java.util.Arrays; import java.util.List; diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeTableSemantics.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeTableSemantics.java index c5d4860d8f4..48a4e0247b7 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeTableSemantics.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeTableSemantics.java @@ -21,6 +21,7 @@ package org.apache.flink.table.runtime.operators.process; import org.apache.flink.table.connector.ChangelogMode; import org.apache.flink.table.functions.ProcessTableFunction; import org.apache.flink.table.functions.TableSemantics; +import org.apache.flink.table.runtime.util.RuntimeChangelogMode; import org.apache.flink.table.types.DataType; import java.io.Serializable; diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeChangelogMode.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/RuntimeChangelogMode.java similarity index 97% rename from flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeChangelogMode.java rename to flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/RuntimeChangelogMode.java index b0718a5fb25..b75b37493c6 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeChangelogMode.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/RuntimeChangelogMode.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.flink.table.runtime.operators.process; +package org.apache.flink.table.runtime.util; import org.apache.flink.annotation.Internal; import org.apache.flink.table.connector.ChangelogMode; 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 53c0b49da50..be4256640dd 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 @@ -21,6 +21,7 @@ package org.apache.flink.table.runtime.operators.join.stream; import org.apache.flink.streaming.api.operators.TwoInputStreamOperator; 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; @@ -79,6 +80,10 @@ class StreamingJoinOperatorTest extends StreamingJoinOperatorTestBase { rightInputSpec, joinTypeSpec[0], joinTypeSpec[1], + // the tests send both retract and upsert inputs + null, + null, + false, new boolean[] {true}, ttl[0], ttl[1]); @@ -810,18 +815,190 @@ class StreamingJoinOperatorTest extends StreamingJoinOperatorTestBase { rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR")); } - /** Views without a unique key never report a stored record, not even an equal one. */ + /** + * A left row replaced without UPDATE_BEFORE is counted by the right row if the old version did + * not match it. + */ + @TestTemplate + void testRightOuterJoinCountsLeftUpsertThatStartsMatching() throws Exception { + assumeFalse(enableAsyncState); + useStreamingJoinOperator( + leftHasUniqueKeySpec(), + rightInputSpec, + false, + true, + knownValuesNonEquiCondition(), + true); + + testHarness.processElement2(insertRecord("LineOrd#1", "AIR")); + assertor.shouldEmit( + testHarness, rowOfKind(RowKind.INSERT, null, null, null, "LineOrd#1", "AIR")); + + testHarness.processElement1(insertRecord("Ord#2", "LineOrd#1", ADDRESS)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, null, null, null, "LineOrd#1", "AIR"), + rowOfKind(RowKind.INSERT, "Ord#2", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR")); + + testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1", UNKNOWN)); + assertor.shouldEmitNothing(testHarness); + + testHarness.processElement1(updateAfterRecord("Ord#1", "LineOrd#1", NEW_ADDRESS)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", NEW_ADDRESS, "LineOrd#1", "AIR")); + + testHarness.processElement1(deleteRecord("Ord#2", "LineOrd#1", ADDRESS)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#2", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR")); + + testHarness.processElement1(deleteRecord("Ord#1", "LineOrd#1", NEW_ADDRESS)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", NEW_ADDRESS, "LineOrd#1", "AIR"), + rowOfKind(RowKind.INSERT, null, null, null, "LineOrd#1", "AIR")); + } + + /** A right row replaced without UPDATE_BEFORE that no longer matches retracts its match. */ + @TestTemplate + void testLeftOuterJoinRetractsLostMatchOfRightUpsert() throws Exception { + assumeFalse(enableAsyncState); + useStreamingJoinOperator( + leftInputSpec, rightInputSpec, true, false, 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, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR"), + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", ADDRESS, null, null)); + + testHarness.processElement2(updateAfterRecord("LineOrd#1", "SHIP")); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", ADDRESS, null, null), + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "SHIP")); + } + + /** + * A null padded left row replaced without UPDATE_BEFORE that now matches retracts its null + * padding. + */ + @TestTemplate + void testLeftOuterJoinRetractsNullPaddingOfLeftUpsert() throws Exception { + assumeFalse(enableAsyncState); + useStreamingJoinOperator( + leftInputSpec, rightInputSpec, true, false, knownValuesNonEquiCondition(), true); + + testHarness.processElement2(insertRecord("LineOrd#1", "AIR")); + testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1", UNKNOWN)); + assertor.shouldEmit( + testHarness, rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", UNKNOWN, null, null)); + + testHarness.processElement1(updateAfterRecord("Ord#1", "LineOrd#1", ADDRESS)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", UNKNOWN, null, null), + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR")); + } + + /** A right row replaced without UPDATE_BEFORE that no longer matches retracts its match. */ @TestTemplate - void testLeftOuterJoinHasRecord() throws Exception { + void testInnerJoinRetractsLostMatchOfRightUpsert() throws Exception { assumeFalse(enableAsyncState); - assertHasRecordAfterInsert(true); + useStreamingJoinOperator( + leftInputSpec, rightInputSpec, false, false, 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, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR")); + } + + /** A left row replaced without UPDATE_BEFORE that no longer matches uncounts its match. */ + @TestTemplate + void testRightOuterJoinUncountsLostMatchOfLeftUpsert() 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", ADDRESS)); + testHarness.getOutput().clear(); + + testHarness.processElement1(updateAfterRecord("Ord#1", "LineOrd#1", UNKNOWN)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR")); + + testHarness.processElement1(deleteRecord("Ord#2", "LineOrd#1", ADDRESS)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#2", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR"), + rowOfKind(RowKind.INSERT, null, null, null, "LineOrd#1", "AIR")); + } + + /** + * A right row replaced by INSERT or UPDATE_AFTER keeps the null padding of both sides in sync + * when it loses and gains its match. + */ + @TestTemplate + void testFullOuterJoinReplacesMatchesOfRightUpsert() throws Exception { + assumeFalse(enableAsyncState); + useStreamingJoinOperator( + leftInputSpec, rightInputSpec, true, true, knownValuesNonEquiCondition(), true); + + testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1", ADDRESS)); + testHarness.processElement2(insertRecord("LineOrd#1", "AIR")); + testHarness.getOutput().clear(); + + testHarness.processElement2(insertRecord("LineOrd#1", UNKNOWN)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "AIR"), + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", ADDRESS, null, null), + rowOfKind(RowKind.INSERT, null, null, null, "LineOrd#1", UNKNOWN)); + + testHarness.processElement2(updateAfterRecord("LineOrd#1", "SHIP")); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, null, null, null, "LineOrd#1", UNKNOWN), + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", ADDRESS, null, null), + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "SHIP")); + + testHarness.processElement1(deleteRecord("Ord#1", "LineOrd#1", ADDRESS)); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", ADDRESS, "LineOrd#1", "SHIP"), + rowOfKind(RowKind.INSERT, null, null, null, "LineOrd#1", "SHIP")); + } + + /** Views without a unique key never return a stored record, not even an equal one. */ + @TestTemplate + void testLeftOuterJoinGetRecord() throws Exception { + assumeFalse(enableAsyncState); + assertGetRecordAfterInsert(true); useStreamingJoinOperator( JoinInputSideSpec.withoutUniqueKey(), JoinInputSideSpec.withoutUniqueKey(), true, false); - assertHasRecordAfterInsert(false); + assertGetRecordAfterInsert(false); } // TODO FLINK-40681: remove once AsyncStateStreamingJoinOperator handles @@ -840,17 +1017,32 @@ class StreamingJoinOperatorTest extends StreamingJoinOperatorTestBase { boolean leftIsOuter, boolean rightIsOuter) throws Exception { + useStreamingJoinOperator( + leftSpec, rightSpec, leftIsOuter, rightIsOuter, joinCondition, false); + } + + private void useStreamingJoinOperator( + JoinInputSideSpec leftSpec, + JoinInputSideSpec rightSpec, + boolean leftIsOuter, + boolean rightIsOuter, + GeneratedJoinCondition condition, + boolean hasNonEquiCondition) + throws Exception { testHarness.close(); testHarness = new KeyedTwoInputStreamOperatorTestHarness<>( new StreamingJoinOperator( leftTypeInfo, rightTypeInfo, - joinCondition, + condition, leftSpec, rightSpec, leftIsOuter, rightIsOuter, + UPSERT, + UPSERT, + hasNonEquiCondition, new boolean[] {true}, 0L, 0L), @@ -860,21 +1052,28 @@ class StreamingJoinOperatorTest extends StreamingJoinOperatorTestBase { testHarness.open(); } - private void assertHasRecordAfterInsert(boolean expected) throws Exception { + private void assertGetRecordAfterInsert(boolean hasUniqueKey) throws Exception { final StreamingJoinOperator operator = (StreamingJoinOperator) testHarness.getOperator(); final RowData left = insertRecord("Ord#1", "LineOrd#1", ADDRESS).getValue(); final RowData right = insertRecord("LineOrd#1", "AIR").getValue(); operator.setCurrentKey(rightKeySelector.getKey(right)); - assertThat(operator.leftRecordStateView.hasRecord(left)).isFalse(); - assertThat(operator.rightRecordStateView.hasRecord(right)).isFalse(); + assertThat(operator.leftRecordStateView.getRecord(left)).isNull(); + assertThat(operator.rightRecordStateView.getRecord(right)).isNull(); testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1", ADDRESS)); testHarness.processElement2(insertRecord("LineOrd#1", "AIR")); - assertThat(operator.leftRecordStateView.hasRecord(left)).isEqualTo(expected); - assertThat(operator.rightRecordStateView.hasRecord(right)).isEqualTo(expected); + assertThat(operator.leftRecordStateView.getRecord(left)) + .isEqualTo(hasUniqueKey ? left : null); + assertThat(operator.rightRecordStateView.getRecord(right)) + .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( 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 34d720bb34d..2af780faa0d 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 @@ -22,12 +22,14 @@ import org.apache.flink.streaming.api.operators.TwoInputStreamOperator; import org.apache.flink.streaming.runtime.operators.asyncprocessing.AsyncKeyOrderedProcessingOperator; import org.apache.flink.streaming.util.KeyedTwoInputStreamOperatorTestHarness; import org.apache.flink.streaming.util.asyncprocessing.AsyncKeyedTwoInputStreamOperatorTestHarness; +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.keyselector.RowDataKeySelector; 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.RowDataHarnessAssertor; +import org.apache.flink.table.runtime.util.RuntimeChangelogMode; import org.apache.flink.table.types.logical.CharType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.RowType; @@ -97,6 +99,30 @@ public abstract class StreamingJoinOperatorTestBase { protected final GeneratedJoinCondition joinCondition = new GeneratedJoinCondition("ConditionFunction", funcCode, new Object[0]); + protected static final String UNKNOWN = "unknown"; + + protected static final RuntimeChangelogMode UPSERT = + RuntimeChangelogMode.serialize(ChangelogMode.upsert()); + + /** Only rows whose last field is not {@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(\"" + + UNKNOWN + + "\");\n" + + " }\n" + + " @Override\n" + + " public boolean apply(org.apache.flink.table.data.RowData in1, org.apache.flink.table.data.RowData in2) {\n" + + " return isKnown(in1) && isKnown(in2);\n" + + " }\n" + + "}\n"; + return new GeneratedJoinCondition("KnownValuesNonEquiCondition", code, new Object[0]); + } + 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/StreamingMiniBatchJoinOperatorTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMiniBatchJoinOperatorTest.java index 98abbe1764a..fde7a318c51 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMiniBatchJoinOperatorTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMiniBatchJoinOperatorTest.java @@ -1285,6 +1285,87 @@ final class StreamingMiniBatchJoinOperatorTest extends StreamingJoinOperatorTest "SHIP")); } + /** + * With a non-equi condition, a -U/+U pair within a bundle may lose or gain the match of the + * left row, so its null padding is updated, but not if both records match it. The same holds + * for a +U without -U. + */ + @Tag("miniBatchSize=10") + @Test + void testLeftJoinHasUniqueKeyNonEquiUpdatesMatchWithinBundle() throws Exception { + final String address = "3 Bellevue Drive, Pottstown, PA 19464"; + testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1", address)); + testHarness.processElement2(insertRecord("Ord#X", "LineOrd#1", "AIR")); + testHarness.prepareSnapshotPreBarrier(1L); + testHarness.getOutput().clear(); + + testHarness.processElement2(updateBeforeRecord("Ord#X", "LineOrd#1", "AIR")); + testHarness.processElement2(updateAfterRecord("Ord#X", "LineOrd#1", UNKNOWN)); + testHarness.prepareSnapshotPreBarrier(2L); + assertor.shouldEmit( + testHarness, + rowOfKind( + RowKind.UPDATE_BEFORE, + "Ord#1", + "LineOrd#1", + address, + "Ord#X", + "LineOrd#1", + "AIR"), + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", address, null, null, null)); + + testHarness.processElement2(updateBeforeRecord("Ord#X", "LineOrd#1", UNKNOWN)); + testHarness.processElement2(updateAfterRecord("Ord#X", "LineOrd#1", "SHIP")); + testHarness.prepareSnapshotPreBarrier(3L); + assertor.shouldEmit( + testHarness, + rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1", address, null, null, null), + rowOfKind( + RowKind.INSERT, + "Ord#1", + "LineOrd#1", + address, + "Ord#X", + "LineOrd#1", + "SHIP")); + + testHarness.processElement2(updateBeforeRecord("Ord#X", "LineOrd#1", "SHIP")); + testHarness.processElement2(updateAfterRecord("Ord#X", "LineOrd#1", "TRUCK")); + testHarness.prepareSnapshotPreBarrier(4L); + assertor.shouldEmit( + testHarness, + rowOfKind( + RowKind.UPDATE_BEFORE, + "Ord#1", + "LineOrd#1", + address, + "Ord#X", + "LineOrd#1", + "SHIP"), + rowOfKind( + RowKind.INSERT, + "Ord#1", + "LineOrd#1", + address, + "Ord#X", + "LineOrd#1", + "TRUCK")); + + testHarness.processElement2(updateAfterRecord("Ord#X", "LineOrd#1", UNKNOWN)); + testHarness.prepareSnapshotPreBarrier(5L); + assertor.shouldEmit( + testHarness, + rowOfKind( + RowKind.DELETE, + "Ord#1", + "LineOrd#1", + address, + "Ord#X", + "LineOrd#1", + "TRUCK"), + rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1", address, null, null, null)); + } + @Tag("miniBatchSize=4") @Test void testLeftJoinJoinKeyContainsUniqueKeyWithUpdateMultipleCases() throws Exception { @@ -1863,16 +1944,21 @@ final class StreamingMiniBatchJoinOperatorTest extends StreamingJoinOperatorTest FlinkJoinType joinType = flinkJoinTypeExtractor.apply(testInfo.getDisplayName()); int batchSize = miniBatchSizeExtractor.apply(testInfo.getTags()); Long[] ttl = STATE_RETENTION_TIME_EXTRACTOR.apply(testInfo.getTags()); + final boolean hasNonEquiCondition = testInfo.getDisplayName().contains("NonEqui"); return MiniBatchStreamingJoinOperator.newMiniBatchStreamJoinOperator( joinType, leftTypeInfo, rightTypeInfo, - joinCondition, + hasNonEquiCondition ? knownValuesNonEquiCondition() : joinCondition, inputSideSpecs[0], inputSideSpecs[1], isOuter[0], isOuter[1], + // the tests send both retract and upsert inputs + null, + null, + hasNonEquiCondition, new boolean[] {true}, ttl[0], ttl[0],
