ESQL: Compute support for filtering grouping aggs (#112476)

Adds support to the compute engine for filtering which positions are
processed by grouping aggs. This should allow syntax like

```
| STATS
       success = COUNT(*) WHERE 200 <= response_code AND response_code < 300,
      redirect = COUNT(*) WHERE 300 <= response_code AND response_code < 400,
    client_err = COUNT(*) WHERE 400 <= response_code AND response_code < 500,
    server_err = COUNT(*) WHERE 500 <= response_code AND response_code < 600,
   total_count = COUNT(*)
  BY hostname
```

We could translate the WHERE expression into an `ExpressionEvaluator`
and run it, then plug it into the filtering support added in this PR.

The actual filtering is done by creating a
`FilteredGroupingAggregatorFunction` which runs wraps a regular
`GroupingAggregatorFunction` first executing the filter against the
incoming `Page` and then `null`ing any positions in the group that don't
match. Then passing the resulting groups into the real aggregator. When
the real grouping aggregator implementation sees `null` value for groups
it skips collecting that position.

We had to make two changes to every agg for this to work: 1. Add a
method to force local group tracking mode on any aggregator. Previously
this was only required if the agg encountered `null` values, but when
we're filtering aggs we can no longer trust the `seen` parameter we get
when building the result. This local group tracking mode let's us track
what we've actually seen locally. 2. Add `Releasable` to the `AddInput`
thing we use to handle chunked pages in grouping aggs. This is required
because the results of the filter must be closed on completion.

Both of these are fairly trivial changes, but require touching every
aggregation.

x-pack/plugin/esql/compute/gen/src/main/java/org/elasticsearch/compute/gen/GroupingAggregatorImplementer.java

@@ -182,6 +182,7 @@ public class GroupingAggregatorImplementer {
         builder.addMethod(addRawInputLoop(INT_VECTOR, valueVectorType(init, combine)));
         builder.addMethod(addRawInputLoop(INT_BLOCK, valueBlockType(init, combine)));
         builder.addMethod(addRawInputLoop(INT_BLOCK, valueVectorType(init, combine)));
+        builder.addMethod(selectedMayContainUnseenGroups());
         builder.addMethod(addIntermediateInput());
         builder.addMethod(addIntermediateRowInput());
         builder.addMethod(evaluateIntermediate());
@@ -338,6 +339,9 @@ public class GroupingAggregatorImplementer {
         addBlock.accept(vector);
         builder.addMethod(vector.build());
 
+        MethodSpec.Builder close = MethodSpec.methodBuilder("close").addAnnotation(Override.class).addModifiers(Modifier.PUBLIC);
+        builder.addMethod(close.build());
+
         return builder.build();
     }
 
@@ -485,6 +489,14 @@ public class GroupingAggregatorImplementer {
         builder.addStatement("$T.combine(state, groupId, $L.getBytesRef($L, scratch))", declarationType, blockVariable, offsetVariable);
     }
 
+    private MethodSpec selectedMayContainUnseenGroups() {
+        MethodSpec.Builder builder = MethodSpec.methodBuilder("selectedMayContainUnseenGroups");
+        builder.addAnnotation(Override.class).addModifiers(Modifier.PUBLIC);
+        builder.addParameter(SEEN_GROUP_IDS, "seenGroupIds");
+        builder.addStatement("state.enableGroupIdTracking(seenGroupIds)");
+        return builder.build();
+    }
+
     private MethodSpec addIntermediateInput() {
         MethodSpec.Builder builder = MethodSpec.methodBuilder("addIntermediateInput");
         builder.addAnnotation(Override.class).addModifiers(Modifier.PUBLIC);

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

@@ -73,6 +73,10 @@ public final class CountDistinctBooleanGroupingAggregatorFunction implements Gro
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -85,6 +89,10 @@ public final class CountDistinctBooleanGroupingAggregatorFunction implements Gro
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -144,6 +152,11 @@ public final class CountDistinctBooleanGroupingAggregatorFunction implements Gro
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -76,6 +76,10 @@ public final class CountDistinctBytesRefGroupingAggregatorFunction implements Gr
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -88,6 +92,10 @@ public final class CountDistinctBytesRefGroupingAggregatorFunction implements Gr
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -151,6 +159,11 @@ public final class CountDistinctBytesRefGroupingAggregatorFunction implements Gr
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -78,6 +78,10 @@ public final class CountDistinctDoubleGroupingAggregatorFunction implements Grou
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -90,6 +94,10 @@ public final class CountDistinctDoubleGroupingAggregatorFunction implements Grou
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -149,6 +157,11 @@ public final class CountDistinctDoubleGroupingAggregatorFunction implements Grou
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -78,6 +78,10 @@ public final class CountDistinctFloatGroupingAggregatorFunction implements Group
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -90,6 +94,10 @@ public final class CountDistinctFloatGroupingAggregatorFunction implements Group
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -149,6 +157,11 @@ public final class CountDistinctFloatGroupingAggregatorFunction implements Group
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -76,6 +76,10 @@ public final class CountDistinctIntGroupingAggregatorFunction implements Groupin
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -88,6 +92,10 @@ public final class CountDistinctIntGroupingAggregatorFunction implements Groupin
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -147,6 +155,11 @@ public final class CountDistinctIntGroupingAggregatorFunction implements Groupin
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -78,6 +78,10 @@ public final class CountDistinctLongGroupingAggregatorFunction implements Groupi
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -90,6 +94,10 @@ public final class CountDistinctLongGroupingAggregatorFunction implements Groupi
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -149,6 +157,11 @@ public final class CountDistinctLongGroupingAggregatorFunction implements Groupi
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -73,6 +73,10 @@ public final class MaxBooleanGroupingAggregatorFunction implements GroupingAggre
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -85,6 +89,10 @@ public final class MaxBooleanGroupingAggregatorFunction implements GroupingAggre
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -144,6 +152,11 @@ public final class MaxBooleanGroupingAggregatorFunction implements GroupingAggre
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -76,6 +76,10 @@ public final class MaxBytesRefGroupingAggregatorFunction implements GroupingAggr
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -88,6 +92,10 @@ public final class MaxBytesRefGroupingAggregatorFunction implements GroupingAggr
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -151,6 +159,11 @@ public final class MaxBytesRefGroupingAggregatorFunction implements GroupingAggr
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -75,6 +75,10 @@ public final class MaxDoubleGroupingAggregatorFunction implements GroupingAggreg
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -87,6 +91,10 @@ public final class MaxDoubleGroupingAggregatorFunction implements GroupingAggreg
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -146,6 +154,11 @@ public final class MaxDoubleGroupingAggregatorFunction implements GroupingAggreg
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -75,6 +75,10 @@ public final class MaxFloatGroupingAggregatorFunction implements GroupingAggrega
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -87,6 +91,10 @@ public final class MaxFloatGroupingAggregatorFunction implements GroupingAggrega
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -146,6 +154,11 @@ public final class MaxFloatGroupingAggregatorFunction implements GroupingAggrega
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -73,6 +73,10 @@ public final class MaxIntGroupingAggregatorFunction implements GroupingAggregato
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -85,6 +89,10 @@ public final class MaxIntGroupingAggregatorFunction implements GroupingAggregato
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -144,6 +152,11 @@ public final class MaxIntGroupingAggregatorFunction implements GroupingAggregato
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -76,6 +76,10 @@ public final class MaxIpGroupingAggregatorFunction implements GroupingAggregator
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -88,6 +92,10 @@ public final class MaxIpGroupingAggregatorFunction implements GroupingAggregator
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -151,6 +159,11 @@ public final class MaxIpGroupingAggregatorFunction implements GroupingAggregator
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -75,6 +75,10 @@ public final class MaxLongGroupingAggregatorFunction implements GroupingAggregat
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -87,6 +91,10 @@ public final class MaxLongGroupingAggregatorFunction implements GroupingAggregat
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -146,6 +154,11 @@ public final class MaxLongGroupingAggregatorFunction implements GroupingAggregat
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -75,6 +75,10 @@ public final class MedianAbsoluteDeviationDoubleGroupingAggregatorFunction imple
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -87,6 +91,10 @@ public final class MedianAbsoluteDeviationDoubleGroupingAggregatorFunction imple
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -146,6 +154,11 @@ public final class MedianAbsoluteDeviationDoubleGroupingAggregatorFunction imple
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -75,6 +75,10 @@ public final class MedianAbsoluteDeviationFloatGroupingAggregatorFunction implem
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -87,6 +91,10 @@ public final class MedianAbsoluteDeviationFloatGroupingAggregatorFunction implem
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -146,6 +154,11 @@ public final class MedianAbsoluteDeviationFloatGroupingAggregatorFunction implem
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -73,6 +73,10 @@ public final class MedianAbsoluteDeviationIntGroupingAggregatorFunction implemen
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -85,6 +89,10 @@ public final class MedianAbsoluteDeviationIntGroupingAggregatorFunction implemen
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -144,6 +152,11 @@ public final class MedianAbsoluteDeviationIntGroupingAggregatorFunction implemen
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -75,6 +75,10 @@ public final class MedianAbsoluteDeviationLongGroupingAggregatorFunction impleme
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -87,6 +91,10 @@ public final class MedianAbsoluteDeviationLongGroupingAggregatorFunction impleme
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -146,6 +154,11 @@ public final class MedianAbsoluteDeviationLongGroupingAggregatorFunction impleme
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -73,6 +73,10 @@ public final class MinBooleanGroupingAggregatorFunction implements GroupingAggre
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -85,6 +89,10 @@ public final class MinBooleanGroupingAggregatorFunction implements GroupingAggre
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -144,6 +152,11 @@ public final class MinBooleanGroupingAggregatorFunction implements GroupingAggre
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -76,6 +76,10 @@ public final class MinBytesRefGroupingAggregatorFunction implements GroupingAggr
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -88,6 +92,10 @@ public final class MinBytesRefGroupingAggregatorFunction implements GroupingAggr
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -151,6 +159,11 @@ public final class MinBytesRefGroupingAggregatorFunction implements GroupingAggr
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -75,6 +75,10 @@ public final class MinDoubleGroupingAggregatorFunction implements GroupingAggreg
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -87,6 +91,10 @@ public final class MinDoubleGroupingAggregatorFunction implements GroupingAggreg
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -146,6 +154,11 @@ public final class MinDoubleGroupingAggregatorFunction implements GroupingAggreg
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -75,6 +75,10 @@ public final class MinFloatGroupingAggregatorFunction implements GroupingAggrega
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -87,6 +91,10 @@ public final class MinFloatGroupingAggregatorFunction implements GroupingAggrega
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -146,6 +154,11 @@ public final class MinFloatGroupingAggregatorFunction implements GroupingAggrega
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -73,6 +73,10 @@ public final class MinIntGroupingAggregatorFunction implements GroupingAggregato
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -85,6 +89,10 @@ public final class MinIntGroupingAggregatorFunction implements GroupingAggregato
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -144,6 +152,11 @@ public final class MinIntGroupingAggregatorFunction implements GroupingAggregato
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -76,6 +76,10 @@ public final class MinIpGroupingAggregatorFunction implements GroupingAggregator
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -88,6 +92,10 @@ public final class MinIpGroupingAggregatorFunction implements GroupingAggregator
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -151,6 +159,11 @@ public final class MinIpGroupingAggregatorFunction implements GroupingAggregator
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -75,6 +75,10 @@ public final class MinLongGroupingAggregatorFunction implements GroupingAggregat
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -87,6 +91,10 @@ public final class MinLongGroupingAggregatorFunction implements GroupingAggregat
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -146,6 +154,11 @@ public final class MinLongGroupingAggregatorFunction implements GroupingAggregat
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -78,6 +78,10 @@ public final class PercentileDoubleGroupingAggregatorFunction implements Groupin
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -90,6 +94,10 @@ public final class PercentileDoubleGroupingAggregatorFunction implements Groupin
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -149,6 +157,11 @@ public final class PercentileDoubleGroupingAggregatorFunction implements Groupin
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -78,6 +78,10 @@ public final class PercentileFloatGroupingAggregatorFunction implements Grouping
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -90,6 +94,10 @@ public final class PercentileFloatGroupingAggregatorFunction implements Grouping
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -149,6 +157,11 @@ public final class PercentileFloatGroupingAggregatorFunction implements Grouping
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -76,6 +76,10 @@ public final class PercentileIntGroupingAggregatorFunction implements GroupingAg
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -88,6 +92,10 @@ public final class PercentileIntGroupingAggregatorFunction implements GroupingAg
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -147,6 +155,11 @@ public final class PercentileIntGroupingAggregatorFunction implements GroupingAg
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -78,6 +78,10 @@ public final class PercentileLongGroupingAggregatorFunction implements GroupingA
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -90,6 +94,10 @@ public final class PercentileLongGroupingAggregatorFunction implements GroupingA
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -149,6 +157,11 @@ public final class PercentileLongGroupingAggregatorFunction implements GroupingA
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -85,6 +85,10 @@ public final class RateDoubleGroupingAggregatorFunction implements GroupingAggre
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock, timestampsVector);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -97,6 +101,10 @@ public final class RateDoubleGroupingAggregatorFunction implements GroupingAggre
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector, timestampsVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -162,6 +170,11 @@ public final class RateDoubleGroupingAggregatorFunction implements GroupingAggre
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -87,6 +87,10 @@ public final class RateFloatGroupingAggregatorFunction implements GroupingAggreg
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock, timestampsVector);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -99,6 +103,10 @@ public final class RateFloatGroupingAggregatorFunction implements GroupingAggreg
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector, timestampsVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -164,6 +172,11 @@ public final class RateFloatGroupingAggregatorFunction implements GroupingAggreg
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -85,6 +85,10 @@ public final class RateIntGroupingAggregatorFunction implements GroupingAggregat
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock, timestampsVector);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -97,6 +101,10 @@ public final class RateIntGroupingAggregatorFunction implements GroupingAggregat
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector, timestampsVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -162,6 +170,11 @@ public final class RateIntGroupingAggregatorFunction implements GroupingAggregat
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -85,6 +85,10 @@ public final class RateLongGroupingAggregatorFunction implements GroupingAggrega
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock, timestampsVector);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -97,6 +101,10 @@ public final class RateLongGroupingAggregatorFunction implements GroupingAggrega
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector, timestampsVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -162,6 +170,11 @@ public final class RateLongGroupingAggregatorFunction implements GroupingAggrega
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -76,6 +76,10 @@ public final class SumDoubleGroupingAggregatorFunction implements GroupingAggreg
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -88,6 +92,10 @@ public final class SumDoubleGroupingAggregatorFunction implements GroupingAggreg
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -147,6 +155,11 @@ public final class SumDoubleGroupingAggregatorFunction implements GroupingAggreg
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -78,6 +78,10 @@ public final class SumFloatGroupingAggregatorFunction implements GroupingAggrega
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -90,6 +94,10 @@ public final class SumFloatGroupingAggregatorFunction implements GroupingAggrega
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -149,6 +157,11 @@ public final class SumFloatGroupingAggregatorFunction implements GroupingAggrega
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -75,6 +75,10 @@ public final class SumIntGroupingAggregatorFunction implements GroupingAggregato
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -87,6 +91,10 @@ public final class SumIntGroupingAggregatorFunction implements GroupingAggregato
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -146,6 +154,11 @@ public final class SumIntGroupingAggregatorFunction implements GroupingAggregato
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -75,6 +75,10 @@ public final class SumLongGroupingAggregatorFunction implements GroupingAggregat
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -87,6 +91,10 @@ public final class SumLongGroupingAggregatorFunction implements GroupingAggregat
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -146,6 +154,11 @@ public final class SumLongGroupingAggregatorFunction implements GroupingAggregat
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -79,6 +79,10 @@ public final class TopBooleanGroupingAggregatorFunction implements GroupingAggre
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -91,6 +95,10 @@ public final class TopBooleanGroupingAggregatorFunction implements GroupingAggre
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -150,6 +158,11 @@ public final class TopBooleanGroupingAggregatorFunction implements GroupingAggre
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -79,6 +79,10 @@ public final class TopDoubleGroupingAggregatorFunction implements GroupingAggreg
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -91,6 +95,10 @@ public final class TopDoubleGroupingAggregatorFunction implements GroupingAggreg
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -150,6 +158,11 @@ public final class TopDoubleGroupingAggregatorFunction implements GroupingAggreg
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -79,6 +79,10 @@ public final class TopFloatGroupingAggregatorFunction implements GroupingAggrega
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -91,6 +95,10 @@ public final class TopFloatGroupingAggregatorFunction implements GroupingAggrega
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -150,6 +158,11 @@ public final class TopFloatGroupingAggregatorFunction implements GroupingAggrega
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -77,6 +77,10 @@ public final class TopIntGroupingAggregatorFunction implements GroupingAggregato
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -89,6 +93,10 @@ public final class TopIntGroupingAggregatorFunction implements GroupingAggregato
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -148,6 +156,11 @@ public final class TopIntGroupingAggregatorFunction implements GroupingAggregato
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -80,6 +80,10 @@ public final class TopIpGroupingAggregatorFunction implements GroupingAggregator
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -92,6 +96,10 @@ public final class TopIpGroupingAggregatorFunction implements GroupingAggregator
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -155,6 +163,11 @@ public final class TopIpGroupingAggregatorFunction implements GroupingAggregator
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -79,6 +79,10 @@ public final class TopLongGroupingAggregatorFunction implements GroupingAggregat
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -91,6 +95,10 @@ public final class TopLongGroupingAggregatorFunction implements GroupingAggregat
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -150,6 +158,11 @@ public final class TopLongGroupingAggregatorFunction implements GroupingAggregat
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -72,6 +72,10 @@ public final class ValuesBooleanGroupingAggregatorFunction implements GroupingAg
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -84,6 +88,10 @@ public final class ValuesBooleanGroupingAggregatorFunction implements GroupingAg
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -143,6 +151,11 @@ public final class ValuesBooleanGroupingAggregatorFunction implements GroupingAg
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -73,6 +73,10 @@ public final class ValuesBytesRefGroupingAggregatorFunction implements GroupingA
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -85,6 +89,10 @@ public final class ValuesBytesRefGroupingAggregatorFunction implements GroupingA
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -148,6 +156,11 @@ public final class ValuesBytesRefGroupingAggregatorFunction implements GroupingA
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -72,6 +72,10 @@ public final class ValuesDoubleGroupingAggregatorFunction implements GroupingAgg
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -84,6 +88,10 @@ public final class ValuesDoubleGroupingAggregatorFunction implements GroupingAgg
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -143,6 +151,11 @@ public final class ValuesDoubleGroupingAggregatorFunction implements GroupingAgg
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -72,6 +72,10 @@ public final class ValuesFloatGroupingAggregatorFunction implements GroupingAggr
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -84,6 +88,10 @@ public final class ValuesFloatGroupingAggregatorFunction implements GroupingAggr
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -143,6 +151,11 @@ public final class ValuesFloatGroupingAggregatorFunction implements GroupingAggr
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -70,6 +70,10 @@ public final class ValuesIntGroupingAggregatorFunction implements GroupingAggreg
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -82,6 +86,10 @@ public final class ValuesIntGroupingAggregatorFunction implements GroupingAggreg
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -141,6 +149,11 @@ public final class ValuesIntGroupingAggregatorFunction implements GroupingAggreg
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -72,6 +72,10 @@ public final class ValuesLongGroupingAggregatorFunction implements GroupingAggre
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -84,6 +88,10 @@ public final class ValuesLongGroupingAggregatorFunction implements GroupingAggre
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -143,6 +151,11 @@ public final class ValuesLongGroupingAggregatorFunction implements GroupingAggre
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

