Create new block when filter OrdinalBytesRefBlock (elastic#136444)

Currently, when filtering OrdinalBytesRefBlock and 
OrdinalBytesRefVector, we reuse the existing dictionary and filter only
the ordinals. Although this is semantically correct, it can break
HashAggregationOperator when ordinal optimizations are enabled in
BlockHash.

Example: given an incoming OrdinalBytesRefBlock after filtering:

dict: [apple, banana, orange]
ordinals: [0, 0, 0, 0, 1, 1]

Here, orange is not referenced. During grouping and hashing, all 
dictionary entries [apple, banana, orange] are hashed with group IDs [1,
2, 3] (0 reserved for null), and the block produces groupings [1, 1, 1,
1, 2, 2]. The output is correct, but when evaluating partial/final
results, we cannot exclude orange (groupId=3) unless we enter slow mode
and track every seen group. This can cause an IndexOutOfBoundsException
if orange is the largest group and the over-allocation is not enough to
cover it, or orange may be included in the result even though it was
excluded previously.

This change enforces the creation of a new Block/Vector when filtering 
OrdinalBytesRefBlock and OrdinalBytesRefVector.

Closes elastic#136423

Author: dnhatn

docs/changelog/136444.yaml

@@ -0,0 +1,6 @@
+pr: 136444
+summary: Create new block when filter `OrdinalBytesRefBlock`
+area: ES|QL
+type: bug
+issues:
+ - 136423

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/data/OrdinalBytesRefBlock.java

@@ -91,41 +91,33 @@ public OrdinalBytesRefBlock asOrdinals() {
 
     @Override
     public BytesRefBlock filter(int... positions) {
-        if (positions.length * ordinals.getTotalValueCount() >= bytes.getPositionCount() * ordinals.getPositionCount()) {
-            OrdinalBytesRefBlock result = null;
-            IntBlock filteredOrdinals = ordinals.filter(positions);
-            try {
-                result = new OrdinalBytesRefBlock(filteredOrdinals, bytes);
-                bytes.incRef();
-            } finally {
-                if (result == null) {
-                    filteredOrdinals.close();
+        // Do not build a filtered block using the same dictionary, because dictionary entries that are not referenced
+        // may reappear when hashing the dictionary in BlockHash.
+        // TODO: merge this BytesRefArrayBlock#filter
+        final OrdinalBytesRefVector vector = asVector();
+        if (vector != null) {
+            return vector.filter(positions).asBlock();
+        }
+        BytesRef scratch = new BytesRef();
+        try (BytesRefBlock.Builder builder = blockFactory().newBytesRefBlockBuilder(positions.length)) {
+            for (int pos : positions) {
+                if (isNull(pos)) {
+                    builder.appendNull();
+                    continue;
                 }
-            }
-            return result;
-        } else {
-            // TODO: merge this BytesRefArrayBlock#filter
-            BytesRef scratch = new BytesRef();
-            try (BytesRefBlock.Builder builder = blockFactory().newBytesRefBlockBuilder(positions.length)) {
-                for (int pos : positions) {
-                    if (isNull(pos)) {
-                        builder.appendNull();
-                        continue;
-                    }
-                    int valueCount = getValueCount(pos);
-                    int first = getFirstValueIndex(pos);
-                    if (valueCount == 1) {
-                        builder.appendBytesRef(getBytesRef(getFirstValueIndex(pos), scratch));
-                    } else {
-                        builder.beginPositionEntry();
-                        for (int c = 0; c < valueCount; c++) {
-                            builder.appendBytesRef(getBytesRef(first + c, scratch));
-                        }
-                        builder.endPositionEntry();
+                int valueCount = getValueCount(pos);
+                int first = getFirstValueIndex(pos);
+                if (valueCount == 1) {
+                    builder.appendBytesRef(getBytesRef(getFirstValueIndex(pos), scratch));
+                } else {
+                    builder.beginPositionEntry();
+                    for (int c = 0; c < valueCount; c++) {
+                        builder.appendBytesRef(getBytesRef(first + c, scratch));
                     }
+                    builder.endPositionEntry();
                 }
-                return builder.mvOrdering(mvOrdering()).build();
             }
+            return builder.mvOrdering(mvOrdering()).build();
         }
     }
 

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/data/OrdinalBytesRefVector.java

@@ -99,26 +99,14 @@ public BytesRefVector getDictionaryVector() {
 
     @Override
     public BytesRefVector filter(int... positions) {
-        if (positions.length >= ordinals.getPositionCount()) {
-            OrdinalBytesRefVector result = null;
-            IntVector filteredOrdinals = ordinals.filter(positions);
-            try {
-                result = new OrdinalBytesRefVector(filteredOrdinals, bytes);
-                bytes.incRef();
-            } finally {
-                if (result == null) {
-                    filteredOrdinals.close();
-                }
-            }
-            return result;
-        } else {
-            final BytesRef scratch = new BytesRef();
-            try (BytesRefVector.Builder builder = blockFactory().newBytesRefVectorBuilder(positions.length)) {
-                for (int p : positions) {
-                    builder.appendBytesRef(getBytesRef(p, scratch));
-                }
-                return builder.build();
+        // Do not build a filtered block using the same dictionary, because dictionary entries that are not referenced
+        // may reappear when hashing the dictionary in BlockHash.
+        final BytesRef scratch = new BytesRef();
+        try (BytesRefVector.Builder builder = blockFactory().newBytesRefVectorBuilder(positions.length)) {
+            for (int p : positions) {
+                builder.appendBytesRef(getBytesRef(p, scratch));
             }
+            return builder.build();
         }
     }
 

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

@@ -51,6 +51,7 @@
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
 import static org.hamcrest.Matchers.sameInstance;
@@ -1518,6 +1519,67 @@ public void testRefCountingDocVector() {
         assertThat(breaker.getUsed(), is(0L));
     }
 
+    public void testFilterOrdinalBytesRefBlock() {
+        int dictSize = between(1, 10);
+        int positionCount = between(1, 100);
+        try (
+            var dictBuilder = blockFactory.newBytesRefVectorBuilder(dictSize);
+            var ordinalBuilder = blockFactory.newIntBlockBuilder(positionCount)
+        ) {
+            for (int i = 0; i < dictSize; i++) {
+                dictBuilder.appendBytesRef(new BytesRef("value" + i));
+            }
+            for (int i = 0; i < positionCount; i++) {
+                int valueCount = randomIntBetween(0, 2);
+                switch (valueCount) {
+                    case 0 -> ordinalBuilder.appendNull();
+                    case 1 -> ordinalBuilder.appendInt(randomIntBetween(0, dictSize - 1));
+                    default -> {
+                        ordinalBuilder.beginPositionEntry();
+                        for (int v = 0; v < valueCount; v++) {
+                            ordinalBuilder.appendInt(randomIntBetween(0, dictSize - 1));
+                        }
+                        ordinalBuilder.endPositionEntry();
+                    }
+                }
+            }
+            try (var block = new OrdinalBytesRefBlock(ordinalBuilder.build(), dictBuilder.build())) {
+                int[] masks = new int[between(1, 100)];
+                for (int i = 0; i < masks.length; i++) {
+                    masks[i] = randomIntBetween(0, positionCount - 1);
+                }
+                try (var filtered = block.filter(masks)) {
+                    assertThat(filtered, not(instanceOf(OrdinalBytesRefBlock.class)));
+                }
+            }
+        }
+    }
+
+    public void testFilterOrdinalBytesRefVector() {
+        int dictSize = between(1, 10);
+        int positionCount = between(1, 100);
+        try (
+            var dictBuilder = blockFactory.newBytesRefVectorBuilder(dictSize);
+            var ordinalBuilder = blockFactory.newIntVectorBuilder(positionCount)
+        ) {
+            for (int i = 0; i < dictSize; i++) {
+                dictBuilder.appendBytesRef(new BytesRef("value" + i));
+            }
+            for (int i = 0; i < positionCount; i++) {
+                ordinalBuilder.appendInt(randomIntBetween(0, dictSize - 1));
+            }
+            try (var vector = new OrdinalBytesRefVector(ordinalBuilder.build(), dictBuilder.build())) {
+                int[] masks = new int[between(1, 100)];
+                for (int i = 0; i < masks.length; i++) {
+                    masks[i] = randomIntBetween(0, positionCount - 1);
+                }
+                try (var filtered = vector.filter(masks)) {
+                    assertThat(filtered, not(instanceOf(OrdinalBytesRefVector.class)));
+                }
+            }
+        }
+    }
+
     /**
      * Take an object with exactly 1 reference and assert that ref counting works fine.
      * Assumes that {@link Releasable#close()} and {@link RefCounted#decRef()} are equivalent.

x-pack/plugin/esql/qa/testFixtures/src/main/resources/stats.csv-spec

@@ -3381,3 +3381,31 @@ VALUES(color):keyword | color:keyword
                yellow | yellow
 // end::mv-group-values-expand-result[]
 ;
+
+filterOrdinalValues
+required_capability: fix_filter_ordinals
+from employees 
+| MV_EXPAND job_positions
+| WHERE job_positions != "Purchase Manager"
+| STATS employees=count(*) BY job_positions
+| SORT job_positions
+| LIMIT 1000
+;
+
+employees:long | job_positions:keyword
+18             | Accountant
+13             | Architect
+11             | Business Analyst
+13             | Data Scientist
+10             | Head Human Resources
+15             | Internship
+14             | Junior Developer
+20             | Principal Support Engineer
+13             | Python Developer
+15             | Reporting Analyst
+20             | Senior Python Developer
+15             | Senior Team Lead
+11             | Support Engineer
+15             | Tech Lead    
+;
+

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

@@ -1509,7 +1509,12 @@ public enum Cap {
         /**
          * Pack dimension values in TS command
          */
-        PACK_DIMENSIONS_IN_TS
+        PACK_DIMENSIONS_IN_TS,
+
+        /**
+         * Create new block when filtering OrdinalBytesRefBlock
+         */
+        FIX_FILTER_ORDINALS,
 
         ;