From ef38250b703b1868830d46a5a42d610cd3ebc621 Mon Sep 17 00:00:00 2001 From: ggjh-159 Date: Thu, 2 Jul 2026 11:12:05 +0800 Subject: [PATCH 1/4] feat(flink): add columnar StreamRecordTimestampInserter operator --- .github/workflows/flink.yml | 4 +- .../gluten/velox/FileSystemSinkFactory.java | 8 +- .../gluten/velox/FuzzerSourceSinkFactory.java | 3 +- .../velox/GlutenRowtimeInserterHelper.java | 147 ++++++++++++++++++ .../apache/gluten/velox/PrintSinkFactory.java | 5 + 5 files changed, 162 insertions(+), 5 deletions(-) create mode 100644 gluten-flink/planner/src/main/java/org/apache/gluten/velox/GlutenRowtimeInserterHelper.java diff --git a/.github/workflows/flink.yml b/.github/workflows/flink.yml index 1df73f31190..30b523e3969 100644 --- a/.github/workflows/flink.yml +++ b/.github/workflows/flink.yml @@ -81,8 +81,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 97fc1edafebd0f505e613d260f77f92f5252d048 + git clone -b feat/stateful-stream-record-timestamp-inserter https://github.com/ggjh-159/velox4j.git + cd velox4j 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/gluten/velox/FileSystemSinkFactory.java b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/FileSystemSinkFactory.java index ff08dcec638..b9ad146c3af 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 @@ -134,9 +134,15 @@ public Transformation buildVeloxSink( "FileSystemInsertTable"); GlutenOneInputOperatorFactory operatorFactory = new GlutenOneInputOperatorFactory(onewInputOperator); + // If rowtime transformation was applied (rowtimeFieldIndex != -1), a native + // StreamRecordTimestampInserter sits at this position. Replace it with a Gluten columnar + // inserter so the chain stays columnar end-to-end; otherwise this is a no-op. + Transformation veloxFileWriterInput = + GlutenRowtimeInserterHelper.processTransformation( + (Transformation) fileWriterTransformation.getInputs().get(0)); Transformation 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 6e9c232f048..9d0c81df68c 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 @@ -138,8 +138,7 @@ public Transformation buildVeloxSink( RowData.class, "FuzzerSink")); DataStream newInputStream = - sinkTransformation - .getInputStream() + GlutenRowtimeInserterHelper.process(sinkTransformation.getInputStream()) .transform("Writer", CommittableMessageTypeInfo.noOutput(), operatorFactory); return new SinkTransformation( 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 00000000000..87a396887c1 --- /dev/null +++ b/gluten-flink/planner/src/main/java/org/apache/gluten/velox/GlutenRowtimeInserterHelper.java @@ -0,0 +1,147 @@ +/* + * 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; + +/** + * Replaces a native Flink {@link StreamRecordTimestampInserter} (per-row timestamp) with a Gluten + * columnar inserter (batch-max timestamp) so the columnar chain stays intact when the downstream + * sink is offloaded to Velox. + * + *

