From f25eb0a225208f15577319b211a8f024fa7a0748 Mon Sep 17 00:00:00 2001 From: Weiqing Yang Date: Sun, 7 Jun 2026 11:04:39 -0700 Subject: [PATCH 1/2] [FLINK-40171][table-runtime] Emit and retract early-fire results in the interval join operator Wire the EARLY_FIRE delay into the interval join operator so an outer join speculatively emits its padded unmatched row after the delay and corrects it when a real match arrives. Covers the natural timer pairings: a row-time join fires on event time, a processing-time join fires on processing time. Processing-time triggering on a row-time join stays rejected at planning. When an unmatched outer row is cached, the operator registers an early-fire timer at rowTime + delay. On that timer it emits the padded row as an INSERT and records that it fired. When the row later matches, it retracts the padded row as UPDATE_BEFORE and emits the matched row as UPDATE_AFTER, matching the update-producing changelog mode inferred for the node. The retraction is tied to the one-time matched-and-emitted flip, so a row that matches several times emits a single correction followed by ordinary inserts. The already-fired marker is a new per-side MapState> kept positionally aligned with the existing row cache, rather than widening the cache tuple, so the cache serializer is unchanged and old savepoints restore the new state empty. The marker is the single gate that keeps a row padded exactly once when the delay is at or beyond the window span. All early-fire work is gated on the hint being set, an outer join, and a non-negative window, so a plain interval join is unchanged and allocates nothing new. EmitAwareCollector carries the changelog stamping so IntervalJoinFunction stays changelog-agnostic, and every padded or matched emit stamps its RowKind explicitly to avoid leaking a kind onto a reused row. --- .../exec/stream/StreamExecIntervalJoin.java | 6 +- .../join/interval/EmitAwareCollector.java | 39 ++- .../join/interval/ProcTimeIntervalJoin.java | 6 +- .../join/interval/RowTimeIntervalJoin.java | 6 +- .../join/interval/TimeIntervalJoin.java | 308 +++++++++++++++--- .../interval/ProcTimeIntervalJoinTest.java | 84 ++++- .../interval/RowTimeIntervalJoinTest.java | 291 ++++++++++++++++- 7 files changed, 682 insertions(+), 58 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 20b676af78562e..3d8d4c7b01b51f 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 @@ -391,7 +391,8 @@ private TwoInputTransformation createProcTimeJoin( minCleanUpIntervalMillis, leftTypeInfo, rightTypeInfo, - joinFunction); + joinFunction, + earlyFireDelay == null ? -1L : earlyFireDelay); // TODO: add async version procJoinFunc to use AsyncKeyedCoProcessOperator return ExecNodeUtil.createTwoInputTransformation( leftInputTransform, @@ -428,7 +429,8 @@ private TwoInputTransformation createRowTimeJoin( rightTypeInfo, joinFunction, windowBounds.getLeftTimeIdx(), - windowBounds.getRightTimeIdx()); + windowBounds.getRightTimeIdx(), + earlyFireDelay == null ? -1L : earlyFireDelay); // TODO: add async version rowJoinFunc to use AsyncKeyedCoProcessOperator return ExecNodeUtil.createTwoInputTransformation( leftInputTransform, diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/EmitAwareCollector.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/EmitAwareCollector.java index e9fe4447f55f11..056ec578731c56 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/EmitAwareCollector.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/EmitAwareCollector.java @@ -19,17 +19,29 @@ package org.apache.flink.table.runtime.operators.join.interval; import org.apache.flink.table.data.RowData; +import org.apache.flink.types.RowKind; import org.apache.flink.util.Collector; /** * Collector to wrap a [[org.apache.flink.table.dataformat.RowData]] and to track whether a row has * been emitted by the inner collector. + * + *

