diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/block/AbstractRowBlock.java b/presto-spi/src/main/java/com/facebook/presto/spi/block/AbstractRowBlock.java
index bb6cf0fdece1..573a9a5cf5ee 100644
--- a/presto-spi/src/main/java/com/facebook/presto/spi/block/AbstractRowBlock.java
+++ b/presto-spi/src/main/java/com/facebook/presto/spi/block/AbstractRowBlock.java
@@ -151,7 +151,7 @@ public <T> T getObject(int position, Class<T> clazz)
         }
         checkReadablePosition(position);
 
-        return clazz.cast(new SingleRowBlock(getFieldBlockOffset(position) * numFields, getFieldBlocks()));
+        return clazz.cast(new SingleRowBlock(getFieldBlockOffset(position), getFieldBlocks()));
     }
 
     @Override
diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/block/AbstractSingleRowBlock.java b/presto-spi/src/main/java/com/facebook/presto/spi/block/AbstractSingleRowBlock.java
index 64ae63428c2a..3ff67584fc27 100644
--- a/presto-spi/src/main/java/com/facebook/presto/spi/block/AbstractSingleRowBlock.java
+++ b/presto-spi/src/main/java/com/facebook/presto/spi/block/AbstractSingleRowBlock.java
@@ -19,137 +19,132 @@
 public abstract class AbstractSingleRowBlock
         implements Block
 {
-    // in AbstractSingleRowBlock, offset is position-based (consider as cell-based), not entry-based.
-    protected final int startOffset;
+    protected final int rowIndex;
 
-    protected final int numFields;
-
-    protected abstract Block getFieldBlock(int fieldIndex);
-
-    protected AbstractSingleRowBlock(int startOffset, int numFields)
+    protected AbstractSingleRowBlock(int rowIndex)
     {
-        this.startOffset = startOffset;
-        this.numFields = numFields;
+        this.rowIndex = rowIndex;
     }
 
-    private int getAbsolutePosition(int position)
+    protected abstract Block getFieldBlock(int fieldIndex);
+
+    private void checkFieldIndex(int position)
     {
         if (position < 0 || position >= getPositionCount()) {
-            throw new IllegalArgumentException("position is not valid");
+            throw new IllegalArgumentException("position is not valid: " + position);
         }
-        return position + startOffset;
     }
 
     @Override
     public boolean isNull(int position)
     {
-        position = getAbsolutePosition(position);
-        return getFieldBlock(position % numFields).isNull(position / numFields);
+        checkFieldIndex(position);
+        return getFieldBlock(position).isNull(rowIndex);
     }
 
     @Override
     public byte getByte(int position, int offset)
     {
-        position = getAbsolutePosition(position);
-        return getFieldBlock(position % numFields).getByte(position / numFields, offset);
+        checkFieldIndex(position);
+        return getFieldBlock(position).getByte(rowIndex, offset);
     }
 
     @Override
     public short getShort(int position, int offset)
     {
-        position = getAbsolutePosition(position);
-        return getFieldBlock(position % numFields).getShort(position / numFields, offset);
+        checkFieldIndex(position);
+        return getFieldBlock(position).getShort(rowIndex, offset);
     }
 
     @Override
     public int getInt(int position, int offset)
     {
-        position = getAbsolutePosition(position);
-        return getFieldBlock(position % numFields).getInt(position / numFields, offset);
+        checkFieldIndex(position);
+        return getFieldBlock(position).getInt(rowIndex, offset);
     }
 
     @Override
     public long getLong(int position, int offset)
     {
-        position = getAbsolutePosition(position);
-        return getFieldBlock(position % numFields).getLong(position / numFields, offset);
+        checkFieldIndex(position);
+        return getFieldBlock(position).getLong(rowIndex, offset);
     }
 
     @Override
     public Slice getSlice(int position, int offset, int length)
     {
-        position = getAbsolutePosition(position);
-        return getFieldBlock(position % numFields).getSlice(position / numFields, offset, length);
+        checkFieldIndex(position);
+        return getFieldBlock(position).getSlice(rowIndex, offset, length);
     }
 
     @Override
     public int getSliceLength(int position)
     {
-        position = getAbsolutePosition(position);
-        return getFieldBlock(position % numFields).getSliceLength(position / numFields);
+        checkFieldIndex(position);
+        return getFieldBlock(position).getSliceLength(rowIndex);
     }
 
     @Override
     public int compareTo(int position, int offset, int length, Block otherBlock, int otherPosition, int otherOffset, int otherLength)
     {
-        position = getAbsolutePosition(position);
-        return getFieldBlock(position % numFields).compareTo(position / numFields, offset, length, otherBlock, otherPosition, otherOffset, otherLength);
+        checkFieldIndex(position);
+        return getFieldBlock(position).compareTo(rowIndex, offset, length, otherBlock, otherPosition, otherOffset, otherLength);
     }
 
     @Override
     public boolean bytesEqual(int position, int offset, Slice otherSlice, int otherOffset, int length)
     {
-        position = getAbsolutePosition(position);
-        return getFieldBlock(position % numFields).bytesEqual(position / numFields, offset, otherSlice, otherOffset, length);
+        checkFieldIndex(position);
+        return getFieldBlock(position).bytesEqual(rowIndex, offset, otherSlice, otherOffset, length);
     }
 
     @Override
     public int bytesCompare(int position, int offset, int length, Slice otherSlice, int otherOffset, int otherLength)
     {
-        position = getAbsolutePosition(position);
-        return getFieldBlock(position % numFields).bytesCompare(position / numFields, offset, length, otherSlice, otherOffset, otherLength);
+        checkFieldIndex(position);
+        return getFieldBlock(position).bytesCompare(rowIndex, offset, length, otherSlice, otherOffset, otherLength);
     }
 
     @Override
     public void writeBytesTo(int position, int offset, int length, BlockBuilder blockBuilder)
     {
-        position = getAbsolutePosition(position);
-        getFieldBlock(position % numFields).writeBytesTo(position / numFields, offset, length, blockBuilder);
+        checkFieldIndex(position);
+        getFieldBlock(position).writeBytesTo(rowIndex, offset, length, blockBuilder);
     }
 
     @Override
     public boolean equals(int position, int offset, Block otherBlock, int otherPosition, int otherOffset, int length)
     {
-        position = getAbsolutePosition(position);
-        return getFieldBlock(position % numFields).equals(position / numFields, offset, otherBlock, otherPosition, otherOffset, length);
+        checkFieldIndex(position);
+        return getFieldBlock(position).equals(rowIndex, offset, otherBlock, otherPosition, otherOffset, length);
     }
 
     @Override
     public long hash(int position, int offset, int length)
     {
-        position = getAbsolutePosition(position);
-        return getFieldBlock(position % numFields).hash(position / numFields, offset, length);
+        checkFieldIndex(position);
+        return getFieldBlock(position).hash(rowIndex, offset, length);
     }
 
     @Override
     public <T> T getObject(int position, Class<T> clazz)
     {
-        position = getAbsolutePosition(position);
-        return getFieldBlock(position % numFields).getObject(position / numFields, clazz);
+        checkFieldIndex(position);
+        return getFieldBlock(position).getObject(rowIndex, clazz);
     }
 
     @Override
     public void writePositionTo(int position, BlockBuilder blockBuilder)
     {
-        position = getAbsolutePosition(position);
-        getFieldBlock(position % numFields).writePositionTo(position / numFields, blockBuilder);
+        checkFieldIndex(position);
+        getFieldBlock(position).writePositionTo(rowIndex, blockBuilder);
     }
 
     @Override
     public Block getSingleValueBlock(int position)
     {
-        position = getAbsolutePosition(position);
-        return getFieldBlock(position % numFields).getSingleValueBlock(position / numFields);
+        checkFieldIndex(position);
+        return getFieldBlock(position).getSingleValueBlock(rowIndex);
     }
 
     @Override
diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/block/RowBlockBuilder.java b/presto-spi/src/main/java/com/facebook/presto/spi/block/RowBlockBuilder.java
index 01708bd1df2a..818e4f5989c7 100644
--- a/presto-spi/src/main/java/com/facebook/presto/spi/block/RowBlockBuilder.java
+++ b/presto-spi/src/main/java/com/facebook/presto/spi/block/RowBlockBuilder.java
@@ -137,7 +137,7 @@ public SingleRowBlockWriter beginBlockEntry()
             throw new IllegalStateException("Expected current entry to be closed but was opened");
         }
         currentEntryOpened = true;
