From f25eb0a225208f15577319b211a8f024fa7a0748 Mon Sep 17 00:00:00 2001 From: Weiqing Yang Date: Sun, 7 Jun 2026 11:04:39 -0700 Subject: [PATCH 1/3] [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 20b676af78562..3d8d4c7b01b51 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 e9fe4447f55f1..056ec578731c5 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 1fefc759fd473..84ad452628978 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 5d972104a6692..57972aff22713 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 4dbf1250dac72..59c389aa92e91 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 071a5669afec3..42a2e1dacbd41 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 d0e6530c190a2..db3ddb5bfc7bd 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 7912a13e609d748962e80993178e58889f979a6f Mon Sep 17 00:00:00 2001 From: weiqingy Date: Mon, 10 Aug 2026 21:01:01 -0700 Subject: [PATCH 2/3] [FLINK-40171][table-runtime] Skip the fired-state read on a non-outer side The early-fire fired bits are only consulted when retracting a speculative pad, which can only happen on an outer side. Gate each per-side read on that side being outer, so a one-sided outer join stops reading state it can never use on the matching path. Extend the full-outer test to also drive the right-side retraction, which the suite did not exercise before. --- .../operators/join/interval/TimeIntervalJoin.java | 6 ++++-- .../join/interval/RowTimeIntervalJoinTest.java | 13 ++++++++++--- 2 files changed, 14 insertions(+), 5 deletions(-) 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 59c389aa92e91..f2abfeb00f22f 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 @@ -226,7 +226,7 @@ public void processElement1(RowData leftRow, Context ctx, Collector out && rightTime <= rightQualifiedUpperBound) { List> rightRows = rightEntry.getValue(); List rightFired = - earlyFireEnabled + earlyFireEnabled && joinType.isRightOuter() ? firedBits(rightFiredState, rightTime, rightRows) : null; boolean entryUpdated = false; @@ -342,7 +342,9 @@ public void processElement2(RowData rightRow, Context ctx, Collector ou if (leftTime >= leftQualifiedLowerBound && leftTime <= leftQualifiedUpperBound) { List> leftRows = leftEntry.getValue(); List leftFired = - earlyFireEnabled ? firedBits(leftFiredState, leftTime, leftRows) : null; + earlyFireEnabled && joinType.isLeftOuter() + ? firedBits(leftFiredState, leftTime, leftRows) + : null; boolean entryUpdated = false; for (int i = 0; i < leftRows.size(); i++) { Tuple2 tuple = leftRows.get(i); 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 db3ddb5bfc7bd..a3748a61563b5 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 @@ -637,7 +637,7 @@ void testRowTimeRightOuterEarlyFireThenMatch() throws Exception { testHarness.close(); } - /** Full outer: both sides early-fire; only the side that later matches is retracted. */ + /** Full outer: both sides early-fire and both are retracted once their match arrives. */ @Test void testRowTimeFullOuterEarlyFireOneMatches() throws Exception { RowTimeIntervalJoin joinProcessFunc = @@ -647,7 +647,7 @@ void testRowTimeFullOuterEarlyFireOneMatches() throws Exception { createTestHarness(joinProcessFunc); testHarness.open(); - // Left row at 10 (will match later), right row at 40 (stays unmatched). + // Left row at 10 and right row at 40, each matched later by a row from the other side. testHarness.processElement1(insertRecord(10L, "k1")); testHarness.processElement2(insertRecord(40L, "k1")); testHarness.processWatermark1(new Watermark(13)); @@ -656,9 +656,14 @@ void testRowTimeFullOuterEarlyFireOneMatches() throws Exception { // Match the left row. testHarness.processElement2(insertRecord(12L, "k1")); - // Fire the right row's early-fire timer (43) and then close everything. + // Fire the right row's early-fire timer (43). testHarness.processWatermark1(new Watermark(43)); testHarness.processWatermark2(new Watermark(43)); + + // A left row in window (40 in [45 - 9, 45 + 5]) matches the early-fired right row. + testHarness.processElement1(insertRecord(45L, "k1")); + + // Close everything. testHarness.processWatermark1(new Watermark(60)); testHarness.processWatermark2(new Watermark(60)); @@ -669,6 +674,8 @@ void testRowTimeFullOuterEarlyFireOneMatches() throws Exception { expectedOutput.add(updateAfterRecord(10L, "k1", 12L, "k1")); expectedOutput.add(insertRecord(null, null, 40L, "k1")); expectedOutput.add(new Watermark(43 - 9)); + expectedOutput.add(updateBeforeRecord(null, null, 40L, "k1")); + expectedOutput.add(updateAfterRecord(45L, "k1", 40L, "k1")); expectedOutput.add(new Watermark(60 - 9)); assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput()); testHarness.close(); From ad2fdc1ea6d26360cfdd5f00a6109132e009cf36 Mon Sep 17 00:00:00 2001 From: weiqingy Date: Mon, 10 Aug 2026 21:01:01 -0700 Subject: [PATCH 3/3] [FLINK-40171][table-runtime] Correct the EmitAwareCollector class javadoc The doc used a Scala-style link to a class that does not exist, and stated that an unarmed collector forwards the existing row kind. It stamps INSERT instead, so a reused row instance cannot leak an UPDATE_AFTER left by an earlier correction. --- .../operators/join/interval/EmitAwareCollector.java | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) 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 056ec578731c5..a4e9c2f465258 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 @@ -23,16 +23,17 @@ 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. + * Collector to wrap a {@link 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}. + * join function knowing about changelogs. When not armed, the collected row is stamped {@link + * RowKind#INSERT}, because the join function reuses a single row instance whose kind may have been + * left at {@link RowKind#UPDATE_AFTER} by an earlier correction. */ class EmitAwareCollector implements Collector {