diff --git a/processing/src/main/java/io/druid/segment/data/BlockLayoutColumnarDoublesSupplier.java b/processing/src/main/java/io/druid/segment/data/BlockLayoutColumnarDoublesSupplier.java
index 62ee6de5bb4..cd6b2229ae6 100644
--- a/processing/src/main/java/io/druid/segment/data/BlockLayoutColumnarDoublesSupplier.java
+++ b/processing/src/main/java/io/druid/segment/data/BlockLayoutColumnarDoublesSupplier.java
@@ -27,7 +27,6 @@ import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.nio.DoubleBuffer;
 
-
 public class BlockLayoutColumnarDoublesSupplier implements Supplier<ColumnarDoubles>
 {
   private final GenericIndexed<ResourceHolder<ByteBuffer>> baseDoubleBuffers;
@@ -42,12 +41,7 @@ public class BlockLayoutColumnarDoublesSupplier implements Supplier<ColumnarDoub
       CompressionStrategy strategy
   )
   {
-
-    baseDoubleBuffers = GenericIndexed.read(
-        fromBuffer,
-        new DecompressingByteBufferObjectStrategy(byteOrder, strategy)
-    );
-
+    baseDoubleBuffers = GenericIndexed.read(fromBuffer, new DecompressingByteBufferObjectStrategy(byteOrder, strategy));
     this.totalSize = totalSize;
     this.sizePer = sizePer;
   }
