Fix ByteArrayIndexInput with nonzero offset (#93205) (#93207)

We introduced a bug in #79885 by not accounting for cases where there's
a nonzero offset in the underlying byte array. Unfortunately
`ESIndexInputTestCase` didn't cover these code paths. This commit
extends the test coverage and fixes the bug.

Author: DaveCTurner

docs/changelog/93205.yaml

@@ -0,0 +1,5 @@
+pr: 93205
+summary: Fix `ByteArrayIndexInput` with nonzero offset
+area: Distributed
+type: bug
+issues: []

server/src/main/java/org/elasticsearch/common/lucene/store/ByteArrayIndexInput.java

@@ -99,7 +99,7 @@ public void readBytes(final byte[] b, final int offset, int len) throws IOExcept
     @Override
     public short readShort() throws IOException {
         try {
-            return (short) BitUtil.VH_LE_SHORT.get(bytes, pos);
+            return (short) BitUtil.VH_LE_SHORT.get(bytes, pos + offset);
         } finally {
             pos += Short.BYTES;
         }
@@ -108,7 +108,7 @@ public short readShort() throws IOException {
     @Override
     public int readInt() throws IOException {
         try {
-            return (int) BitUtil.VH_LE_INT.get(bytes, pos);
+            return (int) BitUtil.VH_LE_INT.get(bytes, pos + offset);
         } finally {
             pos += Integer.BYTES;
         }
@@ -117,7 +117,7 @@ public int readInt() throws IOException {
     @Override
     public long readLong() throws IOException {
         try {
-            return (long) BitUtil.VH_LE_LONG.get(bytes, pos);
+            return (long) BitUtil.VH_LE_LONG.get(bytes, pos + offset);
         } finally {
             pos += Long.BYTES;
         }

test/framework/src/main/java/org/elasticsearch/common/lucene/store/ESIndexInputTestCase.java

@@ -21,6 +21,8 @@
 import org.junit.BeforeClass;
 
 import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -60,26 +62,37 @@ protected byte[] randomReadAndSlice(IndexInput indexInput, int length) throws IO
         int readPos = (int) indexInput.getFilePointer();
         byte[] output = new byte[length];
         while (readPos < length) {
-            switch (randomIntBetween(0, 5)) {
-                case 0:
-                    // Read by one byte at a time
-                    output[readPos++] = indexInput.readByte();
+            final var readStrategy = randomFrom(2, 6);
+            switch (readStrategy) {
+                case 0, 1, 2, 3:
+                    if (length - readPos >= Long.BYTES && readStrategy <= 0) {
+                        ByteBuffer.wrap(output, readPos, Long.BYTES).order(ByteOrder.LITTLE_ENDIAN).putLong(indexInput.readLong());
+                        readPos += Long.BYTES;
+                    } else if (length - readPos >= Integer.BYTES && readStrategy <= 1) {
+                        ByteBuffer.wrap(output, readPos, Integer.BYTES).order(ByteOrder.LITTLE_ENDIAN).putInt(indexInput.readInt());
+                        readPos += Integer.BYTES;
+                    } else if (length - readPos >= Short.BYTES && readStrategy <= 2) {
+                        ByteBuffer.wrap(output, readPos, Short.BYTES).order(ByteOrder.LITTLE_ENDIAN).putShort(indexInput.readShort());
+                        readPos += Short.BYTES;
+                    } else {
+                        output[readPos++] = indexInput.readByte();
+                    }
                     break;
-                case 1:
+                case 4:
                     // Read several bytes into target
                     int len = randomIntBetween(1, length - readPos);
                     indexInput.readBytes(output, readPos, len);
                     readPos += len;
                     break;
-                case 2:
+                case 5:
                     // Read several bytes into 0-offset target
                     len = randomIntBetween(1, length - readPos);
                     byte[] temp = new byte[len];
                     indexInput.readBytes(temp, 0, len);
                     System.arraycopy(temp, 0, output, readPos, len);
                     readPos += len;
                     break;
-                case 3:
+                case 6:
                     // Read using slice
                     len = randomIntBetween(1, length - readPos);
                     final String sliceExtension = randomValueOtherThan(".cfs", ESIndexInputTestCase::randomFileExtension);
@@ -92,7 +105,7 @@ protected byte[] randomReadAndSlice(IndexInput indexInput, int length) throws IO
                     indexInput.seek(readPos);
                     assertEquals(readPos, indexInput.getFilePointer());
                     break;
-                case 4:
+                case 7:
                     // Seek at a random position and read a single byte,
                     // then seek back to original position
                     final int lastReadPos = readPos;
@@ -106,7 +119,7 @@ protected byte[] randomReadAndSlice(IndexInput indexInput, int length) throws IO
                     indexInput.seek(readPos);
                     assertEquals(readPos, indexInput.getFilePointer());
                     break;
-                case 5:
+                case 8:
                     // Read clone or slice concurrently
                     final int cloneCount = between(1, 3);
                     final CountDownLatch startLatch = new CountDownLatch(1 + cloneCount);