diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java index 3d8d4c7b01b51f..c8aeab1ef68503 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java @@ -430,7 +430,11 @@ private TwoInputTransformation createRowTimeJoin( joinFunction, windowBounds.getLeftTimeIdx(), windowBounds.getRightTimeIdx(), - earlyFireDelay == null ? -1L : earlyFireDelay); + earlyFireDelay == null ? -1L : earlyFireDelay, + // Cross-domain flag: an event-time interval join early-fires on the wall + // clock while keeping its event-time cleanup. The operator only acts on it + // once early-firing is enabled (earlyFireDelay >= 0). + earlyFireTimeMode == EarlyFireJoinHintOptions.TimeMode.PROCTIME); // TODO: add async version rowJoinFunc to use AsyncKeyedCoProcessOperator return ExecNodeUtil.createTwoInputTransformation( leftInputTransform, diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalIntervalJoinRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalIntervalJoinRule.java index 5b848e4f20c2de..b0371b8140fcf1 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalIntervalJoinRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalIntervalJoinRule.java @@ -187,12 +187,6 @@ private static EarlyFire extractEarlyFire(List hints, boolean isEventTi "EARLY_FIRE hint requested row-time triggering on a processing-time interval" + " join. Row-time triggering requires a row-time interval join."); } - if (isEventTime && timeMode == TimeMode.PROCTIME) { - // Processing-time triggering on an event-time interval join is not supported. - throw new TableException( - "EARLY_FIRE hint requested processing-time triggering on a row-time interval" - + " join, which is not yet supported."); - } return new EarlyFire(delay == null ? null : delay.toMillis(), timeMode); } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.java index 05879701b33399..d0c3552d96df34 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.java @@ -202,7 +202,7 @@ void testEarlyFireProcTimeOnRowTimeJoin() { + "FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON\n" + " t1.a = t2.a AND\n" + " t1.rowtime BETWEEN t2.rowtime - INTERVAL '10' SECOND AND t2.rowtime + INTERVAL '1' HOUR"; - assertThatThrownBy(() -> verify(sql)).hasStackTraceContaining("not yet supported"); + verify(sql); } @Test diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/IntervalJoinProcTimeEarlyFireRestoreTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/IntervalJoinProcTimeEarlyFireRestoreTest.java new file mode 100644 index 00000000000000..87190d181085ac --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/IntervalJoinProcTimeEarlyFireRestoreTest.java @@ -0,0 +1,67 @@ +/* + * 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.planner.plan.nodes.exec.stream; + +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.table.planner.plan.nodes.exec.testutils.RestoreTestBase; +import org.apache.flink.table.test.program.SourceTestStep; +import org.apache.flink.table.test.program.TableTestProgram; + +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; + +/** + * Restore tests for {@link StreamExecIntervalJoin} early-firing on processing time. + * + *

