[GLUTEN-12426][FLINK] Feat: Add columnar StreamRecordTimestampInserter operator

.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 feature/watermark-status-propagation https://github.com/bigo-sg/velox4j.git
-          cd velox4j && git reset --hard 6e2046f
+          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 ..

gluten-flink/planner/src/main/java/org/apache/gluten/velox/FileSystemSinkFactory.java

@@ -123,9 +123,15 @@ public Transformation<RowData> buildVeloxSink(
             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(),

gluten-flink/planner/src/main/java/org/apache/gluten/velox/FuzzerSourceSinkFactory.java

@@ -139,8 +139,7 @@ public Transformation<RowData> buildVeloxSink(
                 RowData.class,
                 "FuzzerSink"));
     DataStream<RowData> newInputStream =
-        sinkTransformation
-            .getInputStream()
+        GlutenRowtimeInserterHelper.process(sinkTransformation.getInputStream(), false)
             .transform("Writer", CommittableMessageTypeInfo.noOutput(), operatorFactory);
     return new SinkTransformation<RowData, RowData>(
         newInputStream,

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);
+  }
+}

gluten-flink/planner/src/main/java/org/apache/gluten/velox/PrintSinkFactory.java

@@ -108,6 +108,10 @@ public boolean isStdErr() {
   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);

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;
+    }
+  }
+}