@@ -55,16 +49,40 @@ public class BlockLayoutColumnarDoublesSupplier implements Supplier<ColumnarDoub
   @Override
   public ColumnarDoubles get()
   {
-    return new BlockLayoutColumnarDoubles();
+    final int div = Integer.numberOfTrailingZeros(sizePer);
+    final int rem = sizePer - 1;
+    final boolean isPowerOf2 = sizePer == (1 << div);
+    if (isPowerOf2) {
+      return new BlockLayoutColumnarDoubles()
+      {
+        @Override
+        public double get(int index)
+        {
+          // optimize division and remainder for powers of 2
+          final int bufferNum = index >> div;
+
+          if (bufferNum != currBufferNum) {
+            loadBuffer(bufferNum);
+          }
+
+          final int bufferIndex = index & rem;
+          return doubleBuffer.get(bufferIndex);
+        }
+      };
+    } else {
+      return new BlockLayoutColumnarDoubles();
+    }
   }
 
   private class BlockLayoutColumnarDoubles implements ColumnarDoubles
   {
-    final Indexed<ResourceHolder<ByteBuffer>> resourceHolderIndexed = baseDoubleBuffers.singleThreaded();
-    int currIndex = -1;
+    final Indexed<ResourceHolder<ByteBuffer>> singleThreadedDoubleBuffers = baseDoubleBuffers.singleThreaded();
+
+    int currBufferNum = -1;
     ResourceHolder<ByteBuffer> holder;
-    ByteBuffer buffer;
+    /** doubleBuffer's position must be 0 */
     DoubleBuffer doubleBuffer;
+
     @Override
     public int size()
     {
@@ -74,23 +92,24 @@ public class BlockLayoutColumnarDoublesSupplier implements Supplier<ColumnarDoub
     @Override
     public double get(int index)
     {
+      // division + remainder is optimized by the compiler so keep those together
       final int bufferNum = index / sizePer;
       final int bufferIndex = index % sizePer;
 
-      if (bufferNum != currIndex) {
+      if (bufferNum != currBufferNum) {
         loadBuffer(bufferNum);
       }
 
-      return doubleBuffer.get(doubleBuffer.position() + bufferIndex);
+      return doubleBuffer.get(bufferIndex);
     }
 
     protected void loadBuffer(int bufferNum)
     {
       CloseQuietly.close(holder);
-      holder = resourceHolderIndexed.get(bufferNum);
-      buffer = holder.get();
-      doubleBuffer = buffer.asDoubleBuffer();
-      currIndex = bufferNum;
+      holder = singleThreadedDoubleBuffers.get(bufferNum);
+      // asDoubleBuffer() makes the doubleBuffer's position = 0
+      doubleBuffer = holder.get().asDoubleBuffer();
+      currBufferNum = bufferNum;
     }
 
     @Override
@@ -105,9 +124,9 @@ public class BlockLayoutColumnarDoublesSupplier implements Supplier<ColumnarDoub
     public String toString()
     {
       return "BlockCompressedColumnarDoubles_Anonymous{" +
-             "currIndex=" + currIndex +
+             "currBufferNum=" + currBufferNum +
              ", sizePer=" + sizePer +
-             ", numChunks=" + resourceHolderIndexed.size() +
+             ", numChunks=" + singleThreadedDoubleBuffers.size() +
              ", totalSize=" + totalSize +
              '}';
     }
diff --git a/processing/src/main/java/io/druid/segment/data/BlockLayoutColumnarFloatsSupplier.java b/processing/src/main/java/io/druid/segment/data/BlockLayoutColumnarFloatsSupplier.java
index da7d955740b..6e02c1ce177 100644
--- a/processing/src/main/java/io/druid/segment/data/BlockLayoutColumnarFloatsSupplier.java
+++ b/processing/src/main/java/io/druid/segment/data/BlockLayoutColumnarFloatsSupplier.java
@@ -38,14 +38,11 @@ public class BlockLayoutColumnarFloatsSupplier implements Supplier<ColumnarFloat
       int totalSize,
       int sizePer,
       ByteBuffer fromBuffer,
-      ByteOrder order,
+      ByteOrder byteOrder,
       CompressionStrategy strategy
   )
   {
-    baseFloatBuffers = GenericIndexed.read(
-        fromBuffer,
-        new DecompressingByteBufferObjectStrategy(order, strategy)
-    );
+    baseFloatBuffers = GenericIndexed.read(fromBuffer, new DecompressingByteBufferObjectStrategy(byteOrder, strategy));
     this.totalSize = totalSize;
     this.sizePer = sizePer;
   }
@@ -55,8 +52,8 @@ public class BlockLayoutColumnarFloatsSupplier implements Supplier<ColumnarFloat
   {
     final int div = Integer.numberOfTrailingZeros(sizePer);
     final int rem = sizePer - 1;
-    final boolean powerOf2 = sizePer == (1 << div);
-    if (powerOf2) {
+    final boolean isPowerOf2 = sizePer == (1 << div);
+    if (isPowerOf2) {
       return new BlockLayoutColumnarFloats()
       {
         @Override
@@ -65,12 +62,12 @@ public class BlockLayoutColumnarFloatsSupplier implements Supplier<ColumnarFloat
           // optimize division and remainder for powers of 2
           final int bufferNum = index >> div;
 
-          if (bufferNum != currIndex) {
+          if (bufferNum != currBufferNum) {
             loadBuffer(bufferNum);
           }
 
           final int bufferIndex = index & rem;
-          return floatBuffer.get(floatBuffer.position() + bufferIndex);
+          return floatBuffer.get(bufferIndex);
         }
       };
     } else {
@@ -81,9 +78,10 @@ public class BlockLayoutColumnarFloatsSupplier implements Supplier<ColumnarFloat
   private class BlockLayoutColumnarFloats implements ColumnarFloats
   {
     final Indexed<ResourceHolder<ByteBuffer>> singleThreadedFloatBuffers = baseFloatBuffers.singleThreaded();
-    int currIndex = -1;
+
+    int currBufferNum = -1;
     ResourceHolder<ByteBuffer> holder;
-    ByteBuffer buffer;
+    /** floatBuffer's position must be 0 */
     FloatBuffer floatBuffer;
 
     @Override
@@ -99,11 +97,11 @@ public class BlockLayoutColumnarFloatsSupplier implements Supplier<ColumnarFloat
       final int bufferNum = index / sizePer;
       final int bufferIndex = index % sizePer;
 
-      if (bufferNum != currIndex) {
+      if (bufferNum != currBufferNum) {
         loadBuffer(bufferNum);
       }
 
-      return floatBuffer.get(floatBuffer.position() + bufferIndex);
+      return floatBuffer.get(bufferIndex);
     }
 
     @Override
@@ -125,20 +123,9 @@ public class BlockLayoutColumnarFloatsSupplier implements Supplier<ColumnarFloat
     {
       CloseQuietly.close(holder);
       holder = singleThreadedFloatBuffers.get(bufferNum);
-      buffer = holder.get();
-      floatBuffer = buffer.asFloatBuffer();
-      currIndex = bufferNum;
-    }
-
-    @Override
-    public String toString()
-    {
-      return "BlockCompressedColumnarFloats_Anonymous{" +
-             "currIndex=" + currIndex +
-             ", sizePer=" + sizePer +
-             ", numChunks=" + singleThreadedFloatBuffers.size() +
-             ", totalSize=" + totalSize +
-             '}';
+      // asFloatBuffer() makes the floatBuffer's position = 0
+      floatBuffer = holder.get().asFloatBuffer();
+      currBufferNum = bufferNum;
     }
 
     @Override
@@ -148,5 +135,16 @@ public class BlockLayoutColumnarFloatsSupplier implements Supplier<ColumnarFloat
         holder.close();
       }
     }
+
+    @Override
+    public String toString()
+    {
+      return "BlockCompressedColumnarFloats_Anonymous{" +
+             "currBufferNum=" + currBufferNum +
+             ", sizePer=" + sizePer +
+             ", numChunks=" + singleThreadedFloatBuffers.size() +
+             ", totalSize=" + totalSize +
+             '}';
+    }
   }
 }
diff --git a/processing/src/main/java/io/druid/segment/data/BlockLayoutColumnarLongsSupplier.java b/processing/src/main/java/io/druid/segment/data/BlockLayoutColumnarLongsSupplier.java
index 076a225bfc1..6421a3a1e27 100644
--- a/processing/src/main/java/io/druid/segment/data/BlockLayoutColumnarLongsSupplier.java
+++ b/processing/src/main/java/io/druid/segment/data/BlockLayoutColumnarLongsSupplier.java
@@ -30,7 +30,6 @@ import java.nio.LongBuffer;
 
 public class BlockLayoutColumnarLongsSupplier implements Supplier<ColumnarLongs>
 {
-
   private final GenericIndexed<ResourceHolder<ByteBuffer>> baseLongBuffers;
   private final int totalSize;
   private final int sizePer;
@@ -56,8 +55,8 @@ public class BlockLayoutColumnarLongsSupplier implements Supplier<ColumnarLongs>
   {
     final int div = Integer.numberOfTrailingZeros(sizePer);
     final int rem = sizePer - 1;
-    final boolean powerOf2 = sizePer == (1 << div);
-    if (powerOf2) {
+    final boolean isPowerOf2 = sizePer == (1 << div);
+    if (isPowerOf2) {
       // this provide slightly better performance than calling the LongsEncodingReader.read, probably because Java
       // doesn't inline the method call for some reason. This should be removed when test show that performance
       // of using read method is same as directly accessing the longbuffer
@@ -70,12 +69,12 @@ public class BlockLayoutColumnarLongsSupplier implements Supplier<ColumnarLongs>
             // optimize division and remainder for powers of 2
             final int bufferNum = index >> div;
 
-            if (bufferNum != currIndex) {
+            if (bufferNum != currBufferNum) {
               loadBuffer(bufferNum);
             }
 
             final int bufferIndex = index & rem;
-            return longBuffer.get(longBuffer.position() + bufferIndex);
+            return longBuffer.get(bufferIndex);
           }
 
           @Override
@@ -84,8 +83,9 @@ public class BlockLayoutColumnarLongsSupplier implements Supplier<ColumnarLongs>
             CloseQuietly.close(holder);
             holder = singleThreadedLongBuffers.get(bufferNum);
             buffer = holder.get();
+            // asLongBuffer() makes the longBuffer's position = 0
             longBuffer = buffer.asLongBuffer();
-            currIndex = bufferNum;
+            currBufferNum = bufferNum;
           }
         };
       } else {
@@ -97,7 +97,7 @@ public class BlockLayoutColumnarLongsSupplier implements Supplier<ColumnarLongs>
             // optimize division and remainder for powers of 2
             final int bufferNum = index >> div;
 
-            if (bufferNum != currIndex) {
+            if (bufferNum != currBufferNum) {
               loadBuffer(bufferNum);
             }
 
@@ -116,9 +116,11 @@ public class BlockLayoutColumnarLongsSupplier implements Supplier<ColumnarLongs>
   {
     final CompressionFactory.LongEncodingReader reader = baseReader.duplicate();
     final Indexed<ResourceHolder<ByteBuffer>> singleThreadedLongBuffers = baseLongBuffers.singleThreaded();
-    int currIndex = -1;
+
+    int currBufferNum = -1;
     ResourceHolder<ByteBuffer> holder;
     ByteBuffer buffer;
+    /** longBuffer's position must be 0 */
     LongBuffer longBuffer;
 
     @Override
@@ -130,10 +132,11 @@ public class BlockLayoutColumnarLongsSupplier implements Supplier<ColumnarLongs>
     @Override
     public long get(int index)
     {
+      // division + remainder is optimized by the compiler so keep those together
       final int bufferNum = index / sizePer;
       final int bufferIndex = index % sizePer;
 
-      if (bufferNum != currIndex) {
+      if (bufferNum != currBufferNum) {
         loadBuffer(bufferNum);
       }
 
@@ -160,21 +163,10 @@ public class BlockLayoutColumnarLongsSupplier implements Supplier<ColumnarLongs>
       CloseQuietly.close(holder);
       holder = singleThreadedLongBuffers.get(bufferNum);
       buffer = holder.get();
-      currIndex = bufferNum;
+      currBufferNum = bufferNum;
       reader.setBuffer(buffer);
     }
 
-    @Override
-    public String toString()
-    {
-      return "BlockCompressedColumnarLongs_Anonymous{" +
-             "currIndex=" + currIndex +
-             ", sizePer=" + sizePer +
-             ", numChunks=" + singleThreadedLongBuffers.size() +
-             ", totalSize=" + totalSize +
-             '}';
-    }
-
     @Override
     public void close()
     {
@@ -182,5 +174,16 @@ public class BlockLayoutColumnarLongsSupplier implements Supplier<ColumnarLongs>
         holder.close();
       }
     }
+
+    @Override
+    public String toString()
+    {
+      return "BlockCompressedColumnarLongs_Anonymous{" +
+             "currBufferNum=" + currBufferNum +
+             ", sizePer=" + sizePer +
+             ", numChunks=" + singleThreadedLongBuffers.size() +
+             ", totalSize=" + totalSize +
+             '}';
+    }
   }
 }
diff --git a/processing/src/main/java/io/druid/segment/data/CompressedColumnarIntsSupplier.java b/processing/src/main/java/io/druid/segment/data/CompressedColumnarIntsSupplier.java
index cc27355abdf..9b80dd3b725 100644
--- a/processing/src/main/java/io/druid/segment/data/CompressedColumnarIntsSupplier.java
+++ b/processing/src/main/java/io/druid/segment/data/CompressedColumnarIntsSupplier.java
@@ -21,7 +21,6 @@ package io.druid.segment.data;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
-import com.google.common.io.Closeables;
 import com.google.common.primitives.Ints;
 import io.druid.collections.ResourceHolder;
 import io.druid.java.util.common.IAE;
@@ -74,8 +73,8 @@ public class CompressedColumnarIntsSupplier implements WritableSupplier<Columnar
   {
     final int div = Integer.numberOfTrailingZeros(sizePer);
     final int rem = sizePer - 1;
-    final boolean powerOf2 = sizePer == (1 << div);
-    if (powerOf2) {
+    final boolean isPowerOf2 = sizePer == (1 << div);
+    if (isPowerOf2) {
       return new CompressedColumnarInts()
       {
         @Override
@@ -84,12 +83,12 @@ public class CompressedColumnarIntsSupplier implements WritableSupplier<Columnar
           // optimize division and remainder for powers of 2
           final int bufferNum = index >> div;
 
-          if (bufferNum != currIndex) {
+          if (bufferNum != currBufferNum) {
             loadBuffer(bufferNum);
           }
 
           final int bufferIndex = index & rem;
-          return buffer.get(buffer.position() + bufferIndex);
+          return buffer.get(bufferIndex);
         }
       };
     } else {
@@ -272,8 +271,9 @@ public class CompressedColumnarIntsSupplier implements WritableSupplier<Columnar
   {
     final Indexed<ResourceHolder<ByteBuffer>> singleThreadedIntBuffers = baseIntBuffers.singleThreaded();
 
-    int currIndex = -1;
+    int currBufferNum = -1;
     ResourceHolder<ByteBuffer> holder;
+    /** buffer's position must be 0 */
     IntBuffer buffer;
 
     @Override
@@ -285,41 +285,45 @@ public class CompressedColumnarIntsSupplier implements WritableSupplier<Columnar
     @Override
     public int get(int index)
     {
+      // division + remainder is optimized by the compiler so keep those together
       final int bufferNum = index / sizePer;
       final int bufferIndex = index % sizePer;
 
-      if (bufferNum != currIndex) {
+      if (bufferNum != currBufferNum) {
         loadBuffer(bufferNum);
       }
 
-      return buffer.get(buffer.position() + bufferIndex);
+      return buffer.get(bufferIndex);
     }
 
     protected void loadBuffer(int bufferNum)
     {
       CloseQuietly.close(holder);
       holder = singleThreadedIntBuffers.get(bufferNum);
+      // asIntBuffer() makes the buffer's position = 0
       buffer = holder.get().asIntBuffer();
-      currIndex = bufferNum;
+      currBufferNum = bufferNum;
+    }
+
+    @Override
+    public void close()
+    {
+      if (holder != null) {
+        holder.close();
+      }
     }
 
     @Override
     public String toString()
     {
       return "CompressedIntsIndexedSupplier_Anonymous{" +
-             "currIndex=" + currIndex +
+             "currBufferNum=" + currBufferNum +
              ", sizePer=" + sizePer +
              ", numChunks=" + singleThreadedIntBuffers.size() +
              ", totalSize=" + totalSize +
              '}';
     }
 
-    @Override
-    public void close() throws IOException
-    {
-      Closeables.close(holder, false);
-    }
-
     @Override
     public void inspectRuntimeShape(RuntimeShapeInspector inspector)
     {
diff --git a/processing/src/main/java/io/druid/segment/data/CompressedVSizeColumnarIntsSupplier.java b/processing/src/main/java/io/druid/segment/data/CompressedVSizeColumnarIntsSupplier.java
index a59bea5db38..9cda23d9b6c 100644
--- a/processing/src/main/java/io/druid/segment/data/CompressedVSizeColumnarIntsSupplier.java
+++ b/processing/src/main/java/io/druid/segment/data/CompressedVSizeColumnarIntsSupplier.java
@@ -21,7 +21,6 @@ package io.druid.segment.data;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
-import com.google.common.io.Closeables;
 import com.google.common.primitives.Ints;
 import com.google.common.primitives.Shorts;
 import io.druid.collections.ResourceHolder;
@@ -270,7 +269,7 @@ public class CompressedVSizeColumnarIntsSupplier implements WritableSupplier<Col
     @Override
     protected int _get(int index)
     {
-      return intBuffer.get(intBuffer.position() + index);
+      return intBuffer.get(index);
     }
   }
 
@@ -289,7 +288,7 @@ public class CompressedVSizeColumnarIntsSupplier implements WritableSupplier<Col
     protected int _get(int index)
     {
       // removes the need for padding
-      return shortBuffer.get(shortBuffer.position() + index) & 0xFFFF;
+      return shortBuffer.get(index) & 0xFFFF;
     }
   }
 
@@ -299,7 +298,7 @@ public class CompressedVSizeColumnarIntsSupplier implements WritableSupplier<Col
     protected int _get(int index)
     {
       // removes the need for padding
-      return buffer.get(buffer.position() + index) & 0xFF;
+      return buffer.get(index) & 0xFF;
     }
   }
 
@@ -310,8 +309,9 @@ public class CompressedVSizeColumnarIntsSupplier implements WritableSupplier<Col
     final int div = Integer.numberOfTrailingZeros(sizePer);
     final int rem = sizePer - 1;
 
-    int currIndex = -1;
+    int currBufferNum = -1;
     ResourceHolder<ByteBuffer> holder;
+    /** buffer's position must be 0 */
     ByteBuffer buffer;
     boolean bigEndian;
 
@@ -336,7 +336,7 @@ public class CompressedVSizeColumnarIntsSupplier implements WritableSupplier<Col
       // assumes the number of entries in each buffer is a power of 2
       final int bufferNum = index >> div;
 
-      if (bufferNum != currIndex) {
+      if (bufferNum != currBufferNum) {
         loadBuffer(bufferNum);
       }
 
@@ -352,7 +352,7 @@ public class CompressedVSizeColumnarIntsSupplier implements WritableSupplier<Col
      */
     protected int _get(final int index)
     {
-      final int pos = buffer.position() + index * numBytes;
+      final int pos = index * numBytes;
       // example for numBytes = 3
       // big-endian:    0x000c0b0a stored 0c 0b 0a XX, read 0x0c0b0aXX >>> 8
       // little-endian: 0x000c0b0a stored 0a 0b 0c XX, read 0xXX0c0b0a & 0x00FFFFFF
@@ -365,28 +365,33 @@ public class CompressedVSizeColumnarIntsSupplier implements WritableSupplier<Col
     {
       CloseQuietly.close(holder);
       holder = singleThreadedBuffers.get(bufferNum);
-      buffer = holder.get();
-      currIndex = bufferNum;
-      bigEndian = buffer.order().equals(ByteOrder.BIG_ENDIAN);
+      ByteBuffer bb = holder.get();
+      ByteOrder byteOrder = bb.order();
+      // slice() makes the buffer's position = 0
+      buffer = bb.slice().order(byteOrder);
+      currBufferNum = bufferNum;
+      bigEndian = byteOrder.equals(ByteOrder.BIG_ENDIAN);
+    }
+
+    @Override
+    public void close()
+    {
+      if (holder != null) {
+        holder.close();
+      }
     }
 
     @Override
     public String toString()
     {
       return "CompressedVSizedIntsIndexedSupplier{" +
-             "currIndex=" + currIndex +
+             "currBufferNum=" + currBufferNum +
              ", sizePer=" + sizePer +
              ", numChunks=" + singleThreadedBuffers.size() +
              ", totalSize=" + totalSize +
              '}';
     }
 
-    @Override
-    public void close() throws IOException
-    {
-      Closeables.close(holder, false);
-    }
-
     @Override
     public void inspectRuntimeShape(RuntimeShapeInspector inspector)
     {