x-pack/plugin/esql/compute/src/main/generated/org/elasticsearch/compute/aggregation/spatial/SpatialCentroidCartesianPointDocValuesGroupingAggregatorFunction.java

@@ -81,6 +81,10 @@ public final class SpatialCentroidCartesianPointDocValuesGroupingAggregatorFunct
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -93,6 +97,10 @@ public final class SpatialCentroidCartesianPointDocValuesGroupingAggregatorFunct
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -152,6 +160,11 @@ public final class SpatialCentroidCartesianPointDocValuesGroupingAggregatorFunct
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

x-pack/plugin/esql/compute/src/main/generated/org/elasticsearch/compute/aggregation/spatial/SpatialCentroidCartesianPointSourceValuesGroupingAggregatorFunction.java

@@ -84,6 +84,10 @@ public final class SpatialCentroidCartesianPointSourceValuesGroupingAggregatorFu
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -96,6 +100,10 @@ public final class SpatialCentroidCartesianPointSourceValuesGroupingAggregatorFu
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -159,6 +167,11 @@ public final class SpatialCentroidCartesianPointSourceValuesGroupingAggregatorFu
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

x-pack/plugin/esql/compute/src/main/generated/org/elasticsearch/compute/aggregation/spatial/SpatialCentroidGeoPointDocValuesGroupingAggregatorFunction.java

