optimize lookups by caching bytebuffer copy

processing/src/main/java/io/druid/query/aggregation/hyperloglog/HyperUniquesSerde.java

@@ -128,8 +128,9 @@ public class HyperUniquesSerde extends ComplexMetricSerde
       @Override
       public HyperLogLogCollector fromByteBuffer(ByteBuffer buffer, int numBytes)
       {
-        buffer.limit(buffer.position() + numBytes);
-        return HyperLogLogCollector.makeCollector(buffer);
+        final ByteBuffer readOnlyBuffer = buffer.asReadOnlyBuffer();
+        readOnlyBuffer.limit(readOnlyBuffer.position() + numBytes);
+        return HyperLogLogCollector.makeCollector(readOnlyBuffer);
       }
 
       @Override

processing/src/main/java/io/druid/segment/data/ConciseCompressedIndexedInts.java

@@ -123,8 +123,9 @@ public class ConciseCompressedIndexedInts implements IndexedInts, Comparable<Con
     @Override
     public ImmutableConciseSet fromByteBuffer(ByteBuffer buffer, int numBytes)
     {
-      buffer.limit(buffer.position() + numBytes);
-      return new ImmutableConciseSet(buffer.asReadOnlyBuffer());
+      final ByteBuffer readOnlyBuffer = buffer.asReadOnlyBuffer();
+      readOnlyBuffer.limit(readOnlyBuffer.position() + numBytes);
+      return new ImmutableConciseSet(readOnlyBuffer);
     }
 
     @Override

processing/src/main/java/io/druid/segment/data/GenericIndexed.java

@@ -154,6 +154,7 @@ public class GenericIndexed<T> implements Indexed<T>
   private final int size;
 
   private final boolean cacheable;
+  private final ThreadLocal<ByteBuffer> cachedBuffer;
   private final ThreadLocal<SizedLRUMap<Integer, T>> cachedValues = new ThreadLocal<SizedLRUMap<Integer, T>>() {
     @Override
     protected SizedLRUMap<Integer, T> initialValue()
@@ -178,6 +179,15 @@ public class GenericIndexed<T> implements Indexed<T>
     indexOffset = theBuffer.position();
     valuesOffset = theBuffer.position() + (size << 2);
 
+    this.cachedBuffer = new ThreadLocal<ByteBuffer>()
+    {
+      @Override
+      protected ByteBuffer initialValue()
+      {
+        return theBuffer.asReadOnlyBuffer();
+      }
+    };
+
     this.cacheable = strategy instanceof CacheableObjectStrategy;
   }
 
@@ -210,23 +220,24 @@ public class GenericIndexed<T> implements Indexed<T>
       }
     }
 
+    final ByteBuffer copyBuffer = this.cachedBuffer.get();
+
     final int startOffset;
     final int endOffset;
 
     if (index == 0) {
       startOffset = 4;
-      endOffset = theBuffer.getInt(indexOffset);
+      endOffset = copyBuffer.getInt(indexOffset);
     } else {
-      final int position = indexOffset + ((index - 1) * 4);
-      startOffset = theBuffer.getInt(position) + 4;
-      endOffset = theBuffer.getInt(position + Ints.BYTES);
+      copyBuffer.position(indexOffset + ((index - 1) * 4));
+      startOffset = copyBuffer.getInt() + 4;
+      endOffset = copyBuffer.getInt();
     }
 
     if (startOffset == endOffset) {
       return null;
     }
 
-    final ByteBuffer copyBuffer = this.theBuffer.asReadOnlyBuffer();
     copyBuffer.position(valuesOffset + startOffset);
     final int size = endOffset - startOffset;
     final T value = strategy.fromByteBuffer(copyBuffer, size);
@@ -310,9 +321,9 @@ public class GenericIndexed<T> implements Indexed<T>
     }
 
     @Override
-    public String fromByteBuffer(ByteBuffer buffer, int numBytes)
+    public String fromByteBuffer(final ByteBuffer buffer, final int numBytes)
     {
-      byte[] bytes = new byte[numBytes];
+      final byte[] bytes = new byte[numBytes];
       buffer.get(bytes);
       return new String(bytes, Charsets.UTF_8);
     }

processing/src/main/java/io/druid/segment/data/IndexedRTree.java

@@ -81,8 +81,10 @@ public class IndexedRTree implements Comparable<IndexedRTree>
     @Override
     public ImmutableRTree fromByteBuffer(ByteBuffer buffer, int numBytes)
     {
-      buffer.limit(buffer.position() + numBytes);
-      return new ImmutableRTree(buffer.asReadOnlyBuffer());
+
+      final ByteBuffer readOnlyBuffer = buffer.asReadOnlyBuffer();
+      readOnlyBuffer.limit(readOnlyBuffer.position() + numBytes);
+      return new ImmutableRTree(readOnlyBuffer);
     }
 
     @Override

processing/src/main/java/io/druid/segment/data/IntBufferIndexedInts.java

@@ -101,8 +101,9 @@ public class IntBufferIndexedInts implements IndexedInts, Comparable<IntBufferIn
     @Override
     public IntBufferIndexedInts fromByteBuffer(ByteBuffer buffer, int numBytes)
     {
-      buffer.limit(buffer.position() + numBytes);
-      return new IntBufferIndexedInts(buffer);
+      final ByteBuffer readOnlyBuffer = buffer.asReadOnlyBuffer();
+      readOnlyBuffer.limit(readOnlyBuffer.position() + numBytes);
+      return new IntBufferIndexedInts(readOnlyBuffer);
     }
 
     @Override