Fix CompositeBytesReference#slice to not throw AIOOBE with legal offsets.

server/src/main/java/org/elasticsearch/common/bytes/CompositeBytesReference.java

@@ -22,6 +22,7 @@
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.BytesRefBuilder;
 import org.apache.lucene.util.BytesRefIterator;
+import org.apache.lucene.util.FutureObjects;
 import org.apache.lucene.util.RamUsageEstimator;
 
 import java.io.IOException;
@@ -77,10 +78,16 @@ public int length() {
 
     @Override
     public BytesReference slice(int from, int length) {
+        FutureObjects.checkFromIndexSize(from, length, this.length);
+
+        if (length == 0) {
+            return BytesArray.EMPTY;
+        }
+
         // for slices we only need to find the start and the end reference
         // adjust them and pass on the references in between as they are fully contained
         final int to = from + length;
-        final int limit = getOffsetIndex(from + length);
+        final int limit = getOffsetIndex(to - 1);
         final int start = getOffsetIndex(from);
         final BytesReference[] inSlice = new BytesReference[1 + (limit - start)];
         for (int i = 0, j = start; i < inSlice.length; i++) {

server/src/test/java/org/elasticsearch/common/bytes/CompositeBytesReferenceTests.java

@@ -23,6 +23,7 @@
 import org.apache.lucene.util.BytesRefIterator;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.ReleasableBytesStreamOutput;
+import org.hamcrest.Matchers;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -113,4 +114,18 @@ public void testSliceArrayOffset() throws IOException {
     public void testSliceToBytesRef() throws IOException {
         // CompositeBytesReference shifts offsets
     }
+
+    public void testSliceIsNotCompositeIfMatchesSingleSubSlice() {
+        CompositeBytesReference bytesRef = new CompositeBytesReference(
+                new BytesArray(new byte[12]),
+                new BytesArray(new byte[15]),
+                new BytesArray(new byte[13]));
+
+        // Slices that cross boundaries are composite too
+        assertThat(bytesRef.slice(5, 8), Matchers.instanceOf(CompositeBytesReference.class));
+
+        // But not slices that cover a single sub reference
+        assertThat(bytesRef.slice(13, 10), Matchers.not(Matchers.instanceOf(CompositeBytesReference.class))); // strictly within sub
+        assertThat(bytesRef.slice(12, 15), Matchers.not(Matchers.instanceOf(CompositeBytesReference.class))); // equal to sub
+    }
 }

test/framework/src/main/java/org/elasticsearch/common/bytes/AbstractBytesReferenceTestCase.java

@@ -68,20 +68,21 @@ public void testLength() throws IOException {
     }
 
     public void testSlice() throws IOException {
-        int length = randomInt(PAGE_SIZE * 3);
-        BytesReference pbr = newBytesReference(length);
-        int sliceOffset = randomIntBetween(0, length / 2);
-        int sliceLength = Math.max(0, length - sliceOffset - 1);
-        BytesReference slice = pbr.slice(sliceOffset, sliceLength);
-        assertEquals(sliceLength, slice.length());
-        for (int i = 0; i < sliceLength; i++) {
-            assertEquals(pbr.get(i+sliceOffset), slice.get(i));
-        }
-        BytesRef singlePageOrNull = getSinglePageOrNull(slice);
-        if (singlePageOrNull != null) {
-            // we can't assert the offset since if the length is smaller than the refercence
-            // the offset can be anywhere
-            assertEquals(sliceLength, singlePageOrNull.length);
+        for (int length : new int[] {0, 1, randomIntBetween(2, PAGE_SIZE), randomIntBetween(PAGE_SIZE + 1, 3 * PAGE_SIZE)}) {
+            BytesReference pbr = newBytesReference(length);
+            int sliceOffset = randomIntBetween(0, length / 2);
+            int sliceLength = Math.max(0, length - sliceOffset - 1);
+            BytesReference slice = pbr.slice(sliceOffset, sliceLength);
+            assertEquals(sliceLength, slice.length());
+            for (int i = 0; i < sliceLength; i++) {
+                assertEquals(pbr.get(i+sliceOffset), slice.get(i));
+            }
+            BytesRef singlePageOrNull = getSinglePageOrNull(slice);
+            if (singlePageOrNull != null) {
+                // we can't assert the offset since if the length is smaller than the refercence
+                // the offset can be anywhere
+                assertEquals(sliceLength, singlePageOrNull.length);
+            }
         }
     }