Add logic to handle time bucket offset in window functions

Author: sidosera

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/aggregation/WindowGroupingAggregatorFunction.java

@@ -103,7 +103,7 @@ private void evaluateFinalWithWindow(
                 GroupingAggregatorFunction finalAggFunction = finalAgg.aggregatorFunction();
                 boolean hasNullIntermediateState = addInitialIntermediateInput(finalAggFunction, selected, page);
                 for (int i = 0; i < selected.getPositionCount(); i++) {
-                    groupId = selected.getInt(i);
+                    int groupId = selected.getInt(i);
                     mergeBucketsFromWindow(groupId, backwards, page, finalAggFunction, evaluationContext, hasNullIntermediateState);
                 }
             } finally {
@@ -174,6 +174,7 @@ private void mergeBucketsFromWindow(
         }
     }
 
+<<<<<<< HEAD
     /**
      * Adds the non-null positions of {@code page} as intermediate input to {@code fn}, using
      * {@link Block#asVector()} to skip null-checking on dense blocks. Returns whether any null
@@ -189,6 +190,15 @@ private static boolean addInitialIntermediateInput(GroupingAggregatorFunction fn
         if (anyNullable == false) {
             fn.addIntermediateInput(0, groups, page);
             return false;
+=======
+    private static void addNonNullIntermediateInput(GroupingAggregatorFunction fn, IntVector groups, Page page) {
+        int[] positions = positionsWithIntermediateState(page);
+        if (positions.length == 0) {
+            return;
+        }
+        try (IntVector filteredGroups = groups.filter(false, positions); Page filteredPage = filterPage(page, positions)) {
+            fn.addIntermediateInput(0, filteredGroups, filteredPage);
+>>>>>>> eca05d24b96 (Add logic to handle time bucket offset in window functions)
         }
         int[] positions = new int[page.getPositionCount()];
         int count = 0;