Fix off by one in ValuesBytesRefAggregator (#132032)

There are two bugs introduced in #130510 and #131390 affecting the 
VALUES aggregator. The random tests do not cover these edge cases:

1. The check should be firstValues.size() <= group instead of
firstValues.size() < group when reading values from the firstValues
array. We need to inject nulls with repeated values (to simulate
ordinals) to trigger this case.

2. We incorrectly added positionOffset when reading the group ID. We need
to generate more groups to trigger chunking.

Relates #130510 
Relates #131390

Closes #131878

Author: dnhatn

muted-tests.yml

@@ -515,9 +515,6 @@ tests:
 - class: org.elasticsearch.xpack.esql.vector.VectorSimilarityFunctionsIT
   method: testDifferentDimensions {functionName=v_dot_product similarityFunction=DOT_PRODUCT}
   issue: https://github.com/elastic/elasticsearch/issues/131845
-- class: org.elasticsearch.compute.aggregation.ValuesBytesRefGroupingAggregatorFunctionTests
-  method: testSomeFiltered
-  issue: https://github.com/elastic/elasticsearch/issues/131878
 - class: org.elasticsearch.xpack.restart.FullClusterRestartIT
   method: testWatcherWithApiKey {cluster=UPGRADED}
   issue: https://github.com/elastic/elasticsearch/issues/131964

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

@@ -160,8 +160,8 @@ static void addOrdinalInputBlock(
         IntVector hashIds
     ) {
         for (int p = 0; p < groupIds.getPositionCount(); p++) {
+            final int groupId = groupIds.getInt(p);
             final int valuePosition = p + positionOffset;
-            final int groupId = groupIds.getInt(valuePosition);
             final int start = ordinalIds.getFirstValueIndex(valuePosition);
             final int end = start + ordinalIds.getValueCount(valuePosition);
             for (int i = start; i < end; i++) {
@@ -212,8 +212,8 @@ static void combineIntermediateInputValues(
         } else {
             final BytesRef scratch = new BytesRef();
             for (int p = 0; p < groupIds.getPositionCount(); p++) {
+                final int groupId = groupIds.getInt(p);
                 final int valuePosition = p + positionOffset;
-                final int groupId = groupIds.getInt(valuePosition);
                 final int start = values.getFirstValueIndex(valuePosition);
                 final int end = start + values.getValueCount(valuePosition);
                 for (int i = start; i < end; i++) {

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

@@ -525,12 +525,8 @@ $if(BytesRef)$
             try (var builder = blockFactory.newIntBlockBuilder(estimateSize)) {
                 int nextValuesStart = 0;
                 for (int s = 0; s < selected.getPositionCount(); s++) {
-                    int group = selected.getInt(s);
-                    if (firstValues.size() < group) {
-                        builder.appendNull();
-                        continue;
-                    }
-                    int firstValue = firstValues.get(group) - 1;
+                    final int group = selected.getInt(s);
+                    final int firstValue = group >= firstValues.size() ? -1 : firstValues.get(group) - 1;
                     if (firstValue < 0) {
                         builder.appendNull();
                         continue;

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/aggregation/X-ValuesAggregator.java.st

@@ -27,6 +27,9 @@
 import static org.hamcrest.Matchers.nullValue;
 
 public class ValuesBytesRefGroupingAggregatorFunctionTests extends GroupingAggregatorFunctionTestCase {
+    final boolean ordinals = randomBoolean();
+    final boolean withNulls = randomBoolean();
+
     @Override
     protected AggregatorFunctionSupplier aggregatorFunction() {
         return new ValuesBytesRefAggregatorFunctionSupplier();
@@ -39,10 +42,17 @@ protected String expectedDescriptionOfAggregator() {
 
     @Override
     protected SourceOperator simpleInput(BlockFactory blockFactory, int size) {
-        return new LongBytesRefTupleBlockSourceOperator(
-            blockFactory,
-            IntStream.range(0, size).mapToObj(l -> Tuple.tuple(randomLongBetween(0, 4), new BytesRef(randomAlphaOfLengthBetween(0, 100))))
-        );
+
+        return new LongBytesRefTupleBlockSourceOperator(blockFactory, IntStream.range(0, size).mapToObj(l -> {
+            long groupId = randomLongBetween(0, 100);
+            if (withNulls && randomBoolean()) {
+                return Tuple.tuple(groupId, null);
+            }
+            if (ordinals && randomBoolean()) {
+                return Tuple.tuple(groupId, new BytesRef("v" + between(1, 5)));
+            }
+            return Tuple.tuple(groupId, new BytesRef(randomAlphaOfLengthBetween(0, 100)));
+        }));
     }
 
     @Override

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

@@ -26,6 +26,8 @@
 import static org.hamcrest.Matchers.nullValue;
 
 public class ValuesDoubleGroupingAggregatorFunctionTests extends GroupingAggregatorFunctionTestCase {
+    private final boolean withNulls = randomBoolean();
+
     @Override
     protected AggregatorFunctionSupplier aggregatorFunction() {
         return new ValuesDoubleAggregatorFunctionSupplier();
@@ -40,7 +42,8 @@ protected String expectedDescriptionOfAggregator() {
     protected SourceOperator simpleInput(BlockFactory blockFactory, int size) {
         return new LongDoubleTupleBlockSourceOperator(
             blockFactory,
-            LongStream.range(0, size).mapToObj(l -> Tuple.tuple(randomLongBetween(0, 4), randomDouble()))
+            LongStream.range(0, size)
+                .mapToObj(l -> Tuple.tuple(randomLongBetween(0, 100), withNulls && randomBoolean() ? null : randomDouble()))
         );
     }
 

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

@@ -26,6 +26,8 @@
 import static org.hamcrest.Matchers.nullValue;
 
 public class ValuesFloatGroupingAggregatorFunctionTests extends GroupingAggregatorFunctionTestCase {
+    private final boolean withNulls = randomBoolean();
+
     @Override
     protected AggregatorFunctionSupplier aggregatorFunction() {
         return new ValuesFloatAggregatorFunctionSupplier();
@@ -40,7 +42,8 @@ protected String expectedDescriptionOfAggregator() {
     protected SourceOperator simpleInput(BlockFactory blockFactory, int size) {
         return new LongFloatTupleBlockSourceOperator(
             blockFactory,
-            LongStream.range(0, size).mapToObj(l -> Tuple.tuple(randomLongBetween(0, 4), randomFloat()))
+            LongStream.range(0, size)
+                .mapToObj(l -> Tuple.tuple(randomLongBetween(0, 100), withNulls && randomBoolean() ? null : randomFloat()))
         );
     }
 

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

@@ -40,7 +40,7 @@ protected String expectedDescriptionOfAggregator() {
     protected SourceOperator simpleInput(BlockFactory blockFactory, int size) {
         return new LongIntBlockSourceOperator(
             blockFactory,
-            LongStream.range(0, size).mapToObj(l -> Tuple.tuple(randomLongBetween(0, 4), randomInt()))
+            LongStream.range(0, size).mapToObj(l -> Tuple.tuple(randomLongBetween(0, 100), randomInt()))
         );
     }