@@ -81,6 +81,10 @@ public final class SpatialCentroidGeoPointDocValuesGroupingAggregatorFunction im
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -93,6 +97,10 @@ public final class SpatialCentroidGeoPointDocValuesGroupingAggregatorFunction im
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -152,6 +160,11 @@ public final class SpatialCentroidGeoPointDocValuesGroupingAggregatorFunction im
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

x-pack/plugin/esql/compute/src/main/generated/org/elasticsearch/compute/aggregation/spatial/SpatialCentroidGeoPointSourceValuesGroupingAggregatorFunction.java

@@ -84,6 +84,10 @@ public final class SpatialCentroidGeoPointSourceValuesGroupingAggregatorFunction
         public void add(int positionOffset, IntVector groupIds) {
           addRawInput(positionOffset, groupIds, valuesBlock);
         }
+
+        @Override
+        public void close() {
+        }
       };
     }
     return new GroupingAggregatorFunction.AddInput() {
@@ -96,6 +100,10 @@ public final class SpatialCentroidGeoPointSourceValuesGroupingAggregatorFunction
       public void add(int positionOffset, IntVector groupIds) {
         addRawInput(positionOffset, groupIds, valuesVector);
       }
+
+      @Override
+      public void close() {
+      }
     };
   }
 
@@ -159,6 +167,11 @@ public final class SpatialCentroidGeoPointSourceValuesGroupingAggregatorFunction
     }
   }
 
