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

@@ -101,15 +101,14 @@ private void evaluateFinalWithWindow(
                 next.evaluateIntermediate(intermediateBlocks, 0, selected);
                 Page page = new Page(intermediateBlocks);
                 GroupingAggregatorFunction finalAggFunction = finalAgg.aggregatorFunction();
-                int[] positionsWithState = positionsWithIntermediateState(page);
-                boolean hasNullIntermediateState = positionsWithState.length != page.getPositionCount();
+                boolean hasNullIntermediateState = hasNullIntermediateState(page);
                 if (hasNullIntermediateState) {
-                    addNonNullIntermediateInput(finalAggFunction, selected, page, positionsWithState);
+                    addNonNullIntermediateInput(finalAggFunction, selected, page);
                 } else {
                     finalAggFunction.addIntermediateInput(0, 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 {
@@ -186,14 +185,12 @@ private void mergeBucketsFromWindow(
         }
     }
 
-    private static void addNonNullIntermediateInput(GroupingAggregatorFunction fn, IntVector groups, Page page, int[] positionsWithState) {
-        if (positionsWithState.length == 0) {
+    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, positionsWithState);
-            Page filteredPage = filterPage(page, positionsWithState)
-        ) {
+        try (IntVector filteredGroups = groups.filter(false, positions); Page filteredPage = filterPage(page, positions)) {
             fn.addIntermediateInput(0, filteredGroups, filteredPage);
         }
     }
@@ -209,6 +206,10 @@ private static int[] positionsWithIntermediateState(Page page) {
         return Arrays.copyOf(positions, count);
     }
 
+    private static boolean hasNullIntermediateState(Page page) {
+        return positionsWithIntermediateState(page).length != page.getPositionCount();
+    }
+
     private static boolean hasIntermediateState(Page page, int position) {
         for (int i = 0; i < page.getBlockCount(); i++) {
             if (page.getBlock(i).isNull(position)) {