This is an automated email from the ASF dual-hosted git repository.
zhanglistar pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new d8c868b6e3 [GLUTEN-12397][FLINK] Support WatermarkStatus propagation
(#12398)
d8c868b6e3 is described below
commit d8c868b6e362fd1fe465f79b448a7df99bda4092
Author: lgbo <[email protected]>
AuthorDate: Fri Jul 3 14:08:24 2026 +0800
[GLUTEN-12397][FLINK] Support WatermarkStatus propagation (#12398)
* Support watermark status propagation
* Update Flink CI velox4j reference
* Deduplicate watermark status conversion
* Document two-input watermark status handling
* Complete two-input operator checkpoint hooks
* Support Gluten two-input operator base
* Clarify Gluten operator base fork
* Add Flink watermark status tests
* Test records do not reactivate idle watermark status
* Document watermark status test scenarios
* Fix Gluten two-input offload factory
* Update Flink CI velox4j reference
* Fix Gluten offloaded operator factory setup
* Document Gluten one-input factory setup
* fix: Fix spotless import order in OffloadedJobGraphGenerator
* fix: Update velox4j CI reference to latest watermark status commit
* [CORE] Guard watermark handlers against null task and document dead
combined-watermark code
Add null-task guards to GlutenTwoInputOperator.processWatermark/1/2 and
processWatermarkStatus1/2, throwing NullPointerException with lifecycle
context, matching the defensive style already used in snapshotState and
notifyCheckpointComplete. Add @Override to processWatermark(Watermark)
for consistency with the other watermark overrides.
Document that GlutenAbstractStreamOperator.combinedWatermark and the
private indexed processWatermark/processWatermarkStatus overloads are
effectively dead code for current two-input subclasses (which bypass the
Java-side combined watermark), kept only to stay aligned with Flink
1.19.2 AbstractStreamOperator.
---
.github/workflows/flink.yml | 4 +-
.../plan/nodes/exec/stream/StreamExecJoin.java | 35 +-
.../nodes/exec/stream/StreamExecWindowJoin.java | 6 +-
.../gluten/client/OffloadedJobGraphGenerator.java | 11 +-
.../operators/GlutenAbstractStreamOperator.java | 717 +++++++++++++++++++++
.../operators/GlutenTwoInputOperatorFactory.java | 77 +++
.../runtime/operators/GlutenOneInputOperator.java | 10 +
.../runtime/operators/GlutenTwoInputOperator.java | 90 ++-
.../runtime/operators/GlutenWatermarkStatuses.java | 30 +
.../main/java/org/apache/gluten/util/Utils.java | 3 +
.../GlutenStreamTwoInputWatermarkStatusTest.java | 107 +++
.../operators/GlutenStreamWatermarkStatusTest.java | 120 ++++
12 files changed, 1189 insertions(+), 21 deletions(-)
diff --git a/.github/workflows/flink.yml b/.github/workflows/flink.yml
index a222cc7a25..b66bddf427 100644
--- a/.github/workflows/flink.yml
+++ b/.github/workflows/flink.yml
@@ -87,8 +87,8 @@ jobs:
export VELOX_DEPENDENCY_SOURCE=BUNDLED
export fmt_SOURCE=BUNDLED
export folly_SOURCE=BUNDLED
- git clone -b gluten-0530 https://github.com/bigo-sg/velox4j.git
- cd velox4j && git reset --hard
1b1ad0833220476be6a042b83a474cf58f8f076c
+ git clone -b feature/watermark-status-propagation
https://github.com/bigo-sg/velox4j.git
+ cd velox4j && git reset --hard 6e2046f
git apply $GITHUB_WORKSPACE/gluten-flink/patches/fix-velox4j.patch
$GITHUB_WORKSPACE/build/mvn clean install -DskipTests -Dgpg.skip
-Dspotless.skip=true
cd ..
diff --git
a/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java
b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java
index cf66cbbee4..7a638e71fa 100644
---
a/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java
+++
b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java
@@ -19,6 +19,7 @@ package org.apache.flink.table.planner.plan.nodes.exec.stream;
import org.apache.gluten.rexnode.RexConversionContext;
import org.apache.gluten.rexnode.RexNodeConverter;
import org.apache.gluten.rexnode.Utils;
+import org.apache.gluten.streaming.api.operators.GlutenTwoInputOperatorFactory;
import org.apache.gluten.table.runtime.operators.GlutenTwoInputOperator;
import org.apache.gluten.util.LogicalTypeConverter;
import org.apache.gluten.util.PlanNodeIdGenerator;
@@ -40,6 +41,7 @@ import io.github.zhztheplayer.velox4j.type.BooleanType;
import org.apache.flink.FlinkVersion;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
import org.apache.flink.streaming.api.transformations.TwoInputTransformation;
import org.apache.flink.table.data.RowData;
@@ -323,15 +325,30 @@ public class StreamExecJoin extends ExecNodeBase<RowData>
}
final RowType returnType = (RowType) getOutputType();
- final TwoInputTransformation<RowData, RowData, RowData> transform =
- ExecNodeUtil.createTwoInputTransformation(
- leftTransform,
- rightTransform,
- createTransformationMeta(JOIN_TRANSFORMATION, config),
- operator,
- InternalTypeInfo.of(returnType),
- leftTransform.getParallelism(),
- false);
+ final TwoInputTransformation<RowData, RowData, RowData> transform;
+ if (operator instanceof GlutenTwoInputOperator) {
+ transform =
+ ExecNodeUtil.createTwoInputTransformation(
+ leftTransform,
+ rightTransform,
+ createTransformationMeta(JOIN_TRANSFORMATION, config),
+ new GlutenTwoInputOperatorFactory<RowData, RowData, RowData>(
+ (StreamOperator<RowData>) operator),
+ InternalTypeInfo.of(returnType),
+ leftTransform.getParallelism(),
+ 0L,
+ false);
+ } else {
+ transform =
+ ExecNodeUtil.createTwoInputTransformation(
+ leftTransform,
+ rightTransform,
+ createTransformationMeta(JOIN_TRANSFORMATION, config),
+ operator,
+ InternalTypeInfo.of(returnType),
+ leftTransform.getParallelism(),
+ false);
+ }
// set KeyType and Selector for state
RowDataKeySelector leftSelect =
diff --git
a/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecWindowJoin.java
b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecWindowJoin.java
index 400a86ccb0..5a947f1ef5 100644
---
a/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecWindowJoin.java
+++
b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecWindowJoin.java
@@ -17,6 +17,7 @@
package org.apache.flink.table.planner.plan.nodes.exec.stream;
import org.apache.gluten.rexnode.Utils;
+import org.apache.gluten.streaming.api.operators.GlutenTwoInputOperatorFactory;
import org.apache.gluten.table.runtime.operators.GlutenTwoInputOperator;
import org.apache.gluten.util.LogicalTypeConverter;
import org.apache.gluten.util.PlanNodeIdGenerator;
@@ -36,6 +37,7 @@ import io.github.zhztheplayer.velox4j.plan.TableScanNode;
import org.apache.flink.FlinkVersion;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
import org.apache.flink.streaming.api.transformations.TwoInputTransformation;
import org.apache.flink.table.api.TableException;
@@ -245,9 +247,11 @@ public class StreamExecWindowJoin extends
ExecNodeBase<RowData>
leftTransform,
rightTransform,
createTransformationMeta(WINDOW_JOIN_TRANSFORMATION, config),
- operator,
+ new GlutenTwoInputOperatorFactory<RowData, RowData, RowData>(
+ (StreamOperator<RowData>) operator),
InternalTypeInfo.of(returnType),
leftTransform.getParallelism(),
+ 0L,
false);
// set KeyType and Selector for state
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/client/OffloadedJobGraphGenerator.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/client/OffloadedJobGraphGenerator.java
index 27dc0b1933..a9fa29557d 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/client/OffloadedJobGraphGenerator.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/client/OffloadedJobGraphGenerator.java
@@ -16,8 +16,10 @@
*/
package org.apache.gluten.client;
+import org.apache.gluten.streaming.api.operators.GlutenOneInputOperatorFactory;
import org.apache.gluten.streaming.api.operators.GlutenOperator;
import org.apache.gluten.streaming.api.operators.GlutenStreamSource;
+import org.apache.gluten.streaming.api.operators.GlutenTwoInputOperatorFactory;
import org.apache.gluten.table.runtime.keyselector.GlutenKeySelector;
import org.apache.gluten.table.runtime.operators.GlutenOneInputOperator;
import org.apache.gluten.table.runtime.operators.GlutenSourceFunction;
@@ -241,7 +243,9 @@ public class OffloadedJobGraphGenerator {
Class<?> outClass = supportsVectorOutput ? StatefulRecord.class :
RowData.class;
GlutenOneInputOperator<?, ?> newOneInputOp =
sourceOperator.cloneWithInputOutputClasses(planNode, inClass,
outClass);
- offloadedOpConfig.setStreamOperator(newOneInputOp);
+ // setStreamOperator would wrap this in Flink's SimpleOperatorFactory and
skip Gluten-specific
+ // mailbox binding performed by GlutenOneInputOperatorFactory.
+ offloadedOpConfig.setStreamOperatorFactory(new
GlutenOneInputOperatorFactory<>(newOneInputOp));
if (supportsVectorOutput) {
setOffloadedOutputSerializer(offloadedOpConfig, sourceOperator);
}
@@ -279,7 +283,10 @@ public class OffloadedJobGraphGenerator {
sourceOperator.getOutputTypes(),
inClass,
outClass);
- offloadedOpConfig.setStreamOperator(newTwoInputOp);
+ // setStreamOperator would wrap this in Flink's SimpleOperatorFactory,
which only initializes
+ // ProcessingTimeService for Flink AbstractStreamOperator.
GlutenTwoInputOperator uses
+ // GlutenAbstractStreamOperator so it needs the Gluten-specific factory.
+ offloadedOpConfig.setStreamOperatorFactory(new
GlutenTwoInputOperatorFactory<>(newTwoInputOp));
offloadedOpConfig.setStatePartitioner(0, new GlutenKeySelector());
offloadedOpConfig.setStatePartitioner(1, new GlutenKeySelector());
if (supportsVectorOutput) {
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenAbstractStreamOperator.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenAbstractStreamOperator.java
new file mode 100644
index 0000000000..7c8202d32a
--- /dev/null
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenAbstractStreamOperator.java
@@ -0,0 +1,717 @@
+/*
+ * 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.gluten.streaming.api.operators;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.eventtime.IndexedCombinedWatermarkStatus;
+import org.apache.flink.api.common.state.KeyedStateStore;
+import org.apache.flink.api.common.state.State;
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MetricOptions;
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.core.memory.ManagedMemoryUseCase;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.OperatorMetricGroup;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.metrics.groups.InternalOperatorMetricGroup;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+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.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorV2;
+import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.InternalTimeServiceManager;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.streaming.api.operators.KeyContextHandler;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.SetupableStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateContext;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateHandler;
+import
org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.CheckpointedStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamTaskStateInitializer;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+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.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+import org.apache.flink.streaming.util.LatencyStats;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Locale;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * Base class for all stream operators. Operators that contain a user function
should extend the
+ * class {@link AbstractUdfStreamOperator} instead (which is a specialized
subclass of this class).
+ *
+ * <p>For concrete implementations, one of the following two interfaces must
also be implemented, to
+ * mark the operator as unary or binary: {@link OneInputStreamOperator} or
{@link
+ * TwoInputStreamOperator}.
+ *
+ * <p>Methods of {@code StreamOperator} are guaranteed not to be called
concurrently. Also, if using
+ * the timer service, timer callbacks are also guaranteed not to be called
concurrently with methods
+ * on {@code StreamOperator}.
+ *
+ * <p>Note, this class is going to be removed and replaced in the future by
{@link
+ * AbstractStreamOperatorV2}. However as {@link AbstractStreamOperatorV2} is
currently experimental,
+ * {@link AbstractStreamOperator} has not been deprecated just yet.
+ *
+ * <p>This class is a Gluten-local copy of Flink 1.19.2 {@link
AbstractStreamOperator}. Gluten needs
+ * this fork because Flink's watermark status entrypoints are final in both
operator base classes:
+ * {@code AbstractStreamOperator.processWatermarkStatus1/2(...)} and {@code
+ * AbstractStreamOperatorV2.processWatermarkStatus(...)}. Those methods route
two-input {@code
+ * WatermarkStatus} events through Flink's Java-side combined watermark
status. Gluten must instead
+ * forward each input's {@code IDLE}/{@code ACTIVE} status to native
execution, where the Gluten
+ * operator chain keeps the combined watermark/status as the source of truth.
Reusing either Flink
+ * base class would prevent {@code GlutenTwoInputOperator} from overriding the
status handling,
+ * while not using a base class at all would lose the runtime, state,
key-context, timer, latency,
+ * and checkpoint behavior implemented by {@code AbstractStreamOperator}. Keep
this file aligned
+ * with Flink 1.19.2 except for making {@code processWatermarkStatus1/2(...)}
overridable.
+ *
+ * @param <OUT> The output type of the operator.
+ */
+@PublicEvolving
+public abstract class GlutenAbstractStreamOperator<OUT>
+ implements StreamOperator<OUT>,
+ SetupableStreamOperator<OUT>,
+ CheckpointedStreamOperator,
+ KeyContextHandler,
+ Serializable {
+ private static final long serialVersionUID = 1L;
+
+ /** The logger used by the operator class and its subclasses. */
+ protected static final Logger LOG =
LoggerFactory.getLogger(GlutenAbstractStreamOperator.class);
+
+ // ----------- configuration properties -------------
+
+ // A sane default for most operators
+ protected ChainingStrategy chainingStrategy = ChainingStrategy.HEAD;
+
+ // ---------------- runtime fields ------------------
+
+ /** The task that contains this operator (and other operators in the same
chain). */
+ private transient StreamTask<?, ?> container;
+
+ protected transient StreamConfig config;
+
+ protected transient Output<StreamRecord<OUT>> output;
+
+ // Used by the default processWatermark1/2 and processWatermarkStatus1/2
implementations below.
+ // Concrete two-input subclasses (e.g. GlutenTwoInputOperator) override
those entrypoints to
+ // forward each input's watermark/status directly to native, bypassing this
Java-side combined
+ // watermark. The field and the private indexed overloads are kept to stay
aligned with Flink
+ // 1.19.2's AbstractStreamOperator and to serve any future subclass that
wants the default
+ // behavior; they are effectively dead code for the current Gluten operators.
+ private transient IndexedCombinedWatermarkStatus combinedWatermark;
+
+ /** The runtime context for UDFs. */
+ private transient StreamingRuntimeContext runtimeContext;
+
+ // ---------------- key/value state ------------------
+
+ /**
+ * {@code KeySelector} for extracting a key from an element being processed.
This is used to scope
+ * keyed state to a key. This is null if the operator is not a keyed
operator.
+ *
+ * <p>This is for elements from the first input.
+ */
+ private transient KeySelector<?, ?> stateKeySelector1;
+
+ /**
+ * {@code KeySelector} for extracting a key from an element being processed.
This is used to scope
+ * keyed state to a key. This is null if the operator is not a keyed
operator.
+ *
+ * <p>This is for elements from the second input.
+ */
+ private transient KeySelector<?, ?> stateKeySelector2;
+
+ private transient StreamOperatorStateHandler stateHandler;
+
+ private transient InternalTimeServiceManager<?> timeServiceManager;
+
+ // --------------- Metrics ---------------------------
+
+ /** Metric group for the operator. */
+ protected transient InternalOperatorMetricGroup metrics;
+
+ protected transient LatencyStats latencyStats;
+
+ // ---------------- time handler ------------------
+
+ protected transient ProcessingTimeService processingTimeService;
+
+ protected transient RecordAttributes lastRecordAttributes1;
+ protected transient RecordAttributes lastRecordAttributes2;
+
+ // ------------------------------------------------------------------------
+ // Life Cycle
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void setup(
+ StreamTask<?, ?> containingTask, StreamConfig config,
Output<StreamRecord<OUT>> output) {
+ final Environment environment = containingTask.getEnvironment();
+ this.container = containingTask;
+ this.config = config;
+ this.output = output;
+ this.metrics =
+ environment
+ .getMetricGroup()
+ .getOrAddOperator(config.getOperatorID(),
config.getOperatorName());
+ this.combinedWatermark = IndexedCombinedWatermarkStatus.forInputsCount(2);
+
+ try {
+ Configuration taskManagerConfig =
environment.getTaskManagerInfo().getConfiguration();
+ int historySize =
taskManagerConfig.get(MetricOptions.LATENCY_HISTORY_SIZE);
+ if (historySize <= 0) {
+ LOG.warn(
+ "{} has been set to a value equal or below 0: {}. Using default.",
+ MetricOptions.LATENCY_HISTORY_SIZE,
+ historySize);
+ historySize = MetricOptions.LATENCY_HISTORY_SIZE.defaultValue();
+ }
+
+ final String configuredGranularity =
+ taskManagerConfig.get(MetricOptions.LATENCY_SOURCE_GRANULARITY);
+ LatencyStats.Granularity granularity;
+ try {
+ granularity =
+
LatencyStats.Granularity.valueOf(configuredGranularity.toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException iae) {
+ granularity = LatencyStats.Granularity.OPERATOR;
+ LOG.warn(
+ "Configured value {} option for {} is invalid. Defaulting to {}.",
+ configuredGranularity,
+ MetricOptions.LATENCY_SOURCE_GRANULARITY.key(),
+ granularity);
+ }
+ MetricGroup taskMetricGroup = this.metrics.getTaskMetricGroup();
+ this.latencyStats =
+ new LatencyStats(
+ taskMetricGroup.addGroup("latency"),
+ historySize,
+ container.getIndexInSubtaskGroup(),
+ getOperatorID(),
+ granularity);
+ } catch (Exception e) {
+ LOG.warn("An error occurred while instantiating latency metrics.", e);
+ this.latencyStats =
+ new LatencyStats(
+
UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().addGroup("latency"),
+ 1,
+ 0,
+ new OperatorID(),
+ LatencyStats.Granularity.SINGLE);
+ }
+
+ this.runtimeContext =
+ new StreamingRuntimeContext(
+ environment,
+ environment.getAccumulatorRegistry().getUserMap(),
+ getMetricGroup(),
+ getOperatorID(),
+ getProcessingTimeService(),
+ null,
+ environment.getExternalResourceInfoProvider());
+
+ stateKeySelector1 = config.getStatePartitioner(0,
getUserCodeClassloader());
+ stateKeySelector2 = config.getStatePartitioner(1,
getUserCodeClassloader());
+
+ lastRecordAttributes1 = RecordAttributes.EMPTY_RECORD_ATTRIBUTES;
+ lastRecordAttributes2 = RecordAttributes.EMPTY_RECORD_ATTRIBUTES;
+ }
+
+ /**
+ * @deprecated The {@link ProcessingTimeService} instance should be passed
by the operator
+ * constructor and this method will be removed along with {@link
SetupableStreamOperator}.
+ */
+ @Deprecated
+ public void setProcessingTimeService(ProcessingTimeService
processingTimeService) {
+ this.processingTimeService =
Preconditions.checkNotNull(processingTimeService);
+ }
+
+ @Override
+ public OperatorMetricGroup getMetricGroup() {
+ return metrics;
+ }
+
+ @Override
+ public final void initializeState(StreamTaskStateInitializer
streamTaskStateManager)
+ throws Exception {
+
+ final TypeSerializer<?> keySerializer =
config.getStateKeySerializer(getUserCodeClassloader());
+
+ final StreamTask<?, ?> containingTask =
Preconditions.checkNotNull(getContainingTask());
+ final CloseableRegistry streamTaskCloseableRegistry =
+ Preconditions.checkNotNull(containingTask.getCancelables());
+
+ final StreamOperatorStateContext context =
+ streamTaskStateManager.streamOperatorStateContext(
+ getOperatorID(),
+ getClass().getSimpleName(),
+ getProcessingTimeService(),
+ this,
+ keySerializer,
+ streamTaskCloseableRegistry,
+ metrics,
+ config.getManagedMemoryFractionOperatorUseCaseOfSlot(
+ ManagedMemoryUseCase.STATE_BACKEND,
+ runtimeContext.getJobConfiguration(),
+ runtimeContext.getTaskManagerRuntimeInfo().getConfiguration(),
+ runtimeContext.getUserCodeClassLoader()),
+ isUsingCustomRawKeyedState());
+
+ stateHandler =
+ new StreamOperatorStateHandler(context, getExecutionConfig(),
streamTaskCloseableRegistry);
+ timeServiceManager = context.internalTimerServiceManager();
+ stateHandler.initializeOperatorState(this);
+
runtimeContext.setKeyedStateStore(stateHandler.getKeyedStateStore().orElse(null));
+ }
+
+ /**
+ * Indicates whether or not implementations of this class is writing to the
raw keyed state
+ * streams on snapshots, using {@link #snapshotState(StateSnapshotContext)}.
If yes, subclasses
+ * should override this method to return {@code true}.
+ *
+ * <p>Subclasses need to explicitly indicate the use of raw keyed state
because, internally, the
+ * {@link GlutenAbstractStreamOperator} may attempt to read from it as well
to restore heap-based
+ * timers and ultimately fail with read errors. By setting this flag to
{@code true}, this allows
+ * the {@link GlutenAbstractStreamOperator} to know that the data written in
the raw keyed states
+ * were not written by the timer services, and skips the timer restore
attempt.
+ *
+ * <p>Please refer to FLINK-19741 for further details.
+ *
+ * <p>TODO: this method can be removed once all timers are moved to be
managed by state backends.
+ *
+ * @return flag indicating whether or not this operator is writing to raw
keyed state via {@link
+ * #snapshotState(StateSnapshotContext)}.
+ */
+ @Internal
+ protected boolean isUsingCustomRawKeyedState() {
+ return false;
+ }
+
+ /**
+ * This method is called immediately before any elements are processed, it
should contain the
+ * operator's initialization logic, e.g. state initialization.
+ *
+ * <p>The default implementation does nothing.
+ *
+ * @throws Exception An exception in this method causes the operator to fail.
+ */
+ @Override
+ public void open() throws Exception {}
+
+ @Override
+ public void finish() throws Exception {}
+
+ @Override
+ public void close() throws Exception {
+ if (stateHandler != null) {
+ stateHandler.dispose();
+ }
+ }
+
+ @Override
+ public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+ // the default implementation does nothing and accepts the checkpoint
+ // this is purely for subclasses to override
+ }
+
+ @Override
+ public final OperatorSnapshotFutures snapshotState(
+ long checkpointId,
+ long timestamp,
+ CheckpointOptions checkpointOptions,
+ CheckpointStreamFactory factory)
+ throws Exception {
+ return stateHandler.snapshotState(
+ this,
+ Optional.ofNullable(timeServiceManager),
+ getOperatorName(),
+ checkpointId,
+ timestamp,
+ checkpointOptions,
+ factory,
+ isUsingCustomRawKeyedState());
+ }
+
+ /**
+ * Stream operators with state, which want to participate in a snapshot need
to override this hook
+ * method.
+ *
+ * @param context context that provides information and means required for
taking a snapshot
+ */
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {}
+
+ /**
+ * Stream operators with state which can be restored need to override this
hook method.
+ *
+ * @param context context that allows to register different states.
+ */
+ @Override
+ public void initializeState(StateInitializationContext context) throws
Exception {}
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ stateHandler.notifyCheckpointComplete(checkpointId);
+ }
+
+ @Override
+ public void notifyCheckpointAborted(long checkpointId) throws Exception {
+ stateHandler.notifyCheckpointAborted(checkpointId);
+ }
+
+ // ------------------------------------------------------------------------
+ // Properties and Services
+ // ------------------------------------------------------------------------
+
+ /**
+ * Gets the execution config defined on the execution environment of the job
to which this
+ * operator belongs.
+ *
+ * @return The job's execution config.
+ */
+ public ExecutionConfig getExecutionConfig() {
+ return container.getExecutionConfig();
+ }
+
+ public StreamConfig getOperatorConfig() {
+ return config;
+ }
+
+ public StreamTask<?, ?> getContainingTask() {
+ return container;
+ }
+
+ public ClassLoader getUserCodeClassloader() {
+ return container.getUserCodeClassLoader();
+ }
+
+ /**
+ * Return the operator name. If the runtime context has been set, then the
task name with subtask
+ * index is returned. Otherwise, the simple class name is returned.
+ *
+ * @return If runtime context is set, then return task name with subtask
index. Otherwise return
+ * simple class name.
+ */
+ protected String getOperatorName() {
+ if (runtimeContext != null) {
+ return runtimeContext.getTaskInfo().getTaskNameWithSubtasks();
+ } else {
+ return getClass().getSimpleName();
+ }
+ }
+
+ /**
+ * Returns a context that allows the operator to query information about the
execution and also to
+ * interact with systems such as broadcast variables and managed state. This
also allows to
+ * register timers.
+ */
+ @VisibleForTesting
+ public StreamingRuntimeContext getRuntimeContext() {
+ return runtimeContext;
+ }
+
+ public <K> KeyedStateBackend<K> getKeyedStateBackend() {
+ return stateHandler.getKeyedStateBackend();
+ }
+
+ @VisibleForTesting
+ public OperatorStateBackend getOperatorStateBackend() {
+ return stateHandler.getOperatorStateBackend();
+ }
+
+ /**
+ * Returns the {@link ProcessingTimeService} responsible for getting the
current processing time
+ * and registering timers.
+ */
+ @VisibleForTesting
+ public ProcessingTimeService getProcessingTimeService() {
+ return processingTimeService;
+ }
+
+ /**
+ * Creates a partitioned state handle, using the state backend configured
for this task.
+ *
+ * @throws IllegalStateException Thrown, if the key/value state was already
initialized.
+ * @throws Exception Thrown, if the state backend cannot create the
key/value state.
+ */
+ protected <S extends State> S getPartitionedState(StateDescriptor<S, ?>
stateDescriptor)
+ throws Exception {
+ return getPartitionedState(
+ VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE,
stateDescriptor);
+ }
+
+ protected <N, S extends State, T> S getOrCreateKeyedState(
+ TypeSerializer<N> namespaceSerializer, StateDescriptor<S, T>
stateDescriptor)
+ throws Exception {
+ return stateHandler.getOrCreateKeyedState(namespaceSerializer,
stateDescriptor);
+ }
+
+ /**
+ * Creates a partitioned state handle, using the state backend configured
for this task.
+ *
+ * @throws IllegalStateException Thrown, if the key/value state was already
initialized.
+ * @throws Exception Thrown, if the state backend cannot create the
key/value state.
+ */
+ protected <S extends State, N> S getPartitionedState(
+ N namespace, TypeSerializer<N> namespaceSerializer, StateDescriptor<S,
?> stateDescriptor)
+ throws Exception {
+ KeyedStateBackend<?> keyedStateBackend =
stateHandler.getKeyedStateBackend();
+ if (keyedStateBackend != null) {
+ return keyedStateBackend.getPartitionedState(namespace,
namespaceSerializer, stateDescriptor);
+ }
+ throw new RuntimeException(
+ "Cannot create partitioned state. The keyed state "
+ + "backend has not been set. This indicates that the operator is
not "
+ + "partitioned/keyed.");
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public void setKeyContextElement1(StreamRecord record) throws Exception {
+ setKeyContextElement(record, stateKeySelector1);
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public void setKeyContextElement2(StreamRecord record) throws Exception {
+ setKeyContextElement(record, stateKeySelector2);
+ }
+
+ @Internal
+ @Override
+ public boolean hasKeyContext1() {
+ return stateKeySelector1 != null;
+ }
+
+ @Internal
+ @Override
+ public boolean hasKeyContext2() {
+ return stateKeySelector2 != null;
+ }
+
+ private <T> void setKeyContextElement(StreamRecord<T> record, KeySelector<T,
?> selector)
+ throws Exception {
+ if (selector != null) {
+ Object key = selector.getKey(record.getValue());
+ setCurrentKey(key);
+ }
+ }
+
+ public void setCurrentKey(Object key) {
+ stateHandler.setCurrentKey(key);
+ }
+
+ public Object getCurrentKey() {
+ return stateHandler.getCurrentKey();
+ }
+
+ public KeyedStateStore getKeyedStateStore() {
+ if (stateHandler == null) {
+ return null;
+ }
+ return stateHandler.getKeyedStateStore().orElse(null);
+ }
+
+ // ------------------------------------------------------------------------
+ // Context and chaining properties
+ // ------------------------------------------------------------------------
+
+ @Override
+ public final void setChainingStrategy(ChainingStrategy strategy) {
+ this.chainingStrategy = strategy;
+ }
+
+ @Override
+ public final ChainingStrategy getChainingStrategy() {
+ return chainingStrategy;
+ }
+
+ // ------------------------------------------------------------------------
+ // Metrics
+ // ------------------------------------------------------------------------
+
+ // ------- One input stream
+ public void processLatencyMarker(LatencyMarker latencyMarker) throws
Exception {
+ reportOrForwardLatencyMarker(latencyMarker);
+ }
+
+ // ------- Two input stream
+ public void processLatencyMarker1(LatencyMarker latencyMarker) throws
Exception {
+ reportOrForwardLatencyMarker(latencyMarker);
+ }
+
+ public void processLatencyMarker2(LatencyMarker latencyMarker) throws
Exception {
+ reportOrForwardLatencyMarker(latencyMarker);
+ }
+
+ protected void reportOrForwardLatencyMarker(LatencyMarker marker) {
+ // all operators are tracking latencies
+ this.latencyStats.reportLatency(marker);
+
+ // everything except sinks forwards latency markers
+ this.output.emitLatencyMarker(marker);
+ }
+
+ // ------------------------------------------------------------------------
+ // Watermark handling
+ // ------------------------------------------------------------------------
+
+ /**
+ * Returns a {@link InternalTimerService} that can be used to query current
processing time and
+ * event time and to set timers. An operator can have several timer
services, where each has its
+ * own namespace serializer. Timer services are differentiated by the string
key that is given
+ * when requesting them, if you call this method with the same key multiple
times you will get the
+ * same timer service instance in subsequent requests.
+ *
+ * <p>Timers are always scoped to a key, the currently active key of a keyed
stream operation.
+ * When a timer fires, this key will also be set as the currently active key.
+ *
+ * <p>Each timer has attached metadata, the namespace. Different timer
services can have a
+ * different namespace type. If you don't need namespace differentiation you
can use {@link
+ * VoidNamespaceSerializer} as the namespace serializer.
+ *
+ * @param name The name of the requested timer service. If no service exists
under the given name
+ * a new one will be created and returned.
+ * @param namespaceSerializer {@code TypeSerializer} for the timer namespace.
+ * @param triggerable The {@link Triggerable} that should be invoked when
timers fire
+ * @param <N> The type of the timer namespace.
+ */
+ public <K, N> InternalTimerService<N> getInternalTimerService(
+ String name, TypeSerializer<N> namespaceSerializer, Triggerable<K, N>
triggerable) {
+ if (timeServiceManager == null) {
+ throw new RuntimeException("The timer service has not been
initialized.");
+ }
+ @SuppressWarnings("unchecked")
+ InternalTimeServiceManager<K> keyedTimeServiceHandler =
+ (InternalTimeServiceManager<K>) timeServiceManager;
+ KeyedStateBackend<K> keyedStateBackend = getKeyedStateBackend();
+ checkState(keyedStateBackend != null, "Timers can only be used on keyed
operators.");
+ return keyedTimeServiceHandler.getInternalTimerService(
+ name, keyedStateBackend.getKeySerializer(), namespaceSerializer,
triggerable);
+ }
+
+ public void processWatermark(Watermark mark) throws Exception {
+ if (timeServiceManager != null) {
+ timeServiceManager.advanceWatermark(mark);
+ }
+ output.emitWatermark(mark);
+ }
+
+ private void processWatermark(Watermark mark, int index) throws Exception {
+ if (combinedWatermark.updateWatermark(index, mark.getTimestamp())) {
+ processWatermark(new
Watermark(combinedWatermark.getCombinedWatermark()));
+ }
+ }
+
+ public void processWatermark1(Watermark mark) throws Exception {
+ processWatermark(mark, 0);
+ }
+
+ public void processWatermark2(Watermark mark) throws Exception {
+ processWatermark(mark, 1);
+ }
+
+ public void processWatermarkStatus(WatermarkStatus watermarkStatus) throws
Exception {
+ output.emitWatermarkStatus(watermarkStatus);
+ }
+
+ private void processWatermarkStatus(WatermarkStatus watermarkStatus, int
index) throws Exception {
+ boolean wasIdle = combinedWatermark.isIdle();
+ if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
+ processWatermark(new
Watermark(combinedWatermark.getCombinedWatermark()));
+ }
+ if (wasIdle != combinedWatermark.isIdle()) {
+ output.emitWatermarkStatus(watermarkStatus);
+ }
+ }
+
+ public void processWatermarkStatus1(WatermarkStatus watermarkStatus) throws
Exception {
+ processWatermarkStatus(watermarkStatus, 0);
+ }
+
+ public void processWatermarkStatus2(WatermarkStatus watermarkStatus) throws
Exception {
+ processWatermarkStatus(watermarkStatus, 1);
+ }
+
+ @Override
+ public OperatorID getOperatorID() {
+ return config.getOperatorID();
+ }
+
+ protected Optional<InternalTimeServiceManager<?>> getTimeServiceManager() {
+ return Optional.ofNullable(timeServiceManager);
+ }
+
+ @Experimental
+ public void processRecordAttributes(RecordAttributes recordAttributes)
throws Exception {
+ output.emitRecordAttributes(
+ new
RecordAttributesBuilder(Collections.singletonList(recordAttributes)).build());
+ }
+
+ @Experimental
+ public void processRecordAttributes1(RecordAttributes recordAttributes) {
+ lastRecordAttributes1 = recordAttributes;
+ output.emitRecordAttributes(
+ new RecordAttributesBuilder(Arrays.asList(lastRecordAttributes1,
lastRecordAttributes2))
+ .build());
+ }
+
+ @Experimental
+ public void processRecordAttributes2(RecordAttributes recordAttributes) {
+ lastRecordAttributes2 = recordAttributes;
+ output.emitRecordAttributes(
+ new RecordAttributesBuilder(Arrays.asList(lastRecordAttributes1,
lastRecordAttributes2))
+ .build());
+ }
+}
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenTwoInputOperatorFactory.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenTwoInputOperatorFactory.java
new file mode 100644
index 0000000000..95aa7a1087
--- /dev/null
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenTwoInputOperatorFactory.java
@@ -0,0 +1,77 @@
+/*
+ * 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.gluten.streaming.api.operators;
+
+import org.apache.gluten.table.runtime.operators.GlutenMailboxOperatorHelper;
+
+import io.github.zhztheplayer.velox4j.serde.Serde;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.SetupableStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperatorFactory;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Two input operator factory for gluten. */
+public class GlutenTwoInputOperatorFactory<IN1, IN2, OUT> extends
AbstractStreamOperatorFactory<OUT>
+ implements TwoInputStreamOperatorFactory<IN1, IN2, OUT> {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(GlutenTwoInputOperatorFactory.class);
+
+ private final StreamOperator<OUT> operator;
+
+ public GlutenTwoInputOperatorFactory(StreamOperator<OUT> operator) {
+ this.operator = operator;
+ if (!(operator instanceof GlutenOperator)) {
+ throw new RuntimeException("Operator is not gluten operator");
+ }
+ }
+
+ public GlutenOperator getOperator() {
+ return (GlutenOperator) operator;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends StreamOperator<OUT>> T createStreamOperator(
+ StreamOperatorParameters<OUT> parameters) {
+ LOG.debug("Build gluten operator {}",
Serde.toJson(getOperator().getPlanNode()));
+ if (operator instanceof GlutenAbstractStreamOperator) {
+ ((GlutenAbstractStreamOperator<OUT>) operator)
+ .setProcessingTimeService(parameters.getProcessingTimeService());
+ }
+ if (operator instanceof AbstractStreamOperator) {
+ ((AbstractStreamOperator)
operator).setProcessingTimeService(processingTimeService);
+ }
+ if (operator instanceof SetupableStreamOperator) {
+ ((SetupableStreamOperator<OUT>) operator)
+ .setup(
+ parameters.getContainingTask(), parameters.getStreamConfig(),
parameters.getOutput());
+ }
+
GlutenMailboxOperatorHelper.bindAtTaskStartup(getOperator().mailboxHolder(),
parameters);
+ return (T) operator;
+ }
+
+ @Override
+ public Class<? extends StreamOperator> getStreamOperatorClass(ClassLoader
classLoader) {
+ return operator.getClass();
+ }
+}
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java
index ef9d0d845d..5729a60b41 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java
@@ -43,6 +43,7 @@ import org.apache.flink.runtime.state.StateSnapshotContext;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
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.runtime.operators.TableStreamOperator;
import java.util.List;
@@ -223,6 +224,9 @@ public class GlutenOneInputOperator<IN, OUT> extends
TableStreamOperator<OUT>
if (statefulElement.isWatermark()) {
StatefulWatermark watermark = statefulElement.asWatermark();
output.emitWatermark(new Watermark(watermark.getTimestamp()));
+ } else if (statefulElement.isWatermarkStatus()) {
+ output.emitWatermarkStatus(
+
GlutenWatermarkStatuses.toFlinkWatermarkStatus(statefulElement));
} else {
outputBridge.collect(
output, statefulElement.asRecord(),
sessionResource.getAllocator(), outputType);
@@ -249,6 +253,12 @@ public class GlutenOneInputOperator<IN, OUT> extends
TableStreamOperator<OUT>
processElementInternal();
}
+ @Override
+ public void processWatermarkStatus(WatermarkStatus status) throws Exception {
+ task.notifyWatermarkStatus(status.isIdle());
+ processElementInternal();
+ }
+
@Override
public void processWatermark1(Watermark mark) throws Exception {
throw new UnsupportedOperationException("Not implemented for
GlutenOneInputOperator");
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java
index 7845e22a5b..b51ececd9a 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java
@@ -16,6 +16,7 @@
*/
package org.apache.gluten.table.runtime.operators;
+import org.apache.gluten.streaming.api.operators.GlutenAbstractStreamOperator;
import org.apache.gluten.streaming.api.operators.GlutenOperator;
import org.apache.gluten.table.runtime.config.VeloxConnectorConfig;
import org.apache.gluten.table.runtime.config.VeloxQueryConfig;
@@ -36,10 +37,10 @@ import io.github.zhztheplayer.velox4j.type.RowType;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContext;
-import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,8 +50,14 @@ import java.util.Map;
/**
* Two input operator in gluten, which will call Velox to run. It receives
RowVector from upstream
* instead of flink RowData.
+ *
+ * <p>This class intentionally does not extend {@code AbstractStreamOperator}:
in the Flink version
+ * used here, {@code AbstractStreamOperator.processWatermarkStatus1/2(...)}
are final and maintain
+ * Flink's Java-side combined watermark status. Gluten needs to forward each
input's status to
+ * native, which is the source of truth for combined watermark status inside
the Gluten operator
+ * chain.
*/
-public class GlutenTwoInputOperator<IN, OUT> extends
AbstractStreamOperator<OUT>
+public class GlutenTwoInputOperator<IN, OUT> extends
GlutenAbstractStreamOperator<OUT>
implements TwoInputStreamOperator<IN, IN, OUT>, GlutenOperator,
NativeCallbackTarget {
private static final Logger LOG =
LoggerFactory.getLogger(GlutenTwoInputOperator.class);
@@ -75,6 +82,7 @@ public class GlutenTwoInputOperator<IN, OUT> extends
AbstractStreamOperator<OUT>
private VectorOutputBridge<OUT> outputBridge;
private String description;
private final GlutenMailboxHolder mailboxHolder = new GlutenMailboxHolder();
+ private transient boolean stateInitialized;
public GlutenTwoInputOperator(
StatefulPlanNode plan,
@@ -203,6 +211,8 @@ public class GlutenTwoInputOperator<IN, OUT> extends
AbstractStreamOperator<OUT>
if (element.isWatermark()) {
StatefulWatermark watermark = element.asWatermark();
output.emitWatermark(new Watermark(watermark.getTimestamp()));
+ } else if (element.isWatermarkStatus()) {
+
output.emitWatermarkStatus(GlutenWatermarkStatuses.toFlinkWatermarkStatus(element));
} else {
outputBridge.collect(
output, element.asRecord(), sessionResource.getAllocator(),
outputType);
@@ -218,26 +228,73 @@ public class GlutenTwoInputOperator<IN, OUT> extends
AbstractStreamOperator<OUT>
@Override
public void processWatermark(Watermark mark) throws Exception {
+ if (task == null) {
+ throw new NullPointerException(
+ "task is null in processWatermark; expected to be initialized in
open() and "
+ + "cleared only after close()");
+ }
task.notifyWatermark(mark.getTimestamp());
processElementInternal();
}
@Override
public void processWatermark1(Watermark mark) throws Exception {
+ if (task == null) {
+ throw new NullPointerException(
+ "task is null in processWatermark1; expected to be initialized in
open() and "
+ + "cleared only after close()");
+ }
task.notifyWatermark(mark.getTimestamp(), 0);
processElementInternal();
}
@Override
public void processWatermark2(Watermark mark) throws Exception {
+ if (task == null) {
+ throw new NullPointerException(
+ "task is null in processWatermark2; expected to be initialized in
open() and "
+ + "cleared only after close()");
+ }
task.notifyWatermark(mark.getTimestamp(), 1);
processElementInternal();
}
+ @Override
+ public void processWatermarkStatus1(WatermarkStatus status) throws Exception
{
+ if (task == null) {
+ throw new NullPointerException(
+ "task is null in processWatermarkStatus1; expected to be initialized
in open() and "
+ + "cleared only after close()");
+ }
+ task.notifyWatermarkStatus(status.isIdle(), 0);
+ processElementInternal();
+ }
+
+ @Override
+ public void processWatermarkStatus2(WatermarkStatus status) throws Exception
{
+ if (task == null) {
+ throw new NullPointerException(
+ "task is null in processWatermarkStatus2; expected to be initialized
in open() and "
+ + "cleared only after close()");
+ }
+ task.notifyWatermarkStatus(status.isIdle(), 1);
+ processElementInternal();
+ }
+
@Override
public void close() throws Exception {
closing = true;
GlutenCloseables.runWithCleanup(
+ () -> {
+ if (leftInputQueue != null) {
+ leftInputQueue.noMoreInput();
+ }
+ },
+ () -> {
+ if (rightInputQueue != null) {
+ rightInputQueue.noMoreInput();
+ }
+ },
() -> {
if (leftInputQueue != null) {
leftInputQueue.close();
@@ -303,24 +360,39 @@ public class GlutenTwoInputOperator<IN, OUT> extends
AbstractStreamOperator<OUT>
@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
// TODO: notify velox
+ processElementInternal();
super.prepareSnapshotPreBarrier(checkpointId);
}
@Override
public void snapshotState(StateSnapshotContext context) throws Exception {
// TODO: implement it
- task.snapshotState(0);
+ snapshotNativeState(context.getCheckpointId());
super.snapshotState(context);
}
@Override
public void initializeState(StateInitializationContext context) throws
Exception {
- initSession();
// TODO: implement it
- task.initializeState(0, null);
+ initializeNativeState();
super.initializeState(context);
}
+ private void snapshotNativeState(long checkpointId) {
+ if (task != null) {
+ task.snapshotState(checkpointId);
+ }
+ }
+
+ private void initializeNativeState() {
+ if (stateInitialized) {
+ return;
+ }
+ initSession();
+ task.initializeState(0, null);
+ stateInitialized = true;
+ }
+
private void initSession() {
if (sessionResource != null) {
return;
@@ -352,14 +424,18 @@ public class GlutenTwoInputOperator<IN, OUT> extends
AbstractStreamOperator<OUT>
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
// TODO: notify velox
- task.notifyCheckpointComplete(checkpointId);
+ if (task != null) {
+ task.notifyCheckpointComplete(checkpointId);
+ }
super.notifyCheckpointComplete(checkpointId);
}
@Override
public void notifyCheckpointAborted(long checkpointId) throws Exception {
// TODO: notify velox
- task.notifyCheckpointAborted(checkpointId);
+ if (task != null) {
+ task.notifyCheckpointAborted(checkpointId);
+ }
super.notifyCheckpointAborted(checkpointId);
}
}
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenWatermarkStatuses.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenWatermarkStatuses.java
new file mode 100644
index 0000000000..b45def322f
--- /dev/null
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenWatermarkStatuses.java
@@ -0,0 +1,30 @@
+/*
+ * 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.gluten.table.runtime.operators;
+
+import io.github.zhztheplayer.velox4j.stateful.StatefulElement;
+
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+
+final class GlutenWatermarkStatuses {
+
+ private GlutenWatermarkStatuses() {}
+
+ static WatermarkStatus toFlinkWatermarkStatus(StatefulElement element) {
+ return element.asWatermarkStatus().isIdle() ? WatermarkStatus.IDLE :
WatermarkStatus.ACTIVE;
+ }
+}
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/util/Utils.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/Utils.java
index eaa6a7d20e..840abcbf3f 100644
--- a/gluten-flink/runtime/src/main/java/org/apache/gluten/util/Utils.java
+++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/Utils.java
@@ -18,6 +18,7 @@ package org.apache.gluten.util;
import org.apache.gluten.streaming.api.operators.GlutenOneInputOperatorFactory;
import org.apache.gluten.streaming.api.operators.GlutenOperator;
+import org.apache.gluten.streaming.api.operators.GlutenTwoInputOperatorFactory;
import org.apache.flink.connector.file.table.stream.PartitionCommitInfo;
import org.apache.flink.streaming.api.graph.StreamConfig;
@@ -64,6 +65,8 @@ public class Utils {
}
} else if (operatorFactory instanceof GlutenOneInputOperatorFactory) {
return Optional.of(((GlutenOneInputOperatorFactory)
operatorFactory).getOperator());
+ } else if (operatorFactory instanceof GlutenTwoInputOperatorFactory) {
+ return Optional.of(((GlutenTwoInputOperatorFactory)
operatorFactory).getOperator());
}
return Optional.empty();
}
diff --git
a/gluten-flink/ut/src/test/java/org/apache/gluten/streaming/api/operators/GlutenStreamTwoInputWatermarkStatusTest.java
b/gluten-flink/ut/src/test/java/org/apache/gluten/streaming/api/operators/GlutenStreamTwoInputWatermarkStatusTest.java
new file mode 100644
index 0000000000..8306c9d20d
--- /dev/null
+++
b/gluten-flink/ut/src/test/java/org/apache/gluten/streaming/api/operators/GlutenStreamTwoInputWatermarkStatusTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.gluten.streaming.api.operators;
+
+import org.apache.gluten.table.runtime.operators.GlutenTwoInputOperator;
+
+import io.github.zhztheplayer.velox4j.stateful.StatefulRecord;
+
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+import org.apache.flink.streaming.util.TwoInputStreamOperatorTestHarness;
+import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class GlutenStreamTwoInputWatermarkStatusTest extends
GlutenStreamJoinOperatorTestBase {
+
+ @Test
+ public void testWatermarkStatusPropagationThroughNativeJoinOperator() throws
Exception {
+ // Two-input operators combine per-input watermark status in native
execution. The operator
+ // becomes idle only when all inputs are idle, and becomes active when any
input reactivates.
+ GlutenTwoInputOperator operator =
createGlutenJoinOperator(FlinkJoinType.INNER);
+
+ try (TwoInputStreamOperatorTestHarness<StatefulRecord, StatefulRecord,
StatefulRecord> harness =
+ new TwoInputStreamOperatorTestHarness<>(operator)) {
+ harness.setup();
+ harness.open();
+
+ // One idle input is excluded from combined watermark calculation, but
the operator is not
+ // idle yet because the other input is still active.
+ harness.processWatermarkStatus1(WatermarkStatus.IDLE);
+ assertThat(harness.getOutput()).isEmpty();
+
+ // Once both inputs are idle, the combined status changes to IDLE and is
emitted downstream.
+ harness.processWatermarkStatus2(WatermarkStatus.IDLE);
+ assertThat(harness.getOutput()).containsExactly(WatermarkStatus.IDLE);
+
+ // Reactivating either input changes the combined status back to ACTIVE.
+ harness.processWatermarkStatus1(WatermarkStatus.ACTIVE);
+ assertThat(harness.getOutput()).containsExactly(WatermarkStatus.IDLE,
WatermarkStatus.ACTIVE);
+ }
+ }
+
+ @Test
+ public void testWatermarkReactivatesIdleNativeJoinOperator() throws
Exception {
+ // A watermark received after all inputs became idle should implicitly
reactivate that input.
+ // Native execution must emit ACTIVE before the resulting combined
watermark.
+ GlutenTwoInputOperator operator =
createGlutenJoinOperator(FlinkJoinType.INNER);
+
+ try (TwoInputStreamOperatorTestHarness<StatefulRecord, StatefulRecord,
StatefulRecord> harness =
+ new TwoInputStreamOperatorTestHarness<>(operator)) {
+ harness.setup();
+ harness.open();
+
+ harness.processWatermarkStatus1(WatermarkStatus.IDLE);
+ harness.processWatermarkStatus2(WatermarkStatus.IDLE);
+ assertThat(harness.getOutput()).containsExactly(WatermarkStatus.IDLE);
+
+ // Input 1's watermark makes the combined status active again and
advances the watermark.
+ harness.processWatermark1(new Watermark(1L));
+ assertThat(harness.getOutput())
+ .containsExactly(WatermarkStatus.IDLE, WatermarkStatus.ACTIVE, new
Watermark(1L));
+ }
+ }
+
+ @Test
+ public void testWatermarksUseNativeTwoInputMinimum() throws Exception {
+ // While both inputs are active, native execution should combine indexed
input watermarks by
+ // forwarding only monotonically increasing minimum watermarks downstream.
+ GlutenTwoInputOperator operator =
createGlutenJoinOperator(FlinkJoinType.INNER);
+
+ try (TwoInputStreamOperatorTestHarness<StatefulRecord, StatefulRecord,
StatefulRecord> harness =
+ new TwoInputStreamOperatorTestHarness<>(operator)) {
+ harness.setup();
+ harness.open();
+
+ // The first input alone cannot advance the combined watermark because
the second input has no
+ // watermark yet.
+ harness.processWatermark1(new Watermark(100L));
+ assertThat(harness.getOutput()).isEmpty();
+
+ // Once both inputs have watermarks, the lower input watermark is
emitted.
+ harness.processWatermark2(new Watermark(90L));
+ assertThat(harness.getOutput()).containsExactly(new Watermark(90L));
+
+ // Advancing the lower input exposes input 1's watermark as the new
combined minimum.
+ harness.processWatermark2(new Watermark(110L));
+ assertThat(harness.getOutput()).containsExactly(new Watermark(90L), new
Watermark(100L));
+ }
+ }
+}
diff --git
a/gluten-flink/ut/src/test/java/org/apache/gluten/streaming/api/operators/GlutenStreamWatermarkStatusTest.java
b/gluten-flink/ut/src/test/java/org/apache/gluten/streaming/api/operators/GlutenStreamWatermarkStatusTest.java
new file mode 100644
index 0000000000..6370eda2b3
--- /dev/null
+++
b/gluten-flink/ut/src/test/java/org/apache/gluten/streaming/api/operators/GlutenStreamWatermarkStatusTest.java
@@ -0,0 +1,120 @@
+/*
+ * 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.gluten.streaming.api.operators;
+
+import org.apache.gluten.rexnode.Utils;
+import org.apache.gluten.table.runtime.operators.GlutenOneInputOperator;
+import org.apache.gluten.util.PlanNodeIdGenerator;
+
+import io.github.zhztheplayer.velox4j.expression.TypedExpr;
+import io.github.zhztheplayer.velox4j.plan.EmptyNode;
+import io.github.zhztheplayer.velox4j.plan.PlanNode;
+import io.github.zhztheplayer.velox4j.plan.ProjectNode;
+
+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.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.table.data.RowData;
+
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class GlutenStreamWatermarkStatusTest extends
GlutenStreamOperatorTestBase {
+
+ @Test
+ public void testWatermarkStatusPropagationThroughNativeProjectOperator()
throws Exception {
+ // One-input operators should forward explicit upstream IDLE/ACTIVE status
changes through
+ // native execution and emit the corresponding Flink WatermarkStatus
downstream.
+ GlutenOneInputOperator operator = createTestOperator(createProjectPlan(),
typeInfo, typeInfo);
+
+ try (OneInputStreamOperatorTestHarness<RowData, RowData> harness =
+ createTestHarness(operator, typeInfo, typeInfo)) {
+ // IDLE marks the only input idle, so the operator immediately becomes
idle.
+ harness.processWatermarkStatus(WatermarkStatus.IDLE);
+ assertThat(harness.getOutput()).containsExactly(WatermarkStatus.IDLE);
+
+ // An explicit ACTIVE status from upstream should reactivate the
operator.
+ harness.processWatermarkStatus(WatermarkStatus.ACTIVE);
+ assertThat(harness.getOutput()).containsExactly(WatermarkStatus.IDLE,
WatermarkStatus.ACTIVE);
+ }
+ }
+
+ @Test
+ public void testWatermarkReactivatesIdleNativeProjectOperator() throws
Exception {
+ // Flink treats a watermark from an idle input as an implicit
reactivation. Native execution
+ // must emit ACTIVE before forwarding the watermark so downstream
operators see valid ordering.
+ GlutenOneInputOperator operator = createTestOperator(createProjectPlan(),
typeInfo, typeInfo);
+
+ try (OneInputStreamOperatorTestHarness<RowData, RowData> harness =
+ createTestHarness(operator, typeInfo, typeInfo)) {
+ harness.processWatermarkStatus(WatermarkStatus.IDLE);
+ assertThat(harness.getOutput()).containsExactly(WatermarkStatus.IDLE);
+
+ harness.processWatermark(new Watermark(1L));
+ assertThat(harness.getOutput())
+ .containsExactly(WatermarkStatus.IDLE, WatermarkStatus.ACTIVE, new
Watermark(1L));
+ }
+ }
+
+ @Test
+ public void testRecordDoesNotReactivateIdleNativeProjectOperator() throws
Exception {
+ // Records are not watermark/status events. They should still be processed
while the input is
+ // marked idle, but they must not implicitly emit ACTIVE; upstream is
responsible for that.
+ GlutenOneInputOperator operator = createTestOperator(createProjectPlan(),
typeInfo, typeInfo);
+
+ try (OneInputStreamOperatorTestHarness<RowData, RowData> harness =
+ createTestHarness(operator, typeInfo, typeInfo)) {
+ harness.processWatermarkStatus(WatermarkStatus.IDLE);
+ assertThat(harness.getOutput()).containsExactly(WatermarkStatus.IDLE);
+
+ // The projected record is produced, while the watermark status output
remains idle only.
+ harness.processElement(new StreamRecord<>(testData.get(0), 1L));
+ assertThat(harness.getOutput()).contains(WatermarkStatus.IDLE);
+ assertThat(harness.getOutput()).doesNotContain(WatermarkStatus.ACTIVE);
+
assertThat(extractOutputFromHarness(harness)).containsExactly(testData.get(0));
+ }
+ }
+
+ private PlanNode createProjectPlan() {
+ List<String> inputFieldNames = Utils.getNamesFromRowType(rowType);
+ List<TypedExpr> veloxProjections =
+ convertRexListToVelox(createProjectionExpressions(), inputFieldNames);
+
+ return new ProjectNode(
+ PlanNodeIdGenerator.newId(),
+ List.of(new EmptyNode(veloxType)),
+ inputFieldNames,
+ veloxProjections);
+ }
+
+ private List<RexNode> createProjectionExpressions() {
+ RexNode idFieldRef =
rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.INTEGER), 0);
+ RexNode nameFieldRef =
+
rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.VARCHAR), 1);
+ RexNode ageFieldRef =
+
rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.INTEGER), 2);
+
+ return Arrays.asList(idFieldRef, nameFieldRef, ageFieldRef);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
