This is an automated email from the ASF dual-hosted git repository.
RocMarshal pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new a59d93a4407 [FLINK-40173][table-planner] Add restore coverage for
early-fire interval join
a59d93a4407 is described below
commit a59d93a4407ff3393d4d5f887dd89eaa6a6cb576
Author: weiqingy <[email protected]>
AuthorDate: Sat Jul 18 13:05:38 2026 -0700
[FLINK-40173][table-planner] Add restore coverage for early-fire interval
join
Add an INTERVAL_JOIN_EARLY_FIRE restore test program with its plan and
savepoint fixtures, and register it in IntervalJoinRestoreTest, exercising
end-to-end plan and savepoint restore for the early-fire interval join.
Cover both early-fire time modes. The row-time program exercises the fired
bookkeeping restored from a savepoint; a second program with a
processing-time
delay covers the cross-domain schedule state, which only exists in that
mode.
The processing-time program restores with no further input, so its only
output
can come from the restored schedule.
---
.../IntervalJoinProcTimeEarlyFireRestoreTest.java | 67 +++
.../nodes/exec/stream/IntervalJoinRestoreTest.java | 3 +-
.../exec/stream/IntervalJoinTestPrograms.java | 121 +++++
.../plan/interval-join-early-fire.json | 541 +++++++++++++++++++
.../interval-join-early-fire/savepoint/_metadata | Bin 0 -> 19415 bytes
.../plan/interval-join-proc-time-early-fire.json | 595 +++++++++++++++++++++
.../savepoint/_metadata | Bin 0 -> 19078 bytes
7 files changed, 1326 insertions(+), 1 deletion(-)
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 00000000000..87190d18108
--- /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.
+ *
+ * <p>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<TableTestProgram> programs() {
+ return Collections.singletonList(
+ IntervalJoinTestPrograms.INTERVAL_JOIN_PROC_TIME_EARLY_FIRE);
+ }
+
+ @Override
+ protected void awaitSavepointReady(TableTestProgram program,
List<CompletableFuture<?>> 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 4140a1e39ae..91472f42ea1 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 class IntervalJoinRestoreTest extends RestoreTestBase {
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 6cc1c546beb..4bd1d81f9a4 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.SourceTestStep;
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<String, String> 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/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 00000000000..547fdf057e8
--- /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 00000000000..24e4f6290e7
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 00000000000..d3bc9f700ae
--- /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 00000000000..b2b6d2b2204
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
