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..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 @@ -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,12 @@ private TwoInputTransformation createRowTimeJoin( rightTypeInfo, joinFunction, windowBounds.getLeftTimeIdx(), - windowBounds.getRightTimeIdx()); + windowBounds.getRightTimeIdx(), + 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/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..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 @@ -34,7 +34,8 @@ public ProcTimeIntervalJoin( long minCleanUpInterval, InternalTypeInfo leftType, InternalTypeInfo rightType, - IntervalJoinFunction genJoinFunc) { + IntervalJoinFunction genJoinFunc, + long earlyFireDelay) { super( joinType, leftLowerBound, @@ -43,7 +44,10 @@ public ProcTimeIntervalJoin( minCleanUpInterval, leftType, rightType, - genJoinFunc); + genJoinFunc, + 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 5d972104a6692f..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 @@ -40,7 +40,9 @@ public RowTimeIntervalJoin( InternalTypeInfo rightType, IntervalJoinFunction joinFunc, int leftTimeIdx, - int rightTimeIdx) { + int rightTimeIdx, + long earlyFireDelay, + boolean earlyFireCrossDomain) { super( joinType, leftLowerBound, @@ -49,7 +51,9 @@ public RowTimeIntervalJoin( minCleanUpInterval, leftType, rightType, - joinFunc); + joinFunc, + 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 4dbf1250dac725..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,11 +29,13 @@ 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; 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 +66,17 @@ 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; + + // 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 @@ -92,7 +122,9 @@ abstract class TimeIntervalJoin extends KeyedCoProcessFunction leftType, InternalTypeInfo rightType, - IntervalJoinFunction joinFunc) { + IntervalJoinFunction joinFunc, + long earlyFireDelay, + boolean earlyFireCrossDomain) { this.joinType = joinType; this.leftRelativeSize = -leftLowerBound; this.rightRelativeSize = leftUpperBound; @@ -104,6 +136,15 @@ abstract class TimeIntervalJoin extends KeyedCoProcessFunction= 0 + && joinType.isOuter() + && (leftRelativeSize + rightRelativeSize) >= 0; + this.earlyFireCrossDomain = earlyFireCrossDomain; } @Override @@ -129,6 +170,46 @@ 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)); + + 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. ValueStateDescriptor leftValueStateDescriptor = new ValueStateDescriptor<>("IntervalJoinLeftTimerState", Long.class); @@ -178,10 +259,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 +299,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 +332,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. + scheduleEarlyFire(ctx, leftEarlyFireSchedule, timeForLeftRow); + } + } 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 +375,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 +413,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 +446,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. + scheduleEarlyFire(ctx, rightEarlyFireSchedule, timeForRightRow); + } + } 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 +470,40 @@ public void onTimer(long timestamp, OnTimerContext ctx, Collector out) joinFunction.setJoinKey(ctx.getCurrentKey()); joinCollector.setInnerCollector(out); updateOperatorTime(ctx); + + 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); + } + 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,15 +511,105 @@ 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); + } + } + + /** + * 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); } /** @@ -396,6 +665,7 @@ private void removeExpiredRows( Collector collector, long expirationTime, MapState>> rowCache, + MapState> firedState, ValueState timerState, OnTimerContext ctx, boolean removeLeft) @@ -410,28 +680,29 @@ private void removeExpiredRows( Map.Entry>> entry = iterator.next(); Long rowTime = entry.getKey(); if (rowTime <= expirationTime) { - if (removeLeft && joinType.isLeftOuter()) { - 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()) { + 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.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 +718,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..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; @@ -44,6 +45,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 +63,18 @@ 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, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -125,7 +139,18 @@ 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, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -180,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); + FlinkJoinType.LEFT, + -5, + 9, + 0, + 0, + rowType, + rowType, + joinFunction, + 0, + 0, + -1L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -209,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); + FlinkJoinType.LEFT, + -5, + 9, + 0, + 7, + rowType, + rowType, + joinFunction, + 0, + 0, + -1L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -279,7 +326,18 @@ 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, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -350,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); + FlinkJoinType.FULL, + -5, + 9, + 0, + 7, + rowType, + rowType, + joinFunction, + 0, + 0, + -1L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -439,7 +508,9 @@ public void testInterruptibleTimers() throws Exception { rowType, joinFunction, 0, - 0); + 0, + -1L, + false); KeyedTwoInputStreamOperatorTestHarness testHarness = createTestHarness(joinProcessFunc); @@ -512,6 +583,587 @@ 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, + false); + 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, + false); + 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, + false); + 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, + false); + 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, + false); + 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, + false); + 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, + false); + 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, + false); + 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(); + } + + /** + * 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}); + } }