[ESQL] Fix TopN grouping aggregates bug with non-qualifying intermediate states (elastic#129633)

Continuation of elastic#127148

When datanodes send the STATS intermediate states to the coordinator, it aggregates them.
Now, however, the TopN groups sent by a datanode may not be acceptable in the coordinator (Because it has better values already), so it will discard such values.

However, the engine wasn't handling intermediate groups with nulls (TopNBlockHash uses nulls to discard unused groups).
See https://github.com/elastic/elasticsearch/blob/main/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/aggregation/GroupingAggregator.java#L47

_This code isn't connected with the query yet, so there's no bug in production_

Author: ivancea

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

@@ -204,9 +204,9 @@ private TypeSpec type() {
         for (ClassName groupIdClass : GROUP_IDS_CLASSES) {
             builder.addMethod(addRawInputLoop(groupIdClass, blockType(aggParam.type())));
             builder.addMethod(addRawInputLoop(groupIdClass, vectorType(aggParam.type())));
+            builder.addMethod(addIntermediateInput(groupIdClass));
         }
         builder.addMethod(selectedMayContainUnseenGroups());
-        builder.addMethod(addIntermediateInput());
         builder.addMethod(evaluateIntermediate());
         builder.addMethod(evaluateFinal());
         builder.addMethod(toStringMethod());
@@ -583,11 +583,12 @@ private MethodSpec selectedMayContainUnseenGroups() {
         return builder.build();
     }
 
-    private MethodSpec addIntermediateInput() {
+    private MethodSpec addIntermediateInput(TypeName groupsType) {
+        boolean groupsIsBlock = groupsType.toString().endsWith("Block");
         MethodSpec.Builder builder = MethodSpec.methodBuilder("addIntermediateInput");
         builder.addAnnotation(Override.class).addModifiers(Modifier.PUBLIC);
         builder.addParameter(TypeName.INT, "positionOffset");
-        builder.addParameter(INT_VECTOR, "groups");
+        builder.addParameter(groupsType, "groups");
         builder.addParameter(PAGE, "page");
 
         builder.addStatement("state.enableGroupIdTracking(new $T.Empty())", SEEN_GROUP_IDS);
@@ -613,7 +614,18 @@ private MethodSpec addIntermediateInput() {
         }
         builder.beginControlFlow("for (int groupPosition = 0; groupPosition < groups.getPositionCount(); groupPosition++)");
         {
-            builder.addStatement("int groupId = groups.getInt(groupPosition)");
+            if (groupsIsBlock) {
+                builder.beginControlFlow("if (groups.isNull(groupPosition))");
+                builder.addStatement("continue");
+                builder.endControlFlow();
+                builder.addStatement("int groupStart = groups.getFirstValueIndex(groupPosition)");
+                builder.addStatement("int groupEnd = groupStart + groups.getValueCount(groupPosition)");
+                builder.beginControlFlow("for (int g = groupStart; g < groupEnd; g++)");
+                builder.addStatement("int groupId = groups.getInt(g)");
+            } else {
+                builder.addStatement("int groupId = groups.getInt(groupPosition)");
+            }
+
             if (aggState.declaredType().isPrimitive()) {
                 if (warnExceptions.isEmpty()) {
                     assert intermediateState.size() == 2;
@@ -664,6 +676,9 @@ private MethodSpec addIntermediateInput() {
                     declarationType
                 );
             }
+            if (groupsIsBlock) {
+                builder.endControlFlow();
+            }
             builder.endControlFlow();
         }
         return builder.build();