From 8b64241eae889333f2af805b0de197231a28294e Mon Sep 17 00:00:00 2001 From: Mate Czagany Date: Wed, 5 Aug 2026 13:35:28 +0200 Subject: [PATCH] [FLINK-40336][table] Add TemporalRowTimeJoinOperatorV2 to improve join on sorted state backends --- .../exec/stream/StreamExecTemporalJoin.java | 20 +- ...temporal-join-table-join-key-from-map.json | 564 +++++++++++++++ .../savepoint/_metadata | Bin 0 -> 14165 bytes .../temporal-join-table-join-nested-key.json | 594 ++++++++++++++++ .../savepoint/_metadata | Bin 0 -> 14165 bytes .../plan/temporal-join-table-join.json | 543 +++++++++++++++ .../savepoint/_metadata | Bin 0 -> 14128 bytes .../plan/temporal-join-temporal-function.json | 543 +++++++++++++++ .../savepoint/_metadata | Bin 0 -> 14160 bytes .../TemporalRowTimeJoinOperatorV2.java | 645 ++++++++++++++++++ .../LeftTimeIndexKeySerializerTest.java | 122 ++++ .../TemporalRowTimeJoinOperatorV2Test.java | 503 ++++++++++++++ .../TypeSerializerTestCoverageTest.java | 9 +- 13 files changed, 3540 insertions(+), 3 deletions(-) create mode 100644 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 create mode 100644 flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join-key-from-map/savepoint/_metadata create mode 100644 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 create mode 100644 flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join-nested-key/savepoint/_metadata create mode 100644 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 create mode 100644 flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-table-join/savepoint/_metadata create mode 100644 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 create mode 100644 flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-temporal-join_2/temporal-join-temporal-function/savepoint/_metadata create mode 100644 flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/temporal/TemporalRowTimeJoinOperatorV2.java create mode 100644 flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/temporal/LeftTimeIndexKeySerializerTest.java create mode 100644 flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/temporal/TemporalRowTimeJoinOperatorV2Test.java 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 eb6f68a2170903..726fefce0e88eb 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 00000000000000..17372c53d132c6 --- /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 0000000000000000000000000000000000000000..917bea8ad2ea9abf7592297c548c06312599ace0 GIT binary patch literal 14165 zcmeGjYiwIr`PfdAw4;r?MxzfJMQRE~%)Wjkane*8PU1FR9Xr@gx+WCw#y)m#dVQ~Z z?{Vr3sbt#3qm#CYhqMWex2TXh6=^C0!5@Se6{|KhCc!V@M-i_fHi6g#o67gyhkb8t zcP?$ZRP~j7ALsFX-}%mW&i6WwK3wn+LOS7xmqUM$F48^0i`j*M?*!=z_%d1e*u8XN zpZu#MFTWXmp_lyj$|sEGYkHTRDY>j<>B0Ua&!{>js7zK?7XpD3WOE>pQMJl%efl+a z;Oy(PkN@7`U%t5Xb2|hBV-PV0HH^N;U%2qbKKHr5o_qW4ix>alv$x-EVoWGGs;F_Q zGMbC#7W|`t(?4q9488F{;XFr1LrW}9M*7NJ^Q0S|MAl0Zr96Q z?73GTI$~{aatF$6mpe_1C0XIc2W3H82n?Md?!Zun8HJu@DBZiY zJofJ|K!1d`hHz|PyRZbVHSK=OZ@YeXeUQ6nuV25>YxjI`qf^+Xug!s0IN%^($_pYr z$gS`SmzPD{<2cpF4P_SdrzExlR&+uFPtPTD+MoK;M?q?hrFBK0a|i3vsO zWveAxW`d}Clk(|Fp7ANF2)rmfLY0BKl9a@k)Fqh#_zw6v&i|ckQWfHBf&*czUQsRx zpq#vOyfU%GD`_Y+nCoy7N0fAmR4TBgN~>^UIy^fZ=RDaWsPVEW;^<&;soKeVNOGCS+uQ5DwFCj3hZOi-$?-hBe;EhP- zqP4QbNY`9AIWZMZe&*`ctEBhXyU14UH{S$v{lul1B1na6=B`npW%^tW;>`O%Ncu7; z21-U{U=(mAMb1)H1#o(?3Z%FUiXFr~7mm$FGO0*18jeNhBT1-nl8@IBsxpQ0MF0YV zCwS!&Rd_HpRBswF)#S@c4%zxAlqjc#1#Q`F%Wh*~(Pek8PVUU5p~DA4Uv_LMOm_#B8$9Ay)juTSeC8*)3#cB5Ur<>N3 zB^@Hoqa}nF=*Bu`xTeE1P~~h{Z8m4cG+(N5!ZvhWZ8~gf^NV#%wwru4g^X^m*K~rt zri%<_A8Km!loN*To(l}bV6<2qMHx6sS8D`ar^-g_dYYmx!xZJ3j9s(!s;20;P>4#* zZb71{)|w$}vG%mdhv|2cJJta09FHD}#?uD>0xj_m6q*%OL6%~q z<0R}Cj10DBe^+51g16ClM`0&Rho@&U$D-+}#B4f)C7Jm|JOZ-j0*~YdX9&4z zNU7$JoFZpPC%7HS#4&{2_TY5q)Uu}0`?GTF|5W-evOdc0;U+88G=6cX(;fQiin_)% zY*KF7tI-2>k#+Y#JNcpvVF=Qt$fp@>Da199A0qFMHs7QOg(Y*|HTICE=WRE2 zEI!h@>$W>x_d7^;wyY>r%C3?Qh~Cngj((riGuj7=VU{&g&v#Ym!3m7G?=0Nh21m$j zY7$=`!&^7r;Bsh5n!d*E)`+9r^U9$E&~86A?#CNmCZxRnNe&!(rKb!AA*3fXw3u6} zqK%Xl8NMIKwkBGqj&@tab~@Ulm!Q&))`Vo5V3&^SXrWElNpvRXVf*faXgnAZ-{nW!}hjOF2 zoG&yu5+Y>#7*Gf~&gSeu_Blv@VkVLdrxQsdPB|G(rIXR|*>p4!&rByKBQPg3v1mF| ztw2Y+S+1)-o@fuP)ts+Lg;7P~MN62hzQ|^R8=*YIy8BoYs~<6_-opYFb1M3EXb1i4 z`{zIO{G(sbKsq%>w&k(TtWp_q$1nlOp-;%g;Ls1oR`y>Py4m@~r~cXdHRoeRh;GG5 zPga&zAi)ixoe7aw0q}MmFlIJ6t!OMp?h$L*Sw;e_g?=o~H>6!x6tcUXJZ;zznbhTy zxL}t7Y}#1IS24}QNRMTp5HTHFAU#I?jI0U_2xnx+f`#bB;xf%LpxTI3^*10F@p6Dy zi%J1^O9m(t=`IV>7_3X0o8r|aL|-XEYW)8Ck!b06eHsmBVd@>{(HI-1#bcyXqg};# zjbr0Q!)}Yy#OqUt9zgQ1^=82 z4>xqEaFU`@!Y$=A?%oa<&(Mv?1RlaTj4Qe(94-+2996T5P|}6^UW-iYd4H5?2^Ix% zU=O^bxZ!(8r37aVjJOMsax>v_gtygk1|M#KsfUp(__<4y+MTvRUw=OfGr1^a&3$M0 z=Tjg3&m;Q+U~g#foUkQWPElZNW9*45PhWm&-|u0T9weJqc(F`jk+`+#(2iQf%f&^i zeD?fT_TKa5KRkhuN*Yyt4y3RjvRKx`7VDXsFi*+c46pC%)P!xO)+MOwFqcG~fz&o4 zPyx1$h}`&U+laW;2khELgk~eQjfl1p0ahMbKQ(XB;u>mpV>{@7b3y;xX+)r@UJWo^ zF${NWS)f6g8E><;`=l95y7lmC6c9tSM(W%WGo=h?$FnE3L+Rd$>g$YxiAN?-E%XH7 zoKDg`Puw{{&GJfaf|s%soVDCM*=iIW=HakBeD|2&f+VO*){<0tQ7sp-!~~Nnt0f92 zj-aSBFT=l=;AL}J%8ES1gLB}I*Jwizx?#w{7zS5s%9*PR`+z5>F~dQ?A@aAvgp_$k zAG?3!bHDEX&O^OFbQ18(wg8PqUOQRD$WVPaAuB7CqZOJ3Z&sGLDxkyFZl>qPNJFo# z7vc}BHVI+P7iMIt9#x}JTj?GEd!`N&@B*@#!8#@&-QFtQ-Uf7=^3=+i_stXGcp@I1 u2*)6ZWXy1X6($e>?6~Gn3k^iE+5u;w;b^Qhq1U$=jvAz9Q!|nH" + }, { + "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 0000000000000000000000000000000000000000..5a7fd8d1fed17eb705a0f15766260612e40ce580 GIT binary patch literal 14165 zcmeGjZERat_1aF8v|ST-jkdMfLeNq+ME#yW6Q`NBI*GHob?jg}DNQKu%k#DK(&zWw z_nuRyKf08m@iEj*+ZaWNj~{&eDIc1|hs4(p5KMqHF@zBOfFezxF%U?A3XG|o^ZcIQ z^K577)})2{Nq*Pwew=&mIp^Mc?m6eiP8K|bkWTpFzzMS57s#r5 z;kU0mH`Sj#x%TRzJC8lp^NruvL)d3RNGdt1s0pfSG#AS)_`C?iYryz%c(KP>D0l& z=~YY@d|`wZ&Z_Y7+T{zK&f>cty6=$>|KeNTw@)^M=IDT2C+Wlb9))TO$7_(PWl>YH z4=jeq{YfMJijjWbNPkIgGnzIudBv!5wVLkUW2E;v$kTa2qzBm*PGR%1hy-L+FB{4( z=1)u7^2qXO@2cPL4;7Y!xs_ni3r|5}GwgC%5;Tu*c+~3+ghu_Nfv_(S3i^G+K33&c zXh{|%%|mIxJIZI2n(}BZ{w zV)aq5MqM^;wHd3FRurq6p1Dg&Qr3iX=J#`h_O;@24OoG8abzs;I*QJZ4I=@?S#OndQdbOlprQ5m5cVml15yq zB@zs!Uzioh~pr8+b#73G|e7xY1aT>qzczwc{*y5QES*C9fVNxK7Y48Q`s zcbu&KK!x29D&s;SCTaB&N-79%m*X5JZh$!+dpMTJ=(&4!slBUi(Ce>f5lt=% zyt(e|`%wCw|2})K-w85;;GM7}*d!$`!vG(D>B*lyaqn+|OGn7I6;3Qu5GS`KPKk1H zkt&Z}`q(`?pSk=vLMrK44LA_uV1*kf}OO%`|X|D+vrQ7mKoF+kO{0l`<$OHb&Kd+PLI$m{X!bL24@zxB<46 zh}`&UTZw2Z5zM+;h@~C0l?XT#q0*r;rjFZIB9I={!(oHUfLdO?T4CFs+DZiM^6AI9PCz$u=J)^_EuqQm~4F&kz;z%&) zBc%TVV6S^6oMbDG{Tnmf&2h@a5~tMLSipYNDN?DRE!}(uz%4NVbj?Ll6Ne+IN3LGI zT4x2AZ?FQivcp&HVfDJzP_s;%!UgsKXWk1!GLS{9zLHfna6PdlMdqmr9+W~g8+|S= zV=6dEbBDp921Hd&pJa279us6?SwC*8sRWY9_0;?7ju#^(`Q~ zjgBhQl&Q-yGbal<^qQ#P%c7Rl4YkOxlg7Puvx2TrNo!9C`%$;Om^bt0+}Oo@Gbh9j ziB>cZI0z}Y74!$TgQL;o(MiMUoH;%dCEIk5G88*o>=m10+9rL8$aFM46Peh6GyD3j z5gj7Uo4&qg#BfbVX4WA^>Aa7DDS3)#19Cs;J9|DQJ73-lXx!pr5~m?9>}N zQ4U~}Etcw)d1s7lFDaB4FoL2MauD2@1HJ}R4@0;`PXqm#7_4NO$n;G1SS)imIh)C1 zPIf+-z?dBuxFoNULr6`1y@nSCB%o`!QzKmvcYu%$wzp4kZ{GUaV$*k#&DrcuwwR&j z#%7t|YcuL~N5dBRwzV4Vo-VTKCr~3ugLEnK84Z>c;u^|_ZqRPNO%V!9H|2JoeFKJD z+c`!)GHF?X(9ilnfT3oYHh#M{)l6jW&Hty1%$CH7*%z6%s$)c7wpASnTit}n$1O1) zZMmvrJ0orS-uAz~_d7^8Use<<@vEc*thdzKqi@V%dmHKpilN9hJ8j=mrU!`iBzJrS zUNx8bYy3m+4b>8!v1ffstqpb&z4 zLhlxROSNvJWkrVX4`4&HTL@c+uYn$THP)+kd-QoItnb!@WU64#AFXzapqh7C1(m+o z5|kN8e=%J@f-l%HJ5D9|Y)ZK1^;#0ngoMg7J~qNtGvIApvGL)h7@vN{=8F$KR_;4{ z@Y#_UeV_kw&(!ar5&~F*A0ft0A8yb9O=TZ0Y4E4?dnU|8?eC0&DxoG&{l*M{aXLx& zJaOj)mFJY)1SjznV$5^%WS5aO#lig-`0h2o1xZks>^bT3qFOFujtM4RR!bCYL6Fs% zm*Msj{MlZXc#(sdnuBWu7H()IXnIq^7zS%qWlgOV+h7nrJ`MtL+`f$vA#Gl}!RDX+ z;Lp21_ny9QIthf6?*JT&oF3t=kx-*af|nJ_(hB7vd`Xtr8lb6KGc({^BTa+R`{3Gy zW-%e6hwy9UaAQ}^Ms3G?2*NNMAOS7l@E>R(0^-fo@MfCet& 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 0000000000000000000000000000000000000000..0fbacd88a294621e9b87c41bd162b0a5441382d3 GIT binary patch literal 14128 zcmeGjZERat^*lRY&1ju$q0p}=+R~|r`aRpR&nP-OvBYth`ieP^#_0+8(3!(#ubseD|##n>${9UHVPSlaJKab|ggk z*q*+zMVd|ruvH+X;p;Oaw!P23x#L{lA9}BR?e2#zSX-4bV_GoR20gUCQzg2}3p%D+ zQPMSv3oXW{>j!4~OEZ1dO#cHPGn-!Br2FHgD&1}~9WzsrfiqcAB73+wUgffilp(6d zX)ex}N@tJD`mBHUxNFhl@%VDH-prgg@4_c1b4hNtD2uw&jeosE14C|az!h*0x?DqU zPUGiDK@nx$>Gk^k?tnAx^7`@P7l_x#`Mn5?_YXMJL~uF19vANn3=X)R#3Kl4kH?$# z5FbGI!-!xTvjR0J(HQ89jfdjFWGtRaB!kINY9yRU#>4w2lHpi1H5MBQL04it5>8eN zLPoo;$zWt6luCr+;b0^@6^di74JH*e$MOYUm?3OdLN2g6pOy%#7G+(`6INd=kfJV1 z8XH>Bi7N9_f~X=diT4q;zdkD=^95~2(Gj*CHvWUMk2(J)v}!7RdYPVR4;s5)BvM8j z;dMTV|KED%{PrI`@ZFRRmr&}zYMo=I~<^ZqF z(C!xs$j06EBQXS~#72_+i{rl3>QqD4uf zT}PmUb{XAo_NLxU44c&`;D+SWu1ZU?|oZK9&dQVC*1axrf$Ea+g5 z#||M4?=~t?`DKR{NkWz(E4BU!Ij>~ItY{1h<@#O6PWMmWIMroPuSLR~GVOMMFn}uH zE&E{U`7gfRYk*+{9X4AyT&-4O?;TahK~?1!>HJ!F;Kd!^{4M`&J1}8r(QxV`xo~37 z8`B%d<69NoOquA%bU&TZResu0@H0dcRIyMJ>f0?cozPWL&QY0`aZ*qUY{PfTar)k} zP(Zr2@p%OoEJTG{6)OA0Tv*m?B@`AYyp7BWnsgz|ec{iBqe&xoiy^h2tQqvm>se4& z@}f{)ceb8PeCnP1?(*1BMksh2E(sP%i6}V0M=w40yVG~QhFm%bt#iCoBq&Z@RdI?I z^J$_!a{la&+rIwSM=7L&fmJUg=PY5kVDT z(}>9ITWlH;O(TL`(F&57gMRPY1(}RI@~GE$b9MF+n>JhzT>sIi_bki{mPwBV#5kN z07sJK3yR812hq7S?LPtze@ZvAoU@pBG520z>hjy2f4}oD4?k6$k6D}7pwn(*v}C3q zakAxaD+AnmaOLuwy>#pEZ*~lu_%$i88Vant_j|za<1>TA@8knM^nPat{LVl?$T)}G zX;&uTbGrw<8GvnmL>|50)F-_jzKhj)A18Trc!pPN9YyE_YmWQJoH3^ zW1+-&aCj9b@*mwA(ISy-`M|OfG3;Z(@fAo>uQr!Is=x!|>H=1?P?=!?HEQ!rokFg) z9I6VL?t0ns9+)lfkI~yiH8tHP$4uQl>FJN)XlZts%D^aDtRl2RQ4OwZX^Ps-z=-BC z+K2Rjne=O#@*m!pXkGnHa(b4TinY#fI<3%V$roz(Mz%0)D5yjbX;e)m48^;;po|Jk zBQ)g}-CS);hEXvjgJa{VL*e9s*hDf#b5c{WD2?N?<1zI$a)?sXP=EMCgH&j^K${Nj z7>hz7Th?Jatc|i;<9#6o_mENj~xtW_%r>8!ED^;CNLpa%lqUt158`99%_ z3-|28+?|XQ`j}RR={^WqW}^$UcHQ%Wqli`-9!gY#6#zXy|GaehXTM_1eomSFY!4Mx z1>YZH9JC_}665djw|7&#($nZg%&_*PGaLIb>*6TwiRtWs-cz%q(2_Wp9tD_hy>~aj zxkLjjaIhn}SI?FQ`$L3b4E8XQuXV7@)q_P?=JSbim_Z!nUORw++lsn4?ZsdNuNiDW z&i?%IBj4~mfAE!Gc-rUAzm5&5S-WHaFt>k0byi#1ipDDoJ29L74tNw3?@<3 z3Iv@vnAMh5FqQ&eHWXz+;&FCm@Df86H>&O$-qSFL!BwkrWlNH_p;rLt!vIE+yH`Vm zMETMWZT`Klywvgi&-XlQ0}N*GL>%+H5nrx@uYU6@C@SH|0ueBHN|Ct=p#7C*<#2Z$ z8ivC6;H3?{%7mbCfk1}?^<6a@wH5Dv3<9l#gtWkI=0F`0DBf%ZZ?*y61_|+JHJ2ea z3JSaG6z-wn9;O-v-lL(!4{0i(?S AA^-pY literal 0 HcmV?d00001 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 00000000000000..af721eab2aee23 --- /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 0000000000000000000000000000000000000000..7b1df77a75a39c4f229b9fc6d1da62efc2fdf148 GIT binary patch literal 14160 zcmeGjZHyaN@!7uIN7~#u$*JlkZFDG?wm`k>ue)=S($wd38aF=MvV9kFM{w8cXWw4d zyUyaOhX*6wV@^_=k)!K9G zcH?=!gFIgpM7ob%&EsQUm@GkrP{?XDxu&@{`d*LccY?fW9 zN`mGYANP&=eBRNJKN#E_^m=_kAFFbUv?2?V<_UR&;~}4i(tKeokPnUXUdrb4zVU#U z4|#mN*Y61yL(~)6%LP1R#Zb^+2o=ZpVu+Bf`vF1VWqWEM(HvwrnTn>v*5_cvQk7XMlqIx^`OgKIr&1ItLSU4V=ji#a0NoHiF#Bddk zpQlVw1THWdmlr9cR3%L)Q$|~^(5fbgDidAOs3LJ}xQaS2YXI9t zw!Dq(b6ofb*`zAu(`|af9#rO_K*fSO$!S~`-miS`!ad(P^J>mX9C6ZL#Bb&&RU_{B z0=LMqh!z_cRBiu~LiVf*-E92xXa6<$HRsu~lQ`pKfS093;JqMA5s^~?@YWh&k&{ZS zUbH_>?hxgY0Mh``(uh9}{jfE9u{yn2onEXy3f8F0#v5(+FI5&5vZEC}eb{OFB#;O% zmy>isZ~9m#mJ?hVdK<6|bc(C&pCbdt_mr#(8X&Hb?Q<5SllcY8Yk+mvaX~6*WY`8> z?3aap4?_*8*$s~Y$Z6sLj(OJv$U?~mFy}1UEU1a9C}P)f(uQhNqrI~W={t&rfg zqJ(V?w8%C>U{w7mnhZ)16P2oE>&J>lTV!H)*e06a87RxW-1-Ca5u2d|O4EF5lBtFd=`19DhV zxMiGQOJ{y_@A+SFUv&{jjC83m^?_V4u}|vL8^$xbCb}6iF#^wnIHMcZvmGcBbydEu}VRl+zoL`RLgm)eEGsl_uld1pPxoZ6&!4#qz@`X|4wW%& z+>Q}}^r&qPTTBMD@|w*Gi+bu95unNIyUJ_bMBD@;0!?){-tf9yd*HoZ=C>&;+LV3H zbo^M_-XM65-Uz8Vu>yW{M)`Fo)(`K0{l4>0244Txfh(62KXH5pY9lHuwN^&`QLI0z z*Z6IrNv+;q91R{rj!{b>}^|I))wmIuuw7 z1vcDUJzf|u@Wnj!_Cz0^%!mKUqh1uft=08^~^Jbjaxtt$U*@_H7W$q zXhZ5o+LSj*<51JApo>(}IupXWzwRvNb|IvjyO^)%gxD<6CCvj-NeZzH{Q`AxIC?ZX zX+*lSM^jOUfYg(v`Fn*KTe7;2?H(L%3g^asi#@qul<_;OWL#U~7qntE!_l$oe4x`25 z7|Ot5y4*l$l}}B$uBR#LG7=d&VXu$ru`=k_G37tL@6WpW8|3u2nTpNMYMfSJvy>~% zdn0Fz^i>qf3z!8{Pa6rYEs3HIQ%@ziO*dCNvoTN%+3-{!a+}d0$8#ZzHm%;%vkf>LT``t!JBZ zi!MMXttAZJ%U2bJO8heEQRI^v%!Zq!(l;qWY5tbf>A8>KN<*h($|aML6^#~}QxHb% z3N1`=t*fudRM7vY3T8tx)NB+?$KJURL|BiyT$tS&zaOure(Ce*A*T(?ZKn2}U z%w6EZr8iwYqiLWR=9k@;`?m5wK$a=F>nXU^1O3yPNhHo$y!GM@E{A?hW0!rs72>E4 zy!p@rP;U=b9>yD9CL}XW;DLwU9H@fT1~Ce~+h?Foz1wJ2k>UMI*wE}266WD;S3lfZ z>(#nFd=Wl2cWXj2O|XlHYuzF!`z~vsGOx4+Wv1}Il4%~nH?5c*r4qc_63)I}Tf&); z(89&bLbzH1-0wC|0HtA0sfQqmTUM6m0zNng6!016fF>Ou{Oqs){Nod6PyF}`FYMSo z2F(!U3j7E$HkY9mo6xk;;lK!gTtD-|MCyN23L1%;K=s2r0LJMgy|ct!5LBL13K34? zDdf8sX2}+#=pYA2ao~Nxd>14^owt@`s(H0q#u5`urm9vbxQL*ryC}m!C-~D>m3WbZ zxm$oE4Gr8-ncAM(FowZ4s&$c}6-`Y34`!+01ONa4 literal 0 HcmV?d00001 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 00000000000000..1a95d6f9c0a067 --- /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 00000000000000..e46642f79d178f --- /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 00000000000000..6d43bbc4137672 --- /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 0ac026145139e6..72547f53c18f4d 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) {