-        return new SingleRowBlockWriter(fieldBlockBuilders[0].getPositionCount() * numFields, fieldBlockBuilders);
+        return new SingleRowBlockWriter(fieldBlockBuilders[0].getPositionCount(), fieldBlockBuilders);
     }
 
     @Override
diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/block/SingleRowBlock.java b/presto-spi/src/main/java/com/facebook/presto/spi/block/SingleRowBlock.java
index 51d322f2b975..91b8d8596fcf 100644
--- a/presto-spi/src/main/java/com/facebook/presto/spi/block/SingleRowBlock.java
+++ b/presto-spi/src/main/java/com/facebook/presto/spi/block/SingleRowBlock.java
@@ -27,9 +27,9 @@ public class SingleRowBlock
 
     private final Block[] fieldBlocks;
 
-    SingleRowBlock(int cellOffset, Block[] fieldBlocks)
+    SingleRowBlock(int rowIndex, Block[] fieldBlocks)
     {
-        super(cellOffset, fieldBlocks.length);
+        super(rowIndex);
         this.fieldBlocks = fieldBlocks;
     }
 
@@ -42,14 +42,14 @@ protected Block getFieldBlock(int fieldIndex)
     @Override
     public int getPositionCount()
     {
-        return numFields;
+        return fieldBlocks.length;
     }
 
     @Override
     public long getSizeInBytes()
     {
         long sizeInBytes = 0;
-        for (int i = 0; i < numFields; i++) {
+        for (int i = 0; i < fieldBlocks.length; i++) {
             sizeInBytes += getFieldBlock(i).getSizeInBytes();
         }
         return sizeInBytes;
@@ -59,7 +59,7 @@ public long getSizeInBytes()
     public long getRetainedSizeInBytes()
     {
         long retainedSizeInBytes = INSTANCE_SIZE;
-        for (int i = 0; i < numFields; i++) {
+        for (int i = 0; i < fieldBlocks.length; i++) {
             retainedSizeInBytes += getFieldBlock(i).getRetainedSizeInBytes();
         }
         return retainedSizeInBytes;
@@ -68,7 +68,7 @@ public long getRetainedSizeInBytes()
     @Override
     public void retainedBytesForEachPart(BiConsumer<Object, Long> consumer)
     {
-        for (int i = 0; i < numFields; i++) {
+        for (int i = 0; i < fieldBlocks.length; i++) {
             consumer.accept(fieldBlocks[i], fieldBlocks[i].getRetainedSizeInBytes());
         }
         consumer.accept(this, (long) INSTANCE_SIZE);
@@ -77,21 +77,21 @@ public void retainedBytesForEachPart(BiConsumer<Object, Long> consumer)
     @Override
     public BlockEncoding getEncoding()
     {
-        BlockEncoding[] fieldBlockEncodings = new BlockEncoding[numFields];
-        for (int i = 0; i < numFields; i++) {
+        BlockEncoding[] fieldBlockEncodings = new BlockEncoding[fieldBlocks.length];
+        for (int i = 0; i < fieldBlocks.length; i++) {
             fieldBlockEncodings[i] = fieldBlocks[i].getEncoding();
         }
         return new SingleRowBlockEncoding(fieldBlockEncodings);
     }
 
-    public int getOffset()
+    public int getRowIndex()
     {
-        return startOffset;
+        return rowIndex;
     }
 
     @Override
     public String toString()
     {
-        return format("SingleRowBlock{numFields=%d}", numFields);
+        return format("SingleRowBlock{numFields=%d}", fieldBlocks.length);
     }
 }
diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/block/SingleRowBlockEncoding.java b/presto-spi/src/main/java/com/facebook/presto/spi/block/SingleRowBlockEncoding.java
index 351efe6513c2..7f102cea48b7 100644
--- a/presto-spi/src/main/java/com/facebook/presto/spi/block/SingleRowBlockEncoding.java
+++ b/presto-spi/src/main/java/com/facebook/presto/spi/block/SingleRowBlockEncoding.java
@@ -43,9 +43,9 @@ public String getName()
     public void writeBlock(SliceOutput sliceOutput, Block block)
     {
         SingleRowBlock singleRowBlock = (SingleRowBlock) block;
-        int fieldOffset = singleRowBlock.getOffset() / fieldBlockEncodings.length;
+        int rowIndex = singleRowBlock.getRowIndex();
         for (int i = 0; i < fieldBlockEncodings.length; i++) {
-            fieldBlockEncodings[i].writeBlock(sliceOutput, singleRowBlock.getFieldBlock(i).getRegion(fieldOffset, 1));
+            fieldBlockEncodings[i].writeBlock(sliceOutput, singleRowBlock.getFieldBlock(i).getRegion(rowIndex, 1));
         }
     }
 
diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/block/SingleRowBlockWriter.java b/presto-spi/src/main/java/com/facebook/presto/spi/block/SingleRowBlockWriter.java
index e397a3e7926d..69762f0f8dcf 100644
--- a/presto-spi/src/main/java/com/facebook/presto/spi/block/SingleRowBlockWriter.java
+++ b/presto-spi/src/main/java/com/facebook/presto/spi/block/SingleRowBlockWriter.java
@@ -33,9 +33,9 @@ public class SingleRowBlockWriter
     private int currentFieldIndexToWrite;
     private boolean fieldBlockBuilderReturned;
 
-    SingleRowBlockWriter(int startOffset, BlockBuilder[] fieldBlockBuilders)
+    SingleRowBlockWriter(int rowIndex, BlockBuilder[] fieldBlockBuilders)
     {
-        super(startOffset, fieldBlockBuilders.length);
+        super(rowIndex);
         this.fieldBlockBuilders = fieldBlockBuilders;
         long initialBlockBuilderSize = 0;
         for (int i = 0; i < fieldBlockBuilders.length; i++) {
@@ -72,7 +72,7 @@ protected Block getFieldBlock(int fieldIndex)
     public long getSizeInBytes()
     {
         long currentBlockBuilderSize = 0;
-        for (int i = 0; i < numFields; i++) {
+        for (int i = 0; i < fieldBlockBuilders.length; i++) {
             currentBlockBuilderSize += fieldBlockBuilders[i].getSizeInBytes();
         }
         return currentBlockBuilderSize - initialBlockBuilderSize;
@@ -82,7 +82,7 @@ public long getSizeInBytes()
     public long getRetainedSizeInBytes()
     {
         long size = INSTANCE_SIZE;
-        for (int i = 0; i < numFields; i++) {
+        for (int i = 0; i < fieldBlockBuilders.length; i++) {
             size += fieldBlockBuilders[i].getRetainedSizeInBytes();
         }
         return size;
@@ -91,7 +91,7 @@ public long getRetainedSizeInBytes()
     @Override
     public void retainedBytesForEachPart(BiConsumer<Object, Long> consumer)
     {
-        for (int i = 0; i < numFields; i++) {
+        for (int i = 0; i < fieldBlockBuilders.length; i++) {
             consumer.accept(fieldBlockBuilders[i], fieldBlockBuilders[i].getRetainedSizeInBytes());
         }
         consumer.accept(this, (long) INSTANCE_SIZE);
@@ -207,10 +207,10 @@ public BlockBuilder newBlockBuilderLike(BlockBuilderStatus blockBuilderStatus)
     public String toString()
     {
         if (!fieldBlockBuilderReturned) {
-            return format("RowBlock{SingleRowBlockWriter=%d, fieldBlockBuilderReturned=false, positionCount=%d}", numFields, getPositionCount());
+            return format("RowBlock{SingleRowBlockWriter=%d, fieldBlockBuilderReturned=false, positionCount=%d}", fieldBlockBuilders.length, getPositionCount());
         }
         else {
-            return format("RowBlock{SingleRowBlockWriter=%d, fieldBlockBuilderReturned=true}", numFields);
+            return format("RowBlock{SingleRowBlockWriter=%d, fieldBlockBuilderReturned=true}", fieldBlockBuilders.length);
         }
     }
 
@@ -219,7 +219,7 @@ private void checkFieldIndexToWrite()
         if (fieldBlockBuilderReturned) {
             throw new IllegalStateException("cannot do sequential write after getFieldBlockBuilder is called");
         }
-        if (currentFieldIndexToWrite >= numFields) {
+        if (currentFieldIndexToWrite >= fieldBlockBuilders.length) {
             throw new IllegalStateException("currentFieldIndexToWrite is not valid");
         }
     }