Fix BlockHash DirectEncoder (#108283) (#108329)

The DirectEncoder currently returns the incorrect value for the
positionCount() method, which should be the number of positions ready in
the current batch. We need to keep track of whether a position is loaded
via encodeNextBatch() and consumed via the read() method. However, we
can always return 1 for positionCount(), indicating that one position is
already loaded. Our tests failed to catch this because mv_ordering
wasn't enabled when generating test blocks, effectively disabling the
DirectEncoders.

Closes #108268

Author: dnhatn

docs/changelog/108283.yaml

@@ -0,0 +1,6 @@
+pr: 108283
+summary: Fix `BlockHash` `DirectEncoder`
+area: ES|QL
+type: bug
+issues:
+ - 108268

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

@@ -274,13 +274,13 @@ public final void encodeNextBatch() {
 
         @Override
         public final int positionCount() {
-            return Math.max(valueCount, 1);
+            return 1; // always has one position already loaded
         }
 
         @Override
         public final int valueCount(int positionOffset) {
             assert positionOffset == 0 : positionOffset;
-            return positionCount();
+            return Math.max(valueCount, 1);
         }
 
         @Override

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/data/BasicBlockTests.java

@@ -895,6 +895,7 @@ public static RandomBlock randomBlock(
         int maxDupsPerPosition
     ) {
         List<List<Object>> values = new ArrayList<>();
+        Block.MvOrdering mvOrdering = Block.MvOrdering.DEDUPLICATED_AND_SORTED_ASCENDING;
         try (var builder = elementType.newBlockBuilder(positionCount, blockFactory)) {
             boolean bytesRefFromPoints = randomBoolean();
             Supplier<Point> pointSupplier = randomBoolean() ? GeometryTestUtils::randomPoint : ShapeTestUtils::randomPoint;
@@ -955,6 +956,19 @@ public static RandomBlock randomBlock(
                 if (valueCount != 1 || dupCount != 0) {
                     builder.endPositionEntry();
                 }
+                if (dupCount > 0) {
+                    mvOrdering = Block.MvOrdering.UNORDERED;
+                } else if (mvOrdering != Block.MvOrdering.UNORDERED) {
+                    List<Object> dedupedAndSortedList = valuesAtPosition.stream().sorted().distinct().toList();
+                    if (dedupedAndSortedList.size() != valuesAtPosition.size()) {
+                        mvOrdering = Block.MvOrdering.UNORDERED;
+                    } else if (dedupedAndSortedList.equals(valuesAtPosition) == false) {
+                        mvOrdering = Block.MvOrdering.DEDUPLICATED_UNORDERD;
+                    }
+                }
+            }
+            if (randomBoolean()) {
+                builder.mvOrdering(mvOrdering);
             }
             return new RandomBlock(values, builder.build());
         }