The native inserter is constructed by {@code CommonExecSink.applyRowtimeTransformation} and + * sits somewhere on the sink's input chain (position depends on the sink: for a simple sink like + * Fuzzer/DiscardingSink it is the direct input; for FileSystem it sits below the + * StreamingFileWriter and PartitionCommitter). Each sink factory's {@code buildVeloxSink} invokes + * this helper on the inserter-bearing transformation it is about to replace; if no inserter is + * present (e.g., {@code rowtimeFieldIndex == -1}), the helper is a no-op and returns the input + * unchanged. + */ +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; if it + * is a native inserter, rebuilds it as a Gluten columnar inserter and returns a new DataStream + * whose terminal node is the Gluten inserter. Otherwise returns the inputStream unchanged. + */ + public static DataStream process(DataStream inputStream) { + Transformation inputTrans = inputStream.getTransformation(); + Transformation newTrans = processTransformation(inputTrans); + if (newTrans == inputTrans) { + return inputStream; + } + return new DataStream<>(inputStream.getExecutionEnvironment(), newTrans); + } + + /** + * Inspect {@code inputTrans}; if it is a native {@link StreamRecordTimestampInserter}, rebuild it + * as a Gluten columnar inserter whose input is the native inserter's upstream. Returns the new + * transformation, or the original inputTrans when no replacement happened. + */ + public static Transformation processTransformation(Transformation inputTrans) { + 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; + } + int rowtimeIndex = + (int) ReflectUtils.getObjectField(StreamRecordTimestampInserter.class, op, "rowtimeIndex"); + List> inputs = oneInput.getInputs(); + if (inputs.isEmpty()) { + return inputTrans; + } + @SuppressWarnings("unchecked") + Transformation aboveInserter = (Transformation) inputs.get(0); + return buildGlutenInserter(aboveInserter, rowtimeIndex, oneInput.getParallelism()); + } + + private static Transformation buildGlutenInserter( + Transformation aboveInserter, int rowtimeFieldIndex, int parallelism) { + @SuppressWarnings("unchecked") + InternalTypeInfo internalTypeInfo = + (InternalTypeInfo) 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 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 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 e9b9a24623d..c5e19ee1e56 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,11 @@ public boolean isStdErr() { public Transformation buildVeloxSink( Transformation transformation, Map parameters) { Transformation inputTrans = (Transformation) transformation.getInputs().get(0); + // If the upstream chain contains a native StreamRecordTimestampInserter (e.g., when this + // sink is reached via a rowtime-bearing path), replace it with a Gluten columnar inserter. + // For the typical PrintSink path (SinkFunctionProvider), no inserter is present and this is + // a no-op. + inputTrans = GlutenRowtimeInserterHelper.processTransformation(inputTrans); InternalTypeInfo inputTypeInfo = (InternalTypeInfo) inputTrans.getOutputType(); PrintOptions printOpts = extractPrintOptions(transformation); From c2fd991e73a29146cea807aa1e6b783b0fdc3fe3 Mon Sep 17 00:00:00 2001 From: ggjh-159 Date: Thu, 2 Jul 2026 14:32:43 +0800 Subject: [PATCH 2/4] test(flink): add GlutenRowtimeInserterHelper unit tests --- .../GlutenRowtimeInserterHelperTest.java | 117 ++++++++++++++++++ 1 file changed, 117 insertions(+) create mode 100644 gluten-flink/ut/src/test/java/org/apache/gluten/velox/GlutenRowtimeInserterHelperTest.java 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 00000000000..385df49a187 --- /dev/null +++ b/gluten-flink/ut/src/test/java/org/apache/gluten/velox/GlutenRowtimeInserterHelperTest.java @@ -0,0 +1,117 @@ +/* + * 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 newUpstream() { + return new StubTransformation("upstream", InternalTypeInfo.of(UPSTREAM_ROW_TYPE)); + } + + private static OneInputTransformation newNativeInserterTx( + Transformation upstream, int rowtimeIndex) { + StreamRecordTimestampInserter op = new StreamRecordTimestampInserter(rowtimeIndex); + return new OneInputTransformation<>( + upstream, "native-inserter", op, upstream.getOutputType(), 1); + } + + private static OneInputTransformation newOtherOperatorTx( + Transformation upstream) { + StreamMap other = new StreamMap<>(new IdentityMapFunction()); + return new OneInputTransformation<>(upstream, "other-op", other, upstream.getOutputType(), 1); + } + + @Test + void testNoOpForNonOneInputTransformation() { + Transformation stub = + new StubTransformation("stub", InternalTypeInfo.of(UPSTREAM_ROW_TYPE)); + assertSame(stub, GlutenRowtimeInserterHelper.processTransformation(stub)); + } + + @Test + void testNoOpForNonInserterOperator() { + Transformation upstream = newUpstream(); + OneInputTransformation tx = newOtherOperatorTx(upstream); + assertSame(tx, GlutenRowtimeInserterHelper.processTransformation(tx)); + } + + @Test + void testReplacesNativeInserter() { + Transformation upstream = newUpstream(); + OneInputTransformation nativeTx = newNativeInserterTx(upstream, 0); + + Transformation result = GlutenRowtimeInserterHelper.processTransformation(nativeTx); + + assertNotSame(nativeTx, result); + assertTrue(result instanceof OneInputTransformation); + @SuppressWarnings("unchecked") + OneInputTransformation out = + (OneInputTransformation) 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()); + } + + private static final class StubTransformation extends Transformation { + StubTransformation(String name, InternalTypeInfo typeInfo) { + super(name, typeInfo, 1); + } + + @Override + public List> getInputs() { + return Collections.emptyList(); + } + + @Override + protected List> getTransitivePredecessorsInternal() { + return Collections.emptyList(); + } + } + + private static final class IdentityMapFunction implements MapFunction { + @Override + public RowData map(RowData value) { + return value; + } + } +} From cb649b087f019fbd36b1f28d46cbf4b6c1d92224 Mon Sep 17 00:00:00 2001 From: ggjh-159 Date: Mon, 6 Jul 2026 15:57:06 +0800 Subject: [PATCH 3/4] fix review: remove inserter when sink does not consume StreamRecord.timestamp --- .../gluten/velox/FileSystemSinkFactory.java | 8 +-- .../gluten/velox/FuzzerSourceSinkFactory.java | 2 +- .../velox/GlutenRowtimeInserterHelper.java | 69 +++++++++++++------ .../apache/gluten/velox/PrintSinkFactory.java | 9 ++- .../GlutenRowtimeInserterHelperTest.java | 22 ++++-- 5 files changed, 75 insertions(+), 35 deletions(-) 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 b9ad146c3af..7a530d0d504 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 @@ -134,12 +134,12 @@ public Transformation buildVeloxSink( "FileSystemInsertTable"); GlutenOneInputOperatorFactory operatorFactory = new GlutenOneInputOperatorFactory(onewInputOperator); - // If rowtime transformation was applied (rowtimeFieldIndex != -1), a native - // StreamRecordTimestampInserter sits at this position. Replace it with a Gluten columnar - // inserter so the chain stays columnar end-to-end; otherwise this is a no-op. + // 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 veloxFileWriterInput = GlutenRowtimeInserterHelper.processTransformation( - (Transformation) fileWriterTransformation.getInputs().get(0)); + (Transformation) fileWriterTransformation.getInputs().get(0), false); Transformation veloxFileWriterTransformation = new OneInputTransformation( veloxFileWriterInput, 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 9d0c81df68c..2696fb77fbb 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 @@ -138,7 +138,7 @@ public Transformation buildVeloxSink( RowData.class, "FuzzerSink")); DataStream newInputStream = - GlutenRowtimeInserterHelper.process(sinkTransformation.getInputStream()) + GlutenRowtimeInserterHelper.process(sinkTransformation.getInputStream(), false) .transform("Writer", CommittableMessageTypeInfo.noOutput(), operatorFactory); return new SinkTransformation( 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 index 87a396887c1..5f606d204dc 100644 --- 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 @@ -43,17 +43,32 @@ import java.util.Map; /** - * Replaces a native Flink {@link StreamRecordTimestampInserter} (per-row timestamp) with a Gluten - * columnar inserter (batch-max timestamp) so the columnar chain stays intact when the downstream - * sink is offloaded to Velox. + * 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. * - *

The native inserter is constructed by {@code CommonExecSink.applyRowtimeTransformation} and - * sits somewhere on the sink's input chain (position depends on the sink: for a simple sink like - * Fuzzer/DiscardingSink it is the direct input; for FileSystem it sits below the - * StreamingFileWriter and PartitionCommitter). Each sink factory's {@code buildVeloxSink} invokes - * this helper on the inserter-bearing transformation it is about to replace; if no inserter is - * present (e.g., {@code rowtimeFieldIndex == -1}), the helper is a no-op and returns the input - * unchanged. + *

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. + * + *

Callers declare this via {@code requiresTimestamp}: + * + *

    + *
  • {@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. + *
  • {@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. + *
+ * + *

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 { @@ -61,13 +76,14 @@ private GlutenRowtimeInserterHelper() {} /** * Convenience overload that accepts a {@link DataStream} (typical entry point for factories that - * use {@code sinkTransformation.getInputStream()}). Inspects the underlying transformation; if it - * is a native inserter, rebuilds it as a Gluten columnar inserter and returns a new DataStream - * whose terminal node is the Gluten inserter. Otherwise returns the inputStream unchanged. + * 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 process(DataStream inputStream) { + public static DataStream process( + DataStream inputStream, boolean requiresTimestamp) { Transformation inputTrans = inputStream.getTransformation(); - Transformation newTrans = processTransformation(inputTrans); + Transformation newTrans = processTransformation(inputTrans, requiresTimestamp); if (newTrans == inputTrans) { return inputStream; } @@ -75,11 +91,19 @@ public static DataStream process(DataStream inputStream) { } /** - * Inspect {@code inputTrans}; if it is a native {@link StreamRecordTimestampInserter}, rebuild it - * as a Gluten columnar inserter whose input is the native inserter's upstream. Returns the new - * transformation, or the original inputTrans when no replacement happened. + * Inspect {@code inputTrans}. If it is a native {@link StreamRecordTimestampInserter}: + * + *

    + *
  • when {@code requiresTimestamp} is false, return the inserter's upstream, removing the + * inserter from the op chain; + *
  • when {@code requiresTimestamp} is true, rebuild it as a Gluten columnar inserter whose + * input is the native inserter's upstream. + *
+ * + *

Returns the original inputTrans when no inserter is present. */ - public static Transformation processTransformation(Transformation inputTrans) { + public static Transformation processTransformation( + Transformation inputTrans, boolean requiresTimestamp) { if (!(inputTrans instanceof OneInputTransformation)) { return inputTrans; } @@ -92,14 +116,17 @@ public static Transformation processTransformation(Transformation> inputs = oneInput.getInputs(); if (inputs.isEmpty()) { return inputTrans; } @SuppressWarnings("unchecked") Transformation aboveInserter = (Transformation) inputs.get(0); + if (!requiresTimestamp) { + return aboveInserter; + } + int rowtimeIndex = + (int) ReflectUtils.getObjectField(StreamRecordTimestampInserter.class, op, "rowtimeIndex"); return buildGlutenInserter(aboveInserter, rowtimeIndex, oneInput.getParallelism()); } 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 c5e19ee1e56..548e5278e13 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,11 +108,10 @@ public boolean isStdErr() { public Transformation buildVeloxSink( Transformation transformation, Map parameters) { Transformation inputTrans = (Transformation) transformation.getInputs().get(0); - // If the upstream chain contains a native StreamRecordTimestampInserter (e.g., when this - // sink is reached via a rowtime-bearing path), replace it with a Gluten columnar inserter. - // For the typical PrintSink path (SinkFunctionProvider), no inserter is present and this is - // a no-op. - inputTrans = GlutenRowtimeInserterHelper.processTransformation(inputTrans); + // 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 index 385df49a187..e3bcb729d81 100644 --- 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 @@ -63,22 +63,25 @@ private static OneInputTransformation newOtherOperatorTx( void testNoOpForNonOneInputTransformation() { Transformation stub = new StubTransformation("stub", InternalTypeInfo.of(UPSTREAM_ROW_TYPE)); - assertSame(stub, GlutenRowtimeInserterHelper.processTransformation(stub)); + assertSame(stub, GlutenRowtimeInserterHelper.processTransformation(stub, false)); + assertSame(stub, GlutenRowtimeInserterHelper.processTransformation(stub, true)); } @Test void testNoOpForNonInserterOperator() { Transformation upstream = newUpstream(); OneInputTransformation tx = newOtherOperatorTx(upstream); - assertSame(tx, GlutenRowtimeInserterHelper.processTransformation(tx)); + assertSame(tx, GlutenRowtimeInserterHelper.processTransformation(tx, false)); + assertSame(tx, GlutenRowtimeInserterHelper.processTransformation(tx, true)); } @Test - void testReplacesNativeInserter() { + void testReplacesNativeInserterWhenTimestampRequired() { Transformation upstream = newUpstream(); OneInputTransformation nativeTx = newNativeInserterTx(upstream, 0); - Transformation result = GlutenRowtimeInserterHelper.processTransformation(nativeTx); + Transformation result = + GlutenRowtimeInserterHelper.processTransformation(nativeTx, true); assertNotSame(nativeTx, result); assertTrue(result instanceof OneInputTransformation); @@ -92,6 +95,17 @@ void testReplacesNativeInserter() { assertSame(1, out.getParallelism()); } + @Test + void testRemovesNativeInserterWhenTimestampNotRequired() { + Transformation upstream = newUpstream(); + OneInputTransformation nativeTx = newNativeInserterTx(upstream, 0); + + Transformation result = + GlutenRowtimeInserterHelper.processTransformation(nativeTx, false); + + assertSame(upstream, result); + } + private static final class StubTransformation extends Transformation { StubTransformation(String name, InternalTypeInfo typeInfo) { super(name, typeInfo, 1); From a7565e65bd47acf36a4a574b8458111ac5d9582c Mon Sep 17 00:00:00 2001 From: ggjh-159 Date: Mon, 6 Jul 2026 17:11:38 +0800 Subject: [PATCH 4/4] fix: replace em-dash with ASCII hyphen in PrintSinkFactory comment --- .../src/main/java/org/apache/gluten/velox/PrintSinkFactory.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 548e5278e13..c4c883813c5 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 @@ -110,7 +110,7 @@ public Transformation buildVeloxSink( 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. + // chain (if any) is dead weight - remove it. inputTrans = GlutenRowtimeInserterHelper.processTransformation(inputTrans, false); InternalTypeInfo inputTypeInfo = (InternalTypeInfo) inputTrans.getOutputType();