+  @Override
+  public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+    state.enableGroupIdTracking(seenGroupIds);
+  }
+
   @Override
   public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
     state.enableGroupIdTracking(new SeenGroupIds.Empty());

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

@@ -76,6 +76,9 @@ public class CountGroupingAggregatorFunction implements GroupingAggregatorFuncti
                     public void add(int positionOffset, IntVector groupIds) {
                         addRawInput(positionOffset, groupIds, valuesBlock);
                     }
+
+                    @Override
+                    public void close() {}
                 };
             }
         }
@@ -89,6 +92,9 @@ public class CountGroupingAggregatorFunction implements GroupingAggregatorFuncti
             public void add(int positionOffset, IntVector groupIds) {
                 addRawInput(groupIds);
             }
+
+            @Override
+            public void close() {}
         };
     }
 
@@ -149,6 +155,11 @@ public class CountGroupingAggregatorFunction implements GroupingAggregatorFuncti
         }
     }
 
+    @Override
+    public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+        state.enableGroupIdTracking(seenGroupIds);
+    }
+
     @Override
     public void addIntermediateInput(int positionOffset, IntVector groups, Page page) {
         assert channels.size() == intermediateBlockCount();

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

@@ -0,0 +1,46 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.compute.aggregation;
+
+import org.elasticsearch.compute.operator.DriverContext;
+import org.elasticsearch.compute.operator.EvalOperator;
+import org.elasticsearch.core.Releasables;
+
+/**
+ * A {@link AggregatorFunctionSupplier} that wraps another, filtering which positions
+ * are supplied to the aggregator.
+ */
+public record FilteredAggregatorFunctionSupplier(AggregatorFunctionSupplier next, EvalOperator.ExpressionEvaluator.Factory filter)
+    implements
+        AggregatorFunctionSupplier {
+
+    @Override
+    public AggregatorFunction aggregator(DriverContext driverContext) {
+        throw new UnsupportedOperationException("TODO");
+    }
+
+    @Override
+    public GroupingAggregatorFunction groupingAggregator(DriverContext driverContext) {
+        GroupingAggregatorFunction next = this.next.groupingAggregator(driverContext);
+        EvalOperator.ExpressionEvaluator filter = null;
+        try {
+            filter = this.filter.get(driverContext);
+            GroupingAggregatorFunction result = new FilteredGroupingAggregatorFunction(next, filter);
+            next = null;
+            filter = null;
+            return result;
+        } finally {
+            Releasables.closeExpectNoException(next, filter);
+        }
+    }
+
+    @Override
+    public String describe() {
+        return "Filtered[next=" + next.describe() + ", filter=" + filter + "]";
+    }
+}

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

@@ -0,0 +1,117 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.compute.aggregation;
+
+import org.elasticsearch.compute.data.Block;
+import org.elasticsearch.compute.data.BooleanBlock;
+import org.elasticsearch.compute.data.BooleanVector;
+import org.elasticsearch.compute.data.IntBlock;
+import org.elasticsearch.compute.data.IntVector;
+import org.elasticsearch.compute.data.Page;
+import org.elasticsearch.compute.data.ToMask;
+import org.elasticsearch.compute.operator.DriverContext;
+import org.elasticsearch.compute.operator.EvalOperator;
+import org.elasticsearch.core.Releasables;
+
+import java.util.stream.IntStream;
+
+/**
+ * A {@link GroupingAggregatorFunction} that wraps another, filtering which positions
+ * are supplied to the aggregator.
+ */
+record FilteredGroupingAggregatorFunction(GroupingAggregatorFunction next, EvalOperator.ExpressionEvaluator filter)
+    implements
+        GroupingAggregatorFunction {
+
+    FilteredGroupingAggregatorFunction {
+        next.selectedMayContainUnseenGroups(new SeenGroupIds.Empty());
+    }
+
+    @Override
+    public AddInput prepareProcessPage(SeenGroupIds seenGroupIds, Page page) {
+        try (BooleanBlock filterResult = ((BooleanBlock) filter.eval(page))) {
+            ToMask mask = filterResult.toMask();
+            // TODO warn on mv fields
+            AddInput nextAdd = null;
+            try {
+                nextAdd = next.prepareProcessPage(seenGroupIds, page);
+                AddInput result = new FilteredAddInput(mask.mask(), nextAdd, page.getPositionCount());
+                mask = null;
+                nextAdd = null;
+                return result;
+            } finally {
+                Releasables.close(mask, nextAdd);
+            }
+        }
+    }
+
+    private record FilteredAddInput(BooleanVector mask, AddInput nextAdd, int positionCount) implements AddInput {
+        @Override
+        public void add(int positionOffset, IntBlock groupIds) {
+            if (positionOffset == 0) {
+                try (IntBlock filtered = groupIds.keepMask(mask)) {
+                    nextAdd.add(positionOffset, filtered);
+                }
+            } else {
+                try (
+                    BooleanVector offsetMask = mask.filter(
+                        IntStream.range(positionOffset, positionOffset + groupIds.getPositionCount()).toArray()
+                    );
+                    IntBlock filtered = groupIds.keepMask(offsetMask)
+                ) {
+                    nextAdd.add(positionOffset, filtered);
+                }
+            }
+        }
+
+        @Override
+        public void add(int positionOffset, IntVector groupIds) {
+            add(positionOffset, groupIds.asBlock());
+        }
+
+        @Override
+        public void close() {
+            Releasables.close(mask, nextAdd);
+        }
+    }
+
+    @Override
+    public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+        // nothing to do - we already put the underlying agg into this state
+    }
+
+    @Override
+    public void addIntermediateInput(int positionOffset, IntVector groupIdVector, Page page) {
+        next.addIntermediateInput(positionOffset, groupIdVector, page);
+    }
+
+    @Override
+    public void addIntermediateRowInput(int groupId, GroupingAggregatorFunction input, int position) {
+        next.addIntermediateRowInput(groupId, input, position);
+    }
+
+    @Override
+    public void evaluateIntermediate(Block[] blocks, int offset, IntVector selected) {
+        next.evaluateIntermediate(blocks, offset, selected);
+    }
+
+    @Override
+    public void evaluateFinal(Block[] blocks, int offset, IntVector selected, DriverContext driverContext) {
+        next.evaluateFinal(blocks, offset, selected, driverContext);
+    }
+
+    @Override
+    public int intermediateBlockCount() {
+        return next.intermediateBlockCount();
+    }
+
+    @Override
+    public void close() {
+        Releasables.closeExpectNoException(next, filter);
+    }
+}

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

