From 194ee253cdc2e8924d36ea6b02f1a55864e8b40c Mon Sep 17 00:00:00 2001 From: Weiqing Yang Date: Sun, 7 Jun 2026 12:20:32 -0700 Subject: [PATCH 1/2] [FLINK-40172][table-runtime] Support processing-time early fire on a row-time interval join Add the cross-domain timer combination the previous commit left out: an event-time interval join with EARLY_FIRE('time_mode'='proctime') now fires its speculative pads on the wall clock while keeping its event-time cleanup. The temporary "not yet supported" rejection in the planner rule is removed; the row-time-on-processing-time rejection is retained. onTimer distinguishes the two timer kinds by OnTimerContext.timeDomain(): in the cross-domain case early-fire timers are processing-time and cleanup timers are event-time, so a processing-time firing runs early fire and returns while an event-time firing runs cleanup only. The discrimination is gated on a new cross-domain flag, so the natural pairings keep the previous timestamp - delay recovery where early fire and cleanup share a domain. A processing-time firing timestamp cannot be mapped back to an event-time cache bucket arithmetically, so a per-side MapState> keyed by firing processing-time records the event-time bucket keys due to fire then. It is allocated only in the cross-domain case and reuses the existing per-bucket emit and positional fired bit, so the retract-and-correct path is shared. Every scheduled firing time fires and removes its own entry, and a bucket already cleaned by event-time expiry makes the firing a no-op, so nothing accumulates. The schedule is value-typed and order-preserving and processing-time timers are checkpointed, so a timer pending at snapshot fires after restore against the restored schedule and fired bits and emits at most the not-yet-emitted pad. Harness tests cover the wall-clock trigger without watermark advance, a snapshot before the timer fires, and a snapshot after the pad is emitted. --- .../exec/stream/StreamExecIntervalJoin.java | 6 +- .../StreamPhysicalIntervalJoinRule.java | 6 - .../hints/stream/EarlyFireJoinHintTest.java | 2 +- .../hints/stream/EarlyFireJoinHintTest.xml | 32 ++ .../join/interval/ProcTimeIntervalJoin.java | 4 +- .../join/interval/RowTimeIntervalJoin.java | 6 +- .../join/interval/TimeIntervalJoin.java | 117 ++++- .../interval/RowTimeIntervalJoinTest.java | 405 +++++++++++++++++- .../TimeIntervalStreamJoinTestBase.java | 17 +- 9 files changed, 554 insertions(+), 41 deletions(-) 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/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-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}); + } } From c0efadc20586c2e136fae3b3feaab268ea0258d2 Mon Sep 17 00:00:00 2001 From: weiqingy Date: Sat, 18 Jul 2026 13:05:38 -0700 Subject: [PATCH 2/2] [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. --- ...ervalJoinProcTimeEarlyFireRestoreTest.java | 67 ++ .../exec/stream/IntervalJoinRestoreTest.java | 3 +- .../exec/stream/IntervalJoinTestPrograms.java | 121 ++++ .../plan/interval-join-early-fire.json | 541 ++++++++++++++++ .../savepoint/_metadata | Bin 0 -> 19415 bytes .../interval-join-proc-time-early-fire.json | 595 ++++++++++++++++++ .../savepoint/_metadata | Bin 0 -> 19078 bytes 7 files changed, 1326 insertions(+), 1 deletion(-) create mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/IntervalJoinProcTimeEarlyFireRestoreTest.java create mode 100644 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 create mode 100644 flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-interval-join_1/interval-join-early-fire/savepoint/_metadata create mode 100644 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 create mode 100644 flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-interval-join_1/interval-join-proc-time-early-fire/savepoint/_metadata 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/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 0000000000000000000000000000000000000000..24e4f6290e76549439439918b8f01580c596f938 GIT binary patch literal 19415 zcmeHPYm6J!6~4ANn_VF6Lbegf!fFKz1rfi-uW|CD^=`bxjk9a5ouuqGHRJL1?j-h% zGc(@ZL=+ZP0whZM5LE3CT7)81RjH`dR%+GPpOzx^Ux7*mDM0@~QHu~oMXhMlbMDO8 z;~hJA%WguH8OffR`#AU9bI-l^+;cuVbKjz$5Yi7HygbH@43h0rLRDOdA1D3sq6m+< z$1AV@s((5-{pT}JT>si=!7RLn6|7qLY`;d12|874f_gHgY0{!h)rH7$G7>4SSm}Pj z%=pQ<9e2JOsVvQ|eDjUf#}2*#WRU6_lIup^V37kBImY7J%inwS!}o95HSqG=)!+6# z^RQX-vpg$s*9NsL!$s8X&Z;v!$7$#ASx zE^(Z|K~akVcx0C)CQl!{ddLk%({ogSmh(DD+${Btid3chg3E#$tSD7vE2!3kv7($@ zUWv)v%Be(s<-}raDH=H$J;8@Up;av&ECf$9WJ&iYxCj^K;!!Rh;dw3+2_+IiO<1ON zMUr)Yq$2VW!5anrC-`JU5Mt3XEyaMd zx8Va?*oANG8%Sj~m!3-%X6K6eLaLB19?0YibD7B_h0N?s@$l?{G#Sq4vYA3FBhBg! zA5CSCq>K6VTqc#x%%|rd)k{=GT?`0yL0qDNN>!3i!iy9Tm0C@a1N!MYZRk=}3zP(n z1_V_VP6sAW>oliIHA$D2X$q6_RF#CPbcU)E7t74cLS0)@bZFNHpZ04gBf~6N7OI~E z87?G`lc8i0gvpZpCm`+JKi&U!_I&+A?ba{7lz#oWH=sJoV-3kzgO1N+JNq_s%xCxJ zO5iGy!1&I~+tKm=T;BFut$WBNUgT+6j3!}^^3xcX^hYbnu)h=ubN)~wCWg6aq(b>P zA-A3e7MZExB?ZUos|%$n4X6!Smul3u0M98a2LxTPEwzr6b?d9=*}s!*no9O|X#*E` zEl^b!C9rG)u>2Mv!IYlIOUSB9LOoT@8YJ4Qd*{hDk~Y(*RcoxV8 zjXfvG_M9N=b&zosG&u^|$YQ};yM$lQ)1ideqYb@n+mqBi7x53R?vd*k)w{5x?2#Fic+P5 z+}e>Q94R*XJ~A?sf^pBKrW~hxXRDUvU;^qia8!}XGXhLUZ~)kLz=~0u2EZZx@)^)X|Ynbq=*Ooo@lM(uL@sMbXKBHIJ>D=AM z;|{Yssxg084qz!-vNFu#o%!^!Tod3?6kzwJh^KGp%z zHcp-`@yt2*_{}rT#@ZC0E>l@|bsj=Y?QV2fLE9bZZp3zrBeq+q^SFkOorkYsV>NFB zHF-|m|H|lFkTHpw=uqbI0WS;sgs6{k_`Egx%KfJxkqdDle< zwa$`l${z9w8DPNHaOgPM9u6_w=zD?wuf2I_^wW2K>!~kj5pEU=p>D0ghdtnZ{k6*i zc4|vfy#~uPVmo7i?Tne&&Oq|6uPhBN_~(C9jHb{2?lVAu5d{<@MyVwFfV8sn)Li&UWK6@u-g3BFuPR^t9x1m;|n6U%-f87}#0NDN0yG+yQ@B2}+m zZ0B$uJdOLt!FD}~*k1tuZVv{KJR7TLh# z2{!l|K-ZR4vd76D(9;5d<1^Hf_`IYKT4Q~kJdJT#oebGJ3Ye|^3uMU5&nX%}uoYP+ zI~QzB&z4S5Q3tL~0L6F*cCpr`e=j`cL)=rJ23Y@9w9r(2 zEhs3{*FDh2bw(Rt=70+Y+SwfV*t$k-ami-GLzaP6G~32H6CV7&1Myc3n6sM|zo~6l zF23TXbfqSV*2=i&(fnQiK64v~*k7dVg+=QIrV9)!iypg0s^0jQr_X)x&?7uUz-@yc zY&0m$OUXR%eWjF?Pq<^Ijlrt744r)9;?+&+Py4WUGp; zhekDIE6N^iS*xVY+p(3HY8STfvAj|)>e2w)+iZgkBS37+gDcKW6{l?#xiZm=TaA`n zf39*5BWT>XhmrE+Y4hBkp*0da^`Xb|I}k#R%{sW7$SE@!Hw8( zw{%twU7B#==>{*0rt#K^rWINL#PMUX40m&S@{Ya2g6+UB&&8AM4{QQc&BvUI=rCQP znn}v*B?I$hxGZU+pq8fuS)|~+F3*!4X413(|EmMCEqA%jT)v{WbzHI zPUSKt^;Z=5ryzJ3Y{+6&0Q_4S;u9_2(0kl51Y>q1*eV*d)CKE+XPfoFL%?kt?nMZB z>;HVP`qPiRIB@RsBR^w)>T7_ME25&(AYG**c=r`K*d#RFtY&$#b<)wx zJp%C0?uU)N+1gebVM9NK)wS_z5^Zjw)Cl&|4|s6O3lp9 vWTsMC2&$PQJl>=U41hS+{qs- 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 0000000000000000000000000000000000000000..b2b6d2b220409e7a835063955af20e2c5bf89dc6 GIT binary patch literal 19078 zcmeGkZEV}d^&~mAlQgxPc*c@6uGcyNhDD+*OR^jT!(>@m)GD?oDam3ZQz-HzGoeU@ zq#P&6QU@E_0>Kt^SceV6wzLSaAI*TG!+>H~`)5Nopg)$S=$~}h*Pj(dI}~e&0$W?} z-H{YemgLxl?X3w3NFx|j zzV++rYUsT$P8Q$Y`^wk(+g9cUvw)d91i3>r_rmKxed@#a_w;tY{+9fkw&$Lx&)t=k zl+y?Lb@^ zu2*G4H)*-g`s5^e(07WGj1O>gyv7w(S;U&oAYxEyz?sz`=EG878Oh9g6m2RWUe!xdFh48KqiqkPCeB;ueLDh3O& z2kTxxP6e|T~{iMq2>=~T}7kfi0hkH*tSlKE^hlZvNP)5#2^ zdeKo;D+TxpFU;aVQ3fss489=afL2uusf+{0xeBftlB@@k^9I%wUe01o;$`VP)&?4r zvI<|(XH^4W`_Q((lh*NEe+O;SHS~oRmOY^(?T6Ts|*q18lUeW9KILGGOasDtef&jLx7g>-&_dO9l{bpdB*|(UXN!STKO< zw$qX#8mPYsxdbl3J1I#;(wn+L!Rh9|rH0mVCGpcGBDB7gPBvb1K zBITfJ=%R+gt0jq)wa7x-5rSIXPee07Bq)$ns%7?L#X!Ey*h+*XnTr8{1{9 zl#JP>NRP%di9_+sxS^IM z!5$fRJ(d0Zzt2As@`4l-$X*z<9<C*o8}2+y1w!*FL!PWDG!c1mV(yfVXWg}n{0b8leY}lCR*@!XE?noB* z)~4IsqpW0x2bZ06Sr%YA%W7xi4u+k~DB?pC4WkHOkphBRE~^T&A};Vc4nRL=#*n%< zX8iD>=?0js7TV^3xMXl@mbj7dJS#rB-i#T@9Ra;!OC9tgC z+CgMXS?iNQ+B#)3o#DX z>q^Y{=Uau^VwodrH`9p~19w!6I2vuj6I zt!e^3&`)iLI=Ht6PoU1hyiSu}`{#qt-}t!e&vU{F>9)1oW5I_qiq%N52u?9A_ zN5Qc)GkgMd4d*2*$$tvc{+fRI`bTB$<;$1F3(wz-{(~)EM^J|e&<1S1ZTM~A_ZyDV zMd)$>?Y-D_pAf84vZCj!4T|0iKd{b`;ItN%|H*^SC*?*WV|eJFWR>ZKvFzFfrJ<$*0+oM zEd9dm;-9FCyV}{BYG)QZLk*q8bp4Pnrw8ninH+>$=>}h})W-^VQ)qKV!-7QOlxl%J z;WWU!a~x6)CJpxa{|ma=5C`<`Tm@`%uqB9E4&jo>)Odq+M<3odymZRV9&x=jjkM z$xM#}Ez;|-_OeTxj8vL+*EX%2l!Tr+?ODf1cWu)oWVe1g0ZFT$NJw+ti|x8)KxlGBLe(;X0?m zgmX~Ru)SnUc-3eNhNw*NO;_rt8nP_5gQ8D_)hnrYy7lg#X5Wrhqh#G<8Ti3P}38=82cHqkw{ zkim9Ped<0( zrkd{3ZggSlhKBhvK!VNNQb4U>jW<+H5138REEd$>j}80FYa8a)*BR&f+v%?HPa1j3 zw9`8J+Z*+XwKI43)k!~|BYRj~!8Z?8a2$UyjJlTEdDvlXTWUZ3f=`@VeOKVG+Zx!B zU_U^N=a~z)AG!(eV1&SPAn^T6+;v$9G_a`<7AZK2%coC#FZAl8Z~Q9M zF?anfC`EuR(9)LIBG%xMwMS`CdWF;U?3qBK_Bf@H6@ejP>ku5|^`g#c)GbQ7z-wZH zR|E{_PKeWJo0W8sha+*|y~}==6iEkt+DXY)3wpIoQfx5Us$Rj0NRrx%D#SU%!{(|Y z$ULlhL^xW+p^Xf1O;1m;s=+x~IeP&`%8-LO$mKzB;=#WTpo6ty_8}AoYI;CL4;-apIE@ge$9fp( zP$Q~lwPxuafQ^z3kboBusxZ{R1VVS9Mt7hI-IRs!w~`}DHYef}lM|^#JPqgJSS{RN lg9!uxJC>Xn)I=0(1xPD2tqn^PiJ#aE