Copilot commented on code in PR #29794:
URL: https://github.com/apache/flink/pull/29794#discussion_r4475483463
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java:
##########
@@ -370,6 +375,12 @@ private Transformation<RowData>
createNegativeWindowSizeJoin(
}
}
+ /** Whether to use async state, which early firing does not support yet. */
+ private boolean isAsyncStateEnabled(ExecNodeConfig config) {
+ return
config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_STATE_ENABLED)
+ && earlyFireDelay == null;
Review Comment:
For existing ForSt jobs with async state enabled and the default `ForStDB`
timer queues, this automatically switches the interval join on restore, but
incremental restore does not migrate its database-backed timers into the async
backend's heap queues. Restored non-null timer-state values prevent replacement
timers from being registered, so unmatched outer rows on idle keys can remain
un-emitted and uncleaned indefinitely. Migrate the timers or provide a
compatibility gate that retains the synchronous operator for these restores,
and add a ForSt checkpoint-restore regression test.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/asyncprocessing/AsyncStateTimeIntervalJoin.java:
##########
@@ -0,0 +1,550 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.join.interval.asyncprocessing;
+
+import org.apache.flink.api.common.functions.DefaultOpenContext;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.v2.MapState;
+import org.apache.flink.api.common.state.v2.MapStateDescriptor;
+import org.apache.flink.api.common.state.v2.StateFuture;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.api.common.state.v2.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ListTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.core.state.StateFutureUtils;
+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.operators.join.interval.EmitAwareCollector;
+import
org.apache.flink.table.runtime.operators.join.interval.IntervalJoinFunction;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * The async state version of {@code TimeIntervalJoin}, a function to execute
a time interval
+ * (time-bounded) stream join. It keeps the state and the logic of the
synchronous version with two
+ * fixes: rows at time zero or earlier are joined and kept like rows after
time zero, and cleanup
+ * times saturate instead of overflowing. As in the synchronous version, the
event-time operator
+ * time is never below zero, so a row whose window ends at or before time zero
is not cached. Early
+ * firing is not supported.
+ *
+ * <p>State iterators cannot remove entries, so the changes to a scanned cache
are written after the
+ * scan. Records of several keys are in flight at a time, so what the
callbacks of a record need,
+ * such as its join key and the operator time, is taken when its processing
starts. Unlike the
+ * synchronous version, this function does not enable interruptible timers,
which the async state
+ * operators do not support.
+ */
+abstract class AsyncStateTimeIntervalJoin
+ extends KeyedCoProcessFunction<RowData, RowData, RowData, RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG =
LoggerFactory.getLogger(AsyncStateTimeIntervalJoin.class);
+
+ // If all buckets of a cache expire, up to this many are deleted in one
bulk request, which
+ // spares ForSt the scan of clear(). More are deleted by clear(), which
the synchronous state
+ // backends batch instead of running one delete per bucket.
+ private static final int MAX_BULK_DELETE_ON_FULL_EXPIRY = 65_536;
+
+ private final FlinkJoinType joinType;
+ protected final long leftRelativeSize;
+ protected final long rightRelativeSize;
+
+ // Minimum interval by which state is cleaned up
+ private final long minCleanUpInterval;
+ protected final long allowedLateness;
+ private final InternalTypeInfo<RowData> leftType;
+ private final InternalTypeInfo<RowData> rightType;
+ private final IntervalJoinFunction joinFunction;
+ private transient OuterJoinPaddingUtil paddingUtil;
+
+ // cache to store rows from the left stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>> leftCache;
+ // cache to store rows from the right stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>>
rightCache;
+
+ // state to record the timer on the left stream. null means no timer set
+ private transient ValueState<Long> leftTimerState;
+ // state to record the timer on the right stream. null means no timer set
+ private transient ValueState<Long> rightTimerState;
+
+ // Points in time until which the respective cache has been cleaned. They
start at
+ // Long.MIN_VALUE rather than 0, so the first rows at time zero or earlier
are joined, and never
+ // decrease, since a callback may update them with the time of an earlier
timer.
+ private long leftExpirationTime = Long.MIN_VALUE;
+ private long rightExpirationTime = Long.MIN_VALUE;
+
+ AsyncStateTimeIntervalJoin(
+ FlinkJoinType joinType,
+ long leftLowerBound,
+ long leftUpperBound,
+ long allowedLateness,
+ long minCleanUpInterval,
+ InternalTypeInfo<RowData> leftType,
+ InternalTypeInfo<RowData> rightType,
+ IntervalJoinFunction joinFunc) {
+ this.joinType = joinType;
+ this.leftRelativeSize = -leftLowerBound;
+ this.rightRelativeSize = leftUpperBound;
+ this.minCleanUpInterval = Math.max(0, minCleanUpInterval);
+ if (allowedLateness < 0) {
+ throw new IllegalArgumentException("The allowed lateness must be
non-negative.");
+ }
+ this.allowedLateness = allowedLateness;
+ this.leftType = leftType;
+ this.rightType = rightType;
+ this.joinFunction = joinFunc;
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ LOG.info("Interval join is using async state");
+ joinFunction.setRuntimeContext(getRuntimeContext());
+ joinFunction.open(DefaultOpenContext.INSTANCE);
+
+ // Initialize the data caches.
+ ListTypeInfo<Tuple2<RowData, Boolean>> leftRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(leftType,
BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>>
leftMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinLeftCache", BasicTypeInfo.LONG_TYPE_INFO,
leftRowListTypeInfo);
+ leftCache = getRuntimeContext().getMapState(leftMapStateDescriptor);
+
+ ListTypeInfo<Tuple2<RowData, Boolean>> rightRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(rightType,
BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>>
rightMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinRightCache",
+ BasicTypeInfo.LONG_TYPE_INFO,
+ rightRowListTypeInfo);
+ rightCache = getRuntimeContext().getMapState(rightMapStateDescriptor);
+
+ // Initialize the timer states.
+ ValueStateDescriptor<Long> leftValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinLeftTimerState",
Long.class);
+ leftTimerState =
getRuntimeContext().getState(leftValueStateDescriptor);
+
+ ValueStateDescriptor<Long> rightValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinRightTimerState",
Long.class);
+ rightTimerState =
getRuntimeContext().getState(rightValueStateDescriptor);
+
+ paddingUtil = new OuterJoinPaddingUtil(leftType.toRowSize(),
rightType.toRowSize());
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (this.joinFunction != null) {
+ this.joinFunction.close();
+ }
+ }
+
+ @Override
+ public void processElement1(RowData leftRow, Context ctx,
Collector<RowData> out)
+ throws Exception {
+ processElement(leftRow, ctx, out, true);
+ }
+
+ @Override
+ public void processElement2(RowData rightRow, Context ctx,
Collector<RowData> out)
+ throws Exception {
+ processElement(rightRow, ctx, out, false);
+ }
+
+ /**
+ * Joins the row with the cached rows of the other input and caches it for
the rows of the other
+ * input still to come: {@code processElement1} and {@code
processElement2} of the synchronous
+ * version, written once for both inputs.
+ */
+ private void processElement(RowData row, Context ctx, Collector<RowData>
out, boolean isLeft) {
+ final RowData joinKey = ctx.getCurrentKey();
+ final long operatorTime = getOperatorTime(ctx);
Review Comment:
The added tests use synchronously completing state adaptors, so they do not
exercise callback interleaving while state requests are outstanding. Add
deterministic delayed-completion coverage using ForSt or a controllable state
executor, with different keys completing out of order and queued same-key
records and cleanup timers. Verify keyed join results, state updates, and
exactly-once outer padding under that interleaving.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/asyncprocessing/AsyncStateTimeIntervalJoin.java:
##########
@@ -0,0 +1,550 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.join.interval.asyncprocessing;
+
+import org.apache.flink.api.common.functions.DefaultOpenContext;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.v2.MapState;
+import org.apache.flink.api.common.state.v2.MapStateDescriptor;
+import org.apache.flink.api.common.state.v2.StateFuture;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.api.common.state.v2.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ListTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.core.state.StateFutureUtils;
+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.operators.join.interval.EmitAwareCollector;
+import
org.apache.flink.table.runtime.operators.join.interval.IntervalJoinFunction;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * The async state version of {@code TimeIntervalJoin}, a function to execute
a time interval
+ * (time-bounded) stream join. It keeps the state and the logic of the
synchronous version with two
+ * fixes: rows at time zero or earlier are joined and kept like rows after
time zero, and cleanup
+ * times saturate instead of overflowing. As in the synchronous version, the
event-time operator
+ * time is never below zero, so a row whose window ends at or before time zero
is not cached. Early
+ * firing is not supported.
+ *
+ * <p>State iterators cannot remove entries, so the changes to a scanned cache
are written after the
+ * scan. Records of several keys are in flight at a time, so what the
callbacks of a record need,
+ * such as its join key and the operator time, is taken when its processing
starts. Unlike the
+ * synchronous version, this function does not enable interruptible timers,
which the async state
+ * operators do not support.
+ */
+abstract class AsyncStateTimeIntervalJoin
+ extends KeyedCoProcessFunction<RowData, RowData, RowData, RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG =
LoggerFactory.getLogger(AsyncStateTimeIntervalJoin.class);
+
+ // If all buckets of a cache expire, up to this many are deleted in one
bulk request, which
+ // spares ForSt the scan of clear(). More are deleted by clear(), which
the synchronous state
+ // backends batch instead of running one delete per bucket.
+ private static final int MAX_BULK_DELETE_ON_FULL_EXPIRY = 65_536;
+
+ private final FlinkJoinType joinType;
+ protected final long leftRelativeSize;
+ protected final long rightRelativeSize;
+
+ // Minimum interval by which state is cleaned up
+ private final long minCleanUpInterval;
+ protected final long allowedLateness;
+ private final InternalTypeInfo<RowData> leftType;
+ private final InternalTypeInfo<RowData> rightType;
+ private final IntervalJoinFunction joinFunction;
+ private transient OuterJoinPaddingUtil paddingUtil;
+
+ // cache to store rows from the left stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>> leftCache;
+ // cache to store rows from the right stream
+ private transient MapState<Long, List<Tuple2<RowData, Boolean>>>
rightCache;
+
+ // state to record the timer on the left stream. null means no timer set
+ private transient ValueState<Long> leftTimerState;
+ // state to record the timer on the right stream. null means no timer set
+ private transient ValueState<Long> rightTimerState;
+
+ // Points in time until which the respective cache has been cleaned. They
start at
+ // Long.MIN_VALUE rather than 0, so the first rows at time zero or earlier
are joined, and never
+ // decrease, since a callback may update them with the time of an earlier
timer.
+ private long leftExpirationTime = Long.MIN_VALUE;
+ private long rightExpirationTime = Long.MIN_VALUE;
+
+ AsyncStateTimeIntervalJoin(
+ FlinkJoinType joinType,
+ long leftLowerBound,
+ long leftUpperBound,
+ long allowedLateness,
+ long minCleanUpInterval,
+ InternalTypeInfo<RowData> leftType,
+ InternalTypeInfo<RowData> rightType,
+ IntervalJoinFunction joinFunc) {
+ this.joinType = joinType;
+ this.leftRelativeSize = -leftLowerBound;
+ this.rightRelativeSize = leftUpperBound;
+ this.minCleanUpInterval = Math.max(0, minCleanUpInterval);
+ if (allowedLateness < 0) {
+ throw new IllegalArgumentException("The allowed lateness must be
non-negative.");
+ }
+ this.allowedLateness = allowedLateness;
+ this.leftType = leftType;
+ this.rightType = rightType;
+ this.joinFunction = joinFunc;
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ LOG.info("Interval join is using async state");
+ joinFunction.setRuntimeContext(getRuntimeContext());
+ joinFunction.open(DefaultOpenContext.INSTANCE);
+
+ // Initialize the data caches.
+ ListTypeInfo<Tuple2<RowData, Boolean>> leftRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(leftType,
BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>>
leftMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinLeftCache", BasicTypeInfo.LONG_TYPE_INFO,
leftRowListTypeInfo);
+ leftCache = getRuntimeContext().getMapState(leftMapStateDescriptor);
+
+ ListTypeInfo<Tuple2<RowData, Boolean>> rightRowListTypeInfo =
+ new ListTypeInfo<>(new TupleTypeInfo<>(rightType,
BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<RowData, Boolean>>>
rightMapStateDescriptor =
+ new MapStateDescriptor<>(
+ "IntervalJoinRightCache",
+ BasicTypeInfo.LONG_TYPE_INFO,
+ rightRowListTypeInfo);
+ rightCache = getRuntimeContext().getMapState(rightMapStateDescriptor);
+
+ // Initialize the timer states.
+ ValueStateDescriptor<Long> leftValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinLeftTimerState",
Long.class);
+ leftTimerState =
getRuntimeContext().getState(leftValueStateDescriptor);
+
+ ValueStateDescriptor<Long> rightValueStateDescriptor =
+ new ValueStateDescriptor<>("IntervalJoinRightTimerState",
Long.class);
+ rightTimerState =
getRuntimeContext().getState(rightValueStateDescriptor);
+
+ paddingUtil = new OuterJoinPaddingUtil(leftType.toRowSize(),
rightType.toRowSize());
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (this.joinFunction != null) {
+ this.joinFunction.close();
+ }
+ }
+
+ @Override
+ public void processElement1(RowData leftRow, Context ctx,
Collector<RowData> out)
+ throws Exception {
+ processElement(leftRow, ctx, out, true);
+ }
+
+ @Override
+ public void processElement2(RowData rightRow, Context ctx,
Collector<RowData> out)
+ throws Exception {
+ processElement(rightRow, ctx, out, false);
+ }
+
+ /**
+ * Joins the row with the cached rows of the other input and caches it for
the rows of the other
+ * input still to come: {@code processElement1} and {@code
processElement2} of the synchronous
+ * version, written once for both inputs.
+ */
+ private void processElement(RowData row, Context ctx, Collector<RowData>
out, boolean isLeft) {
+ final RowData joinKey = ctx.getCurrentKey();
+ final long operatorTime = getOperatorTime(ctx);
+ final long rowTime =
+ isLeft
+ ? getTimeForLeftStream(ctx, row, operatorTime)
+ : getTimeForRightStream(ctx, row, operatorTime);
+ final long otherQualifiedLowerBound =
+ rowTime - (isLeft ? rightRelativeSize : leftRelativeSize);
+ final long otherQualifiedUpperBound =
+ rowTime + (isLeft ? leftRelativeSize : rightRelativeSize);
+ final boolean thisOuter = isLeft ? joinType.isLeftOuter() :
joinType.isRightOuter();
+ final boolean otherOuter = isLeft ? joinType.isRightOuter() :
joinType.isLeftOuter();
+ final boolean[] emitted = {false};
+
+ // Check if we need to join the current row against cached rows of the
other input.
+ // The condition here should be otherMinimumTime <
otherQualifiedUpperBound.
+ // We use the expiration time of the other cache as an approximation
of the
+ // otherMinimumTime here, since it is always smaller or equal.
+ final StateFuture<Void> joined;
+ if ((isLeft ? rightExpirationTime : leftExpirationTime) <
otherQualifiedUpperBound) {
+ // Upper bound of current join window has not passed the cache
expiration time yet.
+ // There might be qualifying rows in the cache that the current
row needs to be joined
+ // with.
+ final long expirationTime =
+ isLeft
+ ? updateRightExpirationTime(
+ calExpirationTime(operatorTime,
rightRelativeSize))
+ : updateLeftExpirationTime(
+ calExpirationTime(operatorTime,
leftRelativeSize));
+ final MapState<Long, List<Tuple2<RowData, Boolean>>> otherCache =
+ isLeft ? rightCache : leftCache;
+ // A collector per record, as the callbacks of other records run
in between.
+ final EmitAwareCollector joinCollector = new EmitAwareCollector();
+ joinCollector.setInnerCollector(out);
+ final Map<Long, List<Tuple2<RowData, Boolean>>> updatedEntries =
new HashMap<>();
+ final Map<Long, List<Tuple2<RowData, Boolean>>> expiredEntries =
new HashMap<>();
+ final ThrowingConsumer<Map.Entry<Long, List<Tuple2<RowData,
Boolean>>>, Exception>
+ joinEntry =
+ entry -> {
+ final long otherTime = entry.getKey();
+ final List<Tuple2<RowData, Boolean>> otherRows
= entry.getValue();
+ final boolean expired = otherTime <=
expirationTime;
+ if (otherTime >= otherQualifiedLowerBound
+ && otherTime <=
otherQualifiedUpperBound) {
+ boolean entryUpdated = false;
+ for (Tuple2<RowData, Boolean> tuple :
otherRows) {
+ joinCollector.reset();
+ if (isLeft) {
+ joinFunction.join(
+ row, tuple.f0, joinKey,
joinCollector);
+ } else {
+ joinFunction.join(
+ tuple.f0, row, joinKey,
joinCollector);
+ }
+ emitted[0] |=
joinCollector.isEmitted();
+ if (otherOuter && !tuple.f1 &&
joinCollector.isEmitted()) {
+ // Mark the other row as being
successfully joined and
+ // emitted.
+ tuple.f1 = true;
+ entryUpdated = true;
+ }
+ }
+ // An expired entry is removed below
instead.
+ if (entryUpdated && !expired) {
+ updatedEntries.put(otherTime,
otherRows);
Review Comment:
For an outer input, `updatedEntries` retains every newly matched bucket's
row list until the scan finishes, defeating ForSt's paged iteration and risking
heap exhaustion for large windows or concurrent keys. The deletion maps have
the same unbounded buffering at lines 260 and 445–446; the 65,536 cutoff is
checked only after all expired timestamps are retained. Use bounded,
iterator-safe mutation batches and await their completion before caching the
current row or completing cleanup.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
