diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecTemporalJoin.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecTemporalJoin.java index eb6f68a217090..726fefce0e88e 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecTemporalJoin.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecTemporalJoin.java @@ -50,6 +50,7 @@ import org.apache.flink.table.runtime.operators.join.FlinkJoinType; import org.apache.flink.table.runtime.operators.join.temporal.TemporalProcessTimeJoinOperator; import org.apache.flink.table.runtime.operators.join.temporal.TemporalRowTimeJoinOperator; +import org.apache.flink.table.runtime.operators.join.temporal.TemporalRowTimeJoinOperatorV2; import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; import org.apache.flink.table.types.logical.RowType; import org.apache.flink.util.Preconditions; @@ -74,6 +75,12 @@ producedTransformations = StreamExecTemporalJoin.TEMPORAL_JOIN_TRANSFORMATION, minPlanVersion = FlinkVersion.v1_15, minStateVersion = FlinkVersion.v1_15) +@ExecNodeMetadata( + name = "stream-exec-temporal-join", + version = 2, + producedTransformations = StreamExecTemporalJoin.TEMPORAL_JOIN_TRANSFORMATION, + minPlanVersion = FlinkVersion.v2_4, + minStateVersion = FlinkVersion.v2_4) public class StreamExecTemporalJoin extends ExecNodeBase implements StreamExecNode, SingleTransformationTranslator { @@ -265,7 +272,18 @@ private TwoInputStreamOperator createJoinOperator( long minRetentionTime = config.getStateRetentionTime(); long maxRetentionTime = TableConfigUtils.getMaxIdleStateRetentionTime(config); if (rightTimeAttributeIndex >= 0) { - return new TemporalRowTimeJoinOperator( + if (getVersion() == 1) { + return new TemporalRowTimeJoinOperator( + InternalTypeInfo.of(leftInputType), + InternalTypeInfo.of(rightInputType), + generatedJoinCondition, + leftTimeAttributeIndex, + rightTimeAttributeIndex, + minRetentionTime, + maxRetentionTime, + isLeftOuterJoin); + } + return new TemporalRowTimeJoinOperatorV2( InternalTypeInfo.of(leftInputType), InternalTypeInfo.of(rightInputType), generatedJoinCondition, diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join-key-from-map/plan/temporal-join-table-join-key-from-map.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join-key-from-map/plan/temporal-join-table-join-key-from-map.json new file mode 100644 index 0000000000000..17372c53d132c --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join-key-from-map/plan/temporal-join-table-join-key-from-map.json @@ -0,0 +1,564 @@ +{ + "flinkVersion" : "2.4", + "nodes" : [ { + "id" : 19, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`OrdersNestedId`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "amount", + "dataType" : "BIGINT" + }, { + "name" : "nested_row", + "dataType" : "ROW<`currency` VARCHAR(2147483647)>" + }, { + "name" : "nested_map", + "dataType" : "MAP" + }, { + "name" : "order_time", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "TO_TIMESTAMP(`order_time`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "rowtime", + "expression" : { + "rexNode" : { + "kind" : "INPUT_REF", + "inputIndex" : 4, + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`rowtime`" + } + } ] + } + } + }, + "abilities" : [ { + "type" : "ProjectPushDown", + "projectedFields" : [ [ 0 ], [ 3 ], [ 2 ] ], + "producedType" : "ROW<`amount` BIGINT, `order_time` VARCHAR(2147483647), `nested_map` MAP> NOT NULL" + }, { + "type" : "ReadingMetadata", + "metadataKeys" : [ ], + "producedType" : "ROW<`amount` BIGINT, `order_time` VARCHAR(2147483647), `nested_map` MAP> NOT NULL" + }, { + "type" : "WatermarkPushDown", + "watermarkExpr" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "rowtimeExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, + "idleTimeoutMillis" : -1, + "producedType" : "ROW<`amount` BIGINT, `order_time` VARCHAR(2147483647), `nested_map` MAP> NOT NULL", + "watermarkParams" : { + "emitStrategy" : "ON_EVENT", + "alignGroupName" : null, + "alignMaxDrift" : "PT0S", + "alignUpdateInterval" : "PT1S", + "sourceIdleTimeout" : -1 + } + } ] + }, + "outputType" : "ROW<`amount` BIGINT, `order_time` VARCHAR(2147483647), `nested_map` MAP>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, OrdersNestedId, project=[amount, order_time, nested_map], metadata=[], watermark=[TO_TIMESTAMP(order_time)], watermarkEmitStrategy=[on-event]]], fields=[amount, order_time, nested_map])" + }, { + "id" : 20, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "BIGINT" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$ITEM$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "MAP" + }, { + "kind" : "LITERAL", + "value" : "currency", + "type" : "CHAR(8) NOT NULL" + } ], + "type" : "VARCHAR(2147483647)" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "amount", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, { + "name" : "$f5", + "fieldType" : "VARCHAR(2147483647)" + } ] + }, + "description" : "Calc(select=[amount, Reinterpret(TO_TIMESTAMP(order_time)) AS rowtime, ITEM(nested_map, 'currency') AS $f5])" + }, { + "id" : 21, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 2 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "amount", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, { + "name" : "$f5", + "fieldType" : "VARCHAR(2147483647)" + } ] + }, + "description" : "Exchange(distribution=[hash[$f5]])" + }, { + "id" : 22, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`RatesHistory`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "currency", + "dataType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "dataType" : "BIGINT" + }, { + "name" : "rate_time", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "TO_TIMESTAMP(`rate_time`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "rowtime", + "expression" : { + "rexNode" : { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`rowtime`" + } + } ], + "primaryKey" : { + "name" : "PK_currency", + "type" : "PRIMARY_KEY", + "columns" : [ "currency" ] + } + } + } + }, + "abilities" : [ { + "type" : "WatermarkPushDown", + "watermarkExpr" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "rowtimeExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, + "idleTimeoutMillis" : -1, + "producedType" : "ROW<`currency` VARCHAR(2147483647) NOT NULL, `rate` BIGINT, `rate_time` VARCHAR(2147483647)> NOT NULL", + "watermarkParams" : { + "emitStrategy" : "ON_EVENT", + "alignGroupName" : null, + "alignMaxDrift" : "PT0S", + "alignUpdateInterval" : "PT1S", + "sourceIdleTimeout" : -1 + } + } ] + }, + "outputType" : "ROW<`currency` VARCHAR(2147483647) NOT NULL, `rate` BIGINT, `rate_time` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, RatesHistory, watermark=[TO_TIMESTAMP(rate_time)], watermarkEmitStrategy=[on-event]]], fields=[currency, rate, rate_time])" + }, { + "id" : 23, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "VARCHAR(2147483647) NOT NULL" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "BIGINT" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Calc(select=[currency, rate, Reinterpret(TO_TIMESTAMP(rate_time)) AS rowtime])" + }, { + "id" : 24, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Exchange(distribution=[hash[currency]])" + }, { + "id" : 25, + "type" : "stream-exec-temporal-join_2", + "joinSpec" : { + "joinType" : "INNER", + "leftKeys" : [ 2 ], + "rightKeys" : [ 0 ], + "filterNulls" : [ true ], + "nonEquiCondition" : null + }, + "isTemporalFunctionJoin" : false, + "leftTimeAttributeIndex" : 1, + "rightTimeAttributeIndex" : 2, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + }, { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "amount", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, { + "name" : "$f5", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime0", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "TemporalJoin(joinType=[InnerJoin], where=[(($f5 = currency) AND __TEMPORAL_JOIN_CONDITION(rowtime, rowtime0, __TEMPORAL_JOIN_CONDITION_PRIMARY_KEY(currency), __TEMPORAL_JOIN_LEFT_KEY($f5), __TEMPORAL_JOIN_RIGHT_KEY(currency)))], select=[amount, rowtime, $f5, currency, rate, rowtime0])" + }, { + "id" : 26, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "CALL", + "syntax" : "BINARY", + "internalName" : "$*$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "BIGINT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 4, + "type" : "BIGINT" + } ], + "type" : "BIGINT" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`EXPR$0` BIGINT>", + "description" : "Calc(select=[(amount * rate) AS EXPR$0])" + }, { + "id" : 27, + "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" : "amount", + "dataType" : "BIGINT" + } ] + } + } + } + }, + "inputChangelogMode" : [ "INSERT" ], + "upsertMaterializeStrategy" : "MAP", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`EXPR$0` BIGINT>", + "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[EXPR$0])" + } ], + "edges" : [ { + "source" : 19, + "target" : 20, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 20, + "target" : 21, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 22, + "target" : 23, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 23, + "target" : 24, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 21, + "target" : 25, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 24, + "target" : 25, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 25, + "target" : 26, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 26, + "target" : 27, + "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-temporal-join_2/temporal-join-table-join-key-from-map/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join-key-from-map/savepoint/_metadata new file mode 100644 index 0000000000000..917bea8ad2ea9 Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join-key-from-map/savepoint/_metadata differ diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join-nested-key/plan/temporal-join-table-join-nested-key.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join-nested-key/plan/temporal-join-table-join-nested-key.json new file mode 100644 index 0000000000000..fa4507eef0c11 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join-nested-key/plan/temporal-join-table-join-nested-key.json @@ -0,0 +1,594 @@ +{ + "flinkVersion" : "2.4", + "nodes" : [ { + "id" : 10, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`OrdersNestedId`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "amount", + "dataType" : "BIGINT" + }, { + "name" : "nested_row", + "dataType" : "ROW<`currency` VARCHAR(2147483647)>" + }, { + "name" : "nested_map", + "dataType" : "MAP" + }, { + "name" : "order_time", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "TO_TIMESTAMP(`order_time`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "rowtime", + "expression" : { + "rexNode" : { + "kind" : "INPUT_REF", + "inputIndex" : 4, + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`rowtime`" + } + } ] + } + } + }, + "abilities" : [ { + "type" : "ProjectPushDown", + "projectedFields" : [ [ 0 ], [ 3 ], [ 1 ] ], + "producedType" : "ROW<`amount` BIGINT, `order_time` VARCHAR(2147483647), `nested_row` ROW<`currency` VARCHAR(2147483647)>> NOT NULL" + }, { + "type" : "ReadingMetadata", + "metadataKeys" : [ ], + "producedType" : "ROW<`amount` BIGINT, `order_time` VARCHAR(2147483647), `nested_row` ROW<`currency` VARCHAR(2147483647)>> NOT NULL" + }, { + "type" : "WatermarkPushDown", + "watermarkExpr" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "rowtimeExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, + "idleTimeoutMillis" : -1, + "producedType" : "ROW<`amount` BIGINT, `order_time` VARCHAR(2147483647), `nested_row` ROW<`currency` VARCHAR(2147483647)>> NOT NULL", + "watermarkParams" : { + "emitStrategy" : "ON_EVENT", + "alignGroupName" : null, + "alignMaxDrift" : "PT0S", + "alignUpdateInterval" : "PT1S", + "sourceIdleTimeout" : -1 + } + } ] + }, + "outputType" : "ROW<`amount` BIGINT, `order_time` VARCHAR(2147483647), `nested_row` ROW<`currency` VARCHAR(2147483647)>>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, OrdersNestedId, project=[amount, order_time, nested_row], metadata=[], watermark=[TO_TIMESTAMP(order_time)], watermarkEmitStrategy=[on-event]]], fields=[amount, order_time, nested_row])" + }, { + "id" : 11, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "BIGINT" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$CASE$1", + "operands" : [ { + "kind" : "CALL", + "syntax" : "BINARY", + "internalName" : "$=$1", + "operands" : [ { + "kind" : "FIELD_ACCESS", + "name" : "currency", + "expr" : { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "ROW<`currency` VARCHAR(2147483647)>" + } + }, { + "kind" : "LITERAL", + "value" : "usd", + "type" : "VARCHAR(2147483647) NOT NULL" + } ], + "type" : "BOOLEAN" + }, { + "kind" : "CALL", + "internalName" : "$UPPER$1", + "operands" : [ { + "kind" : "FIELD_ACCESS", + "name" : "currency", + "expr" : { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "ROW<`currency` VARCHAR(2147483647)>" + } + } ], + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "FIELD_ACCESS", + "name" : "currency", + "expr" : { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "ROW<`currency` VARCHAR(2147483647)>" + } + } ], + "type" : "VARCHAR(2147483647)" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "amount", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, { + "name" : "$f5", + "fieldType" : "VARCHAR(2147483647)" + } ] + }, + "description" : "Calc(select=[amount, Reinterpret(TO_TIMESTAMP(order_time)) AS rowtime, CASE((nested_row.currency = 'usd'), UPPER(nested_row.currency), nested_row.currency) AS $f5])" + }, { + "id" : 12, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 2 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "amount", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, { + "name" : "$f5", + "fieldType" : "VARCHAR(2147483647)" + } ] + }, + "description" : "Exchange(distribution=[hash[$f5]])" + }, { + "id" : 13, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`RatesHistory`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "currency", + "dataType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "dataType" : "BIGINT" + }, { + "name" : "rate_time", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "TO_TIMESTAMP(`rate_time`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "rowtime", + "expression" : { + "rexNode" : { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`rowtime`" + } + } ], + "primaryKey" : { + "name" : "PK_currency", + "type" : "PRIMARY_KEY", + "columns" : [ "currency" ] + } + } + } + }, + "abilities" : [ { + "type" : "WatermarkPushDown", + "watermarkExpr" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "rowtimeExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, + "idleTimeoutMillis" : -1, + "producedType" : "ROW<`currency` VARCHAR(2147483647) NOT NULL, `rate` BIGINT, `rate_time` VARCHAR(2147483647)> NOT NULL", + "watermarkParams" : { + "emitStrategy" : "ON_EVENT", + "alignGroupName" : null, + "alignMaxDrift" : "PT0S", + "alignUpdateInterval" : "PT1S", + "sourceIdleTimeout" : -1 + } + } ] + }, + "outputType" : "ROW<`currency` VARCHAR(2147483647) NOT NULL, `rate` BIGINT, `rate_time` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, RatesHistory, watermark=[TO_TIMESTAMP(rate_time)], watermarkEmitStrategy=[on-event]]], fields=[currency, rate, rate_time])" + }, { + "id" : 14, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "VARCHAR(2147483647) NOT NULL" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "BIGINT" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Calc(select=[currency, rate, Reinterpret(TO_TIMESTAMP(rate_time)) AS rowtime])" + }, { + "id" : 15, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Exchange(distribution=[hash[currency]])" + }, { + "id" : 16, + "type" : "stream-exec-temporal-join_2", + "joinSpec" : { + "joinType" : "INNER", + "leftKeys" : [ 2 ], + "rightKeys" : [ 0 ], + "filterNulls" : [ true ], + "nonEquiCondition" : null + }, + "isTemporalFunctionJoin" : false, + "leftTimeAttributeIndex" : 1, + "rightTimeAttributeIndex" : 2, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + }, { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "amount", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, { + "name" : "$f5", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime0", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "TemporalJoin(joinType=[InnerJoin], where=[(($f5 = currency) AND __TEMPORAL_JOIN_CONDITION(rowtime, rowtime0, __TEMPORAL_JOIN_CONDITION_PRIMARY_KEY(currency), __TEMPORAL_JOIN_LEFT_KEY($f5), __TEMPORAL_JOIN_RIGHT_KEY(currency)))], select=[amount, rowtime, $f5, currency, rate, rowtime0])" + }, { + "id" : 17, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "CALL", + "syntax" : "BINARY", + "internalName" : "$*$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "BIGINT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 4, + "type" : "BIGINT" + } ], + "type" : "BIGINT" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`EXPR$0` BIGINT>", + "description" : "Calc(select=[(amount * rate) AS EXPR$0])" + }, { + "id" : 18, + "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" : "amount", + "dataType" : "BIGINT" + } ] + } + } + } + }, + "inputChangelogMode" : [ "INSERT" ], + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`EXPR$0` BIGINT>", + "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[EXPR$0])" + } ], + "edges" : [ { + "source" : 10, + "target" : 11, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 11, + "target" : 12, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 13, + "target" : 14, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 14, + "target" : 15, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 12, + "target" : 16, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 15, + "target" : 16, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 16, + "target" : 17, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 17, + "target" : 18, + "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-temporal-join_2/temporal-join-table-join-nested-key/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join-nested-key/savepoint/_metadata new file mode 100644 index 0000000000000..5a7fd8d1fed17 Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join-nested-key/savepoint/_metadata differ diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join/plan/temporal-join-table-join.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join/plan/temporal-join-table-join.json new file mode 100644 index 0000000000000..7125e3be6d4cd --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join/plan/temporal-join-table-join.json @@ -0,0 +1,543 @@ +{ + "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" : "amount", + "dataType" : "BIGINT" + }, { + "name" : "currency", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "order_time", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "TO_TIMESTAMP(`order_time`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "rowtime", + "expression" : { + "rexNode" : { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`rowtime`" + } + } ] + } + } + }, + "abilities" : [ { + "type" : "WatermarkPushDown", + "watermarkExpr" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "rowtimeExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, + "idleTimeoutMillis" : -1, + "producedType" : "ROW<`amount` BIGINT, `currency` VARCHAR(2147483647), `order_time` VARCHAR(2147483647)> NOT NULL", + "watermarkParams" : { + "emitStrategy" : "ON_EVENT", + "alignGroupName" : null, + "alignMaxDrift" : "PT0S", + "alignUpdateInterval" : "PT1S", + "sourceIdleTimeout" : -1 + } + } ] + }, + "outputType" : "ROW<`amount` BIGINT, `currency` VARCHAR(2147483647), `order_time` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, Orders, watermark=[TO_TIMESTAMP(order_time)], watermarkEmitStrategy=[on-event]]], fields=[amount, currency, order_time])" + }, { + "id" : 2, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "BIGINT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "amount", + "fieldType" : "BIGINT" + }, { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Calc(select=[amount, currency, Reinterpret(TO_TIMESTAMP(order_time)) AS rowtime])" + }, { + "id" : 3, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 1 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "amount", + "fieldType" : "BIGINT" + }, { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Exchange(distribution=[hash[currency]])" + }, { + "id" : 4, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`RatesHistory`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "currency", + "dataType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "dataType" : "BIGINT" + }, { + "name" : "rate_time", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "TO_TIMESTAMP(`rate_time`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "rowtime", + "expression" : { + "rexNode" : { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`rowtime`" + } + } ], + "primaryKey" : { + "name" : "PK_currency", + "type" : "PRIMARY_KEY", + "columns" : [ "currency" ] + } + } + } + }, + "abilities" : [ { + "type" : "WatermarkPushDown", + "watermarkExpr" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "rowtimeExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, + "idleTimeoutMillis" : -1, + "producedType" : "ROW<`currency` VARCHAR(2147483647) NOT NULL, `rate` BIGINT, `rate_time` VARCHAR(2147483647)> NOT NULL", + "watermarkParams" : { + "emitStrategy" : "ON_EVENT", + "alignGroupName" : null, + "alignMaxDrift" : "PT0S", + "alignUpdateInterval" : "PT1S", + "sourceIdleTimeout" : -1 + } + } ] + }, + "outputType" : "ROW<`currency` VARCHAR(2147483647) NOT NULL, `rate` BIGINT, `rate_time` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, RatesHistory, watermark=[TO_TIMESTAMP(rate_time)], watermarkEmitStrategy=[on-event]]], fields=[currency, rate, rate_time])" + }, { + "id" : 5, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "VARCHAR(2147483647) NOT NULL" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "BIGINT" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Calc(select=[currency, rate, Reinterpret(TO_TIMESTAMP(rate_time)) AS rowtime])" + }, { + "id" : 6, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Exchange(distribution=[hash[currency]])" + }, { + "id" : 7, + "type" : "stream-exec-temporal-join_2", + "joinSpec" : { + "joinType" : "INNER", + "leftKeys" : [ 1 ], + "rightKeys" : [ 0 ], + "filterNulls" : [ true ], + "nonEquiCondition" : null + }, + "isTemporalFunctionJoin" : false, + "leftTimeAttributeIndex" : 2, + "rightTimeAttributeIndex" : 2, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + }, { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "amount", + "fieldType" : "BIGINT" + }, { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, { + "name" : "currency0", + "fieldType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime0", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "TemporalJoin(joinType=[InnerJoin], where=[((currency = currency0) AND __TEMPORAL_JOIN_CONDITION(rowtime, rowtime0, __TEMPORAL_JOIN_CONDITION_PRIMARY_KEY(currency0), __TEMPORAL_JOIN_LEFT_KEY(currency), __TEMPORAL_JOIN_RIGHT_KEY(currency0)))], select=[amount, currency, rowtime, currency0, rate, rowtime0])" + }, { + "id" : 8, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "CALL", + "syntax" : "BINARY", + "internalName" : "$*$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "BIGINT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 4, + "type" : "BIGINT" + } ], + "type" : "BIGINT" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`EXPR$0` BIGINT>", + "description" : "Calc(select=[(amount * rate) AS EXPR$0])" + }, { + "id" : 9, + "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" : "amount", + "dataType" : "BIGINT" + } ] + } + } + } + }, + "inputChangelogMode" : [ "INSERT" ], + "upsertMaterializeStrategy" : "VALUE", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`EXPR$0` BIGINT>", + "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[EXPR$0])" + } ], + "edges" : [ { + "source" : 1, + "target" : 2, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 2, + "target" : 3, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 4, + "target" : 5, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 5, + "target" : 6, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 3, + "target" : 7, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 6, + "target" : 7, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 7, + "target" : 8, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 8, + "target" : 9, + "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-temporal-join_2/temporal-join-table-join/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join/savepoint/_metadata new file mode 100644 index 0000000000000..0fbacd88a2946 Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join/savepoint/_metadata differ diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-temporal-function/plan/temporal-join-temporal-function.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-temporal-function/plan/temporal-join-temporal-function.json new file mode 100644 index 0000000000000..af721eab2aee2 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-temporal-function/plan/temporal-join-temporal-function.json @@ -0,0 +1,543 @@ +{ + "flinkVersion" : "2.4", + "nodes" : [ { + "id" : 28, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`Orders`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "amount", + "dataType" : "BIGINT" + }, { + "name" : "currency", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "order_time", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "TO_TIMESTAMP(`order_time`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "rowtime", + "expression" : { + "rexNode" : { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`rowtime`" + } + } ] + } + } + }, + "abilities" : [ { + "type" : "WatermarkPushDown", + "watermarkExpr" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "rowtimeExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, + "idleTimeoutMillis" : -1, + "producedType" : "ROW<`amount` BIGINT, `currency` VARCHAR(2147483647), `order_time` VARCHAR(2147483647)> NOT NULL", + "watermarkParams" : { + "emitStrategy" : "ON_EVENT", + "alignGroupName" : null, + "alignMaxDrift" : "PT0S", + "alignUpdateInterval" : "PT1S", + "sourceIdleTimeout" : -1 + } + } ] + }, + "outputType" : "ROW<`amount` BIGINT, `currency` VARCHAR(2147483647), `order_time` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, Orders, watermark=[TO_TIMESTAMP(order_time)], watermarkEmitStrategy=[on-event]]], fields=[amount, currency, order_time])" + }, { + "id" : 29, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "BIGINT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "amount", + "fieldType" : "BIGINT" + }, { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Calc(select=[amount, currency, Reinterpret(TO_TIMESTAMP(order_time)) AS rowtime])" + }, { + "id" : 30, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 1 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "amount", + "fieldType" : "BIGINT" + }, { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Exchange(distribution=[hash[currency]])" + }, { + "id" : 31, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`RatesHistory`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "currency", + "dataType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "dataType" : "BIGINT" + }, { + "name" : "rate_time", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "TO_TIMESTAMP(`rate_time`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "rowtime", + "expression" : { + "rexNode" : { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`rowtime`" + } + } ], + "primaryKey" : { + "name" : "PK_currency", + "type" : "PRIMARY_KEY", + "columns" : [ "currency" ] + } + } + } + }, + "abilities" : [ { + "type" : "WatermarkPushDown", + "watermarkExpr" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "rowtimeExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, + "idleTimeoutMillis" : -1, + "producedType" : "ROW<`currency` VARCHAR(2147483647) NOT NULL, `rate` BIGINT, `rate_time` VARCHAR(2147483647)> NOT NULL", + "watermarkParams" : { + "emitStrategy" : "ON_EVENT", + "alignGroupName" : null, + "alignMaxDrift" : "PT0S", + "alignUpdateInterval" : "PT1S", + "sourceIdleTimeout" : -1 + } + } ] + }, + "outputType" : "ROW<`currency` VARCHAR(2147483647) NOT NULL, `rate` BIGINT, `rate_time` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, RatesHistory, watermark=[TO_TIMESTAMP(rate_time)], watermarkEmitStrategy=[on-event]]], fields=[currency, rate, rate_time])" + }, { + "id" : 32, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "VARCHAR(2147483647) NOT NULL" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "BIGINT" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Calc(select=[currency, rate, Reinterpret(TO_TIMESTAMP(rate_time)) AS rowtime])" + }, { + "id" : 33, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Exchange(distribution=[hash[currency]])" + }, { + "id" : 34, + "type" : "stream-exec-temporal-join_2", + "joinSpec" : { + "joinType" : "INNER", + "leftKeys" : [ 1 ], + "rightKeys" : [ 0 ], + "filterNulls" : [ true ], + "nonEquiCondition" : null + }, + "isTemporalFunctionJoin" : true, + "leftTimeAttributeIndex" : 2, + "rightTimeAttributeIndex" : 2, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + }, { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "amount", + "fieldType" : "BIGINT" + }, { + "name" : "currency", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, { + "name" : "currency0", + "fieldType" : "VARCHAR(2147483647) NOT NULL" + }, { + "name" : "rate", + "fieldType" : "BIGINT" + }, { + "name" : "rowtime0", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "TemporalJoin(joinType=[InnerJoin], where=[(__TEMPORAL_JOIN_CONDITION(rowtime, rowtime0, __TEMPORAL_JOIN_CONDITION_PRIMARY_KEY(currency0)) AND (currency = currency0))], select=[amount, currency, rowtime, currency0, rate, rowtime0])" + }, { + "id" : 35, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "CALL", + "syntax" : "BINARY", + "internalName" : "$*$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "BIGINT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 4, + "type" : "BIGINT" + } ], + "type" : "BIGINT" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`EXPR$0` BIGINT>", + "description" : "Calc(select=[(amount * rate) AS EXPR$0])" + }, { + "id" : 36, + "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" : "amount", + "dataType" : "BIGINT" + } ] + } + } + } + }, + "inputChangelogMode" : [ "INSERT" ], + "upsertMaterializeStrategy" : "VALUE", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`EXPR$0` BIGINT>", + "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[EXPR$0])" + } ], + "edges" : [ { + "source" : 28, + "target" : 29, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 29, + "target" : 30, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 31, + "target" : 32, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 32, + "target" : 33, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 30, + "target" : 34, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 33, + "target" : 34, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 34, + "target" : 35, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 35, + "target" : 36, + "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-temporal-join_2/temporal-join-temporal-function/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-temporal-function/savepoint/_metadata new file mode 100644 index 0000000000000..7b1df77a75a39 Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-temporal-function/savepoint/_metadata differ diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/temporal/TemporalRowTimeJoinOperatorV2.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/temporal/TemporalRowTimeJoinOperatorV2.java new file mode 100644 index 0000000000000..1a95d6f9c0a06 --- /dev/null +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/temporal/TemporalRowTimeJoinOperatorV2.java @@ -0,0 +1,645 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.runtime.operators.join.temporal; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.common.functions.DefaultOpenContext; +import org.apache.flink.api.common.state.MapState; +import org.apache.flink.api.common.state.MapStateDescriptor; +import org.apache.flink.api.common.state.ValueState; +import org.apache.flink.api.common.state.ValueStateDescriptor; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton; +import org.apache.flink.core.memory.DataInputView; +import org.apache.flink.core.memory.DataOutputView; +import org.apache.flink.metrics.Counter; +import org.apache.flink.runtime.state.StateBackendLoader; +import org.apache.flink.runtime.state.VoidNamespace; +import org.apache.flink.runtime.state.VoidNamespaceSerializer; +import org.apache.flink.streaming.api.operators.InternalTimer; +import org.apache.flink.streaming.api.operators.InternalTimerService; +import org.apache.flink.streaming.api.operators.TimestampedCollector; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.util.RowDataUtil; +import org.apache.flink.table.data.utils.JoinedRowData; +import org.apache.flink.table.runtime.generated.GeneratedJoinCondition; +import org.apache.flink.table.runtime.generated.JoinCondition; +import org.apache.flink.table.runtime.operators.sink.SortedLongSerializer; +import org.apache.flink.table.runtime.typeutils.InternalSerializers; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.util.MathUtils; + +import java.io.IOException; +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.Iterator; +import java.util.List; +import java.util.ListIterator; +import java.util.Map; +import java.util.Optional; +import java.util.TreeMap; + +/** + * The operator for temporal join (FOR SYSTEM_TIME AS OF o.rowtime) on row time, it has no + * limitation about message types of the left input and right input, this means the operator deals + * changelog well. + * + *

For Event-time temporal join, its probe side is a regular table, its build side is a versioned + * table, the version of versioned table can extract from the build side state. This operator works + * by keeping on the state collection of probe and build records to process on next watermark. The + * idea is that between watermarks we are collecting those elements and once we are sure that there + * will be no updates we emit the correct result and clean up the expired data in state. + * + *

Probe-side records that arrive late (their event time is less than or equal to the current + * watermark) are dropped on arrival and counted via the {@code numLateRecordsDropped} metric; they + * are not joined or emitted (not even as null-padded results for left outer joins), because the + * matching build-side version may already have been cleaned up. + * + *

Cleaning up the state drops all the "old" values from the probe side, where "old" is defined + * as older than the current watermark. Build side is also cleaned up in the similar fashion, + * however we always keep at least one record - the latest one - even if it's past the last + * watermark. + * + *

One more trick is how the emitting results and cleaning up is triggered. It is achieved by + * registering timers for the keys. We could register a timer for every probe and build side + * element's event time (when watermark exceeds this timer, that's when we are emitting and/or + * cleaning up the state). However this would cause huge number of registered timers. For example + * with following evenTimes of probe records accumulated: {1, 2, 5, 8, 9}, if we had received + * Watermark(10), it would trigger 5 separate timers for the same key. To avoid that we always keep + * only one single registered timer for any given key, registered for the minimal value. Upon + * triggering it, we process all records with event times older then or equal to currentWatermark. + * + *

Compared to {@link TemporalRowTimeJoinOperator}, this version stores the probe side keyed by + * {@link LeftTimeIndexKey} (row time first, then arrival index) and the build side with {@link + * SortedLongSerializer} as key serializer. Both serializations preserve numeric order under + * unsigned lexicographic byte comparison, so state backends that iterate map state in + * serialized-key order (RocksDB, ForSt) return the entries in ascending time order. On such + * backends each watermark firing only reads state entries up to the watermark and stops, instead of + * scanning the whole probe side and materializing plus sorting the whole build side. On unordered + * backends (heap) the behavior is equivalent to {@link TemporalRowTimeJoinOperator}. The two + * operators have incompatible state layouts. + */ +public class TemporalRowTimeJoinOperatorV2 extends BaseTwoInputStreamOperatorWithStateRetention { + + private static final long serialVersionUID = 1L; + + private static final String NEXT_LEFT_INDEX_STATE_NAME = "next-index"; + private static final String LEFT_STATE_NAME = "left"; + private static final String RIGHT_STATE_NAME = "right"; + private static final String REGISTERED_TIMER_STATE_NAME = "timer"; + private static final String TIMERS_STATE_NAME = "timers"; + private static final String LATE_ELEMENTS_DROPPED_METRIC_NAME = "numLateRecordsDropped"; + + private final boolean isLeftOuterJoin; + private final InternalTypeInfo leftType; + private final InternalTypeInfo rightType; + private final GeneratedJoinCondition generatedJoinCondition; + private final int leftTimeAttribute; + private final int rightTimeAttribute; + + private final RowtimeComparator rightRowtimeComparator; + + /** Incremental index generator for the arrival index part of {@link #leftState}'s keys. */ + private transient ValueState nextLeftIndex; + + /** + * Mapping from (row time, arrival index) into the left side `Row`. On ordered state backends + * the entries are iterated in ascending (row time, arrival index) order, which allows stopping + * at the first entry newer than the current watermark. + */ + private transient MapState leftState; + + /** + * Mapping from timestamp to right side `Row`. The key serializer preserves numeric order in + * serialized form, so on ordered state backends the entries are iterated in ascending timestamp + * order. + */ + private transient MapState rightState; + + // Long for correct handling of default null + private transient ValueState registeredTimer; + private transient TimestampedCollector collector; + private transient InternalTimerService timerService; + + private transient JoinCondition joinCondition; + private transient JoinedRowData outRow; + private transient GenericRowData rightNullRow; + + private transient Counter numLateRecordsDropped; + + /** Whether the state backend iterates map state in ascending serialized-key order. */ + private transient boolean isOrderedStateBackend; + + public TemporalRowTimeJoinOperatorV2( + InternalTypeInfo leftType, + InternalTypeInfo rightType, + GeneratedJoinCondition generatedJoinCondition, + int leftTimeAttribute, + int rightTimeAttribute, + long minRetentionTime, + long maxRetentionTime, + boolean isLeftOuterJoin) { + super(minRetentionTime, maxRetentionTime); + this.leftType = leftType; + this.rightType = rightType; + this.generatedJoinCondition = generatedJoinCondition; + this.leftTimeAttribute = leftTimeAttribute; + this.rightTimeAttribute = rightTimeAttribute; + this.rightRowtimeComparator = new RowtimeComparator(rightTimeAttribute); + this.isLeftOuterJoin = isLeftOuterJoin; + } + + @Override + public void open() throws Exception { + super.open(); + joinCondition = + generatedJoinCondition.newInstance(getRuntimeContext().getUserCodeClassLoader()); + joinCondition.setRuntimeContext(getRuntimeContext()); + joinCondition.open(DefaultOpenContext.INSTANCE); + + nextLeftIndex = + getRuntimeContext() + .getState( + new ValueStateDescriptor<>(NEXT_LEFT_INDEX_STATE_NAME, Types.LONG)); + leftState = + getRuntimeContext() + .getMapState( + new MapStateDescriptor<>( + LEFT_STATE_NAME, + LeftTimeIndexKeySerializer.INSTANCE, + InternalSerializers.create(leftType.toRowType()))); + rightState = + getRuntimeContext() + .getMapState( + new MapStateDescriptor<>( + RIGHT_STATE_NAME, + SortedLongSerializer.INSTANCE, + InternalSerializers.create(rightType.toRowType()))); + registeredTimer = + getRuntimeContext() + .getState( + new ValueStateDescriptor<>( + REGISTERED_TIMER_STATE_NAME, Types.LONG)); + + timerService = + getInternalTimerService(TIMERS_STATE_NAME, VoidNamespaceSerializer.INSTANCE, this); + + final String backendId = getKeyedStateBackend().getBackendTypeIdentifier(); + isOrderedStateBackend = + StateBackendLoader.ROCKSDB_STATE_BACKEND_NAME.equals(backendId) + || StateBackendLoader.FORST_STATE_BACKEND_NAME.equals(backendId); + + outRow = new JoinedRowData(); + rightNullRow = new GenericRowData(rightType.toRowType().getFieldCount()); + collector = new TimestampedCollector<>(output); + + numLateRecordsDropped = + getRuntimeContext().getMetricGroup().counter(LATE_ELEMENTS_DROPPED_METRIC_NAME); + } + + @Override + public void processElement1(StreamRecord element) throws Exception { + RowData row = element.getValue(); + long leftTime = getLeftTime(row); + if (leftTime <= timerService.currentWatermark()) { + // The probe-side record is late. Drop it, because the matching build-side version may + // already have been cleaned up. + numLateRecordsDropped.inc(); + return; + } + leftState.put(new LeftTimeIndexKey(leftTime, getNextLeftIndex()), row); + registerSmallestTimer(leftTime); // Timer to emit and clean up the state + + registerProcessingCleanupTimer(); + } + + @Override + public void processElement2(StreamRecord element) throws Exception { + RowData row = element.getValue(); + + long rowTime = getRightTime(row); + rightState.put(rowTime, row); + registerSmallestTimer(rowTime); // Timer to clean up the state + + registerProcessingCleanupTimer(); + } + + @Override + public void onEventTime(InternalTimer timer) throws Exception { + registeredTimer.clear(); + long lastUnprocessedTime = emitResultAndCleanUpState(timerService.currentWatermark()); + if (lastUnprocessedTime < Long.MAX_VALUE) { + registerTimer(lastUnprocessedTime); + } + + // if we have more state at any side, then update the timer, else clean it up. + if (stateCleaningEnabled) { + if (lastUnprocessedTime < Long.MAX_VALUE || !rightState.isEmpty()) { + registerProcessingCleanupTimer(); + } else { + cleanupLastTimer(); + nextLeftIndex.clear(); + } + } + } + + @Override + public void close() throws Exception { + if (joinCondition != null) { + joinCondition.close(); + } + super.close(); + } + + /** + * @return a row time of the oldest unprocessed probe record or Long.MaxValue, if all records + * have been processed. + */ + private long emitResultAndCleanUpState(long currentWatermark) throws Exception { + List rightRowsSorted = getRightRowsSorted(currentWatermark); + long lastUnprocessedTime = Long.MAX_VALUE; + + Iterator> leftIterator = + leftState.entries().iterator(); + // the output records' order should keep same with left input records arrival order + final Map orderedLeftRecords = new TreeMap<>(); + + while (leftIterator.hasNext()) { + Map.Entry entry = leftIterator.next(); + LeftTimeIndexKey leftKey = entry.getKey(); + if (leftKey.timestamp <= currentWatermark) { + orderedLeftRecords.put(leftKey.index, entry.getValue()); + leftIterator.remove(); + } else if (isOrderedStateBackend) { + // Entries are iterated in ascending (timestamp, index) order, so the first entry + // newer than the watermark carries the minimal remaining timestamp. + lastUnprocessedTime = leftKey.timestamp; + break; + } else { + lastUnprocessedTime = Math.min(lastUnprocessedTime, leftKey.timestamp); + } + } + + // iterate the triggered left records in the ascending order of the arrival index, i.e. the + // arrival order. + orderedLeftRecords.forEach( + (leftSeq, leftRow) -> { + long leftTime = getLeftTime(leftRow); + Optional rightRow = latestRightRowToJoin(rightRowsSorted, leftTime); + if (rightRow.isPresent() && RowDataUtil.isAccumulateMsg(rightRow.get())) { + if (joinCondition.apply(leftRow, rightRow.get())) { + collectJoinedRow(leftRow, rightRow.get()); + } else { + if (isLeftOuterJoin) { + collectJoinedRow(leftRow, rightNullRow); + } + } + } else { + if (isLeftOuterJoin) { + collectJoinedRow(leftRow, rightNullRow); + } + } + }); + orderedLeftRecords.clear(); + + cleanupExpiredVersionInState(currentWatermark, rightRowsSorted); + return lastUnprocessedTime; + } + + private void collectJoinedRow(RowData leftSideRow, RowData rightRow) { + outRow.setRowKind(leftSideRow.getRowKind()); + outRow.replace(leftSideRow, rightRow); + collector.collect(outRow); + } + + /** + * Removes all expired version in the versioned table's state according to current watermark. + */ + private void cleanupExpiredVersionInState(long currentWatermark, List rightRowsSorted) + throws Exception { + int i = 0; + int indexToKeep = firstIndexToKeep(currentWatermark, rightRowsSorted); + // clean old version data that behind current watermark + while (i < indexToKeep) { + long rightTime = getRightTime(rightRowsSorted.get(i)); + rightState.remove(rightTime); + i += 1; + } + } + + /** + * The method to be called when a cleanup timer fires. + * + * @param time The timestamp of the fired timer. + */ + @Override + public void cleanupState(long time) { + leftState.clear(); + rightState.clear(); + nextLeftIndex.clear(); + registeredTimer.clear(); + } + + private int firstIndexToKeep(long timerTimestamp, List rightRowsSorted) { + int firstIndexNewerThenTimer = + indexOfFirstElementNewerThanTimer(timerTimestamp, rightRowsSorted); + + if (firstIndexNewerThenTimer < 0) { + return rightRowsSorted.size() - 1; + } else { + return firstIndexNewerThenTimer - 1; + } + } + + private int indexOfFirstElementNewerThanTimer(long timerTimestamp, List list) { + ListIterator iter = list.listIterator(); + while (iter.hasNext()) { + if (getRightTime(iter.next()) > timerTimestamp) { + return iter.previousIndex(); + } + } + return -1; + } + + /** + * Binary search {@code rightRowsSorted} to find the latest right row to join with {@code + * leftTime}. Latest means a right row with largest time that is still smaller or equal to + * {@code leftTime}. For example with: rightState = [1(+I), 4(+U), 7(+U), 9(-D), 12(I)], + * + *

If left time is 6, the valid period should be [4, 7), data 4(+U) should be joined. + * + *

If left time is 10, the valid period should be [9, 12), but data 9(-D) is a DELETE message + * which means the correspond version has no data in period [9, 12), data 9(-D) should not be + * correlated. + * + * @return found element or {@code Optional.empty} If such row was not found (either {@code + * rightRowsSorted} is empty or all {@code rightRowsSorted} are are newer). + */ + private Optional latestRightRowToJoin(List rightRowsSorted, long leftTime) { + return latestRightRowToJoin(rightRowsSorted, 0, rightRowsSorted.size() - 1, leftTime); + } + + private Optional latestRightRowToJoin( + List rightRowsSorted, int low, int high, long leftTime) { + if (low > high) { + // exact value not found, we are returning largest from the values smaller then leftTime + if (low - 1 < 0) { + return Optional.empty(); + } else { + return Optional.of(rightRowsSorted.get(low - 1)); + } + } else { + int mid = (low + high) >>> 1; + RowData midRow = rightRowsSorted.get(mid); + long midTime = getRightTime(midRow); + int cmp = Long.compare(midTime, leftTime); + if (cmp < 0) { + return latestRightRowToJoin(rightRowsSorted, mid + 1, high, leftTime); + } else if (cmp > 0) { + return latestRightRowToJoin(rightRowsSorted, low, mid - 1, leftTime); + } else { + return Optional.of(midRow); + } + } + } + + private void registerSmallestTimer(long timestamp) throws IOException { + Long currentRegisteredTimer = registeredTimer.value(); + if (currentRegisteredTimer == null) { + registerTimer(timestamp); + } else if (currentRegisteredTimer > timestamp) { + timerService.deleteEventTimeTimer(VoidNamespace.INSTANCE, currentRegisteredTimer); + registerTimer(timestamp); + } + } + + private void registerTimer(long timestamp) throws IOException { + registeredTimer.update(timestamp); + timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, timestamp); + } + + private List getRightRowsSorted(long currentWatermark) throws Exception { + List rightRows = new ArrayList<>(); + if (isOrderedStateBackend) { + for (Map.Entry entry : rightState.entries()) { + if (entry.getKey() > currentWatermark) { + break; + } + rightRows.add(entry.getValue()); + } + } else { + for (RowData row : rightState.values()) { + rightRows.add(row); + } + rightRows.sort(rightRowtimeComparator); + } + return rightRows; + } + + private long getNextLeftIndex() throws IOException { + Long index = nextLeftIndex.value(); + if (index == null) { + index = 0L; + } + nextLeftIndex.update(index + 1); + return index; + } + + private long getLeftTime(RowData leftRow) { + return leftRow.getLong(leftTimeAttribute); + } + + private long getRightTime(RowData rightRow) { + return rightRow.getLong(rightTimeAttribute); + } + + // ------------------------------------------------------------------------------------------ + + private static class RowtimeComparator implements Comparator, Serializable { + + private static final long serialVersionUID = 1L; + + private final int timeAttribute; + + private RowtimeComparator(int timeAttribute) { + this.timeAttribute = timeAttribute; + } + + @Override + public int compare(RowData o1, RowData o2) { + long o1Time = o1.getLong(timeAttribute); + long o2Time = o2.getLong(timeAttribute); + return Long.compare(o1Time, o2Time); + } + } + + /** + * Key of {@link #leftState}: the row time of the probe record first, then a per-key arrival + * index to keep records with the same row time distinct and to restore arrival order at + * emission time. + */ + public static final class LeftTimeIndexKey { + + private final long timestamp; + private final long index; + + public LeftTimeIndexKey(long timestamp, long index) { + this.timestamp = timestamp; + this.index = index; + } + + public long getTimestamp() { + return timestamp; + } + + public long getIndex() { + return index; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + LeftTimeIndexKey that = (LeftTimeIndexKey) o; + return timestamp == that.timestamp && index == that.index; + } + + @Override + public int hashCode() { + int result = Long.hashCode(timestamp); + return 31 * result + Long.hashCode(index); + } + + @Override + public String toString() { + return "LeftTimeIndexKey{timestamp=" + timestamp + ", index=" + index + '}'; + } + } + + /** + * Serializer for {@link LeftTimeIndexKey} that produces a lexicographically sortable byte + * representation. + * + * @see SortedLongSerializer + */ + public static final class LeftTimeIndexKeySerializer + extends TypeSerializerSingleton { + + private static final long serialVersionUID = 1L; + + /** Sharable instance of the LeftTimeIndexKeySerializer. */ + public static final LeftTimeIndexKeySerializer INSTANCE = new LeftTimeIndexKeySerializer(); + + private static final LeftTimeIndexKey ZERO = new LeftTimeIndexKey(0L, 0L); + + @Override + public boolean isImmutableType() { + return true; + } + + @Override + public LeftTimeIndexKey createInstance() { + return ZERO; + } + + @Override + public LeftTimeIndexKey copy(LeftTimeIndexKey from) { + return from; + } + + @Override + public LeftTimeIndexKey copy(LeftTimeIndexKey from, LeftTimeIndexKey reuse) { + return from; + } + + @Override + public int getLength() { + return 2 * Long.BYTES; + } + + @Override + public void serialize(LeftTimeIndexKey record, DataOutputView target) throws IOException { + target.writeLong(MathUtils.flipSignBit(record.timestamp)); + target.writeLong(MathUtils.flipSignBit(record.index)); + } + + @Override + public LeftTimeIndexKey deserialize(DataInputView source) throws IOException { + long timestamp = MathUtils.flipSignBit(source.readLong()); + long index = MathUtils.flipSignBit(source.readLong()); + return new LeftTimeIndexKey(timestamp, index); + } + + @Override + public LeftTimeIndexKey deserialize(LeftTimeIndexKey reuse, DataInputView source) + throws IOException { + return deserialize(source); + } + + @Override + public void copy(DataInputView source, DataOutputView target) throws IOException { + target.writeLong(source.readLong()); + target.writeLong(source.readLong()); + } + + @Override + public TypeSerializerSnapshot snapshotConfiguration() { + return new LeftTimeIndexKeySerializerSnapshot(); + } + + /** Serializer configuration snapshot for compatibility and format evolution. */ + @SuppressWarnings("WeakerAccess") + public static final class LeftTimeIndexKeySerializerSnapshot + extends SimpleTypeSerializerSnapshot { + + public LeftTimeIndexKeySerializerSnapshot() { + super(() -> INSTANCE); + } + } + } + + @VisibleForTesting + static String getNextLeftIndexStateName() { + return NEXT_LEFT_INDEX_STATE_NAME; + } + + @VisibleForTesting + static String getRegisteredTimerStateName() { + return REGISTERED_TIMER_STATE_NAME; + } + + @VisibleForTesting + Counter getNumLateRecordsDropped() { + return numLateRecordsDropped; + } + + @VisibleForTesting + boolean isOrderedStateBackend() { + return isOrderedStateBackend; + } +} diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/temporal/LeftTimeIndexKeySerializerTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/temporal/LeftTimeIndexKeySerializerTest.java new file mode 100644 index 0000000000000..e46642f79d178 --- /dev/null +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/temporal/LeftTimeIndexKeySerializerTest.java @@ -0,0 +1,122 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.runtime.operators.join.temporal; + +import org.apache.flink.api.common.typeutils.SerializerTestBase; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.core.memory.DataOutputSerializer; +import org.apache.flink.table.runtime.operators.join.temporal.TemporalRowTimeJoinOperatorV2.LeftTimeIndexKey; +import org.apache.flink.table.runtime.operators.join.temporal.TemporalRowTimeJoinOperatorV2.LeftTimeIndexKeySerializer; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Random; + +import static org.assertj.core.api.Assertions.assertThat; + +/** A test for the {@link LeftTimeIndexKeySerializer}. */ +class LeftTimeIndexKeySerializerTest extends SerializerTestBase { + + @Override + protected TypeSerializer createSerializer() { + return LeftTimeIndexKeySerializer.INSTANCE; + } + + @Override + protected int getLength() { + return 16; + } + + @Override + protected Class getTypeClass() { + return LeftTimeIndexKey.class; + } + + @Override + protected LeftTimeIndexKey[] getTestData() { + Random rnd = new Random(874597969123412341L); + long rndLong = rnd.nextLong(); + + return new LeftTimeIndexKey[] { + new LeftTimeIndexKey(0L, 0L), + new LeftTimeIndexKey(1L, 0L), + new LeftTimeIndexKey(-1L, 0L), + new LeftTimeIndexKey(Long.MAX_VALUE, Long.MAX_VALUE), + new LeftTimeIndexKey(Long.MIN_VALUE, 0L), + new LeftTimeIndexKey(42L, 1L), + new LeftTimeIndexKey(42L, 2L), + new LeftTimeIndexKey(rndLong, 3L), + new LeftTimeIndexKey(-rndLong, 4L) + }; + } + + @Test + void testSerializedByteOrderMatchesNumericOrder() throws IOException { + List keys = new ArrayList<>(); + long[] interestingValues = { + Long.MIN_VALUE, + Long.MIN_VALUE + 1, + -42L, + -1L, + 0L, + 1L, + 42L, + Long.MAX_VALUE - 1, + Long.MAX_VALUE + }; + for (long timestamp : interestingValues) { + for (long index : interestingValues) { + keys.add(new LeftTimeIndexKey(timestamp, index)); + } + } + Random rnd = new Random(42); + for (int i = 0; i < 100; i++) { + keys.add(new LeftTimeIndexKey(rnd.nextLong(), rnd.nextLong())); + } + Collections.shuffle(keys, rnd); + + List numericallySorted = new ArrayList<>(keys); + numericallySorted.sort( + Comparator.comparingLong(LeftTimeIndexKey::getTimestamp) + .thenComparingLong(LeftTimeIndexKey::getIndex)); + + List byteSorted = new ArrayList<>(keys); + byteSorted.sort( + Comparator.comparing( + LeftTimeIndexKeySerializerTest::serializeToBytes, Arrays::compareUnsigned)); + + assertThat(byteSorted).containsExactlyElementsOf(numericallySorted); + } + + private static byte[] serializeToBytes(LeftTimeIndexKey key) { + DataOutputSerializer out = new DataOutputSerializer(16); + try { + LeftTimeIndexKeySerializer.INSTANCE.serialize(key, out); + } catch (IOException e) { + throw new RuntimeException(e); + } + return out.getCopyOfBuffer(); + } +} diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/temporal/TemporalRowTimeJoinOperatorV2Test.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/temporal/TemporalRowTimeJoinOperatorV2Test.java new file mode 100644 index 0000000000000..6d43bbc413767 --- /dev/null +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/temporal/TemporalRowTimeJoinOperatorV2Test.java @@ -0,0 +1,503 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.runtime.operators.join.temporal; + +import org.apache.flink.api.common.state.ValueStateDescriptor; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.runtime.state.StateBackend; +import org.apache.flink.runtime.state.hashmap.HashMapStateBackend; +import org.apache.flink.state.rocksdb.EmbeddedRocksDBStateBackend; +import org.apache.flink.streaming.api.watermark.Watermark; +import org.apache.flink.streaming.util.KeyedTwoInputStreamOperatorTestHarness; +import org.apache.flink.table.data.RowData; + +import org.junit.jupiter.api.Named; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +import java.util.ArrayList; +import java.util.List; +import java.util.stream.Stream; + +import static org.apache.flink.table.runtime.util.StreamRecordUtils.deleteRecord; +import static org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord; +import static org.apache.flink.table.runtime.util.StreamRecordUtils.updateAfterRecord; +import static org.apache.flink.table.runtime.util.StreamRecordUtils.updateBeforeRecord; +import static org.assertj.core.api.Assertions.assertThat; + +/** Harness tests for {@link TemporalRowTimeJoinOperatorV2}. */ +class TemporalRowTimeJoinOperatorV2Test extends TemporalTimeJoinOperatorTestBase { + + private static Stream> stateBackends() { + return Stream.of( + Named.of("heap", new HashMapStateBackend()), + Named.of("rocksdb", new EmbeddedRocksDBStateBackend())); + } + + @ParameterizedTest(name = "backend={0}") + @MethodSource("stateBackends") + void testRowTimeInnerTemporalJoin(StateBackend backend) throws Exception { + List expectedOutput = new ArrayList<>(); + expectedOutput.add(new Watermark(0)); + expectedOutput.add(new Watermark(2)); + expectedOutput.add(insertRecord(3L, "k1", "1a3", 2L, "k1", "1a2")); + expectedOutput.add(new Watermark(5)); + expectedOutput.add(insertRecord(6L, "k2", "2a3", 4L, "k2", "2a4")); + expectedOutput.add(new Watermark(8)); + expectedOutput.add(new Watermark(9)); + expectedOutput.add(insertRecord(11L, "k2", "5a12", 10L, "k2", "2a6")); + expectedOutput.add(new Watermark(13)); + + testRowTimeTemporalJoin(backend, false, expectedOutput); + } + + @ParameterizedTest(name = "backend={0}") + @MethodSource("stateBackends") + void testRowTimeLeftTemporalJoin(StateBackend backend) throws Exception { + List expectedOutput = new ArrayList<>(); + expectedOutput.add(new Watermark(0)); + expectedOutput.add(insertRecord(1L, "k1", "1a1", null, null, null)); + expectedOutput.add(new Watermark(2)); + expectedOutput.add(insertRecord(3L, "k1", "1a3", 2L, "k1", "1a2")); + expectedOutput.add(new Watermark(5)); + expectedOutput.add(insertRecord(6L, "k2", "2a3", 4L, "k2", "2a4")); + expectedOutput.add(new Watermark(8)); + expectedOutput.add(insertRecord(9L, "k2", "5a11", null, null, null)); + expectedOutput.add(new Watermark(9)); + expectedOutput.add(insertRecord(11L, "k2", "5a12", 10L, "k2", "2a6")); + expectedOutput.add(new Watermark(13)); + + testRowTimeTemporalJoin(backend, true, expectedOutput); + } + + private void testRowTimeTemporalJoin( + StateBackend backend, boolean isLeftOuterJoin, List expectedOutput) + throws Exception { + TemporalRowTimeJoinOperatorV2 joinOperator = + new TemporalRowTimeJoinOperatorV2( + rowType, rowType, joinCondition, 0, 0, 0, 0, isLeftOuterJoin); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinOperator, backend); + + testHarness.open(); + assertThat(joinOperator.isOrderedStateBackend()) + .isEqualTo(backend instanceof EmbeddedRocksDBStateBackend); + + testHarness.processWatermark1(new Watermark(0)); + testHarness.processWatermark2(new Watermark(0)); + + testHarness.processElement1(insertRecord(1L, "k1", "1a1")); + testHarness.processElement2(insertRecord(2L, "k1", "1a2")); + + testHarness.processWatermark1(new Watermark(2)); + testHarness.processWatermark2(new Watermark(2)); + + testHarness.processElement1(insertRecord(3L, "k1", "1a3")); + testHarness.processElement2(insertRecord(4L, "k2", "2a4")); + + testHarness.processWatermark1(new Watermark(5)); + testHarness.processWatermark2(new Watermark(5)); + + testHarness.processElement1(insertRecord(6L, "k2", "2a3")); + testHarness.processElement2(updateBeforeRecord(7L, "k2", "2a4")); + testHarness.processElement2(updateAfterRecord(7L, "k2", "2a5")); + + testHarness.processWatermark1(new Watermark(8)); + testHarness.processWatermark2(new Watermark(9)); + + testHarness.processElement1(insertRecord(9L, "k2", "5a11")); + testHarness.processElement1(insertRecord(11L, "k2", "5a12")); + testHarness.processElement2(deleteRecord(9L, "k2", "2a5")); + testHarness.processElement2(insertRecord(10L, "k2", "2a6")); + + testHarness.processWatermark1(new Watermark(13)); + testHarness.processWatermark2(new Watermark(13)); + + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + @ParameterizedTest(name = "backend={0}") + @MethodSource("stateBackends") + void testRowTimeTemporalJoinWithStateRetention(StateBackend backend) throws Exception { + final int minRetentionTime = 4; + final int maxRetentionTime = minRetentionTime * 3 / 2; + TemporalRowTimeJoinOperatorV2 joinOperator = + new TemporalRowTimeJoinOperatorV2( + rowType, + rowType, + joinCondition, + 0, + 0, + minRetentionTime, + maxRetentionTime, + true); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinOperator, backend); + testHarness.open(); + + testHarness.setProcessingTime(3); + testHarness.processElement2(insertRecord(3L, "k1", "0a3")); + testHarness.setProcessingTime(6); + testHarness.processElement1(insertRecord(6L, "k1", "0a6")); + + testHarness.processWatermark1(new Watermark(7)); + testHarness.processWatermark2(new Watermark(7)); + testHarness.processElement2(updateBeforeRecord(3L, "k1", "0a3")); + testHarness.processElement2(updateAfterRecord(3L, "k1", "0a5")); + + testHarness.setProcessingTime(9); + testHarness.processElement1(insertRecord(9L, "k1", "7a9")); + + testHarness.processWatermark1(new Watermark(13)); + testHarness.processWatermark2(new Watermark(13)); + + testHarness.setProcessingTime(9 + maxRetentionTime); + testHarness.processElement1(insertRecord(15L, "k1", "13a15")); + + testHarness.processWatermark1(new Watermark(15)); + testHarness.processWatermark2(new Watermark(16)); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(insertRecord(6L, "k1", "0a6", 3L, "k1", "0a3")); + expectedOutput.add(new Watermark(7)); + expectedOutput.add(insertRecord(9L, "k1", "7a9", 3L, "k1", "0a5")); + expectedOutput.add(new Watermark(13)); + expectedOutput.add(insertRecord(15L, "k1", "13a15", null, null, null)); + expectedOutput.add(new Watermark(15)); + + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + assertThat( + joinOperator + .getKeyedStateStore() + .getState( + new ValueStateDescriptor<>( + TemporalRowTimeJoinOperatorV2 + .getNextLeftIndexStateName(), + Types.LONG)) + .value()) + .isNull(); + assertThat( + joinOperator + .getKeyedStateStore() + .getState( + new ValueStateDescriptor<>( + TemporalRowTimeJoinOperatorV2 + .getRegisteredTimerStateName(), + Types.LONG)) + .value()) + .isNull(); + + testHarness.close(); + } + + @ParameterizedTest(name = "backend={0}") + @MethodSource("stateBackends") + void testRowTimeInnerTemporalJoinOnUpsertSource(StateBackend backend) throws Exception { + List expectedOutput = new ArrayList<>(); + expectedOutput.add(new Watermark(0)); + expectedOutput.add(new Watermark(2)); + expectedOutput.add(updateAfterRecord(3L, "k1", "1a3", 2L, "k1", "1a2")); + expectedOutput.add(new Watermark(5)); + expectedOutput.add(insertRecord(6L, "k2", "2a3", 4L, "k2", "2a4")); + expectedOutput.add(new Watermark(8)); + expectedOutput.add(new Watermark(9)); + expectedOutput.add(insertRecord(11L, "k2", "5a12", 10L, "k2", "2a6")); + expectedOutput.add(new Watermark(13)); + + testRowTimeTemporalJoinOnUpsertSource(backend, false, expectedOutput); + } + + @ParameterizedTest(name = "backend={0}") + @MethodSource("stateBackends") + void testRowTimeLeftTemporalJoinOnUpsertSource(StateBackend backend) throws Exception { + List expectedOutput = new ArrayList<>(); + expectedOutput.add(new Watermark(0)); + expectedOutput.add(insertRecord(1L, "k1", "1a1", null, null, null)); + expectedOutput.add(new Watermark(2)); + expectedOutput.add(updateAfterRecord(3L, "k1", "1a3", 2L, "k1", "1a2")); + expectedOutput.add(new Watermark(5)); + expectedOutput.add(insertRecord(6L, "k2", "2a3", 4L, "k2", "2a4")); + expectedOutput.add(new Watermark(8)); + expectedOutput.add(insertRecord(9L, "k2", "5a11", null, null, null)); + expectedOutput.add(new Watermark(9)); + expectedOutput.add(insertRecord(11L, "k2", "5a12", 10L, "k2", "2a6")); + expectedOutput.add(new Watermark(13)); + + testRowTimeTemporalJoinOnUpsertSource(backend, true, expectedOutput); + } + + private void testRowTimeTemporalJoinOnUpsertSource( + StateBackend backend, boolean isLeftOuterJoin, List expectedOutput) + throws Exception { + TemporalRowTimeJoinOperatorV2 joinOperator = + new TemporalRowTimeJoinOperatorV2( + rowType, rowType, joinCondition, 0, 0, 0, 0, isLeftOuterJoin); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinOperator, backend); + + testHarness.open(); + + testHarness.processWatermark1(new Watermark(0)); + testHarness.processWatermark2(new Watermark(0)); + + testHarness.processElement1(insertRecord(1L, "k1", "1a1")); + testHarness.processElement2(insertRecord(2L, "k1", "1a2")); + + testHarness.processWatermark1(new Watermark(2)); + testHarness.processWatermark2(new Watermark(2)); + + testHarness.processElement1(updateAfterRecord(3L, "k1", "1a3")); + testHarness.processElement2(insertRecord(4L, "k2", "2a4")); + + testHarness.processWatermark1(new Watermark(5)); + testHarness.processWatermark2(new Watermark(5)); + + testHarness.processElement1(insertRecord(6L, "k2", "2a3")); + testHarness.processElement2(updateAfterRecord(7L, "k2", "2a5")); + + testHarness.processWatermark1(new Watermark(8)); + testHarness.processWatermark2(new Watermark(9)); + + testHarness.processElement1(insertRecord(9L, "k2", "5a11")); + testHarness.processElement1(insertRecord(11L, "k2", "5a12")); + testHarness.processElement2(deleteRecord(9L, "k2", "2a5")); + testHarness.processElement2(insertRecord(10L, "k2", "2a6")); + + testHarness.processWatermark1(new Watermark(13)); + testHarness.processWatermark2(new Watermark(13)); + + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + @ParameterizedTest(name = "backend={0}") + @MethodSource("stateBackends") + void testRowTimeInnerTemporalJoinLateRecords(StateBackend backend) throws Exception { + List expectedOutput = new ArrayList<>(); + expectedOutput.add(new Watermark(1)); + expectedOutput.add(insertRecord(3L, "k1", "1a3", 2L, "k1", "2a2")); + expectedOutput.add(new Watermark(5)); + expectedOutput.add(insertRecord(7L, "k1", "1a7", 2L, "k1", "2a2")); + expectedOutput.add(new Watermark(8)); + expectedOutput.add(new Watermark(11)); + expectedOutput.add(insertRecord(13L, "k2", "1a13", 9L, "k2", "2a9")); + expectedOutput.add(new Watermark(13)); + expectedOutput.add(new Watermark(15)); + + testRowTimeTemporalJoinLateRecords(backend, false, expectedOutput); + } + + @ParameterizedTest(name = "backend={0}") + @MethodSource("stateBackends") + void testRowTimeLeftTemporalJoinLateRecords(StateBackend backend) throws Exception { + List expectedOutput = new ArrayList<>(); + expectedOutput.add(new Watermark(1)); + expectedOutput.add(insertRecord(3L, "k1", "1a3", 2L, "k1", "2a2")); + expectedOutput.add(new Watermark(5)); + expectedOutput.add(insertRecord(7L, "k1", "1a7", 2L, "k1", "2a2")); + expectedOutput.add(new Watermark(8)); + expectedOutput.add(insertRecord(10L, "k2", "1a10", null, null, null)); + expectedOutput.add(new Watermark(11)); + expectedOutput.add(insertRecord(13L, "k2", "1a13", 9L, "k2", "2a9")); + expectedOutput.add(new Watermark(13)); + expectedOutput.add(new Watermark(15)); + + testRowTimeTemporalJoinLateRecords(backend, true, expectedOutput); + } + + private void testRowTimeTemporalJoinLateRecords( + StateBackend backend, boolean isLeftOuter, List expectedOutput) + throws Exception { + TemporalRowTimeJoinOperatorV2 joinOperator = + new TemporalRowTimeJoinOperatorV2( + rowType, rowType, joinCondition, 0, 0, 0, 0, isLeftOuter); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinOperator, backend); + + testHarness.open(); + + // initialize watermark to 1 + testHarness.processWatermark1(new Watermark(1)); + testHarness.processWatermark2(new Watermark(1)); + + // Establish a build-side version at time 2 and a non-late probe record at time 3. + testHarness.processElement2(insertRecord(2L, "k1", "2a2")); + testHarness.processElement1(insertRecord(3L, "k1", "1a3")); + testHarness.processWatermark1(new Watermark(5)); + testHarness.processWatermark2(new Watermark(5)); + + // After Watermark(5), any probe record with leftTime <= 5 is late and must be dropped. + testHarness.processElement1(insertRecord(5L, "k1", "1a5")); // leftTime == watermark + testHarness.processElement1(insertRecord(4L, "k1", "1a4")); // leftTime < watermark + testHarness.processElement1(insertRecord(1L, "k1", "1a1")); // leftTime << watermark + // A non-late probe record should still be processed. + testHarness.processElement1(insertRecord(7L, "k1", "1a7")); + testHarness.processWatermark1(new Watermark(8)); + testHarness.processWatermark2(new Watermark(8)); + + // A record for late retraction + testHarness.processElement1(insertRecord(10L, "k2", "1a10")); + testHarness.processWatermark1(new Watermark(11)); + testHarness.processWatermark2(new Watermark(11)); + + // Add a late retraction and a late build-side record + testHarness.processElement1(insertRecord(13L, "k2", "1a13")); + testHarness.processElement2(insertRecord(9L, "k2", "2a9")); + testHarness.processElement1(deleteRecord(10L, "k2", "1a10")); // late -> dropped + testHarness.processWatermark1(new Watermark(13)); + testHarness.processWatermark2(new Watermark(13)); + + // Another late retraction + testHarness.processElement1(deleteRecord(13L, "k2", "1a13")); + testHarness.processWatermark1(new Watermark(15)); + testHarness.processWatermark2(new Watermark(15)); + + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + assertThat(joinOperator.getNumLateRecordsDropped().getCount()).isEqualTo(5L); + + testHarness.close(); + } + + @ParameterizedTest(name = "backend={0}") + @MethodSource("stateBackends") + void testEmissionInArrivalOrder(StateBackend backend) throws Exception { + TemporalRowTimeJoinOperatorV2 joinOperator = + new TemporalRowTimeJoinOperatorV2( + rowType, rowType, joinCondition, 0, 0, 0, 0, false); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinOperator, backend); + + testHarness.open(); + + testHarness.processWatermark1(new Watermark(0)); + testHarness.processWatermark2(new Watermark(0)); + + testHarness.processElement2(insertRecord(1L, "k1", "r1")); + // Probe records arrive out of row-time order; 5 first, then 3 and 4, plus one beyond the + // upcoming watermark. The record with time 5 is exactly at the watermark and must be due. + testHarness.processElement1(insertRecord(5L, "k1", "1a5")); + testHarness.processElement1(insertRecord(3L, "k1", "1a3")); + testHarness.processElement1(insertRecord(4L, "k1", "1a4")); + testHarness.processElement1(insertRecord(8L, "k1", "1a8")); + + testHarness.processWatermark1(new Watermark(5)); + testHarness.processWatermark2(new Watermark(5)); + + testHarness.processWatermark1(new Watermark(9)); + testHarness.processWatermark2(new Watermark(9)); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(new Watermark(0)); + // arrival order 5, 3, 4 - not row-time order 3, 4, 5 + expectedOutput.add(insertRecord(5L, "k1", "1a5", 1L, "k1", "r1")); + expectedOutput.add(insertRecord(3L, "k1", "1a3", 1L, "k1", "r1")); + expectedOutput.add(insertRecord(4L, "k1", "1a4", 1L, "k1", "r1")); + expectedOutput.add(new Watermark(5)); + expectedOutput.add(insertRecord(8L, "k1", "1a8", 1L, "k1", "r1")); + expectedOutput.add(new Watermark(9)); + + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + @ParameterizedTest(name = "backend={0}") + @MethodSource("stateBackends") + void testRightRowAtLeftTimeBoundary(StateBackend backend) throws Exception { + TemporalRowTimeJoinOperatorV2 joinOperator = + new TemporalRowTimeJoinOperatorV2( + rowType, rowType, joinCondition, 0, 0, 0, 0, false); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinOperator, backend); + + testHarness.open(); + + testHarness.processWatermark1(new Watermark(0)); + testHarness.processWatermark2(new Watermark(0)); + + // Build-side version and probe record at the same row time 2 -> must join. + testHarness.processElement2(insertRecord(2L, "k1", "2a2")); + testHarness.processElement1(insertRecord(2L, "k1", "1a2")); + + testHarness.processWatermark1(new Watermark(2)); + testHarness.processWatermark2(new Watermark(2)); + + // DELETE build-side version and probe record at the same row time 4 -> no join. + testHarness.processElement2(deleteRecord(4L, "k1", "2a2")); + testHarness.processElement1(insertRecord(4L, "k1", "1a4")); + + testHarness.processWatermark1(new Watermark(4)); + testHarness.processWatermark2(new Watermark(4)); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(new Watermark(0)); + expectedOutput.add(insertRecord(2L, "k1", "1a2", 2L, "k1", "2a2")); + expectedOutput.add(new Watermark(2)); + expectedOutput.add(new Watermark(4)); + + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + @ParameterizedTest(name = "backend={0}") + @MethodSource("stateBackends") + void testKeepsLatestRightVersionAfterCleanup(StateBackend backend) throws Exception { + TemporalRowTimeJoinOperatorV2 joinOperator = + new TemporalRowTimeJoinOperatorV2( + rowType, rowType, joinCondition, 0, 0, 0, 0, false); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinOperator, backend); + + testHarness.open(); + + // Two build-side versions, no probe records; the watermark triggers cleanup which must + // remove version 2 but keep version 4 (the latest one <= watermark). + testHarness.processElement2(insertRecord(2L, "k1", "2a2")); + testHarness.processElement2(insertRecord(4L, "k1", "2a4")); + + testHarness.processWatermark1(new Watermark(5)); + testHarness.processWatermark2(new Watermark(5)); + + // This probe record joins the surviving version 4. + testHarness.processElement1(insertRecord(6L, "k1", "1a6")); + + testHarness.processWatermark1(new Watermark(7)); + testHarness.processWatermark2(new Watermark(7)); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(new Watermark(5)); + expectedOutput.add(insertRecord(6L, "k1", "1a6", 4L, "k1", "2a4")); + expectedOutput.add(new Watermark(7)); + + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + private KeyedTwoInputStreamOperatorTestHarness + createTestHarness( + TemporalRowTimeJoinOperatorV2 temporalJoinOperator, StateBackend backend) + throws Exception { + + KeyedTwoInputStreamOperatorTestHarness harness = + new KeyedTwoInputStreamOperatorTestHarness<>( + temporalJoinOperator, keySelector, keySelector, keyType); + harness.setStateBackend(backend); + return harness; + } +} diff --git a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java index 0ac026145139e..72547f53c18f4 100644 --- a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java +++ b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java @@ -73,6 +73,7 @@ import org.apache.flink.table.dataview.MapViewSerializer; import org.apache.flink.table.dataview.NullAwareMapSerializer; import org.apache.flink.table.dataview.NullSerializer; +import org.apache.flink.table.runtime.operators.join.temporal.TemporalRowTimeJoinOperatorV2; import org.apache.flink.table.runtime.operators.sink.SortedLongSerializer; import org.apache.flink.table.runtime.operators.window.CountWindow; import org.apache.flink.table.runtime.sequencedmultisetstate.linked.MetaSqnInfoSerializer; @@ -196,7 +197,10 @@ void testTypeSerializerTestCoverage() { SharedBufferEdge.SharedBufferEdgeSerializer.class.getName(), RowDataSerializer.class.getName(), DecimalDataSerializer.class.getName(), - AvroSerializer.class.getName()); + AvroSerializer.class.getName(), + // covered by LeftTimeIndexKeySerializerTest; the nested class name cannot + // match the expected Test pattern + TemporalRowTimeJoinOperatorV2.LeftTimeIndexKeySerializer.class.getName()); // type serializer whitelist for TypeSerializerUpgradeTestBase test coverage final List typeSerializerUpgradeTestBaseWhitelist = @@ -267,7 +271,8 @@ void testTypeSerializerTestCoverage() { RowSqnInfoSerializer.class.getName(), MetaSqnInfoSerializer.class.getName(), SetSerializer.class.getName(), - SortedLongSerializer.class.getName()); + SortedLongSerializer.class.getName(), + TemporalRowTimeJoinOperatorV2.LeftTimeIndexKeySerializer.class.getName()); // check if a test exists for each type serializer for (Class typeSerializer : typeSerializers) {