@@ -51,9 +51,17 @@ public class FromPartialGroupingAggregatorFunction implements GroupingAggregator
             public void add(int positionOffset, IntVector groupIds) {
                 addIntermediateInput(positionOffset, groupIds, page);
             }
+
+            @Override
+            public void close() {}
         };
     }
 
+    @Override
+    public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+        delegate.selectedMayContainUnseenGroups(seenGroupIds);
+    }
+
     @Override
     public void addIntermediateInput(int positionOffset, IntVector groupIdVector, Page page) {
         final CompositeBlock inputBlock = page.getBlock(inputChannel);

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

@@ -49,6 +49,9 @@ public class GroupingAggregator implements Releasable {
                 public void add(int positionOffset, IntVector groupIds) {
                     aggregatorFunction.addIntermediateInput(positionOffset, groupIds, page);
                 }
+
+                @Override
+                public void close() {}
             };
         } else {
             return aggregatorFunction.prepareProcessPage(seenGroupIds, page);

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

@@ -24,7 +24,7 @@ public interface GroupingAggregatorFunction extends Releasable {
      * Consume group ids to cause the {@link GroupingAggregatorFunction}
      * to group values at a particular position into a particular group.
      */
-    interface AddInput {
+    interface AddInput extends Releasable {
         /**
          * Send a batch of group ids to the aggregator. The {@code groupIds}
          * may be offset from the start of the block to allow for sending chunks
@@ -43,6 +43,12 @@ public interface GroupingAggregatorFunction extends Releasable {
          *     {@code groupIds} {@linkplain Block} that contains thousands of
          *     values at a single positions.
          * </p>
+         * <p>
+         *     Finally, it's possible for a single position to be collected into
+         *     <strong>groupIds</strong>. In that case it's positionOffset may
+         *     be skipped entirely or the groupIds block could contain a
+         *     {@code null} value at that position.
+         * </p>
          * @param positionOffset offset into the {@link Page} used to build this
          *                       {@link AddInput} of these ids
          * @param groupIds {@link Block} of group id, some of which may be null
@@ -68,7 +74,7 @@ public interface GroupingAggregatorFunction extends Releasable {
     }
 
     /**
-     * Prepare to process a single page of results.
+     * Prepare to process a single page of input.
      * <p>
      *     This should load the input {@link Block}s and check their types and
      *     select an optimal path and return that path as an {@link AddInput}.
@@ -76,6 +82,16 @@ public interface GroupingAggregatorFunction extends Releasable {
      */
     AddInput prepareProcessPage(SeenGroupIds seenGroupIds, Page page);  // TODO allow returning null to opt out of the callback loop
 
+    /**
+     * Call this to signal to the aggregation that the {@code selected}
+     * parameter that's passed to {@link #evaluateIntermediate} or
+     * {@link #evaluateFinal} may reference groups that haven't been
+     * seen. This puts the underlying storage into a mode where it'll
+     * track which group ids have been seen, even if that increases the
+     * overhead.
+     */
+    void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds);
+
     /**
      * Add data produced by {@link #evaluateIntermediate}.
      */

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

@@ -60,6 +60,11 @@ public class ToPartialGroupingAggregatorFunction implements GroupingAggregatorFu
         return delegate.prepareProcessPage(seenGroupIds, page);
     }
 
+    @Override
+    public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+        delegate.selectedMayContainUnseenGroups(seenGroupIds);
+    }
+
     @Override
     public void addIntermediateInput(int positionOffset, IntVector groupIdVector, Page page) {
         final CompositeBlock inputBlock = page.getBlock(channels.get(0));

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

@@ -13,6 +13,7 @@ import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.IntBlock;
 import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.core.Releasable;
+import org.elasticsearch.core.Releasables;
 
 /**
  * Helper for adding a {@link Page} worth of {@link Block}s to a {@link BlockHash}
@@ -149,6 +150,6 @@ public class AddBlock implements Releasable {
 
     @Override
     public void close() {
-        ords.close();
+        Releasables.closeExpectNoException(ords, addInput);
     }
 }

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

@@ -65,6 +65,9 @@ final class BlockHashRowInTableLookup extends RowInTableLookup {
                         lastOrd = ord;
                     }
                 }
+
+                @Override
+                public void close() {}
             });
             success = true;
         } finally {

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/HashAggregationOperator.java

@@ -150,18 +150,23 @@ public class HashAggregationOperator implements Operator {
                     hashStart = System.nanoTime();
                     aggregationNanos += hashStart - aggStart;
                 }
+
+                @Override
+                public void close() {
+                    Releasables.closeExpectNoException(prepared);
+                }
             }
-            AddInput add = new AddInput();
+            try (AddInput add = new AddInput()) {
+                checkState(needsInput(), "Operator is already finishing");
+                requireNonNull(page, "page is null");
 
-            checkState(needsInput(), "Operator is already finishing");
-            requireNonNull(page, "page is null");
+                for (int i = 0; i < prepared.length; i++) {
+                    prepared[i] = aggregators.get(i).prepareProcessPage(blockHash, page);
+                }
 
-            for (int i = 0; i < prepared.length; i++) {
-                prepared[i] = aggregators.get(i).prepareProcessPage(blockHash, page);
+                blockHash.add(wrapPage(page), add);
+                hashNanos += System.nanoTime() - add.hashStart;
             }
-
-            blockHash.add(wrapPage(page), add);
-            hashNanos += System.nanoTime() - add.hashStart;
         } finally {
             page.releaseBlocks();
             pagesProcessed++;

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/FilteredGroupingAggregatorFunctionTests.java

@@ -0,0 +1,169 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.compute.aggregation;
+
+import org.elasticsearch.compute.data.Block;
+import org.elasticsearch.compute.data.BlockFactory;
+import org.elasticsearch.compute.data.BooleanVector;
+import org.elasticsearch.compute.data.IntBlock;
+import org.elasticsearch.compute.data.LongBlock;
+import org.elasticsearch.compute.data.Page;
+import org.elasticsearch.compute.operator.DriverContext;
+import org.elasticsearch.compute.operator.EvalOperator;
+import org.elasticsearch.compute.operator.LongIntBlockSourceOperator;
+import org.elasticsearch.compute.operator.SourceOperator;
+import org.elasticsearch.core.Tuple;
+import org.junit.After;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.IntStream;
+
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.equalTo;
+
+public class FilteredGroupingAggregatorFunctionTests extends GroupingAggregatorFunctionTestCase {
+    private final List<Exception> unclosed = Collections.synchronizedList(new ArrayList<>());
+
+    // TODO some version of this test that applies across all aggs
+    @Override
+    protected AggregatorFunctionSupplier aggregatorFunction(List<Integer> inputChannels) {
+        return new FilteredAggregatorFunctionSupplier(
+            new SumIntAggregatorFunctionSupplier(inputChannels),
+            new AnyGreaterThanFactory(unclosed, inputChannels)
+        );
+    }
+
+    @Override
+    protected String expectedDescriptionOfAggregator() {
+        return "Filtered[next=sum of ints, filter=any > 0]";
+    }
+
+    @Override
+    protected String expectedToStringOfSimpleAggregator() {
+        return "FilteredGroupingAggregatorFunction[next=SumIntGroupingAggregatorFunction[channels=[1]], filter=any > 0]";
+    }
+
+    @Override
+    protected void assertSimpleGroup(List<Page> input, Block result, int position, Long group) {
+        long sum = 0;
+        for (Page page : input) {
+            LongBlock groups = page.getBlock(0);
+            IntBlock ints = page.getBlock(1);
+            for (int p = 0; p < ints.getPositionCount(); p++) {
+                /*
+                 * Perform the sum on the values *only* if:
+                 * 1. Any of the values is > 0 to line up with the condition
+                 * 2. Any of the groups matches the group we're asserting
+                 */
+                int start = ints.getFirstValueIndex(p);
+                int end = start + ints.getValueCount(p);
+                boolean selected = false;
+                for (int i = start; i < end; i++) {
+                    selected |= ints.getInt(i) > 0;
+                }
+                if (selected == false) {
+                    continue;
+                }
+                selected = false;
+                if (group == null) {
+                    selected = groups.isNull(p);
+                } else {
+                    start = groups.getFirstValueIndex(p);
+                    end = start + groups.getValueCount(p);
+                    for (int i = start; i < end; i++) {
+                        selected |= groups.getLong(i) == group;
+                    }
+                }
+                if (selected == false) {
+                    continue;
+                }
+
+                start = ints.getFirstValueIndex(p);
+                end = start + ints.getValueCount(p);
+                for (int i = start; i < end; i++) {
+                    sum += ints.getInt(i);
+                }
+            }
+        }
+        assertThat(((LongBlock) result).getLong(position), equalTo(sum));
+    }
+
+    @Override
+    protected SourceOperator simpleInput(BlockFactory blockFactory, int size) {
+        int max = between(1, Integer.MAX_VALUE / size / 5);
+        return new LongIntBlockSourceOperator(
+            blockFactory,
+            IntStream.range(0, size).mapToObj(l -> Tuple.tuple(randomLongBetween(0, 4), between(-max, max)))
+        );
+    }
+
+    @After
+    public void checkUnclosed() {
+        for (Exception tracker : unclosed) {
+            logger.error("unclosed", tracker);
+        }
+        assertThat(unclosed, empty());
+    }
+
+    /**
+     * This checks if *any* of the integers are > 0. If so we push the group to
+     * the aggregation.
+     */
+    private record AnyGreaterThanFactory(List<Exception> unclosed, List<Integer> inputChannels)
+        implements
+            EvalOperator.ExpressionEvaluator.Factory {
+        @Override
+        public EvalOperator.ExpressionEvaluator get(DriverContext context) {
+            Exception tracker = new Exception(Integer.toString(unclosed.size()));
+            unclosed.add(tracker);
+            return new AnyGreaterThan(context.blockFactory(), unclosed, tracker, inputChannels);
+        }
+
+        @Override
+        public String toString() {
+            return "any > 0";
+        }
+    }
+
+    private record AnyGreaterThan(BlockFactory blockFactory, List<Exception> unclosed, Exception tracker, List<Integer> inputChannels)
+        implements
+            EvalOperator.ExpressionEvaluator {
+        @Override
+        public Block eval(Page page) {
+            IntBlock ints = page.getBlock(inputChannels.get(0));
+            try (BooleanVector.FixedBuilder result = blockFactory.newBooleanVectorFixedBuilder(ints.getPositionCount())) {
+                position: for (int p = 0; p < ints.getPositionCount(); p++) {
+                    int start = ints.getFirstValueIndex(p);
+                    int end = start + ints.getValueCount(p);
+                    for (int i = start; i < end; i++) {
+                        if (ints.getInt(i) > 0) {
+                            result.appendBoolean(p, true);
+                            continue position;
+                        }
+                    }
+                    result.appendBoolean(p, false);
+                }
+                return result.build().asBlock();
+            }
+        }
+
+        @Override
+        public void close() {
+            if (unclosed.remove(tracker) == false) {
+                throw new IllegalStateException("close failure!");
+            }
+        }
+
+        @Override
+        public String toString() {
+            return "any > 0";
+        }
+    }
+}

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/GroupingAggregatorFunctionTestCase.java