The early-fire timer lives on the wall clock while the join keeps its row-time cleanup, so the + * speculative pad is due at a processing-time instant recorded in the savepoint. The restored job + * ingests no data at all: the pad is emitted purely from restored state once that instant passes, + * which needs a source that stays open rather than one that ends input and closes the window. + */ +public class IntervalJoinProcTimeEarlyFireRestoreTest extends RestoreTestBase { + + private static final long SAVEPOINT_READY_TIMEOUT_MILLIS = TimeUnit.MINUTES.toMillis(5); + + public IntervalJoinProcTimeEarlyFireRestoreTest() { + super(StreamExecIntervalJoin.class, AfterRestoreSource.INFINITE); + } + + @Override + public List programs() { + return Collections.singletonList( + IntervalJoinTestPrograms.INTERVAL_JOIN_PROC_TIME_EARLY_FIRE); + } + + @Override + protected void awaitSavepointReady(TableTestProgram program, List> futures) + throws Exception { + // The join emits nothing before the savepoint, so the default sink-based trigger would fire + // immediately. Gate on the sources instead: stop-with-savepoint then drains their rows into + // keyed state, capturing the still-pending early-fire schedule. + for (SourceTestStep source : program.getSetupSourceTestSteps()) { + final int count = source.dataBeforeRestore.size(); + if (count > 0) { + TestValuesTableFactory.awaitSourceEmitted(source.name, count) + .get(SAVEPOINT_READY_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); + } + } + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/IntervalJoinRestoreTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/IntervalJoinRestoreTest.java index 4140a1e39ae288..91472f42ea1a85 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/IntervalJoinRestoreTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/IntervalJoinRestoreTest.java @@ -36,6 +36,7 @@ public List programs() { return Arrays.asList( IntervalJoinTestPrograms.INTERVAL_JOIN_EVENT_TIME, IntervalJoinTestPrograms.INTERVAL_JOIN_PROC_TIME, - IntervalJoinTestPrograms.INTERVAL_JOIN_NEGATIVE_INTERVAL); + IntervalJoinTestPrograms.INTERVAL_JOIN_NEGATIVE_INTERVAL, + IntervalJoinTestPrograms.INTERVAL_JOIN_EARLY_FIRE); } } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/IntervalJoinTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/IntervalJoinTestPrograms.java index 6cc1c546beb33e..4bd1d81f9a4fed 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/IntervalJoinTestPrograms.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/IntervalJoinTestPrograms.java @@ -23,6 +23,8 @@ import org.apache.flink.table.test.program.TableTestProgram; import org.apache.flink.types.Row; +import java.util.Map; + /** {@link TableTestProgram} definitions for testing {@link StreamExecIntervalJoin}. */ public class IntervalJoinTestPrograms { @@ -152,6 +154,125 @@ public class IntervalJoinTestPrograms { + " WHERE o.proc_time BETWEEN s.proc_time - INTERVAL '5' SECOND AND s.proc_time + INTERVAL '5' SECOND;") .build(); + static final Row[] EARLY_FIRE_ORDER_BEFORE_DATA = { + Row.of(1, "2020-04-15 08:00:01"), Row.of(9, "2020-04-15 08:00:06"), + }; + + static final Row[] EARLY_FIRE_SHIPMENT_BEFORE_DATA = { + Row.of(100, 9, "2020-04-15 08:00:06"), + }; + + static final Row[] EARLY_FIRE_ORDER_AFTER_DATA = { + Row.of(20, "2020-04-15 08:00:20"), + }; + + static final Row[] EARLY_FIRE_SHIPMENT_AFTER_DATA = { + Row.of(101, 1, "2020-04-15 08:00:03"), Row.of(102, 20, "2020-04-15 08:00:20"), + }; + + static final TableTestProgram INTERVAL_JOIN_EARLY_FIRE = + TableTestProgram.of( + "interval-join-early-fire", + "validates the EARLY_FIRE hint on an outer interval join: an unmatched" + + " left row is speculatively padded before the savepoint and" + + " retracted when its match arrives after restore") + .setupTableSource( + SourceTestStep.newBuilder("orders_t") + .addSchema(ORDERS_EVENT_TIME_SCHEMA) + .producedBeforeRestore(EARLY_FIRE_ORDER_BEFORE_DATA) + .producedAfterRestore(EARLY_FIRE_ORDER_AFTER_DATA) + .build()) + .setupTableSource( + SourceTestStep.newBuilder("shipments_t") + .addSchema(SHIPMENTS_EVENT_TIME_SCHEMA) + .producedBeforeRestore(EARLY_FIRE_SHIPMENT_BEFORE_DATA) + .producedAfterRestore(EARLY_FIRE_SHIPMENT_AFTER_DATA) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink_t") + .addSchema(SINK_SCHEMA) + .consumedBeforeRestore( + "+I[1, 2020-04-15 08:00:01, null]", + "+I[9, 2020-04-15 08:00:06, 2020-04-15 08:00:06]") + .consumedAfterRestore( + "-U[1, 2020-04-15 08:00:01, null]", + "+U[1, 2020-04-15 08:00:01, 2020-04-15 08:00:03]", + "+I[20, 2020-04-15 08:00:20, 2020-04-15 08:00:20]") + .build()) + .runSql( + "INSERT INTO sink_t SELECT /*+ EARLY_FIRE('delay'='2s') */\n" + + " o.id AS order_id,\n" + + " o.order_ts_str,\n" + + " s.shipment_ts_str\n" + + " FROM orders_t o LEFT OUTER JOIN shipments_t s\n" + + " ON o.id = s.order_id\n" + + " AND o.order_ts BETWEEN s.shipment_ts - INTERVAL '5' SECOND AND s.shipment_ts + INTERVAL '5' SECOND;") + .build(); + + // Selects the watermark-push-down source runtime, the only values-source runtime that reports + // how many rows it has emitted. The savepoint trigger gates on that count because the query + // below emits nothing before the savepoint. + static final Map PER_RECORD_WATERMARK_SOURCE_OPTIONS = + Map.of( + "disable-lookup", "true", + "enable-watermark-push-down", "true", + "scan.watermark.emit.strategy", "on-event"); + + // The unmatched left row whose speculative pad is scheduled on the wall clock. + static final Row[] PROC_TIME_EARLY_FIRE_ORDER_BEFORE_DATA = { + Row.of(1, "2020-04-15 08:00:01"), + }; + + // A shipment for an order that does not exist: a restore test source must produce at least one + // row, and this one neither joins the left row's key nor produces output on a left outer join. + static final Row[] PROC_TIME_EARLY_FIRE_SHIPMENT_BEFORE_DATA = { + Row.of(100, 99, "2020-04-15 08:00:02"), + }; + + // The 30s delay has to outlast the savepoint trigger so the schedule entry is still pending + // when the snapshot is taken. It also bounds how long a freshly generated savepoint makes the + // restored job wait for its early-fire timer; a savepoint older than the delay fires at once. + static final TableTestProgram INTERVAL_JOIN_PROC_TIME_EARLY_FIRE = + TableTestProgram.of( + "interval-join-proc-time-early-fire", + "validates the EARLY_FIRE hint driving a row-time interval join from" + + " processing time: the savepoint captures the pending" + + " early-fire schedule of an unmatched left row, and the pad" + + " is emitted only once that schedule is restored") + .setupTableSource( + SourceTestStep.newBuilder("orders_t") + .addSchema(ORDERS_EVENT_TIME_SCHEMA) + .addOptions(PER_RECORD_WATERMARK_SOURCE_OPTIONS) + .producedBeforeRestore(PROC_TIME_EARLY_FIRE_ORDER_BEFORE_DATA) + .build()) + .setupTableSource( + SourceTestStep.newBuilder("shipments_t") + .addSchema(SHIPMENTS_EVENT_TIME_SCHEMA) + .addOptions(PER_RECORD_WATERMARK_SOURCE_OPTIONS) + .producedBeforeRestore( + PROC_TIME_EARLY_FIRE_SHIPMENT_BEFORE_DATA) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink_t") + .addSchema(SINK_SCHEMA) + // Nothing is emitted before the savepoint: the left row is + // still cached, its pad is still only scheduled, and the + // watermark has not closed its window. + .consumedBeforeRestore(new String[0]) + // No data is ingested after restore, so this pad can only come + // from the restored schedule and cache. + .consumedAfterRestore("+I[1, 2020-04-15 08:00:01, null]") + .build()) + .runSql( + "INSERT INTO sink_t SELECT /*+ EARLY_FIRE('delay'='30s', 'time-mode'='proctime') */\n" + + " o.id AS order_id,\n" + + " o.order_ts_str,\n" + + " s.shipment_ts_str\n" + + " FROM orders_t o LEFT OUTER JOIN shipments_t s\n" + + " ON o.id = s.order_id\n" + + " AND o.order_ts BETWEEN s.shipment_ts - INTERVAL '5' SECOND AND s.shipment_ts + INTERVAL '5' SECOND;") + .build(); + static final TableTestProgram INTERVAL_JOIN_NEGATIVE_INTERVAL = TableTestProgram.of( "interval-join-negative-interval", diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.xml index 8faa17c8bea3fa..f3866cd3062e7b 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.xml @@ -241,6 +241,38 @@ Calc(select=[a, b], changelogMode=[I,UA]) +- Exchange(distribution=[hash[a]], changelogMode=[I]) +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime], changelogMode=[I]) +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, b, rowtime], metadata=[]]], fields=[a, b, rowtime], changelogMode=[I]) +]]> + + + + + + + + =($4, -($9, 10000:INTERVAL SECOND)), <=($4, +($9, 3600000:INTERVAL HOUR)))], joinType=[left], joinHints=[[[EARLY_FIRE inheritPath:[0] options:{delay=5s, time-mode=proctime}]]]) + :- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[$4]) + : +- LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[PROCTIME()], rowtime=[$3]) + : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]]) + +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[$4]) + +- LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[PROCTIME()], rowtime=[$3]) + +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]]) +]]> + + + = (rowtime0 - 10000:INTERVAL SECOND)) AND (rowtime <= (rowtime0 + 3600000:INTERVAL HOUR)))], select=[a, rowtime, a0, b, rowtime0], earlyFireDelay=[5000], earlyFireTimeMode=[PROCTIME]) + :- Exchange(distribution=[hash[a]]) + : +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime]) + : +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, rowtime], metadata=[]]], fields=[a, rowtime]) + +- Exchange(distribution=[hash[a]]) + +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime]) + +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, b, rowtime], metadata=[]]], fields=[a, b, rowtime]) ]]> diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-interval-join_1/interval-join-early-fire/plan/interval-join-early-fire.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-interval-join_1/interval-join-early-fire/plan/interval-join-early-fire.json new file mode 100644 index 00000000000000..547fdf057e8d9a --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-interval-join_1/interval-join-early-fire/plan/interval-join-early-fire.json @@ -0,0 +1,541 @@ +{ + "flinkVersion" : "2.4", + "nodes" : [ { + "id" : 1, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`orders_t`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "id", + "dataType" : "INT" + }, { + "name" : "order_ts_str", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "order_ts", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "TO_TIMESTAMP(`order_ts_str`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "order_ts", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$-$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "TIMESTAMP(3)" + }, { + "kind" : "LITERAL", + "value" : "1000", + "type" : "INTERVAL SECOND(6) NOT NULL" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`order_ts` - INTERVAL '1' SECOND" + } + } ] + } + } + } + }, + "outputType" : "ROW<`id` INT, `order_ts_str` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, orders_t]], fields=[id, order_ts_str])" + }, { + "id" : 2, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "INT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`id` INT, `order_ts_str` VARCHAR(2147483647), `order_ts` TIMESTAMP(3)>", + "description" : "Calc(select=[id, order_ts_str, TO_TIMESTAMP(order_ts_str) AS order_ts])" + }, { + "id" : 3, + "type" : "stream-exec-watermark-assigner_1", + "watermarkExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$-$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "TIMESTAMP(3)" + }, { + "kind" : "LITERAL", + "value" : "1000", + "type" : "INTERVAL SECOND(6) NOT NULL" + } ], + "type" : "TIMESTAMP(3)" + }, + "rowtimeFieldIndex" : 2, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "id", + "fieldType" : "INT" + }, { + "name" : "order_ts_str", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "order_ts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "WatermarkAssigner(rowtime=[order_ts], watermark=[(order_ts - 1000:INTERVAL SECOND)])" + }, { + "id" : 4, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "id", + "fieldType" : "INT" + }, { + "name" : "order_ts_str", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "order_ts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Exchange(distribution=[hash[id]])" + }, { + "id" : 5, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`shipments_t`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "id", + "dataType" : "INT" + }, { + "name" : "order_id", + "dataType" : "INT" + }, { + "name" : "shipment_ts_str", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "shipment_ts", + "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(`shipment_ts_str`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "shipment_ts", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$-$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "TIMESTAMP(3)" + }, { + "kind" : "LITERAL", + "value" : "1000", + "type" : "INTERVAL SECOND(6) NOT NULL" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`shipment_ts` - INTERVAL '1' SECOND" + } + } ] + } + } + }, + "abilities" : [ { + "type" : "ProjectPushDown", + "projectedFields" : [ [ 1 ], [ 2 ] ], + "producedType" : "ROW<`order_id` INT, `shipment_ts_str` VARCHAR(2147483647)> NOT NULL" + }, { + "type" : "ReadingMetadata", + "metadataKeys" : [ ], + "producedType" : "ROW<`order_id` INT, `shipment_ts_str` VARCHAR(2147483647)> NOT NULL" + } ] + }, + "outputType" : "ROW<`order_id` INT, `shipment_ts_str` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, shipments_t, project=[order_id, shipment_ts_str], metadata=[]]], fields=[order_id, shipment_ts_str])" + }, { + "id" : 6, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "INT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`order_id` INT, `shipment_ts_str` VARCHAR(2147483647), `shipment_ts` TIMESTAMP(3)>", + "description" : "Calc(select=[order_id, shipment_ts_str, TO_TIMESTAMP(shipment_ts_str) AS shipment_ts])" + }, { + "id" : 7, + "type" : "stream-exec-watermark-assigner_1", + "watermarkExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$-$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "TIMESTAMP(3)" + }, { + "kind" : "LITERAL", + "value" : "1000", + "type" : "INTERVAL SECOND(6) NOT NULL" + } ], + "type" : "TIMESTAMP(3)" + }, + "rowtimeFieldIndex" : 2, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "order_id", + "fieldType" : "INT" + }, { + "name" : "shipment_ts_str", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "shipment_ts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "WatermarkAssigner(rowtime=[shipment_ts], watermark=[(shipment_ts - 1000:INTERVAL SECOND)])" + }, { + "id" : 8, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "order_id", + "fieldType" : "INT" + }, { + "name" : "shipment_ts_str", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "shipment_ts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Exchange(distribution=[hash[order_id]])" + }, { + "id" : 9, + "type" : "stream-exec-interval-join_1", + "intervalJoinSpec" : { + "joinSpec" : { + "joinType" : "LEFT", + "leftKeys" : [ 0 ], + "rightKeys" : [ 0 ], + "filterNulls" : [ true ], + "nonEquiCondition" : null + }, + "windowBounds" : { + "isEventTime" : true, + "leftLowerBound" : -5000, + "leftUpperBound" : 5000, + "leftTimeIndex" : 2, + "rightTimeIndex" : 2 + } + }, + "earlyFireDelay" : 2000, + "earlyFireTimeMode" : "ROWTIME", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + }, { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "id", + "fieldType" : "INT" + }, { + "name" : "order_ts_str", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "order_ts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, { + "name" : "order_id", + "fieldType" : "INT" + }, { + "name" : "shipment_ts_str", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "shipment_ts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "IntervalJoin(joinType=[LeftOuterJoin], windowBounds=[isRowTime=true, leftLowerBound=-5000, leftUpperBound=5000, leftTimeIndex=2, rightTimeIndex=2], where=[((id = order_id) AND (order_ts >= (shipment_ts - 5000:INTERVAL SECOND)) AND (order_ts <= (shipment_ts + 5000:INTERVAL SECOND)))], select=[id, order_ts_str, order_ts, order_id, shipment_ts_str, shipment_ts], earlyFireDelay=[2000], earlyFireTimeMode=[ROWTIME])" + }, { + "id" : 10, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "INT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 4, + "type" : "VARCHAR(2147483647)" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`order_id` INT, `order_ts_str` VARCHAR(2147483647), `shipment_ts_str` VARCHAR(2147483647)>", + "description" : "Calc(select=[id AS order_id, order_ts_str, shipment_ts_str])" + }, { + "id" : 11, + "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`.`sink_t`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "order_id", + "dataType" : "INT" + }, { + "name" : "order_ts_str", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "shipment_ts_str", + "dataType" : "VARCHAR(2147483647)" + } ] + } + } + } + }, + "inputChangelogMode" : [ "INSERT", "UPDATE_BEFORE", "UPDATE_AFTER" ], + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`order_id` INT, `order_ts_str` VARCHAR(2147483647), `shipment_ts_str` VARCHAR(2147483647)>", + "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[order_id, order_ts_str, shipment_ts_str])" + } ], + "edges" : [ { + "source" : 1, + "target" : 2, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 2, + "target" : 3, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 3, + "target" : 4, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 5, + "target" : 6, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 6, + "target" : 7, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 7, + "target" : 8, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 4, + "target" : 9, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 8, + "target" : 9, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 9, + "target" : 10, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 10, + "target" : 11, + "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-interval-join_1/interval-join-early-fire/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-interval-join_1/interval-join-early-fire/savepoint/_metadata new file mode 100644 index 00000000000000..24e4f6290e7654 Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-interval-join_1/interval-join-early-fire/savepoint/_metadata differ diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-interval-join_1/interval-join-proc-time-early-fire/plan/interval-join-proc-time-early-fire.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-interval-join_1/interval-join-proc-time-early-fire/plan/interval-join-proc-time-early-fire.json new file mode 100644 index 00000000000000..d3bc9f700aed27 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-interval-join_1/interval-join-proc-time-early-fire/plan/interval-join-proc-time-early-fire.json @@ -0,0 +1,595 @@ +{ + "flinkVersion" : "2.4", + "nodes" : [ { + "id" : 1, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`orders_t`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "id", + "dataType" : "INT" + }, { + "name" : "order_ts_str", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "order_ts", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "TO_TIMESTAMP(`order_ts_str`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "order_ts", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$-$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "TIMESTAMP(3)" + }, { + "kind" : "LITERAL", + "value" : "1000", + "type" : "INTERVAL SECOND(6) NOT NULL" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`order_ts` - INTERVAL '1' SECOND" + } + } ] + } + } + }, + "abilities" : [ { + "type" : "WatermarkPushDown", + "watermarkExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$-$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, { + "kind" : "LITERAL", + "value" : "1000", + "type" : "INTERVAL SECOND(6) NOT NULL" + } ], + "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<`id` INT, `order_ts_str` VARCHAR(2147483647)> NOT NULL", + "watermarkParams" : { + "emitStrategy" : "ON_EVENT", + "alignGroupName" : null, + "alignMaxDrift" : "PT0S", + "alignUpdateInterval" : "PT1S", + "sourceIdleTimeout" : -1 + } + } ] + }, + "outputType" : "ROW<`id` INT, `order_ts_str` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, orders_t, watermark=[-(TO_TIMESTAMP(order_ts_str), 1000:INTERVAL SECOND)], watermarkEmitStrategy=[on-event]]], fields=[id, order_ts_str])" + }, { + "id" : 2, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "INT" + }, { + "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" : 1, + "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" : "id", + "fieldType" : "INT" + }, { + "name" : "order_ts_str", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "order_ts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Calc(select=[id, order_ts_str, Reinterpret(TO_TIMESTAMP(order_ts_str)) AS order_ts])" + }, { + "id" : 3, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "id", + "fieldType" : "INT" + }, { + "name" : "order_ts_str", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "order_ts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Exchange(distribution=[hash[id]])" + }, { + "id" : 4, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`shipments_t`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "id", + "dataType" : "INT" + }, { + "name" : "order_id", + "dataType" : "INT" + }, { + "name" : "shipment_ts_str", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "shipment_ts", + "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(`shipment_ts_str`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "shipment_ts", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$-$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "TIMESTAMP(3)" + }, { + "kind" : "LITERAL", + "value" : "1000", + "type" : "INTERVAL SECOND(6) NOT NULL" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`shipment_ts` - INTERVAL '1' SECOND" + } + } ] + } + } + }, + "abilities" : [ { + "type" : "ProjectPushDown", + "projectedFields" : [ [ 1 ], [ 2 ] ], + "producedType" : "ROW<`order_id` INT, `shipment_ts_str` VARCHAR(2147483647)> NOT NULL" + }, { + "type" : "ReadingMetadata", + "metadataKeys" : [ ], + "producedType" : "ROW<`order_id` INT, `shipment_ts_str` VARCHAR(2147483647)> NOT NULL" + }, { + "type" : "WatermarkPushDown", + "watermarkExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$-$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, { + "kind" : "LITERAL", + "value" : "1000", + "type" : "INTERVAL SECOND(6) NOT NULL" + } ], + "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<`order_id` INT, `shipment_ts_str` VARCHAR(2147483647)> NOT NULL", + "watermarkParams" : { + "emitStrategy" : "ON_EVENT", + "alignGroupName" : null, + "alignMaxDrift" : "PT0S", + "alignUpdateInterval" : "PT1S", + "sourceIdleTimeout" : -1 + } + } ] + }, + "outputType" : "ROW<`order_id` INT, `shipment_ts_str` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, shipments_t, project=[order_id, shipment_ts_str], metadata=[], watermark=[-(TO_TIMESTAMP(shipment_ts_str), 1000:INTERVAL SECOND)], watermarkEmitStrategy=[on-event]]], fields=[order_id, shipment_ts_str])" + }, { + "id" : 5, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "INT" + }, { + "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" : 1, + "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" : "order_id", + "fieldType" : "INT" + }, { + "name" : "shipment_ts_str", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "shipment_ts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Calc(select=[order_id, shipment_ts_str, Reinterpret(TO_TIMESTAMP(shipment_ts_str)) AS shipment_ts])" + }, { + "id" : 6, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "order_id", + "fieldType" : "INT" + }, { + "name" : "shipment_ts_str", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "shipment_ts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Exchange(distribution=[hash[order_id]])" + }, { + "id" : 7, + "type" : "stream-exec-interval-join_1", + "intervalJoinSpec" : { + "joinSpec" : { + "joinType" : "LEFT", + "leftKeys" : [ 0 ], + "rightKeys" : [ 0 ], + "filterNulls" : [ true ], + "nonEquiCondition" : null + }, + "windowBounds" : { + "isEventTime" : true, + "leftLowerBound" : -5000, + "leftUpperBound" : 5000, + "leftTimeIndex" : 2, + "rightTimeIndex" : 2 + } + }, + "earlyFireDelay" : 30000, + "earlyFireTimeMode" : "PROCTIME", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + }, { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "id", + "fieldType" : "INT" + }, { + "name" : "order_ts_str", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "order_ts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, { + "name" : "order_id", + "fieldType" : "INT" + }, { + "name" : "shipment_ts_str", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "shipment_ts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "IntervalJoin(joinType=[LeftOuterJoin], windowBounds=[isRowTime=true, leftLowerBound=-5000, leftUpperBound=5000, leftTimeIndex=2, rightTimeIndex=2], where=[((id = order_id) AND (order_ts >= (shipment_ts - 5000:INTERVAL SECOND)) AND (order_ts <= (shipment_ts + 5000:INTERVAL SECOND)))], select=[id, order_ts_str, order_ts, order_id, shipment_ts_str, shipment_ts], earlyFireDelay=[30000], earlyFireTimeMode=[PROCTIME])" + }, { + "id" : 8, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "INT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 4, + "type" : "VARCHAR(2147483647)" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`order_id` INT, `order_ts_str` VARCHAR(2147483647), `shipment_ts_str` VARCHAR(2147483647)>", + "description" : "Calc(select=[id AS order_id, order_ts_str, shipment_ts_str])" + }, { + "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`.`sink_t`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "order_id", + "dataType" : "INT" + }, { + "name" : "order_ts_str", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "shipment_ts_str", + "dataType" : "VARCHAR(2147483647)" + } ] + } + } + } + }, + "inputChangelogMode" : [ "INSERT", "UPDATE_BEFORE", "UPDATE_AFTER" ], + "upsertMaterializeStrategy" : "MAP", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`order_id` INT, `order_ts_str` VARCHAR(2147483647), `shipment_ts_str` VARCHAR(2147483647)>", + "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[order_id, order_ts_str, shipment_ts_str])" + } ], + "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-interval-join_1/interval-join-proc-time-early-fire/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-interval-join_1/interval-join-proc-time-early-fire/savepoint/_metadata new file mode 100644 index 00000000000000..b2b6d2b220409e Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-interval-join_1/interval-join-proc-time-early-fire/savepoint/_metadata differ diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/ProcTimeIntervalJoin.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/ProcTimeIntervalJoin.java index 84ad4526289786..fc2aa6a5cdc3f4 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/ProcTimeIntervalJoin.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/ProcTimeIntervalJoin.java @@ -45,7 +45,9 @@ public ProcTimeIntervalJoin( leftType, rightType, genJoinFunc, - earlyFireDelay); + earlyFireDelay, + // A proctime join's early fire shares the cleanup domain; never cross-domain. + false); } @Override diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoin.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoin.java index 57972aff22713f..1645ce8d312f5e 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoin.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoin.java @@ -41,7 +41,8 @@ public RowTimeIntervalJoin( IntervalJoinFunction joinFunc, int leftTimeIdx, int rightTimeIdx, - long earlyFireDelay) { + long earlyFireDelay, + boolean earlyFireCrossDomain) { super( joinType, leftLowerBound, @@ -51,7 +52,8 @@ public RowTimeIntervalJoin( leftType, rightType, joinFunc, - earlyFireDelay); + earlyFireDelay, + earlyFireCrossDomain); this.leftTimeIdx = leftTimeIdx; this.rightTimeIdx = rightTimeIdx; } diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalJoin.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalJoin.java index f2abfeb00f22f6..e699dfb7d9a2c1 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalJoin.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalJoin.java @@ -29,6 +29,7 @@ import org.apache.flink.api.java.typeutils.ListTypeInfo; import org.apache.flink.api.java.typeutils.TupleTypeInfo; import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.streaming.api.TimeDomain; import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction; import org.apache.flink.table.data.RowData; import org.apache.flink.table.runtime.operators.join.FlinkJoinType; @@ -71,6 +72,10 @@ abstract class TimeIntervalJoin extends KeyedCoProcessFunction> leftFiredState; private transient MapState> rightFiredState; + // Cross-domain early fire only: maps a firing processing-time to the event-time bucket keys + // whose unmatched outer rows are due to be speculatively padded at that wall-clock instant. + // The event-time bucket key cannot be recovered from a processing-time firing timestamp alone, + // so this index records it at registration and recovers it when the timer fires. Allocated only + // when earlyFireCrossDomain is true. + private transient MapState> leftEarlyFireSchedule; + private transient MapState> rightEarlyFireSchedule; + // state to record the timer on the left stream. 0 means no timer set private transient ValueState leftTimerState; // state to record the timer on the right stream. 0 means no timer set @@ -110,7 +123,8 @@ abstract class TimeIntervalJoin extends KeyedCoProcessFunction leftType, InternalTypeInfo rightType, IntervalJoinFunction joinFunc, - long earlyFireDelay) { + long earlyFireDelay, + boolean earlyFireCrossDomain) { this.joinType = joinType; this.leftRelativeSize = -leftLowerBound; this.rightRelativeSize = leftUpperBound; @@ -130,6 +144,7 @@ abstract class TimeIntervalJoin extends KeyedCoProcessFunction= 0 && joinType.isOuter() && (leftRelativeSize + rightRelativeSize) >= 0; + this.earlyFireCrossDomain = earlyFireCrossDomain; } @Override @@ -174,6 +189,25 @@ public void open(OpenContext openContext) throws Exception { "IntervalJoinRightFired", BasicTypeInfo.LONG_TYPE_INFO, firedListTypeInfo)); + + if (earlyFireCrossDomain) { + ListTypeInfo bucketListTypeInfo = + new ListTypeInfo<>(BasicTypeInfo.LONG_TYPE_INFO); + leftEarlyFireSchedule = + getRuntimeContext() + .getMapState( + new MapStateDescriptor<>( + "IntervalJoinLeftEarlyFireSchedule", + BasicTypeInfo.LONG_TYPE_INFO, + bucketListTypeInfo)); + rightEarlyFireSchedule = + getRuntimeContext() + .getMapState( + new MapStateDescriptor<>( + "IntervalJoinRightEarlyFireSchedule", + BasicTypeInfo.LONG_TYPE_INFO, + bucketListTypeInfo)); + } } // Initialize the timer states. @@ -303,7 +337,7 @@ public void processElement1(RowData leftRow, Context ctx, Collector out appendFired(leftFiredState, timeForLeftRow); if (!emitted) { // Schedule a speculative pad of this unmatched left row after the delay. - registerTimer(ctx, timeForLeftRow + earlyFireDelay); + scheduleEarlyFire(ctx, leftEarlyFireSchedule, timeForLeftRow); } } if (rightTimerState.value() == null) { @@ -419,7 +453,7 @@ public void processElement2(RowData rightRow, Context ctx, Collector ou appendFired(rightFiredState, timeForRightRow); if (!emitted) { // Schedule a speculative pad of this unmatched right row after the delay. - registerTimer(ctx, timeForRightRow + earlyFireDelay); + scheduleEarlyFire(ctx, rightEarlyFireSchedule, timeForRightRow); } } if (leftTimerState.value() == null) { @@ -439,12 +473,30 @@ public void onTimer(long timestamp, OnTimerContext ctx, Collector out) joinCollector.setInnerCollector(out); updateOperatorTime(ctx); - // Early fire runs before cleanup at a shared timestamp so a row that is both due to fire - // and - // due to expire emits its speculative pad here; the cleanup branch's fired-bit gate then - // suppresses a second pad. A cleanup-only timestamp finds no live unfired-unmatched row at - // timestamp - earlyFireDelay and is a cheap no-op. - if (earlyFireEnabled) { + if (earlyFireEnabled && earlyFireCrossDomain) { + // Cross-domain: early-fire timers are processing-time, cleanup timers are event-time. + // timeDomain() is the authoritative discriminator (a processing-time value can + // numerically equal an event-time cleanup value, so timestamp arithmetic is unsafe). + if (ctx.timeDomain() == TimeDomain.PROCESSING_TIME) { + if (joinType.isLeftOuter()) { + fireScheduled( + leftCache, leftFiredState, leftEarlyFireSchedule, timestamp, true); + } + if (joinType.isRightOuter()) { + fireScheduled( + rightCache, rightFiredState, rightEarlyFireSchedule, timestamp, false); + } + // Cleanup is event-time; there is nothing else to do at a processing-time firing. + return; + } + // EVENT_TIME falls through to the cleanup branches below; no early fire in this domain. + } else if (earlyFireEnabled) { + // Natural pairing: the early-fire timer shares its domain with cleanup and fires at + // rowTime + delay, so the bucket key is recovered as timestamp - delay. Early fire runs + // before cleanup at a shared timestamp so a row that is both due to fire and due to + // expire emits its speculative pad here; the cleanup branch's fired-bit gate then + // suppresses a second pad. A cleanup-only timestamp finds no live unfired-unmatched row + // and is a cheap no-op. long rowTime = timestamp - earlyFireDelay; if (joinType.isLeftOuter()) { earlyFire(leftCache, leftFiredState, rowTime, true); @@ -515,6 +567,53 @@ private void earlyFire( } } + /** + * Register the early-fire timer for an unmatched outer row. For the natural pairing the timer + * shares the cleanup domain and fires at {@code bucketKey + earlyFireDelay}, recoverable later + * as {@code timestamp - earlyFireDelay}. For the cross-domain case the timer is a + * processing-time timer at {@code currentProcessingTime() + earlyFireDelay}, and the event-time + * bucket key is recorded in the schedule under that firing time so it can be recovered when the + * processing-time timer fires. + */ + private void scheduleEarlyFire(Context ctx, MapState> schedule, long bucketKey) + throws Exception { + if (earlyFireCrossDomain) { + long firingTime = ctx.timerService().currentProcessingTime() + earlyFireDelay; + List buckets = schedule.get(firingTime); + if (buckets == null) { + buckets = new ArrayList<>(1); + } + buckets.add(bucketKey); + schedule.put(firingTime, buckets); + ctx.timerService().registerProcessingTimeTimer(firingTime); + } else { + registerTimer(ctx, bucketKey + earlyFireDelay); + } + } + + /** + * Recover the event-time buckets scheduled to fire at the given processing-time and early-fire + * each one, then drop the schedule entry. A missing entry is a no-op; a bucket already cleaned + * by event-time expiry resolves to an empty cache lookup inside {@link #earlyFire} and is + * likewise a no-op. + */ + private void fireScheduled( + MapState>> rowCache, + MapState> firedState, + MapState> schedule, + long firingTime, + boolean padLeft) + throws Exception { + List buckets = schedule.get(firingTime); + if (buckets == null) { + return; + } + for (Long bucketKey : buckets) { + earlyFire(rowCache, firedState, bucketKey, padLeft); + } + schedule.remove(firingTime); + } + /** * Calculate the expiration time with the given operator time and relative window size. * diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoinTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoinTest.java index a3748a61563b55..76efcfaad80318 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoinTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoinTest.java @@ -20,6 +20,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.core.execution.CheckpointingMode; +import org.apache.flink.runtime.checkpoint.OperatorSubtaskState; import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator; import org.apache.flink.streaming.api.watermark.Watermark; import org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor; @@ -72,7 +73,8 @@ void testRowTimeInnerJoinWithCommonBounds() throws Exception { joinFunction, 0, 0, - -1L); + -1L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -147,7 +149,8 @@ void testRowTimeInnerJoinWithNegativeBounds() throws Exception { joinFunction, 0, 0, - -1L); + -1L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -202,7 +205,18 @@ void testRowTimeInnerJoinWithNegativeBounds() throws Exception { void testRowTimeInnerJoinRealtimeCleanUp() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.LEFT, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, -1L); + FlinkJoinType.LEFT, + -5, + 9, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + -1L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -231,7 +245,18 @@ void testRowTimeInnerJoinRealtimeCleanUp() throws Exception { void testRowTimeLeftOuterJoin() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.LEFT, -5, 9, 0, 7, rowType, rowType, joinFunction, 0, 0, -1L); + FlinkJoinType.LEFT, + -5, + 9, + 0, + 7, + rowType, + rowType, + joinFunction, + 0, + 0, + -1L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -311,7 +336,8 @@ void testRowTimeRightOuterJoin() throws Exception { joinFunction, 0, 0, - -1L); + -1L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -382,7 +408,18 @@ void testRowTimeRightOuterJoin() throws Exception { void testRowTimeFullOuterJoin() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.FULL, -5, 9, 0, 7, rowType, rowType, joinFunction, 0, 0, -1L); + FlinkJoinType.FULL, + -5, + 9, + 0, + 7, + rowType, + rowType, + joinFunction, + 0, + 0, + -1L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -472,7 +509,8 @@ public void testInterruptibleTimers() throws Exception { joinFunction, 0, 0, - -1L); + -1L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -550,7 +588,18 @@ public void testInterruptibleTimers() throws Exception { void testRowTimeLeftOuterEarlyFire() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.LEFT, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, 3L); + FlinkJoinType.LEFT, + -5, + 9, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + 3L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); testHarness.open(); @@ -580,7 +629,18 @@ void testRowTimeLeftOuterEarlyFire() throws Exception { void testRowTimeLeftOuterEarlyFireThenMatch() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.LEFT, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, 3L); + FlinkJoinType.LEFT, + -5, + 9, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + 3L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); testHarness.open(); @@ -612,7 +672,18 @@ void testRowTimeLeftOuterEarlyFireThenMatch() throws Exception { void testRowTimeRightOuterEarlyFireThenMatch() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.RIGHT, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, 3L); + FlinkJoinType.RIGHT, + -5, + 9, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + 3L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); testHarness.open(); @@ -642,7 +713,18 @@ void testRowTimeRightOuterEarlyFireThenMatch() throws Exception { void testRowTimeFullOuterEarlyFireOneMatches() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.FULL, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, 3L); + FlinkJoinType.FULL, + -5, + 9, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + 3L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); testHarness.open(); @@ -686,7 +768,18 @@ void testRowTimeFullOuterEarlyFireOneMatches() throws Exception { void testRowTimeInnerJoinIgnoresEarlyFire() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.INNER, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, 3L); + FlinkJoinType.INNER, + -5, + 9, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + 3L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); testHarness.open(); @@ -713,7 +806,18 @@ void testRowTimeLeftOuterEarlyFireDelayExceedsSpan() throws Exception { // Window span is 5 + 9 = 14; the delay exceeds it so cleanup may reach the row first. RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.LEFT, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, 20L); + FlinkJoinType.LEFT, + -5, + 9, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + 20L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); testHarness.open(); @@ -735,7 +839,18 @@ void testRowTimeLeftOuterEarlyFireDelayExceedsSpan() throws Exception { void testRowTimeEarlyFireRowKindIsolation() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.LEFT, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, 3L); + FlinkJoinType.LEFT, + -5, + 9, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + 3L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); testHarness.open(); @@ -768,7 +883,18 @@ void testRowTimeEarlyFireRowKindIsolation() throws Exception { void testRowTimeLeftOuterEarlyFireMultiMatch() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.LEFT, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, 3L); + FlinkJoinType.LEFT, + -5, + 9, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + 3L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); testHarness.open(); @@ -796,6 +922,255 @@ void testRowTimeLeftOuterEarlyFireMultiMatch() throws Exception { testHarness.close(); } + /** + * Cross-domain early fire: an event-time interval join firing speculative pads on the wall + * clock. The early-fire timer is a processing-time timer while cleanup stays an event-time + * timer, so the pad fires on a processing-time advance with the watermark unchanged. + */ + @Test + void testRowTimeCrossDomainEarlyFireOnWallClock() throws Exception { + RowTimeIntervalJoin joinProcessFunc = + new RowTimeIntervalJoin( + FlinkJoinType.LEFT, + -5, + 9, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + 3L, + true); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + testHarness.setProcessingTime(0L); + + testHarness.processElement1(insertRecord(10L, "k1")); + // The early-fire timer is processing-time (fires at now + delay = 3); the cleanup timer is + // event-time. The cross-domain split is visible in the per-domain timer counts. + assertThat(testHarness.numProcessingTimeTimers()).isEqualTo(1); + assertThat(testHarness.numEventTimeTimers()).isEqualTo(1); + + // Advance the wall clock past the firing time without advancing the watermark: the pad + // fires purely on processing time. + testHarness.setProcessingTime(3L); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(insertRecord(10L, "k1", null, null)); + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + /** + * Cross-domain early fire followed by an in-window event-time match: the wall-clock pad is + * retracted via -U/+U and the later event-time cleanup emits nothing. + */ + @Test + void testRowTimeCrossDomainEarlyFireThenMatch() throws Exception { + RowTimeIntervalJoin joinProcessFunc = + new RowTimeIntervalJoin( + FlinkJoinType.LEFT, + -5, + 9, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + 3L, + true); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + testHarness.setProcessingTime(0L); + + testHarness.processElement1(insertRecord(10L, "k1")); + // Wall-clock pad fires. + testHarness.setProcessingTime(3L); + + // A right row arrives in window (10 in [12 - 5, 12 + 9]) and matches the padded left row. + testHarness.processElement2(insertRecord(12L, "k1")); + + // Cross cleanup on the event-time clock: no further pad, the row already matched. + testHarness.processWatermark1(new Watermark(30)); + testHarness.processWatermark2(new Watermark(30)); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(insertRecord(10L, "k1", null, null)); + expectedOutput.add(updateBeforeRecord(10L, "k1", null, null)); + expectedOutput.add(updateAfterRecord(10L, "k1", 12L, "k1")); + expectedOutput.add(new Watermark(30 - 9)); + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + /** + * Cross-domain restore safety: snapshot after the early-fire timer is registered but before it + * fires, then restore and advance the wall clock. Exactly one pad is emitted after restore - + * none lost, none duplicated. + */ + @Test + void testRowTimeCrossDomainSnapshotBeforeFire() throws Exception { + RowTimeIntervalJoin joinProcessFunc = + new RowTimeIntervalJoin( + FlinkJoinType.LEFT, + -5, + 9, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + 3L, + true); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + testHarness.setProcessingTime(0L); + + testHarness.processElement1(insertRecord(10L, "k1")); + // Snapshot with the processing-time early-fire timer pending (firing time is 3). + testHarness.prepareSnapshotPreBarrier(0L); + OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0); + testHarness.close(); + + // Nothing was emitted before the snapshot. + assertThat(testHarness.getOutput()).isEmpty(); + + RowTimeIntervalJoin restoredFunc = + new RowTimeIntervalJoin( + FlinkJoinType.LEFT, + -5, + 9, + 0, + 0, + rowType, + rowType, + newJoinFunction(), + 0, + 0, + 3L, + true); + testHarness = createTestHarness(restoredFunc); + testHarness.setup(); + testHarness.initializeState(snapshot); + testHarness.open(); + + // The restored processing-time timer fires once on the wall-clock advance. + testHarness.setProcessingTime(3L); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(insertRecord(10L, "k1", null, null)); + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + /** + * Cross-domain restore safety: snapshot after the pad has already been emitted, then restore + * and let a match arrive. The positional fired bit survives the restore so the post-restore + * match still retracts the pad via -U/+U. + */ + @Test + void testRowTimeCrossDomainSnapshotAfterFire() throws Exception { + RowTimeIntervalJoin joinProcessFunc = + new RowTimeIntervalJoin( + FlinkJoinType.LEFT, + -5, + 9, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + 3L, + true); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + testHarness.setProcessingTime(0L); + + testHarness.processElement1(insertRecord(10L, "k1")); + // Fire the wall-clock pad before snapshotting. + testHarness.setProcessingTime(3L); + testHarness.prepareSnapshotPreBarrier(0L); + OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0); + testHarness.close(); + + RowTimeIntervalJoin restoredFunc = + new RowTimeIntervalJoin( + FlinkJoinType.LEFT, + -5, + 9, + 0, + 0, + rowType, + rowType, + newJoinFunction(), + 0, + 0, + 3L, + true); + testHarness = createTestHarness(restoredFunc); + testHarness.setup(); + testHarness.initializeState(snapshot); + testHarness.open(); + testHarness.setProcessingTime(3L); + + // A match arrives after restore; the restored fired bit drives the -U/+U correction. + testHarness.processElement2(insertRecord(12L, "k1")); + testHarness.processWatermark1(new Watermark(30)); + testHarness.processWatermark2(new Watermark(30)); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(updateBeforeRecord(10L, "k1", null, null)); + expectedOutput.add(updateAfterRecord(10L, "k1", 12L, "k1")); + expectedOutput.add(new Watermark(30 - 9)); + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + /** An inner join with the cross-domain hint must not early-fire: the no-op guard holds. */ + @Test + void testRowTimeCrossDomainInnerJoinIgnoresEarlyFire() throws Exception { + RowTimeIntervalJoin joinProcessFunc = + new RowTimeIntervalJoin( + FlinkJoinType.INNER, + -5, + 9, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + 3L, + true); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + testHarness.setProcessingTime(0L); + + testHarness.processElement1(insertRecord(10L, "k1")); + // No early-fire processing-time timer for an inner join: only the event-time cleanup timer. + assertThat(testHarness.numProcessingTimeTimers()).isEqualTo(0); + assertThat(testHarness.numEventTimeTimers()).isEqualTo(1); + + testHarness.setProcessingTime(3L); + + assertThat(testHarness.getOutput()).isEmpty(); + testHarness.close(); + } + private KeyedTwoInputStreamOperatorTestHarness createTestHarness(RowTimeIntervalJoin intervalJoinFunc) throws Exception { KeyedCoProcessOperator operator = diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalStreamJoinTestBase.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalStreamJoinTestBase.java index 034fe71f37c23b..1a1ea06847ea01 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalStreamJoinTestBase.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalStreamJoinTestBase.java @@ -50,10 +50,15 @@ abstract class TimeIntervalStreamJoinTestBase { + " return true;\n" + " }\n" + "}\n"; - protected IntervalJoinFunction joinFunction = - new IntervalJoinFunction( - new GeneratedJoinCondition( - "TestIntervalJoinCondition", funcCode, new Object[0]), - outputRowType, - new boolean[] {true}); + protected IntervalJoinFunction joinFunction = newJoinFunction(); + + // IntervalJoinFunction.open() consumes its generated-code field (sets it to null), so a single + // instance cannot be opened twice. Restore tests that open a second operator must build a fresh + // function for the restored harness. + protected IntervalJoinFunction newJoinFunction() { + return new IntervalJoinFunction( + new GeneratedJoinCondition("TestIntervalJoinCondition", funcCode, new Object[0]), + outputRowType, + new boolean[] {true}); + } }