Avoid memory allocation when searching index summary

patch by benedict; reviewed by ariel for CASSANDRA-8793

CHANGES.txt

@@ -1,4 +1,5 @@
 3.0
+ * Avoid memory allocation when searching index summary (CASSANDRA-8793)
  * Optimise (Time)?UUIDType Comparisons (CASSANDRA-8730)
  * Make CRC32Ex into a separate maven dependency (CASSANDRA-8836)
  * Use preloaded jemalloc w/ Unsafe (CASSANDRA-8714)

src/java/org/apache/cassandra/io/sstable/IndexSummary.java

@@ -30,6 +30,7 @@ import org.apache.cassandra.io.util.Memory;
 import org.apache.cassandra.io.util.MemoryOutputStream;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.concurrent.WrappedSharedCloseable;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 import static org.apache.cassandra.io.sstable.Downsampling.BASE_SAMPLING_LEVEL;
 
@@ -105,11 +106,13 @@ public class IndexSummary extends WrappedSharedCloseable
     // Harmony's Collections implementation
     public int binarySearch(RowPosition key)
     {
+        ByteBuffer hollow = MemoryUtil.getHollowDirectByteBuffer();
         int low = 0, mid = offsetCount, high = mid - 1, result = -1;
         while (low <= high)
         {
             mid = (low + high) >> 1;
-            result = -DecoratedKey.compareTo(partitioner, ByteBuffer.wrap(getKey(mid)), key);
+            fillTemporaryKey(mid, hollow);
+            result = -DecoratedKey.compareTo(partitioner, hollow, key);
             if (result > 0)
             {
                 low = mid + 1;
@@ -147,6 +150,13 @@ public class IndexSummary extends WrappedSharedCloseable
         return key;
     }
 
+    private void fillTemporaryKey(int index, ByteBuffer buffer)
+    {
+        long start = getPositionInSummary(index);
+        int keySize = (int) (calculateEnd(index) - start - 8L);
+        entries.setByteBuffer(buffer, start, keySize);
+    }
+
     public long getPosition(int index)
     {
         return entries.getLong(calculateEnd(index) - 8);

src/java/org/apache/cassandra/io/sstable/IndexSummaryBuilder.java

@@ -107,8 +107,8 @@ public class IndexSummaryBuilder implements AutoCloseable
 
         // for initializing data structures, adjust our estimates based on the sampling level
         maxExpectedEntries = Math.max(1, (maxExpectedEntries * samplingLevel) / BASE_SAMPLING_LEVEL);
-        offsets = new SafeMemoryWriter(4 * maxExpectedEntries).withByteOrder(ByteOrder.nativeOrder());
+        offsets = new SafeMemoryWriter(4 * maxExpectedEntries).order(ByteOrder.nativeOrder());
-        entries = new SafeMemoryWriter(40 * maxExpectedEntries).withByteOrder(ByteOrder.nativeOrder());
+        entries = new SafeMemoryWriter(40 * maxExpectedEntries).order(ByteOrder.nativeOrder());
         setNextSamplePosition(-minIndexInterval);
     }
 

src/java/org/apache/cassandra/io/util/Memory.java

@@ -182,7 +182,7 @@ public class Memory implements AutoCloseable
 
     public void setShort(long offset, short l)
     {
-        checkBounds(offset, offset + 4);
+        checkBounds(offset, offset + 2);
         if (unaligned)
         {
             unsafe.putShort(peer + offset, l);
@@ -245,9 +245,7 @@ public class Memory implements AutoCloseable
         else if (count == 0)
             return;
 
-        long end = memoryOffset + count;
+        checkBounds(memoryOffset, memoryOffset + count);
-        checkBounds(memoryOffset, end);
-
         unsafe.copyMemory(buffer, BYTE_ARRAY_BASE_OFFSET + bufferOffset, null, peer + memoryOffset, count);
     }
 
@@ -343,6 +341,8 @@ public class Memory implements AutoCloseable
 
     public void put(long trgOffset, Memory memory, long srcOffset, long size)
     {
+        checkBounds(trgOffset, trgOffset + size);
+        memory.checkBounds(srcOffset, srcOffset + size);
         unsafe.copyMemory(memory.peer + srcOffset, peer + trgOffset, size);
     }
 
@@ -401,6 +401,19 @@ public class Memory implements AutoCloseable
         return result;
     }
 
+    public ByteBuffer asByteBuffer(long offset, int length)
+    {
+        checkBounds(offset, offset + length);
+        return MemoryUtil.getByteBuffer(peer + offset, length);
+    }
+
+    // MUST provide a buffer created via MemoryUtil.getHollowDirectByteBuffer()
+    public void setByteBuffer(ByteBuffer buffer, long offset, int length)
+    {
+        checkBounds(offset, offset + length);
+        MemoryUtil.setByteBuffer(buffer, peer + offset, length);
+    }
+
     public String toString()
     {
         return toString(peer, size);

src/java/org/apache/cassandra/io/util/SafeMemoryWriter.java

@@ -33,6 +33,7 @@ public class SafeMemoryWriter extends AbstractDataOutput implements DataOutputPl
         buffer = new SafeMemory(initialCapacity);
     }
 
+    @Override
     public void write(byte[] buffer, int offset, int count)
     {
         long newLength = ensureCapacity(count);
@@ -40,6 +41,7 @@ public class SafeMemoryWriter extends AbstractDataOutput implements DataOutputPl
         this.length = newLength;
     }
 
+    @Override
     public void write(int oneByte)
     {
         long newLength = ensureCapacity(1);
@@ -47,6 +49,7 @@ public class SafeMemoryWriter extends AbstractDataOutput implements DataOutputPl
         length = newLength;
     }
 
+    @Override
     public void writeShort(int val) throws IOException
     {
         if (order != ByteOrder.nativeOrder())
@@ -56,6 +59,7 @@ public class SafeMemoryWriter extends AbstractDataOutput implements DataOutputPl
         length = newLength;
     }
 
+    @Override
     public void writeInt(int val)
     {
         if (order != ByteOrder.nativeOrder())
@@ -65,6 +69,7 @@ public class SafeMemoryWriter extends AbstractDataOutput implements DataOutputPl
         length = newLength;
     }
 
+    @Override
     public void writeLong(long val)
     {
         if (order != ByteOrder.nativeOrder())
@@ -74,6 +79,7 @@ public class SafeMemoryWriter extends AbstractDataOutput implements DataOutputPl
         length = newLength;
     }
 
+    @Override
     public void write(ByteBuffer buffer)
     {
         long newLength = ensureCapacity(buffer.remaining());
@@ -81,10 +87,11 @@ public class SafeMemoryWriter extends AbstractDataOutput implements DataOutputPl
         length = newLength;
     }
 
-    public void write(Memory memory)
+    @Override
+    public void write(Memory memory, long offset, long size)
     {
-        long newLength = ensureCapacity(memory.size());
+        long newLength = ensureCapacity(size);
-        buffer.put(length, memory, 0, memory.size());
+        buffer.put(length, memory, offset, size);
         length = newLength;
     }
 
@@ -128,7 +135,7 @@ public class SafeMemoryWriter extends AbstractDataOutput implements DataOutputPl
 
     // TODO: consider hoisting this into DataOutputPlus, since most implementations can copy with this gracefully
     // this would simplify IndexSummary.IndexSummarySerializer.serialize()
-    public SafeMemoryWriter withByteOrder(ByteOrder order)
+    public SafeMemoryWriter order(ByteOrder order)
     {
         this.order = order;
         return this;

src/java/org/apache/cassandra/utils/memory/MemoryUtil.java

@@ -123,6 +123,13 @@ public abstract class MemoryUtil
     }
 
     public static ByteBuffer getByteBuffer(long address, int length)
+    {
+        ByteBuffer instance = getHollowDirectByteBuffer();
+        setByteBuffer(instance, address, length);
+        return instance;
+    }
+
+    public static ByteBuffer getHollowDirectByteBuffer()
     {
         ByteBuffer instance;
         try
@@ -133,12 +140,15 @@ public abstract class MemoryUtil
         {
             throw new AssertionError(e);
         }
+        instance.order(ByteOrder.nativeOrder());
+        return instance;
+    }
 
+    public static void setByteBuffer(ByteBuffer instance, long address, int length)
+    {
         unsafe.putLong(instance, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET, address);
         unsafe.putInt(instance, DIRECT_BYTE_BUFFER_CAPACITY_OFFSET, length);
         unsafe.putInt(instance, DIRECT_BYTE_BUFFER_LIMIT_OFFSET, length);
-        instance.order(ByteOrder.nativeOrder());
-        return instance;
     }
 
     public static long getLongByByte(long address)