@@ -52,11 +52,14 @@ import static org.elasticsearch.compute.data.BlockTestUtils.append;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 
+/**
+ * Shared tests for testing grouped aggregations.
+ */
 public abstract class GroupingAggregatorFunctionTestCase extends ForkingOperatorTestCase {
     protected abstract AggregatorFunctionSupplier aggregatorFunction(List<Integer> inputChannels);
 
     protected final int aggregatorIntermediateBlockCount() {
-        try (var agg = aggregatorFunction(List.of()).aggregator(driverContext())) {
+        try (var agg = aggregatorFunction(List.of()).groupingAggregator(driverContext())) {
             return agg.intermediateBlockCount();
         }
     }
@@ -101,16 +104,20 @@ public abstract class GroupingAggregatorFunctionTestCase extends ForkingOperator
     @Override
     protected final Matcher<String> expectedToStringOfSimple() {
         String hash = "blockHash=LongBlockHash{channel=0, entries=0, seenNull=false}";
-        String type = getClass().getSimpleName().replace("Tests", "");
         return equalTo(
             "HashAggregationOperator["
                 + hash
                 + ", aggregators=[GroupingAggregator[aggregatorFunction="
-                + type
-                + "[channels=[1]], mode=SINGLE]]]"
+                + expectedToStringOfSimpleAggregator()
+                + ", mode=SINGLE]]]"
         );
     }
 
+    protected String expectedToStringOfSimpleAggregator() {
+        String type = getClass().getSimpleName().replace("Tests", "");
+        return type + "[channels=[1]]";
+    }
+
     private SeenGroups seenGroups(List<Page> input) {
         boolean seenNullGroup = false;
         SortedSet<Long> seenGroups = new TreeSet<>();
@@ -544,7 +551,7 @@ public abstract class GroupingAggregatorFunctionTestCase extends ForkingOperator
                     @Override
                     public AddInput prepareProcessPage(SeenGroupIds ignoredSeenGroupIds, Page page) {
                         return new AddInput() {
-                            AddInput delegateAddInput = delegate.prepareProcessPage(bigArrays -> {
+                            final AddInput delegateAddInput = delegate.prepareProcessPage(bigArrays -> {
                                 BitArray seen = new BitArray(0, bigArrays);
                                 seen.or(seenGroupIds);
                                 return seen;
@@ -595,9 +602,19 @@ public abstract class GroupingAggregatorFunctionTestCase extends ForkingOperator
                                     delegateAddInput.add(positionOffset + offset, blockFactory.newIntArrayVector(chunk, count));
                                 }
                             }
+
+                            @Override
+                            public void close() {
+                                delegateAddInput.close();
+                            }
                         };
                     }
 
+                    @Override
+                    public void selectedMayContainUnseenGroups(SeenGroupIds seenGroupIds) {
+                        delegate.selectedMayContainUnseenGroups(seenGroupIds);
+                    }
+
                     @Override
                     public void addIntermediateInput(int positionOffset, IntVector groupIds, Page page) {
                         int[] chunk = new int[emitChunkSize];

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/blockhash/AddBlockTests.java

@@ -39,6 +39,7 @@ public class AddBlockTests extends ESTestCase {
         }
         expected.add(added(3, 4));
         assertThat(result.added, equalTo(expected));
+        assertThat(result.closed, equalTo(true));
     }
 
     public void testMvBlockEndsOnBatchBoundary() {
@@ -62,6 +63,7 @@ public class AddBlockTests extends ESTestCase {
         // We uselessly flush an empty position if emitBatchSize lines up with the total count
         expected.add(new Added(1, List.of(List.of())));
         assertThat(result.added, equalTo(expected));
+        assertThat(result.closed, equalTo(true));
     }
 
     public void testMvPositionEndOnBatchBoundary() {
@@ -83,6 +85,7 @@ public class AddBlockTests extends ESTestCase {
         // Because the first position ended on a block boundary we uselessly emit an empty position there
         expected.add(new Added(0, List.of(List.of(), List.of(0, 2))));
         assertThat(result.added, equalTo(expected));
+        assertThat(result.closed, equalTo(true));
     }
 
     public void testMv() {
@@ -103,6 +106,7 @@ public class AddBlockTests extends ESTestCase {
         }
         expected.add(new Added(1, List.of(List.of(2))));
         assertThat(result.added, equalTo(expected));
+        assertThat(result.closed, equalTo(true));
     }
 
     @After
@@ -117,6 +121,8 @@ public class AddBlockTests extends ESTestCase {
     }
 
     private class TestAddInput implements GroupingAggregatorFunction.AddInput {
+        private boolean closed = false;
+
         private final List<Added> added = new ArrayList<>();
 
         @Override
@@ -139,5 +145,10 @@ public class AddBlockTests extends ESTestCase {
         public void add(int positionOffset, IntVector groupIds) {
             add(positionOffset, groupIds.asBlock());
         }
+
+        @Override
+        public void close() {
+            closed = true;
+        }
     }
 }

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/blockhash/BlockHashTests.java

@@ -1166,6 +1166,9 @@ public class BlockHashTests extends ESTestCase {
                     groupIds.incRef();
                     output1.add(new Output(positionOffset, null, groupIds));
                 }
+
+                @Override
+                public void close() {}
             });
             hash2.add(page, new GroupingAggregatorFunction.AddInput() {
                 @Override
@@ -1179,6 +1182,9 @@ public class BlockHashTests extends ESTestCase {
                     groupIds.incRef();
                     output2.add(new Output(positionOffset, null, groupIds));
                 }
+
+                @Override
+                public void close() {}
             });
             assertThat(output1.size(), equalTo(output1.size()));
             for (int i = 0; i < output1.size(); i++) {
@@ -1297,6 +1303,9 @@ public class BlockHashTests extends ESTestCase {
             public void add(int positionOffset, IntVector groupIds) {
                 add(positionOffset, groupIds.asBlock());
             }
+
+            @Override
+            public void close() {}
         });
         if (blockHash instanceof LongLongBlockHash == false
             && blockHash instanceof BytesRefLongBlockHash == false

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/AbstractAggregationTestCase.java

@@ -11,6 +11,7 @@ import org.elasticsearch.compute.aggregation.Aggregator;
 import org.elasticsearch.compute.aggregation.AggregatorFunctionSupplier;
 import org.elasticsearch.compute.aggregation.AggregatorMode;
 import org.elasticsearch.compute.aggregation.GroupingAggregator;
+import org.elasticsearch.compute.aggregation.GroupingAggregatorFunction;
 import org.elasticsearch.compute.aggregation.SeenGroupIds;
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.ElementType;
@@ -455,22 +456,26 @@ public abstract class AbstractAggregationTestCase extends AbstractFunctionTestCa
         for (int currentGroupOffset = 0; currentGroupOffset < groupCount;) {
             int groupSliceRemainingSize = Math.min(groupSliceSize, groupCount - currentGroupOffset);
             var seenGroupIds = new SeenGroupIds.Range(0, allValuesNull ? 0 : currentGroupOffset + groupSliceRemainingSize);
-            var addInput = aggregator.prepareProcessPage(seenGroupIds, inputPage);
-
-            var positionCount = inputPage.getPositionCount();
-            var dataSliceSize = 1;
-            // Divide data in chunks
-            for (int currentDataOffset = 0; currentDataOffset < positionCount;) {
-                int dataSliceRemainingSize = Math.min(dataSliceSize, positionCount - currentDataOffset);
-                try (
-                    var groups = makeGroupsVector(currentGroupOffset, currentGroupOffset + groupSliceRemainingSize, dataSliceRemainingSize)
-                ) {
-                    addInput.add(currentDataOffset, groups);
-                }
+            try (GroupingAggregatorFunction.AddInput addInput = aggregator.prepareProcessPage(seenGroupIds, inputPage)) {
+                var positionCount = inputPage.getPositionCount();
+                var dataSliceSize = 1;
+                // Divide data in chunks
+                for (int currentDataOffset = 0; currentDataOffset < positionCount;) {
+                    int dataSliceRemainingSize = Math.min(dataSliceSize, positionCount - currentDataOffset);
+                    try (
+                        var groups = makeGroupsVector(
+                            currentGroupOffset,
+                            currentGroupOffset + groupSliceRemainingSize,
+                            dataSliceRemainingSize
+                        )
+                    ) {
+                        addInput.add(currentDataOffset, groups);
+                    }
 
-                currentDataOffset += dataSliceSize;
-                if (positionCount > currentDataOffset) {
-                    dataSliceSize = randomIntBetween(1, Math.min(100, positionCount - currentDataOffset));
+                    currentDataOffset += dataSliceSize;
+                    if (positionCount > currentDataOffset) {
+                        dataSliceSize = randomIntBetween(1, Math.min(100, positionCount - currentDataOffset));
+                    }
                 }
             }