This is an automated email from the ASF dual-hosted git repository.
KevinyhZou 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 522b45d62b [GLUTEN-12426][FLINK] Feat: Add columnar
StreamRecordTimestampInserter operator (#12428)
522b45d62b is described below
commit 522b45d62b27a0df5b880e82d0b23427bbe9559f
Author: GGboom <[email protected]>
AuthorDate: Thu Jul 16 14:04:03 2026 +0800
[GLUTEN-12426][FLINK] Feat: Add columnar StreamRecordTimestampInserter
operator (#12428)
* feat(flink): add columnar StreamRecordTimestampInserter operator
---
.github/workflows/flink.yml | 2 +-
gluten-flink/docs/Flink.md | 2 +-
.../apache/gluten/velox/FileSystemSinkFactory.java | 8 +-
.../gluten/velox/FuzzerSourceSinkFactory.java | 3 +-
.../gluten/velox/GlutenRowtimeInserterHelper.java | 174 +++++++++++++++++++++
.../org/apache/gluten/velox/PrintSinkFactory.java | 4 +
.../velox/GlutenRowtimeInserterHelperTest.java | 131 ++++++++++++++++
7 files changed, 319 insertions(+), 5 deletions(-)
diff --git a/.github/workflows/flink.yml b/.github/workflows/flink.yml
index 0fbaa01897..878b47145d 100644
--- a/.github/workflows/flink.yml
+++ b/.github/workflows/flink.yml
@@ -88,7 +88,7 @@ jobs:
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
edffdc6404e942e1eb7b848c6517fa763bb91c7e
+ cd velox4j && git reset --hard
95e9afe64c38d2e5ba962eee7949ef629fe50197
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/docs/Flink.md b/gluten-flink/docs/Flink.md
index 7ccafbe33e..1da1209f20 100644
--- a/gluten-flink/docs/Flink.md
+++ b/gluten-flink/docs/Flink.md
@@ -48,7 +48,7 @@ As some features have not been committed to upstream, you
have to use the follow
## fetch velox4j code
git clone -b gluten-0530 https://github.com/bigo-sg/velox4j.git
cd velox4j
-git reset --hard b3987293d08c8b4b13a37162ba7238b45df7feb0
+git reset --hard 95e9afe64c38d2e5ba962eee7949ef629fe50197
mvn clean install -DskipTests -Dgpg.skip -Dspotless.skip=true
```
**Get gluten**
diff --git
a/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FileSystemSinkFactory.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FileSystemSinkFactory.java
index 57c9576322..c3fffcfaaa 100644
---
a/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FileSystemSinkFactory.java
+++
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FileSystemSinkFactory.java
@@ -123,9 +123,15 @@ public class FileSystemSinkFactory implements
VeloxSourceSinkFactory {
getSinkDescription());
GlutenOneInputOperatorFactory<RowData, ?> operatorFactory =
new GlutenOneInputOperatorFactory<>(onewInputOperator);
+ // StreamingFileWriter is fully offloaded to the velox file writer, which
never consults
+ // StreamRecord.timestamp for partition / roll / commit. Remove any native
+ // StreamRecordTimestampInserter from the input chain.
+ Transformation<RowData> veloxFileWriterInput =
+ GlutenRowtimeInserterHelper.processTransformation(
+ (Transformation<RowData>)
fileWriterTransformation.getInputs().get(0), false);
OneInputTransformation<RowData, ?> veloxFileWriterTransformation =
new OneInputTransformation(
- fileWriterTransformation.getInputs().get(0),
+ veloxFileWriterInput,
fileWriterTransformation.getName(),
operatorFactory,
fileWriterTransformation.getOutputType(),
diff --git
a/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FuzzerSourceSinkFactory.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FuzzerSourceSinkFactory.java
index b16d0e405a..c283ade027 100644
---
a/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FuzzerSourceSinkFactory.java
+++
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FuzzerSourceSinkFactory.java
@@ -139,8 +139,7 @@ public class FuzzerSourceSinkFactory implements
VeloxSourceSinkFactory {
RowData.class,
"FuzzerSink"));
DataStream<RowData> newInputStream =
- sinkTransformation
- .getInputStream()
+
GlutenRowtimeInserterHelper.process(sinkTransformation.getInputStream(), false)
.transform("Writer", CommittableMessageTypeInfo.noOutput(),
operatorFactory);
return new SinkTransformation<RowData, RowData>(
newInputStream,
diff --git
a/gluten-flink/planner/src/main/java/org/apache/gluten/velox/GlutenRowtimeInserterHelper.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/GlutenRowtimeInserterHelper.java
new file mode 100644
index 0000000000..5f606d204d
--- /dev/null
+++
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/GlutenRowtimeInserterHelper.java
@@ -0,0 +1,174 @@
+/*
+ * 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.velox;
+
+import org.apache.gluten.table.runtime.operators.GlutenOneInputOperator;
+import org.apache.gluten.util.LogicalTypeConverter;
+import org.apache.gluten.util.PlanNodeIdGenerator;
+import org.apache.gluten.util.ReflectUtils;
+
+import io.github.zhztheplayer.velox4j.expression.FieldAccessTypedExpr;
+import io.github.zhztheplayer.velox4j.expression.InputTypedExpr;
+import io.github.zhztheplayer.velox4j.plan.EmptyNode;
+import io.github.zhztheplayer.velox4j.plan.ProjectNode;
+import io.github.zhztheplayer.velox4j.plan.StatefulPlanNode;
+import io.github.zhztheplayer.velox4j.plan.StreamRecordTimestampInserterNode;
+import io.github.zhztheplayer.velox4j.stateful.StatefulRecord;
+import io.github.zhztheplayer.velox4j.type.RowType;
+
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.table.data.RowData;
+import
org.apache.flink.table.runtime.operators.sink.StreamRecordTimestampInserter;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Inspects a sink input chain for the native Flink {@link
StreamRecordTimestampInserter} (per-row
+ * timestamp) added by {@code CommonExecSink.applyRowtimeTransformation}, and
either removes or
+ * replaces it depending on whether the downstream sink actually consumes the
timestamp.
+ *
+ * <p>The native inserter stamps each row's rowtime onto the surrounding
{@code StreamRecord} so
+ * that downstream {@code SinkFunction.Context.timestamp()} readers (or sinks
that read {@code
+ * StreamRecord.timestamp} directly) can access it. Sinks that never read the
timestamp therefore
+ * don't need the inserter at all.
+ *
+ * <p>Callers declare this via {@code requiresTimestamp}:
+ *
+ * <ul>
+ * <li>{@code false}: the sink does not consume {@code
StreamRecord.timestamp}. The inserter is
+ * removed from the op chain and its upstream is wired directly to the
sink. This is the
+ * correct behavior for every sink currently wired through the helper
(Print, Fuzzer/Discard,
+ * FileSystem): Print reads rowtime from RowData via {@code
SinkOperator.timestamp()};
+ * SinkV2-based sinks (DiscardingSink, etc.) cannot read the timestamp
at all; the velox file
+ * writer doesn't consult StreamRecord.timestamp for
partition/roll/commit.
+ * <li>{@code true}: the sink does consume {@code StreamRecord.timestamp}.
The inserter is rebuilt
+ * as a Gluten columnar inserter (batch-max timestamp on a {@link
StatefulRecord}) so the
+ * columnar chain stays intact end-to-end. No current caller uses this
branch; it's kept for
+ * future sinks that read the timestamp directly.
+ * </ul>
+ *
+ * <p>If no inserter is present (e.g., {@code rowtimeFieldIndex == -1}), the
helper is a no-op and
+ * returns the input unchanged regardless of {@code requiresTimestamp}.
+ */
+public final class GlutenRowtimeInserterHelper {
+
+ private GlutenRowtimeInserterHelper() {}
+
+ /**
+ * Convenience overload that accepts a {@link DataStream} (typical entry
point for factories that
+ * use {@code sinkTransformation.getInputStream()}). Inspects the underlying
transformation and
+ * removes or replaces the inserter per {@code requiresTimestamp}; returns a
new DataStream whose
+ * terminal node reflects the result, or the inputStream unchanged when no
replacement happened.
+ */
+ public static DataStream<RowData> process(
+ DataStream<RowData> inputStream, boolean requiresTimestamp) {
+ Transformation<RowData> inputTrans = inputStream.getTransformation();
+ Transformation<RowData> newTrans = processTransformation(inputTrans,
requiresTimestamp);
+ if (newTrans == inputTrans) {
+ return inputStream;
+ }
+ return new DataStream<>(inputStream.getExecutionEnvironment(), newTrans);
+ }
+
+ /**
+ * Inspect {@code inputTrans}. If it is a native {@link
StreamRecordTimestampInserter}:
+ *
+ * <ul>
+ * <li>when {@code requiresTimestamp} is false, return the inserter's
upstream, removing the
+ * inserter from the op chain;
+ * <li>when {@code requiresTimestamp} is true, rebuild it as a Gluten
columnar inserter whose
+ * input is the native inserter's upstream.
+ * </ul>
+ *
+ * <p>Returns the original inputTrans when no inserter is present.
+ */
+ public static Transformation<RowData> processTransformation(
+ Transformation<RowData> inputTrans, boolean requiresTimestamp) {
+ if (!(inputTrans instanceof OneInputTransformation)) {
+ return inputTrans;
+ }
+ OneInputTransformation<?, ?> oneInput = (OneInputTransformation<?, ?>)
inputTrans;
+ if (!(oneInput.getOperatorFactory() instanceof SimpleOperatorFactory)) {
+ return inputTrans;
+ }
+ @SuppressWarnings("rawtypes")
+ Object op = ((SimpleOperatorFactory)
oneInput.getOperatorFactory()).getOperator();
+ if (!(op instanceof StreamRecordTimestampInserter)) {
+ return inputTrans;
+ }
+ List<Transformation<?>> inputs = oneInput.getInputs();
+ if (inputs.isEmpty()) {
+ return inputTrans;
+ }
+ @SuppressWarnings("unchecked")
+ Transformation<RowData> aboveInserter = (Transformation<RowData>)
inputs.get(0);
+ if (!requiresTimestamp) {
+ return aboveInserter;
+ }
+ int rowtimeIndex =
+ (int) ReflectUtils.getObjectField(StreamRecordTimestampInserter.class,
op, "rowtimeIndex");
+ return buildGlutenInserter(aboveInserter, rowtimeIndex,
oneInput.getParallelism());
+ }
+
+ private static Transformation<RowData> buildGlutenInserter(
+ Transformation<RowData> aboveInserter, int rowtimeFieldIndex, int
parallelism) {
+ @SuppressWarnings("unchecked")
+ InternalTypeInfo<RowData> internalTypeInfo =
+ (InternalTypeInfo<RowData>) aboveInserter.getOutputType();
+ final org.apache.flink.table.types.logical.RowType inputRowType =
+ (org.apache.flink.table.types.logical.RowType)
internalTypeInfo.toLogicalType();
+ final RowType vlInputType = (RowType)
LogicalTypeConverter.toVLType(inputRowType);
+ final List<String> fieldNames = inputRowType.getFieldNames();
+ final String rowtimeFieldName = fieldNames.get(rowtimeFieldIndex);
+ final InputTypedExpr inputExpr = new InputTypedExpr(vlInputType);
+ final ProjectNode project =
+ new ProjectNode(
+ PlanNodeIdGenerator.newId(),
+ List.of(new EmptyNode(vlInputType)),
+ List.of(rowtimeFieldName),
+ List.of(FieldAccessTypedExpr.create(inputExpr, rowtimeFieldName)));
+ final StreamRecordTimestampInserterNode inserterNode =
+ new StreamRecordTimestampInserterNode(
+ PlanNodeIdGenerator.newId(), null, project, rowtimeFieldIndex);
+ final StatefulPlanNode statefulPlan = new
StatefulPlanNode(inserterNode.getId(), inserterNode);
+
+ final GlutenOneInputOperator<StatefulRecord, StatefulRecord> operator =
+ new GlutenOneInputOperator<>(
+ statefulPlan,
+ PlanNodeIdGenerator.newId(),
+ vlInputType,
+ Map.of(inserterNode.getId(), vlInputType),
+ StatefulRecord.class,
+ StatefulRecord.class,
+ "StreamRecordTimestampInserter");
+
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ final OneInputStreamOperator rawOperator = (OneInputStreamOperator)
operator;
+ return new OneInputTransformation<>(
+ aboveInserter,
+ "StreamRecordTimestampInserter",
+ SimpleOperatorFactory.of(rawOperator),
+ aboveInserter.getOutputType(),
+ parallelism);
+ }
+}
diff --git
a/gluten-flink/planner/src/main/java/org/apache/gluten/velox/PrintSinkFactory.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/PrintSinkFactory.java
index e9b9a24623..c4c883813c 100644
---
a/gluten-flink/planner/src/main/java/org/apache/gluten/velox/PrintSinkFactory.java
+++
b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/PrintSinkFactory.java
@@ -108,6 +108,10 @@ public class PrintSinkFactory implements
VeloxSourceSinkFactory {
public Transformation buildVeloxSink(
Transformation<RowData> transformation, Map<String, Object> parameters) {
Transformation inputTrans = (Transformation)
transformation.getInputs().get(0);
+ // PrintSink reads rowtime directly from RowData via
SinkOperator.timestamp() and never
+ // inspects StreamRecord.timestamp, so a native
StreamRecordTimestampInserter on the input
+ // chain (if any) is dead weight - remove it.
+ inputTrans = GlutenRowtimeInserterHelper.processTransformation(inputTrans,
false);
InternalTypeInfo inputTypeInfo = (InternalTypeInfo)
inputTrans.getOutputType();
PrintOptions printOpts = extractPrintOptions(transformation);
diff --git
a/gluten-flink/ut/src/test/java/org/apache/gluten/velox/GlutenRowtimeInserterHelperTest.java
b/gluten-flink/ut/src/test/java/org/apache/gluten/velox/GlutenRowtimeInserterHelperTest.java
new file mode 100644
index 0000000000..e3bcb729d8
--- /dev/null
+++
b/gluten-flink/ut/src/test/java/org/apache/gluten/velox/GlutenRowtimeInserterHelperTest.java
@@ -0,0 +1,131 @@
+/*
+ * 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.velox;
+
+import org.apache.gluten.table.runtime.operators.GlutenOneInputOperator;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamMap;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.table.data.RowData;
+import
org.apache.flink.table.runtime.operators.sink.StreamRecordTimestampInserter;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class GlutenRowtimeInserterHelperTest {
+
+ private static final RowType UPSTREAM_ROW_TYPE = RowType.of(new
BigIntType());
+
+ private static Transformation<RowData> newUpstream() {
+ return new StubTransformation("upstream",
InternalTypeInfo.of(UPSTREAM_ROW_TYPE));
+ }
+
+ private static OneInputTransformation<RowData, RowData> newNativeInserterTx(
+ Transformation<RowData> upstream, int rowtimeIndex) {
+ StreamRecordTimestampInserter op = new
StreamRecordTimestampInserter(rowtimeIndex);
+ return new OneInputTransformation<>(
+ upstream, "native-inserter", op, upstream.getOutputType(), 1);
+ }
+
+ private static OneInputTransformation<RowData, RowData> newOtherOperatorTx(
+ Transformation<RowData> upstream) {
+ StreamMap<RowData, RowData> other = new StreamMap<>(new
IdentityMapFunction());
+ return new OneInputTransformation<>(upstream, "other-op", other,
upstream.getOutputType(), 1);
+ }
+
+ @Test
+ void testNoOpForNonOneInputTransformation() {
+ Transformation<RowData> stub =
+ new StubTransformation("stub", InternalTypeInfo.of(UPSTREAM_ROW_TYPE));
+ assertSame(stub, GlutenRowtimeInserterHelper.processTransformation(stub,
false));
+ assertSame(stub, GlutenRowtimeInserterHelper.processTransformation(stub,
true));
+ }
+
+ @Test
+ void testNoOpForNonInserterOperator() {
+ Transformation<RowData> upstream = newUpstream();
+ OneInputTransformation<RowData, RowData> tx = newOtherOperatorTx(upstream);
+ assertSame(tx, GlutenRowtimeInserterHelper.processTransformation(tx,
false));
+ assertSame(tx, GlutenRowtimeInserterHelper.processTransformation(tx,
true));
+ }
+
+ @Test
+ void testReplacesNativeInserterWhenTimestampRequired() {
+ Transformation<RowData> upstream = newUpstream();
+ OneInputTransformation<RowData, RowData> nativeTx =
newNativeInserterTx(upstream, 0);
+
+ Transformation<RowData> result =
+ GlutenRowtimeInserterHelper.processTransformation(nativeTx, true);
+
+ assertNotSame(nativeTx, result);
+ assertTrue(result instanceof OneInputTransformation);
+ @SuppressWarnings("unchecked")
+ OneInputTransformation<RowData, RowData> out =
+ (OneInputTransformation<RowData, RowData>) result;
+ assertTrue(out.getOperatorFactory() instanceof SimpleOperatorFactory);
+ Object op = ((SimpleOperatorFactory<?>)
out.getOperatorFactory()).getOperator();
+ assertTrue(op instanceof GlutenOneInputOperator);
+ assertSame(upstream, out.getInputs().get(0));
+ assertSame(1, out.getParallelism());
+ }
+
+ @Test
+ void testRemovesNativeInserterWhenTimestampNotRequired() {
+ Transformation<RowData> upstream = newUpstream();
+ OneInputTransformation<RowData, RowData> nativeTx =
newNativeInserterTx(upstream, 0);
+
+ Transformation<RowData> result =
+ GlutenRowtimeInserterHelper.processTransformation(nativeTx, false);
+
+ assertSame(upstream, result);
+ }
+
+ private static final class StubTransformation extends
Transformation<RowData> {
+ StubTransformation(String name, InternalTypeInfo<RowData> typeInfo) {
+ super(name, typeInfo, 1);
+ }
+
+ @Override
+ public List<Transformation<?>> getInputs() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
+ return Collections.emptyList();
+ }
+ }
+
+ private static final class IdentityMapFunction implements
MapFunction<RowData, RowData> {
+ @Override
+ public RowData map(RowData value) {
+ return value;
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
