[GLUTEN-10246][FLINK] Support processing time window(tumble) for nextmark q12 (apache#10898)

* support proctime window

* remove useless changes

* remove useless changes

* support kafka data source

* fix reviews

* optimize code

* fix memory leak

* merge master

* xxx

* support proc time window

* remove useless code

* fix ut

* restore q5,q7,q8 tests

* Fix tz download failure

* fix ci

* update velox4j version

Author: KevinyhZou

.github/workflows/flink.yml

@@ -59,10 +59,18 @@ jobs:
         run: |
           source /opt/rh/gcc-toolset-11/enable
           sudo dnf install -y patchelf
-          sudo yum install https://mirror.stream.centos.org/9-stream/BaseOS/x86_64/os/Packages/tzdata-2025a-1.el9.noarch.rpm -y
+          sudo yum install https://mirror.stream.centos.org/9-stream/BaseOS/x86_64/os/Packages/tzdata-2026a-1.el9.noarch.rpm -y
           sudo .github/workflows/util/install-flink-deps.sh
+          # Remove system fmt v11 from the CI image to avoid header conflicts with
+          # velox bundled fmt 10.x during folly/velox native compilation.
+          sudo rm -rf /usr/local/include/fmt
+          sudo rm -rf /usr/local/lib/cmake/fmt
+          sudo rm -f /usr/local/lib/libfmt* /usr/local/lib64/libfmt*
+          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 889bafcf2fa04e8c31a30edbdf40fe203ef58484
+          cd velox4j && git reset --hard 115edf79d265a61c30d45dfcc6ce932ad92378ca
           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/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 889bafcf2fa04e8c31a30edbdf40fe203ef58484
+git reset --hard 115edf79d265a61c30d45dfcc6ce932ad92378ca
 mvn clean install -DskipTests -Dgpg.skip -Dspotless.skip=true
 ```
 **Get gluten**

gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecGlobalWindowAggregate.java

@@ -19,7 +19,6 @@
 import org.apache.gluten.rexnode.AggregateCallConverter;
 import org.apache.gluten.rexnode.Utils;
 import org.apache.gluten.rexnode.WindowUtils;
-import org.apache.gluten.table.runtime.operators.GlutenOneInputOperator;
 import org.apache.gluten.util.LogicalTypeConverter;
 import org.apache.gluten.util.PlanNodeIdGenerator;
 
@@ -37,14 +36,13 @@
 
 import org.apache.flink.FlinkVersion;
 import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.tuple.Tuple5;
 import org.apache.flink.configuration.ReadableConfig;
 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.planner.codegen.CodeGeneratorContext;
-import org.apache.flink.table.planner.codegen.agg.AggsHandlerCodeGenerator;
 import org.apache.flink.table.planner.delegation.PlannerBase;
 import org.apache.flink.table.planner.plan.logical.WindowingStrategy;
 import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
@@ -55,24 +53,21 @@
 import org.apache.flink.table.planner.plan.nodes.exec.InputProperty;
 import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
 import org.apache.flink.table.planner.plan.utils.AggregateInfoList;
+import org.apache.flink.table.planner.plan.utils.AggregateUtil;
 import org.apache.flink.table.planner.plan.utils.KeySelectorUtil;
 import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
 import org.apache.flink.table.planner.utils.TableConfigUtils;
-import org.apache.flink.table.runtime.generated.GeneratedNamespaceAggsHandleFunction;
 import org.apache.flink.table.runtime.groupwindow.NamedWindowProperty;
-import org.apache.flink.table.runtime.groupwindow.WindowProperty;
 import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
-import org.apache.flink.table.runtime.operators.window.tvf.slicing.SliceAssigner;
 import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
 import org.apache.flink.table.runtime.util.TimeWindowUtil;
-import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.RowType;
 
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 import org.apache.calcite.rel.core.AggregateCall;
-import org.apache.calcite.tools.RelBuilder;
 import org.apache.commons.math3.util.ArithmeticUtils;
 
 import javax.annotation.Nullable;
@@ -184,6 +179,14 @@ protected Transformation<RowData> translateToPlanInternal(
     final ZoneId shiftTimeZone =
         TimeWindowUtil.getShiftTimeZone(
             windowing.getTimeAttributeType(), TableConfigUtils.getLocalTimeZone(config));
+    final AggregateInfoList globalAggInfoList =
+        AggregateUtil.deriveStreamWindowAggregateInfoList(
+            planner.getTypeFactory(),
+            localAggInputRowType, // should use original input here
+            JavaScalaConversionUtil.toScala(Arrays.asList(aggCalls)),
+            needRetraction,
+            windowing.getWindow(),
+            true);
 
     // --- Begin Gluten-specific code changes ---
     // TODO: velox window not equal to flink window.
@@ -205,11 +208,15 @@ protected Transformation<RowData> translateToPlanInternal(
     // TODO: support more window types.
     Tuple5<Long, Long, Long, Integer, Integer> windowSpecParams =
         WindowUtils.extractWindowParameters(windowing);
+    Tuple2<Integer, Integer> windowStartAndEndIndexes =
+        WindowUtils.getWindowStartAndEndIndexes(namedWindowProperties, (RowType) getOutputType());
     long size = windowSpecParams.f0;
     long slide = windowSpecParams.f1;
     long offset = windowSpecParams.f2;
     int rowtimeIndex = windowSpecParams.f3;
     int windowType = windowSpecParams.f4;
+    int windowStartIndex = windowStartAndEndIndexes.f0;
+    int windowEndIndex = windowStartAndEndIndexes.f1;
     PartitionFunctionSpec sliceAssignerSpec =
         new StreamWindowPartitionFunctionSpec(
             inputType, rowtimeIndex, size, slide, offset, windowType);
@@ -252,24 +259,30 @@ protected Transformation<RowData> translateToPlanInternal(
             offset,
             windowType,
             outputType,
-            rowtimeIndex);
-    final OneInputStreamOperator windowOperator =
-        new GlutenOneInputOperator(
+            true,
+            rowtimeIndex,
+            windowStartIndex,
+            windowEndIndex);
+    final LogicalType[] accTypes = convertToLogicalTypes(globalAggInfoList.getAccTypes());
+    final RowDataKeySelector selector =
+        KeySelectorUtil.getRowDataSelector(
+            planner.getFlinkContext().getClassLoader(),
+            grouping,
+            InternalTypeInfo.of(inputRowType));
+    final OneInputStreamOperator<RowData, RowData> windowOperator =
+        new org.apache.gluten.table.runtime.operators.WindowAggOperator<RowData, RowData, Long>(
             new StatefulPlanNode(windowAgg.getId(), windowAgg),
             PlanNodeIdGenerator.newId(),
             inputType,
             Map.of(windowAgg.getId(), outputType),
             RowData.class,
             RowData.class,
-            "StreamExecGlobalWindowAggregate");
+            "StreamExecWindowAggregate",
+            selector.getProducedType(),
+            globalAggInfoList.getAggNames(),
+            accTypes);
     // --- End Gluten-specific code changes ---
 
-    final RowDataKeySelector selector =
-        KeySelectorUtil.getRowDataSelector(
-            planner.getFlinkContext().getClassLoader(),
-            grouping,
-            InternalTypeInfo.of(inputRowType));
-
     final OneInputTransformation<RowData, RowData> transform =
         ExecNodeUtil.createOneInputTransformation(
             inputTransform,
@@ -285,40 +298,4 @@ protected Transformation<RowData> translateToPlanInternal(
     transform.setStateKeyType(selector.getProducedType());
     return transform;
   }
-
-  private GeneratedNamespaceAggsHandleFunction<Long> createAggsHandler(
-      String name,
-      SliceAssigner sliceAssigner,
-      AggregateInfoList aggInfoList,
-      int mergedAccOffset,
-      boolean mergedAccIsOnHeap,
-      DataType[] mergedAccExternalTypes,
-      ExecNodeConfig config,
-      ClassLoader classLoader,
-      RelBuilder relBuilder,
-      ZoneId shifTimeZone) {
-    final AggsHandlerCodeGenerator generator =
-        new AggsHandlerCodeGenerator(
-                new CodeGeneratorContext(config, classLoader),
-                relBuilder,
-                JavaScalaConversionUtil.toScala(localAggInputRowType.getChildren()),
-                true) // copyInputField
-            .needAccumulate()
-            .needMerge(mergedAccOffset, mergedAccIsOnHeap, mergedAccExternalTypes);
-
-    final List<WindowProperty> windowProperties =
-        Arrays.asList(
-            Arrays.stream(namedWindowProperties)
-                .map(NamedWindowProperty::getProperty)
-                .toArray(WindowProperty[]::new));
-
-    return generator.generateNamespaceAggsHandler(
-        name,
-        aggInfoList,
-        JavaScalaConversionUtil.toScala(windowProperties),
-        sliceAssigner,
-        // we use window end timestamp to indicate a slicing window, see SliceAssigner
-        Long.class,
-        shifTimeZone);
-  }
 }

gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecLocalWindowAggregate.java

@@ -220,7 +220,10 @@ protected Transformation<RowData> translateToPlanInternal(
             offset,
             windowType,
             outputType,
-            rowtimeIndex);
+            false,
+            rowtimeIndex,
+            -1,
+            -1);
     final OneInputStreamOperator localAggOperator =
         new GlutenOneInputOperator(
             new StatefulPlanNode(windowAgg.getId(), windowAgg),

gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecWindowAggregate.java

@@ -36,6 +36,7 @@
 
 import org.apache.flink.FlinkVersion;
 import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.tuple.Tuple5;
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
@@ -208,11 +209,16 @@ protected Transformation<RowData> translateToPlanInternal(
     // TODO: support more window types.
     Tuple5<Long, Long, Long, Integer, Integer> windowSpecParams =
         WindowUtils.extractWindowParameters(windowing);
+    Tuple2<Integer, Integer> windowStartAndEndIndexes =
+        WindowUtils.getWindowStartAndEndIndexes(namedWindowProperties, (RowType) getOutputType());
     long size = windowSpecParams.f0;
     long slide = windowSpecParams.f1;
     long offset = windowSpecParams.f2;
     int rowtimeIndex = windowSpecParams.f3;
     int windowType = windowSpecParams.f4;
+    boolean isRowTime = windowing.isRowtime();
+    int windowStartIndex = windowStartAndEndIndexes.f0;
+    int windowEndIndex = windowStartAndEndIndexes.f1;
     PartitionFunctionSpec sliceAssignerSpec =
         new StreamWindowPartitionFunctionSpec(
             inputType, rowtimeIndex, size, slide, offset, windowType);
@@ -221,25 +227,29 @@ protected Transformation<RowData> translateToPlanInternal(
             PlanNodeIdGenerator.newId(),
             AggregateStep.SINGLE,
             groupingKeys,
-            groupingKeys,
+            isRowTime ? groupingKeys : List.of(),
             aggNames,
             aggregates,
             false,
             List.of(new EmptyNode(inputType)),
             null,
             List.of());
+    // processing time window can not apply to local-global aggregate optimization, so here we need
+    // to set local aggregtate as null when it is not event time window.
     PlanNode localAgg =
-        new AggregationNode(
-            PlanNodeIdGenerator.newId(),
-            AggregateStep.SINGLE,
-            groupingKeys,
-            groupingKeys,
-            aggNames,
-            aggregates,
-            false,
-            List.of(new EmptyNode(inputType)),
-            null,
-            List.of());
+        isRowTime
+            ? new AggregationNode(
+                PlanNodeIdGenerator.newId(),
+                AggregateStep.SINGLE,
+                groupingKeys,
+                groupingKeys,
+                aggNames,
+                aggregates,
+                false,
+                List.of(new EmptyNode(inputType)),
+                null,
+                List.of())
+            : null;
     PlanNode windowAgg =
         new StreamWindowAggregationNode(
             PlanNodeIdGenerator.newId(),
@@ -255,7 +265,10 @@ protected Transformation<RowData> translateToPlanInternal(
             offset,
             windowType,
             outputType,
-            rowtimeIndex);
+            windowing.isRowtime(),
+            rowtimeIndex,
+            windowStartIndex,
+            windowEndIndex);
     final RowDataKeySelector selector =
         KeySelectorUtil.getRowDataSelector(
             planner.getFlinkContext().getClassLoader(),
@@ -266,6 +279,9 @@ protected Transformation<RowData> translateToPlanInternal(
             .map(x -> x.getLogicalType())
             .collect(Collectors.toList())
             .toArray(new LogicalType[] {});
+    // For TVF windows (Tumbling, Hopping, Cumulative, Session), the window namespace
+    // is identified by the window end timestamp (Long). If count-based windows are
+    // supported in the future, a different serializer may be needed.
     final OneInputStreamOperator<RowData, RowData> windowOperator =
         new org.apache.gluten.table.runtime.operators.WindowAggOperator<RowData, RowData, Long>(
             new StatefulPlanNode(windowAgg.getId(), windowAgg),

gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/WindowUtils.java

@@ -16,16 +16,24 @@
  */
 package org.apache.gluten.rexnode;
 
+import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.tuple.Tuple5;
+import org.apache.flink.table.planner.plan.logical.CumulativeWindowSpec;
 import org.apache.flink.table.planner.plan.logical.HoppingWindowSpec;
+import org.apache.flink.table.planner.plan.logical.SessionWindowSpec;
 import org.apache.flink.table.planner.plan.logical.SliceAttachedWindowingStrategy;
 import org.apache.flink.table.planner.plan.logical.TimeAttributeWindowingStrategy;
 import org.apache.flink.table.planner.plan.logical.TumblingWindowSpec;
 import org.apache.flink.table.planner.plan.logical.WindowAttachedWindowingStrategy;
 import org.apache.flink.table.planner.plan.logical.WindowSpec;
 import org.apache.flink.table.planner.plan.logical.WindowingStrategy;
+import org.apache.flink.table.runtime.groupwindow.NamedWindowProperty;
+import org.apache.flink.table.runtime.groupwindow.WindowEnd;
+import org.apache.flink.table.runtime.groupwindow.WindowStart;
+import org.apache.flink.table.types.logical.RowType;
 
 import java.time.Duration;
+import java.util.List;
 
 /** Utility to store some useful functions. */
 public class WindowUtils {
@@ -53,24 +61,27 @@ public static Tuple5<Long, Long, Long, Integer, Integer> extractWindowParameters
       if (windowOffset != null) {
         offset = windowOffset.toMillis();
       }
+      windowType = 0;
     } else if (windowSpec instanceof TumblingWindowSpec) {
       size = ((TumblingWindowSpec) windowSpec).getSize().toMillis();
       Duration windowOffset = ((TumblingWindowSpec) windowSpec).getOffset();
       if (windowOffset != null) {
         offset = windowOffset.toMillis();
       }
+      windowType = 1;
+    } else if (windowSpec instanceof CumulativeWindowSpec) {
+      windowType = 2;
+    } else if (windowSpec instanceof SessionWindowSpec) {
+      windowType = 3;
     } else {
       throw new RuntimeException("Not support window spec " + windowSpec);
     }
-
     if (windowing instanceof TimeAttributeWindowingStrategy) {
       if (windowing.isRowtime()) {
         rowtimeIndex = ((TimeAttributeWindowingStrategy) windowing).getTimeAttributeIndex();
       }
-      windowType = 0;
     } else if (windowing instanceof WindowAttachedWindowingStrategy) {
       rowtimeIndex = ((WindowAttachedWindowingStrategy) windowing).getWindowEnd();
-      windowType = 1;
     } else if (windowing instanceof SliceAttachedWindowingStrategy) {
       rowtimeIndex = ((SliceAttachedWindowingStrategy) windowing).getSliceEnd();
     } else {
@@ -79,4 +90,18 @@ public static Tuple5<Long, Long, Long, Integer, Integer> extractWindowParameters
     return new Tuple5<Long, Long, Long, Integer, Integer>(
         size, slide, offset, rowtimeIndex, windowType);
   }
+
+  public static Tuple2<Integer, Integer> getWindowStartAndEndIndexes(
+      NamedWindowProperty[] props, RowType outputType) {
+    int windowStartIndex = -1, windowEndIndex = -1;
+    List<String> outputNames = outputType.getFieldNames();
+    for (NamedWindowProperty prop : props) {
+      if (prop.getProperty() instanceof WindowStart) {
+        windowStartIndex = outputNames.indexOf(prop.getName());
+      } else if (prop.getProperty() instanceof WindowEnd) {
+        windowEndIndex = outputNames.indexOf(prop.getName());
+      }
+    }
+    return new Tuple2<Integer, Integer>(windowStartIndex, windowEndIndex);
+  }
 }