gustavodemorais commented on code in PR #28329: URL: https://github.com/apache/flink/pull/28329#discussion_r3545710161
########## flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/snapshot/LateralSnapshotJoinOperator.java: ########## @@ -0,0 +1,1078 @@ +/* + * 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.snapshot; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.common.functions.DefaultOpenContext; +import org.apache.flink.api.common.state.ListState; +import org.apache.flink.api.common.state.ListStateDescriptor; +import org.apache.flink.api.common.state.MapState; +import org.apache.flink.api.common.state.MapStateDescriptor; +import org.apache.flink.api.common.state.ValueState; +import org.apache.flink.api.common.state.ValueStateDescriptor; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.api.common.typeutils.base.EnumSerializer; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Gauge; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.runtime.state.StateBackendLoader; +import org.apache.flink.runtime.state.StateInitializationContext; +import org.apache.flink.runtime.state.StateSnapshotContext; +import org.apache.flink.runtime.state.VoidNamespace; +import org.apache.flink.runtime.state.VoidNamespaceSerializer; +import org.apache.flink.streaming.api.operators.AbstractStreamOperator; +import org.apache.flink.streaming.api.operators.InternalTimer; +import org.apache.flink.streaming.api.operators.InternalTimerService; +import org.apache.flink.streaming.api.operators.TimestampedCollector; +import org.apache.flink.streaming.api.operators.Triggerable; +import org.apache.flink.streaming.api.operators.TwoInputStreamOperator; +import org.apache.flink.streaming.api.watermark.Watermark; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.utils.JoinedRowData; +import org.apache.flink.table.runtime.generated.GeneratedJoinCondition; +import org.apache.flink.table.runtime.generated.JoinCondition; +import org.apache.flink.table.runtime.operators.join.JoinConditionWithNullFilters; +import org.apache.flink.table.runtime.operators.metrics.SimpleGauge; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.types.RowKind; +import org.apache.flink.util.Preconditions; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ScheduledFuture; + +/** + * Stream operator implementing the {@code LATERAL SNAPSHOT} processing-time temporal table join. + * + * <p>The operator runs in two operator-wide phases that progress only forward, LOAD then JOIN: LOAD + * bootstraps the build-side table from its changelog up to {@code loadCompletedTime}, buffering + * probe rows meanwhile; JOIN then continuously joins probe rows against the materialized build-side + * state. How each input is handled depends on the current phase: + * + * <ul> + * <li>Build-side (input2 / right) changes are handled the same way in both phases: they are + * buffered in {@code buildChangeBuffer} and applied lazily to a per-key multi-set in {@code + * buildTableState} on the next per-key access (build- or probe-side) once the build-side + * watermark has advanced past the buffer's tag, or at the flip. The buffer is keyed by the + * build-side row-time attribute ({@code buildRowtimeIndex}) and applied in row-time order + * once the build watermark reaches it; changes sharing a row-time are applied in arrival + * order, which mirrors the upstream emit order (a retraction directly precedes its + * accumulation). Gating application on the row-time preserves atomic update visibility across + * {@code -U}/{@code +U} pairs in JOIN phase. + * <li>Probe-side (input1 / left) records are handled differently per phase. During LOAD each is + * buffered in {@code probeBuffer} under a synthetic, increasing timestamp with one event-time + * timer registered per record. At the flip these timers fire (one record each, interruptibly) + * and join the buffered probes against the materialized build-side state in insertion order. + * During JOIN records are joined immediately, unless the key's flip drain has not finished + * yet, in which case the record is buffered behind the pending ones to preserve order. + * </ul> + * + * <p>Watermark forwarding rules: + * + * <ul> + * <li>Build-side watermarks are never forwarded downstream. + * <li>Probe-side watermarks are held back during LOAD and forwarded during JOIN phase. + * </ul> + * + * <p>The flip from LOAD to JOIN phase is triggered by either: + * + * <ul> + * <li>the build-side watermark reaching {@code loadCompletedTime} (event-time gate), or + * <li>the {@code loadCompletedIdleTimeoutMs} processing-time timer firing without any build-side + * watermark advance. + * </ul> + * + * <p>State TTL eviction happens during JOIN phase and is implemented with keyed processing-time + * timers (matching the semantics of Flink's standard {@code StateTtlConfig}). + * + * <p>Streaming only: The operator joins data with processing-time semantics which can't be (easily) + * done in batch. Also, the operator's phase is held in union operator state, which is incompatible + * with finished operators. Joins against LATERAL SNAPSHOT functions should be translated to regular + * joins. The changes on the build-side would be consolidated into a final table and then be joined + * with the probe-side. All probe-side records are then joined against the same (and final) version + * of the build-side input. + */ +@Internal +public class LateralSnapshotJoinOperator extends AbstractStreamOperator<RowData> + implements TwoInputStreamOperator<RowData, RowData, RowData>, + Triggerable<RowData, VoidNamespace> { + + // -------------------------- static final definitions -------------------------- + + private static final long serialVersionUID = 1L; + + private static final Logger LOG = LoggerFactory.getLogger(LateralSnapshotJoinOperator.class); + + /** Operator state names. */ + private static final String OPERATOR_PHASE_STATE_NAME = "lateral-snapshot-phase"; + + /** Keyed state names. */ + @VisibleForTesting static final String BUILD_TABLE_STATE_NAME = "build-table"; + + @VisibleForTesting static final String BUILD_CHANGE_BUFFER_STATE_NAME = "build-change-buffer"; + @VisibleForTesting static final String BUFFERED_AT_WM_STATE_NAME = "buffered-at-wm"; + @VisibleForTesting static final String PROBE_BUFFER_STATE_NAME = "probe-buffer"; + private static final String PROBE_BUFFER_SEQ_STATE_NAME = "probe-buffer-seq"; + private static final String TTL_EXPIRY_STATE_NAME = "ttl-expiry"; + + /** + * Single timer service for two kinds of per-key timers, distinguished by timer type (not by + * namespace): event-time timers drain a key's LOAD-buffered probes at the flip ({@link + * #onEventTime}), processing-time timers evict keyed state ({@link #onProcessingTime}). Both + * use {@link VoidNamespace} so no namespace is serialized per timer. + */ + private static final String TIMER_SERVICE_NAME = "lateral-snapshot-timers"; + + /** Gauge: probe-side records currently buffered during LOAD. */ + @VisibleForTesting + static final String NUM_PROBE_BUFFERED_METRIC_NAME = "numProbeSideRecordsBuffered"; + + /** Counter: keys evicted from state by TTL. */ + @VisibleForTesting + static final String NUM_STATE_TTL_EVICTIONS_METRIC_NAME = "numStateTtlEvictions"; + + /** Gauge: latest build-side watermark observed. */ + @VisibleForTesting + static final String CURRENT_BUILD_WM_METRIC_NAME = "currentBuildSideWatermark"; + + /** Gauge: latest probe-side watermark observed. */ + @VisibleForTesting + static final String CURRENT_PROBE_WM_METRIC_NAME = "currentProbeSideWatermark"; + + /** Gauge: current phase ordinal, 0 = LOAD, 1 = JOIN. */ + @VisibleForTesting static final String CURRENT_PHASE_METRIC_NAME = "currentPhase"; + + /** Gauge: max joined records emitted for a probe-side record. */ + @VisibleForTesting static final String MAX_JOIN_FAN_OUT_METRIC_NAME = "maxJoinFanOut"; + + /** Gauge: average joined records emitted per probe-side record. */ + @VisibleForTesting static final String AVG_JOIN_FAN_OUT_METRIC_NAME = "avgJoinFanOut"; + + /** Counter: probe-side records without a match. */ + @VisibleForTesting + static final String NUM_UNMATCHED_PROBE_METRIC_NAME = "numUnmatchedProbeRecords"; + + /** Counter: build-side retractions for a row not present in state. */ + @VisibleForTesting + static final String NUM_UNMATCHED_BUILD_RETRACTIONS_METRIC_NAME = + "numUnmatchedBuildRetractions"; + + /** Two-phase state machine. */ + enum Phase { + LOAD, + JOIN + } + + /** What triggered the {@link Phase#LOAD} to {@link Phase#JOIN} flip (for logging). */ + private enum FlipTrigger { + BUILD_WATERMARK, + IDLE_TIMEOUT + } + + // -------------------------- constructor args -------------------------- + + private final boolean isLeftOuterJoin; + private final InternalTypeInfo<RowData> leftType; + private final InternalTypeInfo<RowData> rightType; + + /** Field index of the build-side (right) row-time attribute. */ + private final int buildRowtimeIndex; + + private final GeneratedJoinCondition generatedJoinCondition; + + /** + * Per-equi-key flag indicating whether rows with a NULL in that key position must be filtered + * before the join condition runs (SQL semantics: {@code NULL = NULL} is not true). + */ + private final boolean[] filterNullKeys; + + /** + * Timestamp at which the build-side watermark must arrive for the operator to flip from {@code + * LOAD} to JOIN. + */ + private final long loadCompletedTime; + + /** + * Processing-time idle timeout duration (millis) on build-side watermarks. When configured, the + * operator flips to JOIN if no build-side watermark advance is seen for this duration. + */ + @Nullable private final Long loadCompletedIdleTimeoutMs; + + /** + * State TTL (millis) to clean up any keyed state during JOIN phase. We schedule TTL timers + * maxStateTtlMs ahead and check on minStateTtlMs before scheduling a new timer. This avoids + * rescheduling timers on every state access while still ensuring that keyed state is evicted + * after at most maxStateTtlMs of key inactivity during JOIN phase. If minStateTtlMs is set to + * 0, state TTL is disabled. + */ + private final long minStateTtlMs; + + private final long maxStateTtlMs; + + // -------------------------- transient runtime fields -------------------------- + + private transient JoinConditionWithNullFilters joinCondition; + private transient GenericRowData nullPaddedBuild; + private transient TimestampedCollector<RowData> collector; + + private transient InternalTimerService<VoidNamespace> timerService; + + private transient Phase phase; + + /** + * Processing-time wall clock at which the operator transitioned from {@link Phase#LOAD} to + * {@link Phase#JOIN}. {@code null} while still in LOAD. Used by the TTL handler to reschedule + * state TTL timers that fire too early. + */ + @Nullable private transient Long flipProcTime; + + /** Highest build-side watermark observed. */ + private transient long currentBuildSideWm; + + /** Latest probe-side watermark observed. */ + private transient long currentProbeSideWm; + + /** Non-keyed processing-time idle-flip timer. */ + @Nullable private transient ScheduledFuture<?> idleFlipTimer; + + /** + * True if the keyed state backend iterates map entries in serialized-key order (RocksDB/ForSt), + * so the row-time-keyed {@link #buildChangeBuffer} is already scanned in event-time order. + */ + private transient boolean sortedStateBackend; + + // -------------------------- keyed state -------------------------- + + /** + * Build-side table as multi-set (row → reference count). Holds the committed snapshot, i.e. all + * build changes up to the last applied build watermark; probes join against this. + */ + private transient MapState<RowData, Long> buildTableState; + + /** + * Build-side changes not yet visible in {@link #buildTableState}, keyed by their build row-time + * (multiple changes with the same row-time share a list value). Changes are applied to {@code + * buildTableState} once the build watermark reaches their row-time. + */ + private transient MapState<Long, List<RowData>> buildChangeBuffer; + + /** Build-side watermark to ensure atomic application of build changes. */ + private transient ValueState<Long> bufferedAtWmState; + + /** + * Buffer for probe-side records during LOAD (and for the rare JOIN-phase record that arrives + * while a key's buffer is still draining). Keyed by a synthetic, monotonically increasing + * timestamp so the per-record flip timers fire in insertion order. Each buffered row has one + * event-time timer registered on its key; the timer drains exactly that row. + */ + private transient MapState<Long, RowData> probeBuffer; + + /** + * Next sequence number for the current key, used to derive distinct, ordered synthetic + * timestamps for {@link #probeBuffer}. {@code null} iff the buffer is empty, so it doubles as a + * cheap "buffer non-empty" flag on the JOIN-phase fast path. + */ + private transient ValueState<Long> probeBufferSeq; + + /** Most recently registered TTL timer deadline; used to advance TTL timer. */ + private transient ValueState<Long> ttlExpiryState; + + // -------------------------- operator state -------------------------- + + private transient ListState<Phase> operatorPhaseState; + + // -------------------------- metrics -------------------------- + + private transient Counter numStateTtlEvictions; + private transient Counter numUnmatchedProbeRecords; + private transient Counter numUnmatchedBuildRetractions; + + private transient SimpleGauge<Long> probeBufferedGauge; + private transient SimpleGauge<Long> buildWmGauge; + private transient SimpleGauge<Long> probeWmGauge; + private transient SimpleGauge<Long> maxFanOutGauge; + private transient SimpleGauge<Double> avgFanOutGauge; + private transient Gauge<Integer> phaseGauge; + + /** Backing accumulators for the push-model gauges (in-memory, not persisted, best-effort). */ + private transient long probeBufferedCount; + + private transient long maxJoinFanOut; + private transient long totalJoinFanOut; + private transient long totalProbeJoins; + + public LateralSnapshotJoinOperator( + boolean isLeftOuterJoin, + InternalTypeInfo<RowData> leftType, + InternalTypeInfo<RowData> rightType, + int buildRowtimeIndex, + GeneratedJoinCondition generatedJoinCondition, + boolean[] filterNullKeys, + Long loadCompletedTime, Review Comment: You changes the field to the long primitive ins the class long. Small follow-up: the constructor param is still Long loadCompletedTime. Since it's mandatory (checkNotNull, no @Nullable), a primitive long here would match the field -- 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]
