ESQL: Fix rare bug with empty string (#102350)

This fixes a rare bug that occurs when an empty value lands on the page
boundary of the last page. We only allocate the last page if we need
some bytes from it. So if you add an empty string to the as the last
entry in a BytesRefBlock we don't allocate a whole new slab of bytes to
hold the empty string. But, without this change, we attempt to read from
the unallocated array. We try to read 0 bytes from it, but still. That's
a read past the end of the array.

Closes #101969

docs/changelog/102350.yaml

@@ -0,0 +1,6 @@
+pr: 102350
+summary: "ESQL: Fix rare bug with empty string"
+area: ES|QL
+type: bug
+issues:
+ - 101969

server/src/main/java/org/elasticsearch/common/util/BigByteArray.java

@@ -64,6 +64,10 @@ public byte set(long index, byte value) {
     @Override
     public boolean get(long index, int len, BytesRef ref) {
         assert index + len <= size();
+        if (len == 0) {
+            ref.length = 0;
+            return false;
+        }
         int pageIndex = pageIndex(index);
         final int indexInPage = indexInPage(index);
         if (indexInPage + len <= pageSize()) {

server/src/test/java/org/elasticsearch/common/util/BytesRefArrayTests.java

@@ -17,6 +17,9 @@
 import org.elasticsearch.test.ESTestCase;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.IntStream;
 
 import static org.hamcrest.Matchers.equalTo;
 
@@ -115,6 +118,64 @@ public void testLookup() throws IOException {
         }
     }
 
+    public void testReadWritten() {
+        testReadWritten(false);
+    }
+
+    public void testReadWrittenHalfEmpty() {
+        testReadWritten(true);
+    }
+
+    private void testReadWritten(boolean halfEmpty) {
+        List<BytesRef> values = new ArrayList<>();
+        int bytes = PageCacheRecycler.PAGE_SIZE_IN_BYTES * between(2, 20);
+        int used = 0;
+        while (used < bytes) {
+            String str = halfEmpty && randomBoolean() ? "" : randomAlphaOfLengthBetween(0, 200);
+            BytesRef v = new BytesRef(str);
+            used += v.length;
+            values.add(v);
+        }
+        testReadWritten(values, randomBoolean() ? bytes : between(0, bytes));
+    }
+
+    public void testReadWrittenRepeated() {
+        testReadWrittenRepeated(false, between(2, 3000));
+    }
+
+    public void testReadWrittenRepeatedPowerOfTwo() {
+        testReadWrittenRepeated(false, 1024);
+    }
+
+    public void testReadWrittenRepeatedHalfEmpty() {
+        testReadWrittenRepeated(true, between(1, 3000));
+    }
+
+    public void testReadWrittenRepeatedHalfEmptyPowerOfTwo() {
+        testReadWrittenRepeated(true, 1024);
+    }
+
+    public void testReadWrittenRepeated(boolean halfEmpty, int listSize) {
+        List<BytesRef> values = randomList(2, 10, () -> {
+            String str = halfEmpty && randomBoolean() ? "" : randomAlphaOfLengthBetween(0, 10);
+            return new BytesRef(str);
+        });
+        testReadWritten(IntStream.range(0, listSize).mapToObj(i -> values).flatMap(List::stream).toList(), 10);
+    }
+
+    private void testReadWritten(List<BytesRef> values, int initialCapacity) {
+        try (BytesRefArray array = new BytesRefArray(initialCapacity, mockBigArrays())) {
+            for (BytesRef v : values) {
+                array.append(v);
+            }
+            BytesRef scratch = new BytesRef();
+            for (int i = 0; i < values.size(); i++) {
+                array.get(i, scratch);
+                assertThat(scratch, equalTo(values.get(i)));
+            }
+        }
+    }
+
     private static BigArrays mockBigArrays() {
         return new MockBigArrays(new MockPageCacheRecycler(Settings.EMPTY), new NoneCircuitBreakerService());
     }