The collector can be armed with a correction before a single matched row is collected. When + * armed, the next collected row is treated as the corrected result of a previously emitted + * speculative outer-join pad: the pending pad is emitted first stamped {@link + * RowKind#UPDATE_BEFORE}, then the matched row is stamped {@link RowKind#UPDATE_AFTER}. This turns + * the join function's single {@code INSERT} emit into the {@code -U}/{@code +U} pair without the + * join function knowing about changelogs. When not armed, collected rows are forwarded with their + * existing {@link RowKind}. */ class EmitAwareCollector implements Collector { private boolean emitted = false; private Collector innerCollector; + // The pad to retract before the next matched row, or null when no correction is armed. + private RowData pendingRetraction; + void reset() { emitted = false; } @@ -42,10 +54,35 @@ void setInnerCollector(Collector innerCollector) { this.innerCollector = innerCollector; } + /** + * Arms the collector so the next collected matched row is corrected into a {@code -U}/{@code + * +U} pair against the given padded row. + */ + void armRetraction(RowData retractionPad) { + retractionPad.setRowKind(RowKind.UPDATE_BEFORE); + this.pendingRetraction = retractionPad; + } + + /** Clears an armed correction that was never consumed (the join condition did not match). */ + void disarm() { + this.pendingRetraction = null; + } + @Override public void collect(RowData record) { emitted = true; - innerCollector.collect(record); + if (pendingRetraction != null) { + innerCollector.collect(pendingRetraction); + pendingRetraction = null; + record.setRowKind(RowKind.UPDATE_AFTER); + innerCollector.collect(record); + } else { + // The matched row reuses a single instance whose kind may have been left as + // UPDATE_AFTER by a previous correction; force INSERT so a later ordinary match is not + // mis-emitted as an update. + record.setRowKind(RowKind.INSERT); + innerCollector.collect(record); + } } @Override 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 1fefc759fd473e..84ad4526289786 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 @@ -34,7 +34,8 @@ public ProcTimeIntervalJoin( long minCleanUpInterval, InternalTypeInfo leftType, InternalTypeInfo rightType, - IntervalJoinFunction genJoinFunc) { + IntervalJoinFunction genJoinFunc, + long earlyFireDelay) { super( joinType, leftLowerBound, @@ -43,7 +44,8 @@ public ProcTimeIntervalJoin( minCleanUpInterval, leftType, rightType, - genJoinFunc); + genJoinFunc, + earlyFireDelay); } @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 5d972104a6692f..57972aff22713f 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 @@ -40,7 +40,8 @@ public RowTimeIntervalJoin( InternalTypeInfo rightType, IntervalJoinFunction joinFunc, int leftTimeIdx, - int rightTimeIdx) { + int rightTimeIdx, + long earlyFireDelay) { super( joinType, leftLowerBound, @@ -49,7 +50,8 @@ public RowTimeIntervalJoin( minCleanUpInterval, leftType, rightType, - joinFunc); + joinFunc, + earlyFireDelay); 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 4dbf1250dac725..59c389aa92e917 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 @@ -34,6 +34,7 @@ import org.apache.flink.table.runtime.operators.join.FlinkJoinType; import org.apache.flink.table.runtime.operators.join.OuterJoinPaddingUtil; import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.types.RowKind; import org.apache.flink.util.Collector; import org.slf4j.Logger; @@ -64,6 +65,13 @@ abstract class TimeIntervalJoin extends KeyedCoProcessFunction>> rightCache; + // For each cached outer row, whether its speculative early-fire pad has already been emitted. + // The list is positionally aligned 1:1 with the row-time bucket in leftCache / rightCache, so + // firedState.get(t).get(i) corresponds to cache.get(t).get(i). It is kept as a parallel list + // rather than a third tuple field so the existing cache serializer stays unchanged. The bit + // gates both the unmatched window-close pad (it must not be emitted twice) and the retraction + // on a later match (only a row that was speculatively padded needs correcting). + private transient MapState> leftFiredState; + private transient MapState> rightFiredState; + // 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 @@ -92,7 +109,8 @@ abstract class TimeIntervalJoin extends KeyedCoProcessFunction leftType, InternalTypeInfo rightType, - IntervalJoinFunction joinFunc) { + IntervalJoinFunction joinFunc, + long earlyFireDelay) { this.joinType = joinType; this.leftRelativeSize = -leftLowerBound; this.rightRelativeSize = leftUpperBound; @@ -104,6 +122,14 @@ abstract class TimeIntervalJoin extends KeyedCoProcessFunction= 0 + && joinType.isOuter() + && (leftRelativeSize + rightRelativeSize) >= 0; } @Override @@ -129,6 +155,27 @@ public void open(OpenContext openContext) throws Exception { rightRowListTypeInfo); rightCache = getRuntimeContext().getMapState(rightMapStateDescriptor); + // Early-fire bookkeeping, aligned with the caches above. New descriptor names restore as + // empty state from savepoints taken before early firing existed. + if (earlyFireEnabled) { + ListTypeInfo firedListTypeInfo = + new ListTypeInfo<>(BasicTypeInfo.BOOLEAN_TYPE_INFO); + leftFiredState = + getRuntimeContext() + .getMapState( + new MapStateDescriptor<>( + "IntervalJoinLeftFired", + BasicTypeInfo.LONG_TYPE_INFO, + firedListTypeInfo)); + rightFiredState = + getRuntimeContext() + .getMapState( + new MapStateDescriptor<>( + "IntervalJoinRightFired", + BasicTypeInfo.LONG_TYPE_INFO, + firedListTypeInfo)); + } + // Initialize the timer states. ValueStateDescriptor leftValueStateDescriptor = new ValueStateDescriptor<>("IntervalJoinLeftTimerState", Long.class); @@ -178,10 +225,28 @@ public void processElement1(RowData leftRow, Context ctx, Collector out if (rightTime >= rightQualifiedLowerBound && rightTime <= rightQualifiedUpperBound) { List> rightRows = rightEntry.getValue(); + List rightFired = + earlyFireEnabled + ? firedBits(rightFiredState, rightTime, rightRows) + : null; boolean entryUpdated = false; - for (Tuple2 tuple : rightRows) { + for (int i = 0; i < rightRows.size(); i++) { + Tuple2 tuple = rightRows.get(i); joinCollector.reset(); + boolean retract = + rightFired != null + && joinType.isRightOuter() + && !tuple.f1 + && rightFired.get(i); + if (retract) { + // The speculative pad for this right row was already emitted as an + // insert; arm the collector so the match becomes -U(pad)/+U(match). + joinCollector.armRetraction(paddingUtil.padRight(tuple.f0)); + } joinFunction.join(leftRow, tuple.f0, joinCollector); + if (retract && !joinCollector.isEmitted()) { + joinCollector.disarm(); + } emitted = emitted || joinCollector.isEmitted(); if (joinType.isRightOuter()) { if (!tuple.f1 && joinCollector.isEmitted()) { @@ -200,17 +265,24 @@ public void processElement1(RowData leftRow, Context ctx, Collector out if (rightTime <= rightExpirationTime) { if (joinType.isRightOuter()) { List> rightRows = rightEntry.getValue(); - rightRows.forEach( - (Tuple2 tuple) -> { - if (!tuple.f1) { - // Emit a null padding result if the right row has never - // been successfully joined. - joinCollector.collect(paddingUtil.padRight(tuple.f0)); - } - }); + List rightFired = + earlyFireEnabled + ? firedBits(rightFiredState, rightTime, rightRows) + : null; + for (int i = 0; i < rightRows.size(); i++) { + Tuple2 tuple = rightRows.get(i); + // Skip a row whose speculative pad already fired: it is correct as + // emitted and must not be padded a second time. + if (!tuple.f1 && (rightFired == null || !rightFired.get(i))) { + collectPad(paddingUtil.padRight(tuple.f0)); + } + } } // eager remove rightIterator.remove(); + if (earlyFireEnabled) { + removeFired(rightFiredState, rightTime); + } } // We could do the short-cutting optimization here once we get a state with // ordered keys. } @@ -226,13 +298,21 @@ public void processElement1(RowData leftRow, Context ctx, Collector out } leftRowList.add(Tuple2.of(leftRow, emitted)); leftCache.put(timeForLeftRow, leftRowList); + if (earlyFireEnabled && joinType.isLeftOuter()) { + // The new tuple has not been speculatively padded yet, so its bit starts false. + appendFired(leftFiredState, timeForLeftRow); + if (!emitted) { + // Schedule a speculative pad of this unmatched left row after the delay. + registerTimer(ctx, timeForLeftRow + earlyFireDelay); + } + } if (rightTimerState.value() == null) { // Register a timer on the RIGHT stream to remove rows. registerCleanUpTimer(ctx, timeForLeftRow, true); } } else if (!emitted && joinType.isLeftOuter()) { // Emit a null padding result if the left row is not cached and successfully joined. - joinCollector.collect(paddingUtil.padLeft(leftRow)); + collectPad(paddingUtil.padLeft(leftRow)); } } @@ -261,10 +341,26 @@ public void processElement2(RowData rightRow, Context ctx, Collector ou Long leftTime = leftEntry.getKey(); if (leftTime >= leftQualifiedLowerBound && leftTime <= leftQualifiedUpperBound) { List> leftRows = leftEntry.getValue(); + List leftFired = + earlyFireEnabled ? firedBits(leftFiredState, leftTime, leftRows) : null; boolean entryUpdated = false; - for (Tuple2 tuple : leftRows) { + for (int i = 0; i < leftRows.size(); i++) { + Tuple2 tuple = leftRows.get(i); joinCollector.reset(); + boolean retract = + leftFired != null + && joinType.isLeftOuter() + && !tuple.f1 + && leftFired.get(i); + if (retract) { + // The speculative pad for this left row was already emitted as an + // insert; arm the collector so the match becomes -U(pad)/+U(match). + joinCollector.armRetraction(paddingUtil.padLeft(tuple.f0)); + } joinFunction.join(tuple.f0, rightRow, joinCollector); + if (retract && !joinCollector.isEmitted()) { + joinCollector.disarm(); + } emitted = emitted || joinCollector.isEmitted(); if (joinType.isLeftOuter()) { if (!tuple.f1 && joinCollector.isEmitted()) { @@ -283,17 +379,24 @@ public void processElement2(RowData rightRow, Context ctx, Collector ou if (leftTime <= leftExpirationTime) { if (joinType.isLeftOuter()) { List> leftRows = leftEntry.getValue(); - leftRows.forEach( - (Tuple2 tuple) -> { - if (!tuple.f1) { - // Emit a null padding result if the left row has never been - // successfully joined. - joinCollector.collect(paddingUtil.padLeft(tuple.f0)); - } - }); + List leftFired = + earlyFireEnabled + ? firedBits(leftFiredState, leftTime, leftRows) + : null; + for (int i = 0; i < leftRows.size(); i++) { + Tuple2 tuple = leftRows.get(i); + // Skip a row whose speculative pad already fired: it is correct as + // emitted and must not be padded a second time. + if (!tuple.f1 && (leftFired == null || !leftFired.get(i))) { + collectPad(paddingUtil.padLeft(tuple.f0)); + } + } } // eager remove leftIterator.remove(); + if (earlyFireEnabled) { + removeFired(leftFiredState, leftTime); + } } // We could do the short-cutting optimization here once we get a state with // ordered keys. } @@ -309,13 +412,21 @@ public void processElement2(RowData rightRow, Context ctx, Collector ou } rightRowList.add(Tuple2.of(rightRow, emitted)); rightCache.put(timeForRightRow, rightRowList); + if (earlyFireEnabled && joinType.isRightOuter()) { + // The new tuple has not been speculatively padded yet, so its bit starts false. + appendFired(rightFiredState, timeForRightRow); + if (!emitted) { + // Schedule a speculative pad of this unmatched right row after the delay. + registerTimer(ctx, timeForRightRow + earlyFireDelay); + } + } if (leftTimerState.value() == null) { // Register a timer on the LEFT stream to remove rows. registerCleanUpTimer(ctx, timeForRightRow, false); } } else if (!emitted && joinType.isRightOuter()) { // Emit a null padding result if the right row is not cached and successfully joined. - joinCollector.collect(paddingUtil.padRight(rightRow)); + collectPad(paddingUtil.padRight(rightRow)); } } @@ -325,6 +436,22 @@ public void onTimer(long timestamp, OnTimerContext ctx, Collector out) joinFunction.setJoinKey(ctx.getCurrentKey()); 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) { + long rowTime = timestamp - earlyFireDelay; + if (joinType.isLeftOuter()) { + earlyFire(leftCache, leftFiredState, rowTime, true); + } + if (joinType.isRightOuter()) { + earlyFire(rightCache, rightFiredState, rowTime, false); + } + } + // In the future, we should separate the left and right watermarks. Otherwise, the // registered timer of the faster stream will be delayed, even if the watermarks have // already been emitted by the source. @@ -332,14 +459,57 @@ public void onTimer(long timestamp, OnTimerContext ctx, Collector out) if (leftCleanUpTime != null && timestamp == leftCleanUpTime) { rightExpirationTime = calExpirationTime(leftOperatorTime, rightRelativeSize); removeExpiredRows( - joinCollector, rightExpirationTime, rightCache, leftTimerState, ctx, false); + joinCollector, + rightExpirationTime, + rightCache, + rightFiredState, + leftTimerState, + ctx, + false); } Long rightCleanUpTime = rightTimerState.value(); if (rightCleanUpTime != null && timestamp == rightCleanUpTime) { leftExpirationTime = calExpirationTime(rightOperatorTime, leftRelativeSize); removeExpiredRows( - joinCollector, leftExpirationTime, leftCache, rightTimerState, ctx, true); + joinCollector, + leftExpirationTime, + leftCache, + leftFiredState, + rightTimerState, + ctx, + true); + } + } + + /** + * Emit the speculative null-padding result for every cached outer row at the given row time + * that is still unmatched and has not yet had its pad emitted, flipping its fired bit so + * neither this path nor the later window-close pad emits it again. + */ + private void earlyFire( + MapState>> rowCache, + MapState> firedState, + long rowTime, + boolean padLeft) + throws Exception { + List> rows = rowCache.get(rowTime); + if (rows == null) { + return; + } + List fired = firedBits(firedState, rowTime, rows); + boolean changed = false; + for (int i = 0; i < rows.size(); i++) { + Tuple2 tuple = rows.get(i); + if (!tuple.f1 && !fired.get(i)) { + collectPad( + padLeft ? paddingUtil.padLeft(tuple.f0) : paddingUtil.padRight(tuple.f0)); + fired.set(i, true); + changed = true; + } + } + if (changed) { + firedState.put(rowTime, fired); } } @@ -396,6 +566,7 @@ private void removeExpiredRows( Collector collector, long expirationTime, MapState>> rowCache, + MapState> firedState, ValueState timerState, OnTimerContext ctx, boolean removeLeft) @@ -410,28 +581,29 @@ private void removeExpiredRows( Map.Entry>> entry = iterator.next(); Long rowTime = entry.getKey(); if (rowTime <= expirationTime) { - if (removeLeft && joinType.isLeftOuter()) { + boolean removeOuter = + (removeLeft && joinType.isLeftOuter()) + || (!removeLeft && joinType.isRightOuter()); + if (removeOuter) { List> rows = entry.getValue(); - rows.forEach( - (Tuple2 tuple) -> { - if (!tuple.f1) { - // Emit a null padding result if the row has never been - // successfully joined. - collector.collect(paddingUtil.padLeft(tuple.f0)); - } - }); - } else if (!removeLeft && joinType.isRightOuter()) { - List> rows = entry.getValue(); - rows.forEach( - (Tuple2 tuple) -> { - if (!tuple.f1) { - // Emit a null padding result if the row has never been - // successfully joined. - collector.collect(paddingUtil.padRight(tuple.f0)); - } - }); + List fired = + earlyFireEnabled ? firedBits(firedState, rowTime, rows) : null; + for (int i = 0; i < rows.size(); i++) { + Tuple2 tuple = rows.get(i); + // Emit a null padding result only if the row was never matched and its + // speculative pad has not already been emitted. + if (!tuple.f1 && (fired == null || !fired.get(i))) { + collectPad( + removeLeft + ? paddingUtil.padLeft(tuple.f0) + : paddingUtil.padRight(tuple.f0)); + } + } } iterator.remove(); + if (earlyFireEnabled) { + removeFired(firedState, rowTime); + } } else { // We find the earliest timestamp that is still valid. if (rowTime < earliestTimestamp || earliestTimestamp < 0) { @@ -447,6 +619,58 @@ private void removeExpiredRows( // No rows left in the cache. Clear the states and the timerState will be 0. timerState.clear(); rowCache.clear(); + if (earlyFireEnabled && firedState != null) { + firedState.clear(); + } + } + } + + /** + * Emit a padded outer-join row as an insert, overriding any leaked row kind on the reused row. + */ + private void collectPad(RowData paddedRow) { + paddedRow.setRowKind(RowKind.INSERT); + joinCollector.collect(paddedRow); + } + + /** + * Return the fired-bit list aligned with the given cache bucket. Only called when early firing + * is enabled. When the stored list is absent or its length no longer matches the bucket (e.g. + * after a restore), a fresh all-false list of the right length is rebuilt so no row is ever + * treated as already fired. + */ + private List firedBits( + MapState> firedState, + long rowTime, + List> rows) + throws Exception { + if (firedState != null) { + List fired = firedState.get(rowTime); + if (fired != null && fired.size() == rows.size()) { + return fired; + } + } + List fired = new ArrayList<>(rows.size()); + for (int i = 0; i < rows.size(); i++) { + fired.add(Boolean.FALSE); + } + return fired; + } + + private void appendFired(MapState> firedState, long rowTime) + throws Exception { + List fired = firedState.get(rowTime); + if (fired == null) { + fired = new ArrayList<>(1); + } + fired.add(Boolean.FALSE); + firedState.put(rowTime, fired); + } + + private void removeFired(MapState> firedState, long rowTime) + throws Exception { + if (firedState != null) { + firedState.remove(rowTime); } } diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/ProcTimeIntervalJoinTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/ProcTimeIntervalJoinTest.java index 071a5669afec3c..42a2e1dacbd412 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/ProcTimeIntervalJoinTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/ProcTimeIntervalJoinTest.java @@ -33,6 +33,8 @@ import java.util.List; import static org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord; +import static org.apache.flink.table.runtime.util.StreamRecordUtils.updateAfterRecord; +import static org.apache.flink.table.runtime.util.StreamRecordUtils.updateBeforeRecord; import static org.assertj.core.api.Assertions.assertThat; /** Test for {@link ProcTimeIntervalJoin}. */ @@ -49,7 +51,7 @@ class ProcTimeIntervalJoinTest extends TimeIntervalStreamJoinTestBase { void testProcTimeInnerJoinWithCommonBounds() throws Exception { ProcTimeIntervalJoin joinProcessFunc = new ProcTimeIntervalJoin( - FlinkJoinType.INNER, -10, 20, 15, rowType, rowType, joinFunction); + FlinkJoinType.INNER, -10, 20, 15, rowType, rowType, joinFunction, -1L); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); testHarness.open(); @@ -108,7 +110,7 @@ void testProcTimeInnerJoinWithCommonBounds() throws Exception { void testProcTimeInnerJoinWithNegativeBounds() throws Exception { ProcTimeIntervalJoin joinProcessFunc = new ProcTimeIntervalJoin( - FlinkJoinType.INNER, -10, -5, 2, rowType, rowType, joinFunction); + FlinkJoinType.INNER, -10, -5, 2, rowType, rowType, joinFunction, -1L); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -168,6 +170,84 @@ void testProcTimeInnerJoinWithNegativeBounds() throws Exception { testHarness.close(); } + /** Early fire on processing time, then a match retracts the speculative pad. */ + @Test + void testProcTimeLeftOuterEarlyFireThenMatch() throws Exception { + ProcTimeIntervalJoin joinProcessFunc = + new ProcTimeIntervalJoin( + FlinkJoinType.LEFT, -5, 9, 0, rowType, rowType, joinFunction, 3L); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + + testHarness.setProcessingTime(10); + testHarness.processElement1(insertRecord(1L, "a")); + // One cleanup timer plus one early-fire timer at 10 + 3 = 13. + assertThat(testHarness.numProcessingTimeTimers()).isEqualTo(2); + + // Fire the early-fire timer: the unmatched left row is speculatively padded. + testHarness.setProcessingTime(13); + + // A right row matches the early-fired left row. + testHarness.setProcessingTime(14); + testHarness.processElement2(insertRecord(1L, "b")); + + // Advance past cleanup: no further pad. + testHarness.setProcessingTime(40); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(insertRecord(1L, "a", null, null)); + expectedOutput.add(updateBeforeRecord(1L, "a", null, null)); + expectedOutput.add(updateAfterRecord(1L, "a", 1L, "b")); + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + /** With early fire disabled the processing-time inner join behaves exactly as before. */ + @Test + void testProcTimeInnerJoinIgnoresEarlyFire() throws Exception { + ProcTimeIntervalJoin joinProcessFunc = + new ProcTimeIntervalJoin( + FlinkJoinType.INNER, -5, 9, 0, rowType, rowType, joinFunction, 3L); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + + testHarness.setProcessingTime(10); + testHarness.processElement1(insertRecord(1L, "a")); + // No early-fire timer for an inner join. + assertThat(testHarness.numProcessingTimeTimers()).isEqualTo(1); + + testHarness.setProcessingTime(13); + testHarness.setProcessingTime(40); + + List expectedOutput = new ArrayList<>(); + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + /** Delay larger than the window span still pads an unmatched row exactly once. */ + @Test + void testProcTimeLeftOuterEarlyFireDelayExceedsSpan() throws Exception { + // Window span is 5 + 9 = 14; the delay exceeds it so cleanup may reach the row first. + ProcTimeIntervalJoin joinProcessFunc = + new ProcTimeIntervalJoin( + FlinkJoinType.LEFT, -5, 9, 0, rowType, rowType, joinFunction, 20L); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + + testHarness.setProcessingTime(10); + testHarness.processElement1(insertRecord(1L, "a")); + // Cleanup at 16, early fire at 30: advancing past both must still emit a single pad. + testHarness.setProcessingTime(35); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(insertRecord(1L, "a", null, null)); + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + private KeyedTwoInputStreamOperatorTestHarness createTestHarness(ProcTimeIntervalJoin intervalJoinFunc) throws Exception { KeyedCoProcessOperator operator = 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 d0e6530c190a21..db3ddb5bfc7bdf 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 @@ -44,6 +44,8 @@ import static org.apache.flink.configuration.CheckpointingOptions.ENABLE_UNALIGNED; import static org.apache.flink.configuration.CheckpointingOptions.ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS; import static org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord; +import static org.apache.flink.table.runtime.util.StreamRecordUtils.updateAfterRecord; +import static org.apache.flink.table.runtime.util.StreamRecordUtils.updateBeforeRecord; import static org.assertj.core.api.Assertions.assertThat; /** Test for {@link RowTimeIntervalJoin}. */ @@ -60,7 +62,17 @@ class RowTimeIntervalJoinTest extends TimeIntervalStreamJoinTestBase { void testRowTimeInnerJoinWithCommonBounds() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.INNER, -10, 20, 0, 15, rowType, rowType, joinFunction, 0, 0); + FlinkJoinType.INNER, + -10, + 20, + 0, + 15, + rowType, + rowType, + joinFunction, + 0, + 0, + -1L); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -125,7 +137,17 @@ void testRowTimeInnerJoinWithCommonBounds() throws Exception { void testRowTimeInnerJoinWithNegativeBounds() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.INNER, -10, -7, 0, 0, rowType, rowType, joinFunction, 0, 0); + FlinkJoinType.INNER, + -10, + -7, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + -1L); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -180,7 +202,7 @@ void testRowTimeInnerJoinWithNegativeBounds() throws Exception { void testRowTimeInnerJoinRealtimeCleanUp() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.LEFT, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0); + FlinkJoinType.LEFT, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, -1L); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -209,7 +231,7 @@ void testRowTimeInnerJoinRealtimeCleanUp() throws Exception { void testRowTimeLeftOuterJoin() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.LEFT, -5, 9, 0, 7, rowType, rowType, joinFunction, 0, 0); + FlinkJoinType.LEFT, -5, 9, 0, 7, rowType, rowType, joinFunction, 0, 0, -1L); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -279,7 +301,17 @@ void testRowTimeLeftOuterJoin() throws Exception { void testRowTimeRightOuterJoin() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.RIGHT, -5, 9, 0, 7, rowType, rowType, joinFunction, 0, 0); + FlinkJoinType.RIGHT, + -5, + 9, + 0, + 7, + rowType, + rowType, + joinFunction, + 0, + 0, + -1L); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -350,7 +382,7 @@ void testRowTimeRightOuterJoin() throws Exception { void testRowTimeFullOuterJoin() throws Exception { RowTimeIntervalJoin joinProcessFunc = new RowTimeIntervalJoin( - FlinkJoinType.FULL, -5, 9, 0, 7, rowType, rowType, joinFunction, 0, 0); + FlinkJoinType.FULL, -5, 9, 0, 7, rowType, rowType, joinFunction, 0, 0, -1L); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -439,7 +471,8 @@ public void testInterruptibleTimers() throws Exception { rowType, joinFunction, 0, - 0); + 0, + -1L); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -512,6 +545,250 @@ public void testInterruptibleTimers() throws Exception { testHarness.close(); } + /** Early fire: an unmatched left outer row is speculatively padded once the delay elapses. */ + @Test + void testRowTimeLeftOuterEarlyFire() throws Exception { + RowTimeIntervalJoin joinProcessFunc = + new RowTimeIntervalJoin( + FlinkJoinType.LEFT, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, 3L); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + + testHarness.processElement1(insertRecord(10L, "k1")); + // One cleanup timer plus one early-fire timer at 10 + 3 = 13. + assertThat(testHarness.numEventTimeTimers()).isEqualTo(2); + + // Cross the early-fire time but not the cleanup time (16): the speculative pad is emitted. + testHarness.processWatermark1(new Watermark(13)); + testHarness.processWatermark2(new Watermark(13)); + + // Cross the cleanup time: the already-fired row must not be padded again. + testHarness.processWatermark1(new Watermark(20)); + testHarness.processWatermark2(new Watermark(20)); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(insertRecord(10L, "k1", null, null)); + expectedOutput.add(new Watermark(13 - 9)); + expectedOutput.add(new Watermark(20 - 9)); + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + /** Early fire then a match: the speculative pad is retracted and replaced by the joined row. */ + @Test + void testRowTimeLeftOuterEarlyFireThenMatch() throws Exception { + RowTimeIntervalJoin joinProcessFunc = + new RowTimeIntervalJoin( + FlinkJoinType.LEFT, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, 3L); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + + testHarness.processElement1(insertRecord(10L, "k1")); + testHarness.processWatermark1(new Watermark(13)); + testHarness.processWatermark2(new Watermark(13)); + + // A right row arrives in window (10 in [12 - 5, 12 + 9]) and matches the early-fired left + // row. + testHarness.processElement2(insertRecord(12L, "k1")); + + // Cross cleanup: 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(new Watermark(13 - 9)); + 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(); + } + + /** Symmetric retraction for a right outer join. */ + @Test + void testRowTimeRightOuterEarlyFireThenMatch() throws Exception { + RowTimeIntervalJoin joinProcessFunc = + new RowTimeIntervalJoin( + FlinkJoinType.RIGHT, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, 3L); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + + testHarness.processElement2(insertRecord(10L, "k1")); + testHarness.processWatermark1(new Watermark(13)); + testHarness.processWatermark2(new Watermark(13)); + + // A left row in window matches the early-fired right row. + testHarness.processElement1(insertRecord(12L, "k1")); + + testHarness.processWatermark1(new Watermark(30)); + testHarness.processWatermark2(new Watermark(30)); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(insertRecord(null, null, 10L, "k1")); + expectedOutput.add(new Watermark(13 - 9)); + expectedOutput.add(updateBeforeRecord(null, null, 10L, "k1")); + expectedOutput.add(updateAfterRecord(12L, "k1", 10L, "k1")); + expectedOutput.add(new Watermark(30 - 9)); + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + /** Full outer: both sides early-fire; only the side that later matches is retracted. */ + @Test + void testRowTimeFullOuterEarlyFireOneMatches() throws Exception { + RowTimeIntervalJoin joinProcessFunc = + new RowTimeIntervalJoin( + FlinkJoinType.FULL, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, 3L); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + + // Left row at 10 (will match later), right row at 40 (stays unmatched). + testHarness.processElement1(insertRecord(10L, "k1")); + testHarness.processElement2(insertRecord(40L, "k1")); + testHarness.processWatermark1(new Watermark(13)); + testHarness.processWatermark2(new Watermark(13)); + + // Match the left row. + testHarness.processElement2(insertRecord(12L, "k1")); + + // Fire the right row's early-fire timer (43) and then close everything. + testHarness.processWatermark1(new Watermark(43)); + testHarness.processWatermark2(new Watermark(43)); + testHarness.processWatermark1(new Watermark(60)); + testHarness.processWatermark2(new Watermark(60)); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(insertRecord(10L, "k1", null, null)); + expectedOutput.add(new Watermark(13 - 9)); + expectedOutput.add(updateBeforeRecord(10L, "k1", null, null)); + expectedOutput.add(updateAfterRecord(10L, "k1", 12L, "k1")); + expectedOutput.add(insertRecord(null, null, 40L, "k1")); + expectedOutput.add(new Watermark(43 - 9)); + expectedOutput.add(new Watermark(60 - 9)); + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + /** With early fire disabled the operator output is identical to a plain interval join. */ + @Test + void testRowTimeInnerJoinIgnoresEarlyFire() throws Exception { + RowTimeIntervalJoin joinProcessFunc = + new RowTimeIntervalJoin( + FlinkJoinType.INNER, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, 3L); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + + testHarness.processElement1(insertRecord(10L, "k1")); + // No early-fire timer for an inner join: only the cleanup timer is registered. + assertThat(testHarness.numEventTimeTimers()).isEqualTo(1); + + testHarness.processWatermark1(new Watermark(13)); + testHarness.processWatermark2(new Watermark(13)); + testHarness.processWatermark1(new Watermark(30)); + testHarness.processWatermark2(new Watermark(30)); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(new Watermark(13 - 9)); + expectedOutput.add(new Watermark(30 - 9)); + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + /** Delay larger than the window span still pads an unmatched row exactly once. */ + @Test + 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); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + + testHarness.processElement1(insertRecord(10L, "k1")); + // Cleanup at 16, early fire at 30: advancing past both must still emit a single pad. + testHarness.processWatermark1(new Watermark(35)); + testHarness.processWatermark2(new Watermark(35)); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(insertRecord(10L, "k1", null, null)); + expectedOutput.add(new Watermark(35 - 9)); + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + /** A normal pad emitted after a retraction must be an insert, not a leaked update-before. */ + @Test + void testRowTimeEarlyFireRowKindIsolation() throws Exception { + RowTimeIntervalJoin joinProcessFunc = + new RowTimeIntervalJoin( + FlinkJoinType.LEFT, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, 3L); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + + // Row A early-fires then matches, producing a retraction that leaves the reused pad row at + // UPDATE_BEFORE. + testHarness.processElement1(insertRecord(10L, "k1")); + testHarness.processWatermark1(new Watermark(13)); + testHarness.processWatermark2(new Watermark(13)); + testHarness.processElement2(insertRecord(12L, "k1")); + + // Row B early-fires and never matches; its window-close pad must be an insert. + testHarness.processElement1(insertRecord(40L, "k2")); + testHarness.processWatermark1(new Watermark(60)); + testHarness.processWatermark2(new Watermark(60)); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(insertRecord(10L, "k1", null, null)); + expectedOutput.add(new Watermark(13 - 9)); + expectedOutput.add(updateBeforeRecord(10L, "k1", null, null)); + expectedOutput.add(updateAfterRecord(10L, "k1", 12L, "k1")); + expectedOutput.add(insertRecord(40L, "k2", null, null)); + expectedOutput.add(new Watermark(60 - 9)); + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + + /** Multiple matches of an early-fired row produce exactly one retraction. */ + @Test + void testRowTimeLeftOuterEarlyFireMultiMatch() throws Exception { + RowTimeIntervalJoin joinProcessFunc = + new RowTimeIntervalJoin( + FlinkJoinType.LEFT, -5, 9, 0, 0, rowType, rowType, joinFunction, 0, 0, 3L); + KeyedTwoInputStreamOperatorTestHarness testHarness = + createTestHarness(joinProcessFunc); + testHarness.open(); + + testHarness.processElement1(insertRecord(10L, "k1")); + testHarness.processWatermark1(new Watermark(13)); + testHarness.processWatermark2(new Watermark(13)); + + // First match: corrected via -U/+U. + testHarness.processElement2(insertRecord(12L, "k1")); + // Second match of the same left row: an ordinary insert, no second retraction. + testHarness.processElement2(insertRecord(14L, "k1")); + + testHarness.processWatermark1(new Watermark(30)); + testHarness.processWatermark2(new Watermark(30)); + + List expectedOutput = new ArrayList<>(); + expectedOutput.add(insertRecord(10L, "k1", null, null)); + expectedOutput.add(new Watermark(13 - 9)); + expectedOutput.add(updateBeforeRecord(10L, "k1", null, null)); + expectedOutput.add(updateAfterRecord(10L, "k1", 12L, "k1")); + expectedOutput.add(insertRecord(10L, "k1", 14L, "k1")); + expectedOutput.add(new Watermark(30 - 9)); + assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); + testHarness.close(); + } + private KeyedTwoInputStreamOperatorTestHarness createTestHarness(RowTimeIntervalJoin intervalJoinFunc) throws Exception { KeyedCoProcessOperator operator = From 6e742715c675ded0a60040c0d4125fcdc3e3d9bf Mon Sep 17 00:00:00 2001 From: Weiqing Yang Date: Sun, 7 Jun 2026 12:20:32 -0700 Subject: [PATCH 2/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 59c389aa92e917..041dd50f8589fc 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) { @@ -417,7 +451,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) { @@ -437,12 +471,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); @@ -513,6 +565,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 db3ddb5bfc7bdf..c93b85c06a7b52 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(); @@ -679,7 +761,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(); @@ -706,7 +799,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(); @@ -728,7 +832,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(); @@ -761,7 +876,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(); @@ -789,6 +915,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}); + } }