diff --git a/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/IntegerPointable.java b/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/IntegerPointable.java
index 34bf494..a3b9f44 100644
--- a/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/IntegerPointable.java
+++ b/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/IntegerPointable.java
@@ -45,7 +45,7 @@
         private static final long serialVersionUID = 1L;
 
         @Override
-        public IPointable createPointable() {
+        public IntegerPointable createPointable() {
             return new IntegerPointable();
         }
 
diff --git a/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/LongPointable.java b/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/LongPointable.java
index fee0667..41af193 100644
--- a/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/LongPointable.java
+++ b/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/LongPointable.java
@@ -142,4 +142,10 @@
     public double doubleValue() {
         return getLong();
     }
+
+    public static byte[] toByteArray(long value) {
+        byte[] bytes = new byte[Long.BYTES];
+        setLong(bytes, 0, value);
+        return bytes;
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/util/ArrayBackedValueStorage.java b/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/util/ArrayBackedValueStorage.java
index 250d21a..33aa58b 100644
--- a/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/util/ArrayBackedValueStorage.java
+++ b/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/util/ArrayBackedValueStorage.java
@@ -65,4 +65,8 @@
         reset();
         append(value);
     }
+
+    public void setSize(int bytesRequired) {
+        data.setSize(bytesRequired);
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/util/ByteArrayAccessibleOutputStream.java b/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/util/ByteArrayAccessibleOutputStream.java
index c1558ba..260312b 100644
--- a/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/util/ByteArrayAccessibleOutputStream.java
+++ b/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/util/ByteArrayAccessibleOutputStream.java
@@ -79,8 +79,9 @@
 
     private void ensureCapacity(int minCapacity) {
         // overflow-conscious code
-        if (minCapacity - buf.length > 0)
+        if (minCapacity - buf.length > 0) {
             grow(minCapacity);
+        }
     }
 
     /**
@@ -94,11 +95,13 @@
         // overflow-conscious code
         int oldCapacity = buf.length;
         int newCapacity = oldCapacity << 1;
-        if (newCapacity - minCapacity < 0)
+        if (newCapacity - minCapacity < 0) {
             newCapacity = minCapacity;
+        }
         if (newCapacity < 0) {
-            if (minCapacity < 0) // overflow
+            if (minCapacity < 0) {
                 throw new OutOfMemoryError();
+            }
             newCapacity = Integer.MAX_VALUE;
         }
         buf = Arrays.copyOf(buf, newCapacity);
@@ -110,4 +113,9 @@
     public int getLength() {
         return count;
     }
+
+    public void setSize(int bytesRequired) {
+        ensureCapacity(bytesRequired);
+        count = bytesRequired;
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/util/GrowableArray.java b/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/util/GrowableArray.java
index 6e329ab..39c9095 100644
--- a/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/util/GrowableArray.java
+++ b/hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/util/GrowableArray.java
@@ -67,4 +67,8 @@
     public void append(IValueReference value) throws IOException {
         dos.write(value.getByteArray(), value.getStartOffset(), value.getLength());
     }
+
+    public void setSize(int bytesRequired) {
+        baaos.setSize(bytesRequired);
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-hdfs/hyracks-hdfs-core/src/main/java/org/apache/hyracks/hdfs/lib/RawBinaryComparatorFactory.java b/hyracks-fullstack/hyracks/hyracks-hdfs/hyracks-hdfs-core/src/main/java/org/apache/hyracks/hdfs/lib/RawBinaryComparatorFactory.java
index fa0ecbb..01acddd 100644
--- a/hyracks-fullstack/hyracks/hyracks-hdfs/hyracks-hdfs-core/src/main/java/org/apache/hyracks/hdfs/lib/RawBinaryComparatorFactory.java
+++ b/hyracks-fullstack/hyracks/hyracks-hdfs/hyracks-hdfs-core/src/main/java/org/apache/hyracks/hdfs/lib/RawBinaryComparatorFactory.java
@@ -24,27 +24,25 @@
 public class RawBinaryComparatorFactory implements IBinaryComparatorFactory {
 
     private static final long serialVersionUID = 1L;
-    public static IBinaryComparatorFactory INSTANCE = new RawBinaryComparatorFactory();
+    private static final IBinaryComparator comparator = RawBinaryComparatorFactory::compare;
+    public static final IBinaryComparatorFactory INSTANCE = new RawBinaryComparatorFactory();
 
     private RawBinaryComparatorFactory() {
     }
 
     @Override
     public IBinaryComparator createBinaryComparator() {
-        return new IBinaryComparator() {
+        return comparator;
+    }
 
-            @Override
-            public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
-                int commonLength = Math.min(l1, l2);
-                for (int i = 0; i < commonLength; i++) {
-                    if (b1[s1 + i] != b2[s2 + i]) {
-                        return b1[s1 + i] - b2[s2 + i];
-                    }
-                }
-                int difference = l1 - l2;
-                return difference == 0 ? 0 : (difference > 0 ? 1 : -1);
+    public static final int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
+        int commonLength = Math.min(l1, l2);
+        for (int i = 0; i < commonLength; i++) {
+            if (b1[s1 + i] != b2[s2 + i]) {
+                return b1[s1 + i] - b2[s2 + i];
             }
-
-        };
+        }
+        int difference = l1 - l2;
+        return difference == 0 ? 0 : (difference > 0 ? 1 : -1);
     }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeFieldPrefixNSMLeafFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeFieldPrefixNSMLeafFrame.java
index 73447dd..157c663 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeFieldPrefixNSMLeafFrame.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeFieldPrefixNSMLeafFrame.java
@@ -56,15 +56,12 @@
  */
 public class BTreeFieldPrefixNSMLeafFrame implements IBTreeLeafFrame {
 
-    protected static final int pageLsnOff = 0; // 0
-    protected static final int tupleCountOff = pageLsnOff + 8; // 8
-    protected static final int freeSpaceOff = tupleCountOff + 4; // 12
-    protected static final int totalFreeSpaceOff = freeSpaceOff + 4; // 16
-    protected static final int levelOff = totalFreeSpaceOff + 4; // 20
-    protected static final int smFlagOff = levelOff + 1; // 21
-    protected static final int uncompressedTupleCountOff = smFlagOff + 1; // 22
-    protected static final int prefixTupleCountOff = uncompressedTupleCountOff + 4; // 26
-    protected static final int nextLeafOff = prefixTupleCountOff + 4; // 30
+    protected static final int PAGE_LSN_OFFSET = ITreeIndexFrame.Constants.RESERVED_HEADER_SIZE;
+    protected static final int TOTAL_FREE_SPACE_OFFSET = PAGE_LSN_OFFSET + 8;
+    protected static final int SM_FLAG_OFFSET = TOTAL_FREE_SPACE_OFFSET + 4;
+    protected static final int UNCOMPRESSED_TUPLE_COUNT_OFFSET = SM_FLAG_OFFSET + 1;
+    protected static final int PREFIX_TUPLE_COUNT_OFFSET = UNCOMPRESSED_TUPLE_COUNT_OFFSET + 4;
+    protected static final int NEXT_LEAF_OFFSET = PREFIX_TUPLE_COUNT_OFFSET + 4;
 
     private final IPrefixSlotManager slotManager;
     private final ITreeIndexFrameCompressor compressor;
@@ -132,11 +129,11 @@
     public boolean compact() {
         resetSpaceParams();
 
-        int tupleCount = buf.getInt(tupleCountOff);
+        int tupleCount = buf.getInt(ITreeIndexFrame.Constants.TUPLE_COUNT_OFFSET);
 
         // determine start of target free space (depends on assumptions stated above)
-        int freeSpace = buf.getInt(freeSpaceOff);
-        int prefixTupleCount = buf.getInt(prefixTupleCountOff);
+        int freeSpace = buf.getInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET);
+        int prefixTupleCount = buf.getInt(PREFIX_TUPLE_COUNT_OFFSET);
         if (prefixTupleCount > 0) {
 
             // debug
@@ -177,14 +174,16 @@
             int tupleLength = tupleEndOff - tupleOff;
             System.arraycopy(buf.array(), tupleOff, buf.array(), freeSpace, tupleLength);
 
-            slotManager.setSlot(sortedTupleOffs.get(i).slotOff, slotManager.encodeSlotFields(prefixSlotNum, freeSpace));
+            slotManager.setSlot(sortedTupleOffs.get(i).slotOff,
+                    slotManager.encodeSlotFields(prefixSlotNum, freeSpace));
             freeSpace += tupleLength;
         }
 
-        buf.putInt(freeSpaceOff, freeSpace);
-        int totalFreeSpace = buf.capacity() - buf.getInt(freeSpaceOff)
-                - ((buf.getInt(tupleCountOff) + buf.getInt(prefixTupleCountOff)) * slotManager.getSlotSize());
-        buf.putInt(totalFreeSpaceOff, totalFreeSpace);
+        buf.putInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET, freeSpace);
+        int totalFreeSpace = buf.capacity() - buf.getInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET)
+                - ((buf.getInt(ITreeIndexFrame.Constants.TUPLE_COUNT_OFFSET) + buf.getInt(PREFIX_TUPLE_COUNT_OFFSET))
+                        * slotManager.getSlotSize());
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, totalFreeSpace);
 
         return false;
     }
@@ -205,7 +204,7 @@
         int suffixFieldStart = 0;
         if (prefixSlotNum == FieldPrefixSlotManager.TUPLE_UNCOMPRESSED) {
             suffixFieldStart = 0;
-            buf.putInt(uncompressedTupleCountOff, buf.getInt(uncompressedTupleCountOff) - 1);
+            buf.putInt(UNCOMPRESSED_TUPLE_COUNT_OFFSET, buf.getInt(UNCOMPRESSED_TUPLE_COUNT_OFFSET) - 1);
         } else {
             int prefixSlot = buf.getInt(slotManager.getPrefixSlotOff(prefixSlotNum));
             suffixFieldStart = slotManager.decodeFirstSlotField(prefixSlot);
@@ -215,14 +214,17 @@
         tupleSize = tupleWriter.bytesRequired(frameTuple, suffixFieldStart, frameTuple.getFieldCount()
                 - suffixFieldStart);
 
-        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) - 1);
-        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + tupleSize + slotManager.getSlotSize());
+        buf.putInt(ITreeIndexFrame.Constants.TUPLE_COUNT_OFFSET, buf.getInt(
+                ITreeIndexFrame.Constants.TUPLE_COUNT_OFFSET) - 1);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.getInt(TOTAL_FREE_SPACE_OFFSET) + tupleSize
+                + slotManager.getSlotSize());
     }
 
     @Override
     public FrameOpSpaceStatus hasSpaceInsert(ITupleReference tuple) throws HyracksDataException {
-        int freeContiguous = buf.capacity() - buf.getInt(freeSpaceOff)
-                - ((buf.getInt(tupleCountOff) + buf.getInt(prefixTupleCountOff)) * slotManager.getSlotSize());
+        int freeContiguous = buf.capacity() - buf.getInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET)
+                - ((buf.getInt(ITreeIndexFrame.Constants.TUPLE_COUNT_OFFSET) + buf.getInt(PREFIX_TUPLE_COUNT_OFFSET))
+                        * slotManager.getSlotSize());
 
         int bytesRequired = tupleWriter.bytesRequired(tuple);
 
@@ -232,7 +234,7 @@
         }
 
         // See if tuple would fit into remaining space after compaction.
-        if (bytesRequired + slotManager.getSlotSize() <= buf.getInt(totalFreeSpaceOff)) {
+        if (bytesRequired + slotManager.getSlotSize() <= buf.getInt(TOTAL_FREE_SPACE_OFFSET)) {
             return FrameOpSpaceStatus.SUFFICIENT_SPACE;
         }
 
@@ -255,7 +257,7 @@
 
     @Override
     public void insert(ITupleReference tuple, int tupleIndex) {
-        int slot = slotManager.insertSlot(tupleIndex, buf.getInt(freeSpaceOff));
+        int slot = slotManager.insertSlot(tupleIndex, buf.getInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET));
         int prefixSlotNum = slotManager.decodeFirstSlotField(slot);
         int numPrefixFields = 0;
         if (prefixSlotNum != FieldPrefixSlotManager.TUPLE_UNCOMPRESSED) {
@@ -263,16 +265,19 @@
             int prefixSlot = buf.getInt(prefixSlotOff);
             numPrefixFields = slotManager.decodeFirstSlotField(prefixSlot);
         } else {
-            buf.putInt(uncompressedTupleCountOff, buf.getInt(uncompressedTupleCountOff) + 1);
+            buf.putInt(UNCOMPRESSED_TUPLE_COUNT_OFFSET, buf.getInt(UNCOMPRESSED_TUPLE_COUNT_OFFSET) + 1);
         }
 
-        int freeSpace = buf.getInt(freeSpaceOff);
+        int freeSpace = buf.getInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET);
         int bytesWritten = tupleWriter.writeTupleFields(tuple, numPrefixFields,
                 tuple.getFieldCount() - numPrefixFields, buf.array(), freeSpace);
 
-        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
-        buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + bytesWritten);
-        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());
+        buf.putInt(ITreeIndexFrame.Constants.TUPLE_COUNT_OFFSET, buf.getInt(
+                ITreeIndexFrame.Constants.TUPLE_COUNT_OFFSET) + 1);
+        buf.putInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET, buf.getInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET)
+                + bytesWritten);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.getInt(TOTAL_FREE_SPACE_OFFSET) - bytesWritten
+                - slotManager.getSlotSize());
     }
 
     @Override
@@ -301,15 +306,16 @@
             return FrameOpSpaceStatus.SUFFICIENT_INPLACE_SPACE;
         }
 
-        int freeContiguous = buf.capacity() - buf.getInt(freeSpaceOff)
-                - ((buf.getInt(tupleCountOff) + buf.getInt(prefixTupleCountOff)) * slotManager.getSlotSize());
+        int freeContiguous = buf.capacity() - buf.getInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET)
+                - ((buf.getInt(ITreeIndexFrame.Constants.TUPLE_COUNT_OFFSET) + buf.getInt(PREFIX_TUPLE_COUNT_OFFSET))
+                        * slotManager.getSlotSize());
 
         // Enough space if we delete the old tuple and insert the new one without compaction?
         if (newTupleBytes <= freeContiguous) {
             return FrameOpSpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
         }
         // Enough space if we delete the old tuple and compact?
-        if (additionalBytesRequired <= buf.getInt(totalFreeSpaceOff)) {
+        if (additionalBytesRequired <= buf.getInt(TOTAL_FREE_SPACE_OFFSET)) {
             return FrameOpSpaceStatus.SUFFICIENT_SPACE;
         }
         return FrameOpSpaceStatus.INSUFFICIENT_SPACE;
@@ -336,41 +342,41 @@
         } else {
             // Insert the new tuple suffix at the end of the free space, and change
             // the slot value (effectively "deleting" the old tuple).
-            int newSuffixTupleStartOff = buf.getInt(freeSpaceOff);
+            int newSuffixTupleStartOff = buf.getInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET);
             bytesWritten = tupleWriter.writeTupleFields(newTuple, numPrefixFields, fieldCount - numPrefixFields,
                     buf.array(), newSuffixTupleStartOff);
             // Update slot value using the same prefix slot num.
             slotManager.setSlot(tupleSlotOff, slotManager.encodeSlotFields(prefixSlotNum, newSuffixTupleStartOff));
             // Update contiguous free space pointer.
-            buf.putInt(freeSpaceOff, newSuffixTupleStartOff + bytesWritten);
+            buf.putInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET, newSuffixTupleStartOff + bytesWritten);
         }
-        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + oldTupleBytes - bytesWritten);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.getInt(TOTAL_FREE_SPACE_OFFSET) + oldTupleBytes - bytesWritten);
     }
 
     protected void resetSpaceParams() {
-        buf.putInt(freeSpaceOff, getOrigFreeSpaceOff());
-        buf.putInt(totalFreeSpaceOff, getOrigTotalFreeSpace());
+        buf.putInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET, getOrigFreeSpaceOff());
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, getOrigTotalFreeSpace());
     }
 
     @Override
     public void initBuffer(byte level) {
-        buf.putLong(pageLsnOff, 0);
+        buf.putLong(PAGE_LSN_OFFSET, 0);
         // during creation
-        buf.putInt(tupleCountOff, 0);
+        buf.putInt(ITreeIndexFrame.Constants.TUPLE_COUNT_OFFSET, 0);
         resetSpaceParams();
-        buf.putInt(uncompressedTupleCountOff, 0);
-        buf.putInt(prefixTupleCountOff, 0);
-        buf.put(levelOff, level);
-        buf.put(smFlagOff, (byte) 0);
-        buf.putInt(nextLeafOff, -1);
+        buf.putInt(UNCOMPRESSED_TUPLE_COUNT_OFFSET, 0);
+        buf.putInt(PREFIX_TUPLE_COUNT_OFFSET, 0);
+        buf.put(ITreeIndexFrame.Constants.LEVEL_OFFSET, level);
+        buf.put(SM_FLAG_OFFSET, (byte) 0);
+        buf.putInt(NEXT_LEAF_OFFSET, -1);
     }
 
     public void setTotalFreeSpace(int totalFreeSpace) {
-        buf.putInt(totalFreeSpaceOff, totalFreeSpace);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, totalFreeSpace);
     }
 
     public int getOrigTotalFreeSpace() {
-        return buf.capacity() - (nextLeafOff + 4);
+        return buf.capacity() - (NEXT_LEAF_OFFSET + 4);
     }
 
     @Override
@@ -437,7 +443,8 @@
         int tupleIndex = slotManager.decodeSecondSlotField(slot);
         // Error indicator is set if there is no exact match.
         if (tupleIndex == slotManager.getErrorIndicator()) {
-            throw new TreeIndexNonExistentKeyException("Trying to update a tuple with a nonexistent key in leaf node.");
+            throw new TreeIndexNonExistentKeyException(
+                    "Trying to update a tuple with a nonexistent key in leaf node.");
         }
         return slot;
     }
@@ -454,7 +461,8 @@
         int tupleIndex = slotManager.decodeSecondSlotField(slot);
         // Error indicator is set if there is no exact match.
         if (tupleIndex == slotManager.getErrorIndicator()) {
-            throw new TreeIndexNonExistentKeyException("Trying to delete a tuple with a nonexistent key in leaf node.");
+            throw new TreeIndexNonExistentKeyException(
+                    "Trying to delete a tuple with a nonexistent key in leaf node.");
         }
         return slot;
     }
@@ -462,21 +470,21 @@
     @Override
     public String printHeader() {
         StringBuilder strBuilder = new StringBuilder();
-        strBuilder.append("pageLsnOff:                " + pageLsnOff + "\n");
-        strBuilder.append("tupleCountOff:             " + tupleCountOff + "\n");
-        strBuilder.append("freeSpaceOff:              " + freeSpaceOff + "\n");
-        strBuilder.append("totalFreeSpaceOff:         " + totalFreeSpaceOff + "\n");
-        strBuilder.append("levelOff:                  " + levelOff + "\n");
-        strBuilder.append("smFlagOff:                 " + smFlagOff + "\n");
-        strBuilder.append("uncompressedTupleCountOff: " + uncompressedTupleCountOff + "\n");
-        strBuilder.append("prefixTupleCountOff:       " + prefixTupleCountOff + "\n");
-        strBuilder.append("nextLeafOff:               " + nextLeafOff + "\n");
+        strBuilder.append("pageLsnOff:                " + PAGE_LSN_OFFSET + "\n");
+        strBuilder.append("tupleCountOff:             " + ITreeIndexFrame.Constants.TUPLE_COUNT_OFFSET + "\n");
+        strBuilder.append("freeSpaceOff:              " + ITreeIndexFrame.Constants.FREE_SPACE_OFFSET + "\n");
+        strBuilder.append("totalFreeSpaceOff:         " + TOTAL_FREE_SPACE_OFFSET + "\n");
+        strBuilder.append("levelOff:                  " + ITreeIndexFrame.Constants.LEVEL_OFFSET + "\n");
+        strBuilder.append("smFlagOff:                 " + SM_FLAG_OFFSET + "\n");
+        strBuilder.append("uncompressedTupleCountOff: " + UNCOMPRESSED_TUPLE_COUNT_OFFSET + "\n");
+        strBuilder.append("prefixTupleCountOff:       " + PREFIX_TUPLE_COUNT_OFFSET + "\n");
+        strBuilder.append("nextLeafOff:               " + NEXT_LEAF_OFFSET + "\n");
         return strBuilder.toString();
     }
 
     @Override
     public int getTupleCount() {
-        return buf.getInt(tupleCountOff);
+        return buf.getInt(ITreeIndexFrame.Constants.TUPLE_COUNT_OFFSET);
     }
 
     @Override
@@ -493,50 +501,50 @@
 
     @Override
     public long getPageLsn() {
-        return buf.getLong(pageLsnOff);
+        return buf.getLong(PAGE_LSN_OFFSET);
     }
 
     @Override
     public void setPageLsn(long pageLsn) {
-        buf.putLong(pageLsnOff, pageLsn);
+        buf.putLong(PAGE_LSN_OFFSET, pageLsn);
     }
 
     @Override
     public int getTotalFreeSpace() {
-        return buf.getInt(totalFreeSpaceOff);
+        return buf.getInt(TOTAL_FREE_SPACE_OFFSET);
     }
 
     @Override
     public boolean isLeaf() {
-        return buf.get(levelOff) == 0;
+        return buf.get(ITreeIndexFrame.Constants.LEVEL_OFFSET) == 0;
     }
 
     @Override
     public boolean isInterior() {
-        return buf.get(levelOff) > 0;
+        return buf.get(ITreeIndexFrame.Constants.LEVEL_OFFSET) > 0;
     }
 
     @Override
     public byte getLevel() {
-        return buf.get(levelOff);
+        return buf.get(ITreeIndexFrame.Constants.LEVEL_OFFSET);
     }
 
     @Override
     public void setLevel(byte level) {
-        buf.put(levelOff, level);
+        buf.put(ITreeIndexFrame.Constants.LEVEL_OFFSET, level);
     }
 
     @Override
     public boolean getSmFlag() {
-        return buf.get(smFlagOff) != 0;
+        return buf.get(SM_FLAG_OFFSET) != 0;
     }
 
     @Override
     public void setSmFlag(boolean smFlag) {
         if (smFlag) {
-            buf.put(smFlagOff, (byte) 1);
+            buf.put(SM_FLAG_OFFSET, (byte) 1);
         } else {
-            buf.put(smFlagOff, (byte) 0);
+            buf.put(SM_FLAG_OFFSET, (byte) 0);
         }
     }
 
@@ -551,21 +559,21 @@
     }
 
     public int getPrefixTupleCount() {
-        return buf.getInt(prefixTupleCountOff);
+        return buf.getInt(PREFIX_TUPLE_COUNT_OFFSET);
     }
 
     public void setPrefixTupleCount(int prefixTupleCount) {
-        buf.putInt(prefixTupleCountOff, prefixTupleCount);
+        buf.putInt(PREFIX_TUPLE_COUNT_OFFSET, prefixTupleCount);
     }
 
     @Override
     public void insertSorted(ITupleReference tuple) throws HyracksDataException {
-        int freeSpace = buf.getInt(freeSpaceOff);
+        int freeSpace = buf.getInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET);
         int fieldsToTruncate = 0;
 
         // check if tuple matches last prefix tuple
-        if (buf.getInt(prefixTupleCountOff) > 0) {
-            framePrefixTuple.resetByTupleIndex(this, buf.getInt(prefixTupleCountOff) - 1);
+        if (buf.getInt(PREFIX_TUPLE_COUNT_OFFSET) > 0) {
+            framePrefixTuple.resetByTupleIndex(this, buf.getInt(PREFIX_TUPLE_COUNT_OFFSET) - 1);
             if (cmp.fieldRangeCompare(tuple, framePrefixTuple, 0, framePrefixTuple.getFieldCount()) == 0) {
                 fieldsToTruncate = framePrefixTuple.getFieldCount();
             }
@@ -577,17 +585,20 @@
         // insert slot
         int prefixSlotNum = FieldPrefixSlotManager.TUPLE_UNCOMPRESSED;
         if (fieldsToTruncate > 0) {
-            prefixSlotNum = buf.getInt(prefixTupleCountOff) - 1;
+            prefixSlotNum = buf.getInt(PREFIX_TUPLE_COUNT_OFFSET) - 1;
         } else {
-            buf.putInt(uncompressedTupleCountOff, buf.getInt(uncompressedTupleCountOff) + 1);
+            buf.putInt(UNCOMPRESSED_TUPLE_COUNT_OFFSET, buf.getInt(UNCOMPRESSED_TUPLE_COUNT_OFFSET) + 1);
         }
         int insSlot = slotManager.encodeSlotFields(prefixSlotNum, FieldPrefixSlotManager.GREATEST_KEY_INDICATOR);
         slotManager.insertSlot(insSlot, freeSpace);
 
         // update page metadata
-        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
-        buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + bytesWritten);
-        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());
+        buf.putInt(ITreeIndexFrame.Constants.TUPLE_COUNT_OFFSET, buf.getInt(
+                ITreeIndexFrame.Constants.TUPLE_COUNT_OFFSET) + 1);
+        buf.putInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET, buf.getInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET)
+                + bytesWritten);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.getInt(TOTAL_FREE_SPACE_OFFSET) - bytesWritten
+                - slotManager.getSlotSize());
     }
 
     @Override
@@ -688,8 +699,8 @@
         int length = rf.slotManager.getSlotSize() * tuplesToRight;
         System.arraycopy(right.array(), src, right.array(), dest, length);
 
-        right.putInt(tupleCountOff, tuplesToRight);
-        right.putInt(prefixTupleCountOff, prefixesToRight);
+        right.putInt(ITreeIndexFrame.Constants.TUPLE_COUNT_OFFSET, tuplesToRight);
+        right.putInt(PREFIX_TUPLE_COUNT_OFFSET, prefixesToRight);
 
         // on left page move slots to reflect possibly removed prefixes
         src = slotManager.getTupleSlotEndOff() + tuplesToRight * slotManager.getSlotSize();
@@ -698,8 +709,8 @@
         length = slotManager.getSlotSize() * tuplesToLeft;
         System.arraycopy(buf.array(), src, buf.array(), dest, length);
 
-        buf.putInt(tupleCountOff, tuplesToLeft);
-        buf.putInt(prefixTupleCountOff, prefixesToLeft);
+        buf.putInt(ITreeIndexFrame.Constants.TUPLE_COUNT_OFFSET, tuplesToLeft);
+        buf.putInt(PREFIX_TUPLE_COUNT_OFFSET, prefixesToLeft);
 
         // compact both pages
         compact();
@@ -721,39 +732,39 @@
         int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());
         splitKey.initData(splitKeySize);
         tupleWriter.writeTupleFields(frameTuple, 0, cmp.getKeyFieldCount(), splitKey.getBuffer().array(), 0);
-        splitKey.getTuple().resetByTupleOffset(splitKey.getBuffer(), 0);
+        splitKey.getTuple().resetByTupleOffset(splitKey.getBuffer().array(), 0);
     }
 
     @Override
     public int getFreeSpaceOff() {
-        return buf.getInt(freeSpaceOff);
+        return buf.getInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET);
     }
 
     public int getOrigFreeSpaceOff() {
-        return nextLeafOff + 4;
+        return NEXT_LEAF_OFFSET + 4;
     }
 
     @Override
     public void setFreeSpaceOff(int freeSpace) {
-        buf.putInt(freeSpaceOff, freeSpace);
+        buf.putInt(ITreeIndexFrame.Constants.FREE_SPACE_OFFSET, freeSpace);
     }
 
     @Override
     public void setNextLeaf(int page) {
-        buf.putInt(nextLeafOff, page);
+        buf.putInt(NEXT_LEAF_OFFSET, page);
     }
 
     @Override
     public int getNextLeaf() {
-        return buf.getInt(nextLeafOff);
+        return buf.getInt(NEXT_LEAF_OFFSET);
     }
 
     public int getUncompressedTupleCount() {
-        return buf.getInt(uncompressedTupleCountOff);
+        return buf.getInt(UNCOMPRESSED_TUPLE_COUNT_OFFSET);
     }
 
     public void setUncompressedTupleCount(int uncompressedTupleCount) {
-        buf.putInt(uncompressedTupleCountOff, uncompressedTupleCount);
+        buf.putInt(UNCOMPRESSED_TUPLE_COUNT_OFFSET, uncompressedTupleCount);
     }
 
     @Override
@@ -787,7 +798,7 @@
 
     @Override
     public int getPageHeaderSize() {
-        return nextLeafOff;
+        return NEXT_LEAF_OFFSET;
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeNSMInteriorFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeNSMInteriorFrame.java
index 9505f8a..e59523c 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeNSMInteriorFrame.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeNSMInteriorFrame.java
@@ -45,8 +45,8 @@
 
 public class BTreeNSMInteriorFrame extends TreeIndexNSMFrame implements IBTreeInteriorFrame {
 
-    private static final int rightLeafOff = flagOff + 1; // 22
-    private static final int childPtrSize = 4;
+    private static final int RIGHT_LEAF_OFFSET = TreeIndexNSMFrame.RESERVED_HEADER_SIZE;
+    private static final int CHILD_PTR_SIZE = 4;
 
     private final ITreeIndexTupleReference cmpFrameTuple;
     private final ITreeIndexTupleReference previousFt;
@@ -61,13 +61,13 @@
 
     @Override
     public int getBytesRequiredToWriteTuple(ITupleReference tuple) {
-        return tupleWriter.bytesRequired(tuple) + childPtrSize + slotManager.getSlotSize();
+        return tupleWriter.bytesRequired(tuple) + CHILD_PTR_SIZE + slotManager.getSlotSize();
     }
 
     @Override
     public void initBuffer(byte level) {
         super.initBuffer(level);
-        buf.putInt(rightLeafOff, -1);
+        buf.putInt(RIGHT_LEAF_OFFSET, -1);
     }
 
     @Override
@@ -82,7 +82,7 @@
 
     @Override
     public FrameOpSpaceStatus hasSpaceInsert(ITupleReference tuple) throws HyracksDataException {
-        int tupleSize = tupleWriter.bytesRequired(tuple) + childPtrSize;
+        int tupleSize = tupleWriter.bytesRequired(tuple) + CHILD_PTR_SIZE;
         if (tupleSize > getMaxTupleSize(buf.capacity())) {
             return FrameOpSpaceStatus.TOO_LARGE;
         }
@@ -99,28 +99,29 @@
 
     @Override
     public void insert(ITupleReference tuple, int tupleIndex) {
-        int slotOff = slotManager.insertSlot(tupleIndex, buf.getInt(freeSpaceOff));
-        int freeSpace = buf.getInt(freeSpaceOff);
+        int slotOff = slotManager.insertSlot(tupleIndex, buf.getInt(Constants.FREE_SPACE_OFFSET));
+        int freeSpace = buf.getInt(Constants.FREE_SPACE_OFFSET);
         int bytesWritten = tupleWriter.writeTupleFields(tuple, 0, tuple.getFieldCount(), buf.array(), freeSpace);
         System.arraycopy(tuple.getFieldData(tuple.getFieldCount() - 1), getLeftChildPageOff(tuple), buf.array(),
-                freeSpace + bytesWritten, childPtrSize);
-        int tupleSize = bytesWritten + childPtrSize;
-        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
-        buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + tupleSize);
-        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - tupleSize - slotManager.getSlotSize());
+                freeSpace + bytesWritten, CHILD_PTR_SIZE);
+        int tupleSize = bytesWritten + CHILD_PTR_SIZE;
+        buf.putInt(Constants.TUPLE_COUNT_OFFSET, buf.getInt(Constants.TUPLE_COUNT_OFFSET) + 1);
+        buf.putInt(Constants.FREE_SPACE_OFFSET, buf.getInt(Constants.FREE_SPACE_OFFSET) + tupleSize);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.getInt(TOTAL_FREE_SPACE_OFFSET) - tupleSize - slotManager
+                .getSlotSize());
         // Did we insert into the rightmost slot?
         if (slotOff == slotManager.getSlotEndOff()) {
-            System.arraycopy(tuple.getFieldData(tuple.getFieldCount() - 1), getLeftChildPageOff(tuple) + childPtrSize,
-                    buf.array(), rightLeafOff, childPtrSize);
+            System.arraycopy(tuple.getFieldData(tuple.getFieldCount() - 1), getLeftChildPageOff(tuple) + CHILD_PTR_SIZE,
+                    buf.array(), RIGHT_LEAF_OFFSET, CHILD_PTR_SIZE);
         } else {
             // If slotOff has a right (slot-)neighbor then update its child pointer.
             // The only time when this is NOT the case, is when this is the very first tuple
             // (or when the splitkey goes into the rightmost slot but that case is handled in the if above).
-            if (buf.getInt(tupleCountOff) > 1) {
+            if (buf.getInt(Constants.TUPLE_COUNT_OFFSET) > 1) {
                 int rightNeighborOff = slotOff - slotManager.getSlotSize();
-                frameTuple.resetByTupleOffset(buf, slotManager.getTupleOff(rightNeighborOff));
-                System.arraycopy(tuple.getFieldData(0), getLeftChildPageOff(tuple) + childPtrSize, buf.array(),
-                        getLeftChildPageOff(frameTuple), childPtrSize);
+                frameTuple.resetByTupleOffset(buf.array(), slotManager.getTupleOff(rightNeighborOff));
+                System.arraycopy(tuple.getFieldData(0), getLeftChildPageOff(tuple) + CHILD_PTR_SIZE, buf.array(),
+                        getLeftChildPageOff(frameTuple), CHILD_PTR_SIZE);
             }
         }
     }
@@ -142,13 +143,13 @@
         int keySize;
         if (tupleIndex == slotManager.getGreatestKeyIndicator()) {
             tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
-            frameTuple.resetByTupleOffset(buf, tupleOff);
+            frameTuple.resetByTupleOffset(buf.array(), tupleOff);
             keySize = frameTuple.getTupleSize();
             // Copy new rightmost pointer.
-            System.arraycopy(buf.array(), tupleOff + keySize, buf.array(), rightLeafOff, childPtrSize);
+            System.arraycopy(buf.array(), tupleOff + keySize, buf.array(), RIGHT_LEAF_OFFSET, CHILD_PTR_SIZE);
         } else {
             tupleOff = slotManager.getTupleOff(slotOff);
-            frameTuple.resetByTupleOffset(buf, tupleOff);
+            frameTuple.resetByTupleOffset(buf.array(), tupleOff);
             keySize = frameTuple.getTupleSize();
             // Perform deletion (we just do a memcpy to overwrite the slot).
             int slotStartOff = slotManager.getSlotEndOff();
@@ -156,25 +157,25 @@
             System.arraycopy(buf.array(), slotStartOff, buf.array(), slotStartOff + slotManager.getSlotSize(), length);
         }
         // Maintain space information.
-        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) - 1);
-        buf.putInt(totalFreeSpaceOff,
-                buf.getInt(totalFreeSpaceOff) + keySize + childPtrSize + slotManager.getSlotSize());
+        buf.putInt(Constants.TUPLE_COUNT_OFFSET, buf.getInt(Constants.TUPLE_COUNT_OFFSET) - 1);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET,
+                buf.getInt(TOTAL_FREE_SPACE_OFFSET) + keySize + CHILD_PTR_SIZE + slotManager.getSlotSize());
     }
 
     @Override
     public void deleteGreatest() {
         int slotOff = slotManager.getSlotEndOff();
         int tupleOff = slotManager.getTupleOff(slotOff);
-        frameTuple.resetByTupleOffset(buf, tupleOff);
+        frameTuple.resetByTupleOffset(buf.array(), tupleOff);
         int keySize = tupleWriter.bytesRequired(frameTuple);
-        System.arraycopy(buf.array(), tupleOff + keySize, buf.array(), rightLeafOff, childPtrSize);
+        System.arraycopy(buf.array(), tupleOff + keySize, buf.array(), RIGHT_LEAF_OFFSET, CHILD_PTR_SIZE);
         // Maintain space information.
-        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) - 1);
-        buf.putInt(totalFreeSpaceOff,
-                buf.getInt(totalFreeSpaceOff) + keySize + childPtrSize + slotManager.getSlotSize());
-        int freeSpace = buf.getInt(freeSpaceOff);
-        if (freeSpace == tupleOff + keySize + childPtrSize) {
-            buf.putInt(freeSpace, freeSpace - (keySize + childPtrSize));
+        buf.putInt(Constants.TUPLE_COUNT_OFFSET, buf.getInt(Constants.TUPLE_COUNT_OFFSET) - 1);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET,
+                buf.getInt(TOTAL_FREE_SPACE_OFFSET) + keySize + CHILD_PTR_SIZE + slotManager.getSlotSize());
+        int freeSpace = buf.getInt(Constants.FREE_SPACE_OFFSET);
+        if (freeSpace == tupleOff + keySize + CHILD_PTR_SIZE) {
+            buf.putInt(freeSpace, freeSpace - (keySize + CHILD_PTR_SIZE));
         }
     }
 
@@ -185,17 +186,19 @@
 
     @Override
     public void insertSorted(ITupleReference tuple) {
-        int freeSpace = buf.getInt(freeSpaceOff);
+        int freeSpace = buf.getInt(Constants.FREE_SPACE_OFFSET);
         slotManager.insertSlot(slotManager.getGreatestKeyIndicator(), freeSpace);
         int bytesWritten = tupleWriter.writeTuple(tuple, buf, freeSpace);
         System.arraycopy(tuple.getFieldData(tuple.getFieldCount() - 1), getLeftChildPageOff(tuple), buf.array(),
-                freeSpace + bytesWritten, childPtrSize);
-        int tupleSize = bytesWritten + childPtrSize;
-        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
-        buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + tupleSize);
-        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - tupleSize - slotManager.getSlotSize());
-        System.arraycopy(tuple.getFieldData(0), getLeftChildPageOff(tuple) + childPtrSize, buf.array(), rightLeafOff,
-                childPtrSize);
+                freeSpace + bytesWritten, CHILD_PTR_SIZE);
+        int tupleSize = bytesWritten + CHILD_PTR_SIZE;
+        buf.putInt(Constants.TUPLE_COUNT_OFFSET, buf.getInt(Constants.TUPLE_COUNT_OFFSET) + 1);
+        buf.putInt(Constants.FREE_SPACE_OFFSET, buf.getInt(Constants.FREE_SPACE_OFFSET) + tupleSize);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.getInt(TOTAL_FREE_SPACE_OFFSET) - tupleSize - slotManager
+                .getSlotSize());
+        System.arraycopy(tuple.getFieldData(0), getLeftChildPageOff(tuple) + CHILD_PTR_SIZE, buf.array(),
+                RIGHT_LEAF_OFFSET,
+                CHILD_PTR_SIZE);
     }
 
     @Override
@@ -219,7 +222,7 @@
             int i;
             for (i = 0; i < tupleCount; ++i) {
                 frameTuple.resetByTupleIndex(this, i);
-                totalSize += tupleWriter.bytesRequired(frameTuple) + childPtrSize + slotManager.getSlotSize();
+                totalSize += tupleWriter.bytesRequired(frameTuple) + CHILD_PTR_SIZE + slotManager.getSlotSize();
                 if (totalSize >= halfPageSize) {
                     break;
                 }
@@ -243,11 +246,11 @@
                     + tuplesToLeft * rightFrame.getSlotManager().getSlotSize();
             int length = rightFrame.getSlotManager().getSlotSize() * tuplesToRight;
             System.arraycopy(right.array(), src, right.array(), dest, length);
-            right.putInt(tupleCountOff, tuplesToRight);
+            right.putInt(Constants.TUPLE_COUNT_OFFSET, tuplesToRight);
 
             // On the left page, remove the highest key and make its child pointer
             // the rightmost child pointer.
-            buf.putInt(tupleCountOff, tuplesToLeft);
+            buf.putInt(Constants.TUPLE_COUNT_OFFSET, tuplesToLeft);
         }
         // Copy the split key to be inserted.
         // We must do so because setting the new split key will overwrite the
@@ -256,16 +259,16 @@
 
         // Set split key to be highest value in left page.
         int tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
-        frameTuple.resetByTupleOffset(buf, tupleOff);
+        frameTuple.resetByTupleOffset(buf.array(), tupleOff);
         int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());
         splitKey.initData(splitKeySize);
         tupleWriter.writeTuple(frameTuple, splitKey.getBuffer(), 0);
-        splitKey.getTuple().resetByTupleOffset(splitKey.getBuffer(), 0);
+        splitKey.getTuple().resetByTupleOffset(splitKey.getBuffer().array(), 0);
 
         int deleteTupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
-        frameTuple.resetByTupleOffset(buf, deleteTupleOff);
-        buf.putInt(rightLeafOff, buf.getInt(getLeftChildPageOff(frameTuple)));
-        buf.putInt(tupleCountOff, tuplesToLeft - 1);
+        frameTuple.resetByTupleOffset(buf.array(), deleteTupleOff);
+        buf.putInt(RIGHT_LEAF_OFFSET, buf.getInt(getLeftChildPageOff(frameTuple)));
+        buf.putInt(Constants.TUPLE_COUNT_OFFSET, tuplesToLeft - 1);
 
         // Compact both pages.
         rightFrame.compact();
@@ -284,10 +287,10 @@
     @Override
     public boolean compact() {
         resetSpaceParams();
-        int tupleCount = buf.getInt(tupleCountOff);
-        int freeSpace = buf.getInt(freeSpaceOff);
+        int tupleCount = buf.getInt(Constants.TUPLE_COUNT_OFFSET);
+        int freeSpace = buf.getInt(Constants.FREE_SPACE_OFFSET);
         // Sort the slots by the tuple offset they point to.
-        ArrayList<SlotOffTupleOff> sortedTupleOffs = new ArrayList<SlotOffTupleOff>();
+        ArrayList<SlotOffTupleOff> sortedTupleOffs = new ArrayList<>();
         sortedTupleOffs.ensureCapacity(tupleCount);
         for (int i = 0; i < tupleCount; i++) {
             int slotOff = slotManager.getSlotOff(i);
@@ -299,25 +302,25 @@
         // the left, reclaiming free space.
         for (int i = 0; i < sortedTupleOffs.size(); i++) {
             int tupleOff = sortedTupleOffs.get(i).tupleOff;
-            frameTuple.resetByTupleOffset(buf, tupleOff);
+            frameTuple.resetByTupleOffset(buf.array(), tupleOff);
             int tupleEndOff = frameTuple.getFieldStart(frameTuple.getFieldCount() - 1)
                     + frameTuple.getFieldLength(frameTuple.getFieldCount() - 1);
-            int tupleLength = tupleEndOff - tupleOff + childPtrSize;
+            int tupleLength = tupleEndOff - tupleOff + CHILD_PTR_SIZE;
             System.arraycopy(buf.array(), tupleOff, buf.array(), freeSpace, tupleLength);
             slotManager.setSlot(sortedTupleOffs.get(i).slotOff, freeSpace);
             freeSpace += tupleLength;
         }
         // Update contiguous free space pointer and total free space indicator.
-        buf.putInt(freeSpaceOff, freeSpace);
-        buf.putInt(totalFreeSpaceOff, buf.capacity() - freeSpace - tupleCount * slotManager.getSlotSize());
+        buf.putInt(Constants.FREE_SPACE_OFFSET, freeSpace);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.capacity() - freeSpace - tupleCount * slotManager.getSlotSize());
         return false;
     }
 
     @Override
     public int getChildPageId(RangePredicate pred) throws HyracksDataException {
         // Trivial case where there is only a child pointer (and no key).
-        if (buf.getInt(tupleCountOff) == 0) {
-            return buf.getInt(rightLeafOff);
+        if (buf.getInt(Constants.TUPLE_COUNT_OFFSET) == 0) {
+            return buf.getInt(RIGHT_LEAF_OFFSET);
         }
         // Trivial cases where no low key or high key was given (e.g. during an
         // index scan).
@@ -340,7 +343,7 @@
         int slotOff = slotManager.getSlotOff(tupleIndex);
         // Follow the rightmost (greatest) child pointer.
         if (tupleIndex == slotManager.getGreatestKeyIndicator()) {
-            return buf.getInt(rightLeafOff);
+            return buf.getInt(RIGHT_LEAF_OFFSET);
         }
         // Deal with prefix searches.
         // slotManager.findTupleIndex() will return an arbitrary tuple matching
@@ -348,7 +351,7 @@
         // To make sure we traverse the right path, we must find the
         // leftmost or rightmost tuple that matches the prefix.
         int origTupleOff = slotManager.getTupleOff(slotOff);
-        cmpFrameTuple.resetByTupleOffset(buf, origTupleOff);
+        cmpFrameTuple.resetByTupleOffset(buf.array(), origTupleOff);
         int cmpTupleOff = origTupleOff;
         // The answer set begins with the lowest key matching the prefix.
         // We must follow the child pointer of the lowest (leftmost) key
@@ -357,14 +360,14 @@
         slotOff += slotManager.getSlotSize();
         while (slotOff < maxSlotOff) {
             cmpTupleOff = slotManager.getTupleOff(slotOff);
-            frameTuple.resetByTupleOffset(buf, cmpTupleOff);
+            frameTuple.resetByTupleOffset(buf.array(), cmpTupleOff);
             if (targetCmp.compare(cmpFrameTuple, frameTuple) != 0) {
                 break;
             }
             slotOff += slotManager.getSlotSize();
         }
         slotOff -= slotManager.getSlotSize();
-        frameTuple.resetByTupleOffset(buf, slotManager.getTupleOff(slotOff));
+        frameTuple.resetByTupleOffset(buf.array(), slotManager.getTupleOff(slotOff));
         int childPageOff = getLeftChildPageOff(frameTuple);
 
         return buf.getInt(childPageOff);
@@ -373,24 +376,24 @@
     @Override
     public int getLeftmostChildPageId() {
         int tupleOff = slotManager.getTupleOff(slotManager.getSlotStartOff());
-        frameTuple.resetByTupleOffset(buf, tupleOff);
+        frameTuple.resetByTupleOffset(buf.array(), tupleOff);
         int childPageOff = getLeftChildPageOff(frameTuple);
         return buf.getInt(childPageOff);
     }
 
     @Override
     public int getRightmostChildPageId() {
-        return buf.getInt(rightLeafOff);
+        return buf.getInt(RIGHT_LEAF_OFFSET);
     }
 
     @Override
     public void setRightmostChildPageId(int pageId) {
-        buf.putInt(rightLeafOff, pageId);
+        buf.putInt(RIGHT_LEAF_OFFSET, pageId);
     }
 
     @Override
     public int getPageHeaderSize() {
-        return rightLeafOff + 4;
+        return RIGHT_LEAF_OFFSET + 4;
     }
 
     private int getLeftChildPageOff(ITupleReference tuple) {
@@ -414,25 +417,27 @@
 
     // For debugging.
     public ArrayList<Integer> getChildren(MultiComparator cmp) {
-        ArrayList<Integer> ret = new ArrayList<Integer>();
+        ArrayList<Integer> ret = new ArrayList<>();
         frameTuple.setFieldCount(cmp.getKeyFieldCount());
-        int tupleCount = buf.getInt(tupleCountOff);
+        int tupleCount = buf.getInt(Constants.TUPLE_COUNT_OFFSET);
         for (int i = 0; i < tupleCount; i++) {
             int tupleOff = slotManager.getTupleOff(slotManager.getSlotOff(i));
-            frameTuple.resetByTupleOffset(buf, tupleOff);
+            frameTuple.resetByTupleOffset(buf.array(), tupleOff);
             int intVal = IntegerPointable.getInteger(buf.array(),
                     frameTuple.getFieldStart(frameTuple.getFieldCount() - 1)
                             + frameTuple.getFieldLength(frameTuple.getFieldCount() - 1));
             ret.add(intVal);
         }
         if (!isLeaf()) {
-            int rightLeaf = buf.getInt(rightLeafOff);
-            if (rightLeaf > 0)
-                ret.add(buf.getInt(rightLeafOff));
+            int rightLeaf = buf.getInt(RIGHT_LEAF_OFFSET);
+            if (rightLeaf > 0) {
+                ret.add(buf.getInt(RIGHT_LEAF_OFFSET));
+            }
         }
         return ret;
     }
 
+    @Override
     public void validate(PageValidationInfo pvi) throws HyracksDataException {
         int tupleCount = getTupleCount();
         for (int i = 0; i < tupleCount; i++) {
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeNSMLeafFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeNSMLeafFrame.java
index f7af7ad..6a957b5 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeNSMLeafFrame.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeNSMLeafFrame.java
@@ -41,7 +41,7 @@
 import org.apache.hyracks.storage.common.buffercache.IExtraPageBlockHelper;
 
 public class BTreeNSMLeafFrame extends TreeIndexNSMFrame implements IBTreeLeafFrame {
-    protected static final int nextLeafOff = flagOff + 1; // 22
+    protected static final int NEXT_LEAF_OFFSET = TreeIndexNSMFrame.RESERVED_HEADER_SIZE;
 
     private MultiComparator cmp;
 
@@ -54,7 +54,7 @@
 
     @Override
     public int getPageHeaderSize() {
-        return nextLeafOff + 4;
+        return NEXT_LEAF_OFFSET + 4;
     }
 
     @Override
@@ -65,17 +65,17 @@
     @Override
     public void initBuffer(byte level) {
         super.initBuffer(level);
-        buf.putInt(nextLeafOff, -1);
+        buf.putInt(NEXT_LEAF_OFFSET, -1);
     }
 
     @Override
     public void setNextLeaf(int page) {
-        buf.putInt(nextLeafOff, page);
+        buf.putInt(NEXT_LEAF_OFFSET, page);
     }
 
     @Override
     public int getNextLeaf() {
-        return buf.getInt(nextLeafOff);
+        return buf.getInt(NEXT_LEAF_OFFSET);
     }
 
     @Override
@@ -161,12 +161,13 @@
 
     @Override
     public void insert(ITupleReference tuple, int tupleIndex) {
-        int freeSpace = buf.getInt(freeSpaceOff);
+        int freeSpace = buf.getInt(Constants.FREE_SPACE_OFFSET);
         slotManager.insertSlot(tupleIndex, freeSpace);
         int bytesWritten = tupleWriter.writeTuple(tuple, buf.array(), freeSpace);
-        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
-        buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + bytesWritten);
-        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());
+        buf.putInt(Constants.TUPLE_COUNT_OFFSET, buf.getInt(Constants.TUPLE_COUNT_OFFSET) + 1);
+        buf.putInt(Constants.FREE_SPACE_OFFSET, buf.getInt(Constants.FREE_SPACE_OFFSET) + bytesWritten);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.getInt(TOTAL_FREE_SPACE_OFFSET) - bytesWritten - slotManager
+                .getSlotSize());
     }
 
     @Override
@@ -261,10 +262,10 @@
                     + tuplesToLeft * rightFrame.getSlotManager().getSlotSize();
             int length = rightFrame.getSlotManager().getSlotSize() * tuplesToRight;
             System.arraycopy(right.array(), src, right.array(), dest, length);
-            right.putInt(tupleCountOff, tuplesToRight);
+            right.putInt(Constants.TUPLE_COUNT_OFFSET, tuplesToRight);
 
             // On left page only change the tupleCount indicator.
-            buf.putInt(tupleCountOff, tuplesToLeft);
+            buf.putInt(Constants.TUPLE_COUNT_OFFSET, tuplesToLeft);
 
             // Compact both pages.
             rightFrame.compact();
@@ -288,13 +289,14 @@
 
         // Set the split key to be highest key in the left page.
         int tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
-        frameTuple.resetByTupleOffset(buf, tupleOff);
+        frameTuple.resetByTupleOffset(buf.array(), tupleOff);
         int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());
         splitKey.initData(splitKeySize);
         tupleWriter.writeTupleFields(frameTuple, 0, cmp.getKeyFieldCount(), splitKey.getBuffer().array(), 0);
-        splitKey.getTuple().resetByTupleOffset(splitKey.getBuffer(), 0);
+        splitKey.getTuple().resetByTupleOffset(splitKey.getBuffer().array(), 0);
     }
 
+    @Override
     public void ensureCapacity(IBufferCache bufferCache, ITupleReference tuple,
                                IExtraPageBlockHelper extraPageBlockHelper) throws HyracksDataException {
         // we call ensureCapacity() for large tuples- ensure large flag is set
@@ -323,7 +325,8 @@
         System.arraycopy(buf.array(), oldSlotEnd, buf.array(), slotManager.getSlotEndOff(), oldSlotStart - oldSlotEnd);
 
         // fixup total free space counter
-        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + (bufferCache.getPageSize() * deltaPages));
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.getInt(TOTAL_FREE_SPACE_OFFSET) + (bufferCache.getPageSize()
+                * deltaPages));
     }
 
     @Override
@@ -342,6 +345,7 @@
         this.cmp = cmp;
     }
 
+    @Override
     public void validate(PageValidationInfo pvi) throws HyracksDataException {
         int tupleCount = getTupleCount();
         for (int i = 0; i < tupleCount; i++) {
@@ -364,7 +368,7 @@
     @Override
     public String printHeader() {
         StringBuilder strBuilder = new StringBuilder(super.printHeader());
-        strBuilder.append("nextLeafOff:       " + nextLeafOff + "\n");
+        strBuilder.append("nextLeafOff:       " + NEXT_LEAF_OFFSET + "\n");
         return strBuilder.toString();
     }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTree.java
index b6a64fa..5fdd943 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTree.java
@@ -1041,7 +1041,7 @@
                         splitKey.initData(splitKeySize);
                         tupleWriter.writeTupleFields(leafFrontier.lastTuple, 0, cmp.getKeyFieldCount(),
                                 splitKey.getBuffer().array(), 0);
-                        splitKey.getTuple().resetByTupleOffset(splitKey.getBuffer(), 0);
+                        splitKey.getTuple().resetByTupleOffset(splitKey.getBuffer().array(), 0);
                         splitKey.setLeftPage(leafFrontier.pageId);
 
                         propagateBulk(1, pagesToWrite);
@@ -1137,7 +1137,7 @@
                 splitKey.initData(splitKeySize);
                 tupleWriter.writeTupleFields(frontier.lastTuple, 0, cmp.getKeyFieldCount(),
                         splitKey.getBuffer().array(), 0);
-                splitKey.getTuple().resetByTupleOffset(splitKey.getBuffer(), 0);
+                splitKey.getTuple().resetByTupleOffset(splitKey.getBuffer().array(), 0);
 
                 ((IBTreeInteriorFrame) interiorFrame).deleteGreatest();
                 int finalPageId = freePageManager.takePage(metaFrame);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeOpContext.java
index 3de9a0c..3c5d0e5 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeOpContext.java
@@ -38,7 +38,7 @@
 import org.apache.hyracks.storage.am.common.api.ISearchOperationCallback;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexCursor;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrame;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexTupleReference;
 import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
 import org.apache.hyracks.storage.am.common.ophelpers.MultiComparator;
@@ -56,7 +56,7 @@
     public IBTreeLeafFrame leafFrame;
     public IBTreeInteriorFrame interiorFrame;
     public final IPageManager freePageManager;
-    public final ITreeIndexMetaDataFrame metaFrame;
+    public final ITreeIndexMetadataFrame metaFrame;
     public IndexOperation op;
     public ITreeIndexCursor cursor;
     public BTreeCursorInitialState cursorInitialState;
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeSplitKey.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeSplitKey.java
index f500a1a..4700ea3 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeSplitKey.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeSplitKey.java
@@ -35,6 +35,7 @@
         this.tuple = tuple;
     }
 
+    @Override
     public void initData(int keySize) {
         // try to reuse existing memory from a lower-level split if possible
         this.keySize = keySize;
@@ -48,49 +49,58 @@
             buf = ByteBuffer.wrap(data);
         }
 
-        tuple.resetByTupleOffset(buf, 0);
+        tuple.resetByTupleOffset(buf.array(), 0);
     }
 
+    @Override
     public void reset() {
         data = null;
         buf = null;
     }
 
+    @Override
     public ByteBuffer getBuffer() {
         return buf;
     }
 
+    @Override
     public ITreeIndexTupleReference getTuple() {
         return tuple;
     }
 
+    @Override
     public int getLeftPage() {
         return buf.getInt(keySize);
     }
 
+    @Override
     public int getRightPage() {
         return buf.getInt(keySize + 4);
     }
 
+    @Override
     public void setLeftPage(int leftPage) {
         buf.putInt(keySize, leftPage);
     }
 
+    @Override
     public void setRightPage(int rightPage) {
         buf.putInt(keySize + 4, rightPage);
     }
 
+    @Override
     public void setPages(int leftPage, int rightPage) {
         buf.putInt(keySize, leftPage);
         buf.putInt(keySize + 4, rightPage);
     }
 
+    @Override
     public BTreeSplitKey duplicate(ITreeIndexTupleReference copyTuple) {
         BTreeSplitKey copy = new BTreeSplitKey(copyTuple);
         copy.data = data.clone();
         copy.buf = ByteBuffer.wrap(copy.data);
         copy.tuple.setFieldCount(tuple.getFieldCount());
-        copy.tuple.resetByTupleOffset(copy.buf, 0);
+        copy.tuple.resetByTupleOffset(copy.buf.array(), 0);
         return copy;
     }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/FieldPrefixPrefixTupleReference.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/FieldPrefixPrefixTupleReference.java
index e110370..b767537 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/FieldPrefixPrefixTupleReference.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/FieldPrefixPrefixTupleReference.java
@@ -40,7 +40,7 @@
         int prefixSlot = concreteFrame.getBuffer().getInt(prefixSlotOff);
         setFieldCount(slotManager.decodeFirstSlotField(prefixSlot));
         tupleStartOff = slotManager.decodeSecondSlotField(prefixSlot);
-        buf = concreteFrame.getBuffer();
+        buf = concreteFrame.getBuffer().array();
         resetByTupleOffset(buf, tupleStartOff);
     }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/FieldPrefixTupleReference.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/FieldPrefixTupleReference.java
index a249fbb..566947d 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/FieldPrefixTupleReference.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/FieldPrefixTupleReference.java
@@ -18,8 +18,6 @@
  */
 package org.apache.hyracks.storage.am.btree.impls;
 
-import java.nio.ByteBuffer;
-
 import org.apache.hyracks.storage.am.btree.api.IPrefixSlotManager;
 import org.apache.hyracks.storage.am.btree.frames.BTreeFieldPrefixNSMLeafFrame;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
@@ -84,11 +82,11 @@
     public int getFieldLength(int fIdx) {
         if (fIdx < numPrefixFields) {
             helperTuple.setFieldCount(numPrefixFields);
-            helperTuple.resetByTupleOffset(frame.getBuffer(), prefixTupleStartOff);
+            helperTuple.resetByTupleOffset(frame.getBuffer().array(), prefixTupleStartOff);
             return helperTuple.getFieldLength(fIdx);
         } else {
             helperTuple.setFieldCount(numPrefixFields, fieldCount - numPrefixFields);
-            helperTuple.resetByTupleOffset(frame.getBuffer(), suffixTupleStartOff);
+            helperTuple.resetByTupleOffset(frame.getBuffer().array(), suffixTupleStartOff);
             return helperTuple.getFieldLength(fIdx - numPrefixFields);
         }
     }
@@ -97,18 +95,18 @@
     public int getFieldStart(int fIdx) {
         if (fIdx < numPrefixFields) {
             helperTuple.setFieldCount(numPrefixFields);
-            helperTuple.resetByTupleOffset(frame.getBuffer(), prefixTupleStartOff);
+            helperTuple.resetByTupleOffset(frame.getBuffer().array(), prefixTupleStartOff);
             return helperTuple.getFieldStart(fIdx);
         } else {
             helperTuple.setFieldCount(numPrefixFields, fieldCount - numPrefixFields);
-            helperTuple.resetByTupleOffset(frame.getBuffer(), suffixTupleStartOff);
+            helperTuple.resetByTupleOffset(frame.getBuffer().array(), suffixTupleStartOff);
             return helperTuple.getFieldStart(fIdx - numPrefixFields);
         }
     }
 
     // unsupported operation
     @Override
-    public void resetByTupleOffset(ByteBuffer buf, int tupleStartOffset) {
+    public void resetByTupleOffset(byte[] buf, int tupleStartOffset) {
         throw new UnsupportedOperationException("Resetting this type of frame by offset is not supported.");
     }
 
@@ -119,15 +117,16 @@
 
     public int getSuffixTupleSize() {
         helperTuple.setFieldCount(numPrefixFields, fieldCount - numPrefixFields);
-        helperTuple.resetByTupleOffset(frame.getBuffer(), suffixTupleStartOff);
+        helperTuple.resetByTupleOffset(frame.getBuffer().array(), suffixTupleStartOff);
         return helperTuple.getTupleSize();
     }
 
     public int getPrefixTupleSize() {
-        if (numPrefixFields == 0)
+        if (numPrefixFields == 0) {
             return 0;
+        }
         helperTuple.setFieldCount(numPrefixFields);
-        helperTuple.resetByTupleOffset(frame.getBuffer(), prefixTupleStartOff);
+        helperTuple.resetByTupleOffset(frame.getBuffer().array(), prefixTupleStartOff);
         return helperTuple.getTupleSize();
     }
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/pom.xml b/hyracks-fullstack/hyracks/hyracks-storage-am-common/pom.xml
index 7ebd27f..1d69439 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/pom.xml
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/pom.xml
@@ -80,5 +80,10 @@
       <artifactId>hyracks-data-std</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 </project>
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IMetadataPageManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IMetadataPageManager.java
index 02390c8..58c837b 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IMetadataPageManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IMetadataPageManager.java
@@ -19,55 +19,38 @@
 package org.apache.hyracks.storage.am.common.api;
 
 import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.common.buffercache.ICachedPage;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.data.std.api.IValueReference;
 
 /**
  * Used to read from and write to index metadata.
- * The index metadata contains information such as:
- * --The LSN of the index.
- * --Free page information {Set of free pages}
- * --Filter information.
- * TODO: This interface needs to change to have calls to request memory space and write to those bytes
+ * The index metadata contains key-value pairs
  */
 public interface IMetadataPageManager extends IPageManager {
-    public static class Constants {
-        public static final long INVALID_LSN_OFFSET = -1;
-        private Constants() {
-        }
-    }
-
     /**
-     * Locate the filter page in an index file
-     *
-     * @return The offset of the filter page if it exists, or less than zero if no filter page exists yet
+     * put the key value pair in the metadata page using the passed frame
+     * @param frame
+     * @param key
+     * @param value
      * @throws HyracksDataException
      */
-    int getFilterPageId() throws HyracksDataException;
-    void setFilterPageId(int filterPageId) throws HyracksDataException;
-    long getLSN() throws HyracksDataException;
-    void setLSN(long lsn) throws HyracksDataException;
+    void put(ITreeIndexMetadataFrame frame, IValueReference key, IValueReference value) throws HyracksDataException;
 
     /**
-     * Set the cached page to manage for filter data
-     *
-     * @param page
-     *            The page to manage
+     * get the value of the key from the metadata page using the passed frame
+     * @param frame
+     * @param key
+     * @param value
      * @throws HyracksDataException
      */
-    void setFilterPage(ICachedPage page) throws HyracksDataException;
+    void get(ITreeIndexMetadataFrame frame, IValueReference key, IPointable value) throws HyracksDataException;
 
     /**
-     * Get filter page if exists, create and return a new one if it doesn't
-     * @return
+     * @param frame
+     * @param key
+     * @return The byte offset in the index file for the entry with the passed key if the index is valid and the key
+     *         exists, returns -1 otherwise. use the passed frame to read the metadata page
      * @throws HyracksDataException
      */
-    ICachedPage getFilterPage() throws HyracksDataException;
-
-    /**
-     * @return The LSN byte offset in the LSM disk component if the index is valid,
-     *         otherwise {@link #INVALID_LSN_OFFSET}.
-     * @throws HyracksDataException
-     */
-    long getLSNOffset() throws HyracksDataException;
-    long getLastMarkerLSN() throws HyracksDataException;
+    long getFileOffset(ITreeIndexMetadataFrame frame, IValueReference key) throws HyracksDataException;
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IPageManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IPageManager.java
index 2aee403..b7987f8 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IPageManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IPageManager.java
@@ -50,7 +50,7 @@
      * Create a metadata frame to be used for reading and writing to metadata pages
      * @return a new metadata frame
      */
-    ITreeIndexMetaDataFrame createMetadataFrame();
+    ITreeIndexMetadataFrame createMetadataFrame();
 
     /**
      * Determines where the main metadata page is located in an index file
@@ -82,7 +82,7 @@
      * @return A page location, or -1 if no free page could be found or allocated
      * @throws HyracksDataException
      */
-    int takePage(ITreeIndexMetaDataFrame frame) throws HyracksDataException;
+    int takePage(ITreeIndexMetadataFrame frame) throws HyracksDataException;
 
     /**
      * Get the location of a block of free pages to use for index operations
@@ -92,7 +92,7 @@
      * @return The starting page location, or -1 if a block of free pages could be found or allocated
      * @throws HyracksDataException
      */
-    int takeBlock(ITreeIndexMetaDataFrame frame, int count) throws HyracksDataException;
+    int takeBlock(ITreeIndexMetadataFrame frame, int count) throws HyracksDataException;
 
     /**
      * Add a page back to the pool of free pages within an index file
@@ -103,7 +103,7 @@
      *            The page to be returned to the set of free pages
      * @throws HyracksDataException
      */
-    void releasePage(ITreeIndexMetaDataFrame frame, int page) throws HyracksDataException;
+    void releasePage(ITreeIndexMetadataFrame frame, int page) throws HyracksDataException;
 
     /**
      * Add a page back to the pool of free pages within an index file
@@ -115,7 +115,7 @@
      *            the number of regular sized pages in the free pages block
      * @throws HyracksDataException
      */
-    void releaseBlock(ITreeIndexMetaDataFrame frame, int page, int count) throws HyracksDataException;
+    void releaseBlock(ITreeIndexMetadataFrame frame, int page, int count) throws HyracksDataException;
 
     /**
      * Gets the highest page offset according to the metadata
@@ -125,7 +125,7 @@
      * @return The locaiton of the highest offset page
      * @throws HyracksDataException
      */
-    int getMaxPageId(ITreeIndexMetaDataFrame frame) throws HyracksDataException;
+    int getMaxPageId(ITreeIndexMetadataFrame frame) throws HyracksDataException;
 
     /**
      * Check whether the index is empty or not.
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexFrame.java
index 037c183..8e7834f 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexFrame.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexFrame.java
@@ -31,6 +31,21 @@
 
 public interface ITreeIndexFrame {
 
+    public static class Constants {
+        /*
+         * Storage version #. Change this if you alter any tree frame formats to stop
+         * possible corruption from old versions reading new formats.
+         */
+        public static final int VERSION = 5;
+        public static final int TUPLE_COUNT_OFFSET = 0;
+        public static final int FREE_SPACE_OFFSET = TUPLE_COUNT_OFFSET + 4;
+        public static final int LEVEL_OFFSET = FREE_SPACE_OFFSET + 4;
+        public static final int RESERVED_HEADER_SIZE = LEVEL_OFFSET + 1;
+
+        private Constants() {
+        }
+    }
+
     public void initBuffer(byte level);
 
     public FrameOpSpaceStatus hasSpaceInsert(ITupleReference tuple) throws HyracksDataException;
@@ -73,7 +88,7 @@
     public String printHeader();
 
     public void split(ITreeIndexFrame rightFrame, ITupleReference tuple, ISplitKey splitKey,
-                      IExtraPageBlockHelper extraPageBlockHelper, IBufferCache bufferCache)
+            IExtraPageBlockHelper extraPageBlockHelper, IBufferCache bufferCache)
             throws HyracksDataException, TreeIndexException;
 
     public ISlotManager getSlotManager();
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetaDataFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetaDataFrame.java
deleted file mode 100644
index f49972c..0000000
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetaDataFrame.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.hyracks.storage.am.common.api;
-
-import org.apache.hyracks.storage.common.buffercache.ICachedPage;
-
-public interface ITreeIndexMetaDataFrame {
-
-    // Storage version #. Change this if you alter any tree frame formats to stop
-    // possible corruption from old versions reading new formats.
-    int VERSION = 4;
-
-    public void initBuffer();
-
-    public void setPage(ICachedPage page);
-
-    public ICachedPage getPage();
-
-    public byte getLevel();
-
-    public void setLevel(byte level);
-
-    public int getNextMetadataPage();
-
-    public void setNextPage(int nextPage);
-
-    public int getMaxPage();
-
-    public void setMaxPage(int maxPage);
-
-    public int getFreePage();
-
-    public boolean hasSpace();
-
-    public void addFreePage(int freePage);
-
-    // Special flag for LSM-Components to mark whether they are valid or not.
-    public boolean isValid();
-
-    // Set special validity flag.
-    public void setValid(boolean isValid);
-
-    // Return the lsm component filter page id.
-    public int getLSMComponentFilterPageId();
-
-    // Set the lsm component filter page id.
-    public void setLSMComponentFilterPageId(int filterPage);
-
-    // Special placeholder for LSN information. Used for transactional LSM indexes.
-    public long getLSN();
-
-    public void setLSN(long lsn);
-
-    public int getVersion();
-
-    // Special placeholder for LSN information of a marker log. used for rollback information
-    public long getLastMarkerLSN();
-
-    public void setLastMarkerLSN(long lsn);
-
-    void setRootPageNumber(int rootPage);
-
-    int getRootPageNumber();
-
-    boolean isMetadataPage();
-
-    boolean isFreePage();
-}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetadataFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetadataFrame.java
new file mode 100644
index 0000000..7efc469
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetadataFrame.java
@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hyracks.storage.am.common.api;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.hyracks.storage.common.buffercache.ICachedPage;
+
+/**
+ * A frame for reading and writing metadata pages of an index
+ * It can also be used to read metadata of non metadata pages
+ */
+public interface ITreeIndexMetadataFrame {
+    /**
+     * initialize the metadata page
+     */
+    void init();
+
+    /**
+     * Set the page in the frame
+     * @param page
+     */
+    void setPage(ICachedPage page);
+
+    /**
+     * @return the page set in this frame
+     */
+    ICachedPage getPage();
+
+    /**
+     * The page level.
+     * This method can be run on any type of page
+     */
+    byte getLevel();
+
+    /**
+     * Set the page level
+     * @param level
+     */
+    void setLevel(byte level);
+
+    /**
+     * Get the next metadata page if this page is linked to other metadata pages
+     * Return a negative value otherwise
+     * @return
+     */
+    int getNextMetadataPage();
+
+    /**
+     * Link this metadata page to another one
+     * @param nextPage
+     */
+    void setNextMetadataPage(int nextPage);
+
+    /**
+     * @return the max index file page as indicated by the current metadata page
+     */
+    int getMaxPage();
+
+    /**
+     * Set the max page of the file
+     * @param maxPage
+     */
+    void setMaxPage(int maxPage);
+
+    /**
+     * Get a free page from the page
+     * @return
+     */
+    int getFreePage();
+
+    /**
+     * Get the remaining space in the metadata page
+     * @return
+     */
+    int getSpace();
+
+    /**
+     * add a new free page to the metadata page
+     * @param freePage
+     */
+    void addFreePage(int freePage);
+
+    /**
+     * get the value with the key = key
+     * @param key
+     * @param value
+     */
+    void get(IValueReference key, IPointable value);
+
+    /**
+     * set the value with the key = key
+     * @param key
+     * @param value
+     * @throws HyracksDataException
+     */
+    void put(IValueReference key, IValueReference value) throws HyracksDataException;
+
+    /**
+     * @return true if the index is valid according to the metadata page, false otherwise
+     */
+    boolean isValid();
+
+    /**
+     * Sets the index to be valid in the metadata page
+     * @param valid
+     */
+    void setValid(boolean valid);
+
+    /**
+     * Get the storage version associated with this index
+     * @return
+     */
+    int getVersion();
+
+    /**
+     * Set the index root page id
+     * @param rootPage
+     */
+    void setRootPageId(int rootPage);
+
+    /**
+     * @return the index root page id
+     */
+    int getRootPageId();
+
+    /**
+     * @return the number of key value pairs
+     */
+    int getTupleCount();
+
+    /**
+     * return the offset to the entry of the passed key, -1, otherwise
+     * @param key
+     */
+    int getOffset(IValueReference key);
+
+    /**
+     * @return true if the inspected page is a metadata page, false otherwise
+     */
+    boolean isMetadataPage();
+
+    /**
+     * @return true if the inspected page is a free page, false otherwise
+     */
+    boolean isFreePage();
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetaDataFrameFactory.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetadataFrameFactory.java
similarity index 88%
rename from hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetaDataFrameFactory.java
rename to hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetadataFrameFactory.java
index ff7deaf..7018743 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetaDataFrameFactory.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetadataFrameFactory.java
@@ -19,6 +19,7 @@
 
 package org.apache.hyracks.storage.am.common.api;
 
-public interface ITreeIndexMetaDataFrameFactory {
-    public ITreeIndexMetaDataFrame createFrame();
+@FunctionalInterface
+public interface ITreeIndexMetadataFrameFactory {
+    ITreeIndexMetadataFrame createFrame();
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexTupleReference.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexTupleReference.java
index 6b041eb..42f0ed5 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexTupleReference.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexTupleReference.java
@@ -19,8 +19,6 @@
 
 package org.apache.hyracks.storage.am.common.api;
 
-import java.nio.ByteBuffer;
-
 import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
 
 public interface ITreeIndexTupleReference extends ITupleReference {
@@ -28,7 +26,7 @@
 
     public void setFieldCount(int fieldStartIndex, int fieldCount);
 
-    public void resetByTupleOffset(ByteBuffer buf, int tupleStartOffset);
+    public void resetByTupleOffset(byte[] buf, int tupleStartOffset);
 
     public void resetByTupleIndex(ITreeIndexFrame frame, int tupleIndex);
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexDataflowHelper.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexDataflowHelper.java
index 5099df1..c6d4e35 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexDataflowHelper.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexDataflowHelper.java
@@ -30,7 +30,7 @@
 import org.apache.hyracks.storage.am.common.api.IIndexDataflowHelper;
 import org.apache.hyracks.storage.am.common.api.IPageManagerFactory;
 import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
-import org.apache.hyracks.storage.am.common.frames.LIFOMetaDataFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
 import org.apache.hyracks.storage.common.file.ILocalResourceFactory;
 import org.apache.hyracks.storage.common.file.ILocalResourceRepository;
 import org.apache.hyracks.storage.common.file.IResourceIdFactory;
@@ -99,7 +99,7 @@
                         .getLocalResourceFactory();
                 localResourceRepository.insert(localResourceFactory.createLocalResource(resourceID, resourceRef
                         .getRelativePath(),
-                        LIFOMetaDataFrame.VERSION, partition));
+                        ITreeIndexFrame.Constants.VERSION, partition));
             } catch (IOException e) {
                 throw new HyracksDataException(e);
             }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/LIFOMetaDataFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/LIFOMetaDataFrame.java
index 7d33519..de890c4 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/LIFOMetaDataFrame.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/LIFOMetaDataFrame.java
@@ -16,44 +16,40 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-
 package org.apache.hyracks.storage.am.common.frames;
 
 import java.nio.ByteBuffer;
 
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame.Constants;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrame;
 import org.apache.hyracks.storage.common.buffercache.ICachedPage;
 
-// all meta pages of this kind have a negative level
-// the first meta page has level -1, all other meta pages have level -2
-// the first meta page is special because it guarantees to have a correct max page
-// other meta pages (i.e., with level -2) have junk in the max page field
-
-public class LIFOMetaDataFrame implements ITreeIndexMetaDataFrame {
+/**
+ * Frame content
+ * [ Headers defined in {@link ITreeIndexFrame}][max page][next page][valid]
+ * [storage version][root page][free page count][k1 length][k1][v1 length][v1]
+ * [k2 length][k2][v2 length][v2].......
+ * ....
+ * ....
+ * [free page 5][free page 4][free page 3][free page 2][free page 1]
+ *
+ */
+public class LIFOMetaDataFrame implements ITreeIndexMetadataFrame {
 
     private static final byte META_PAGE_LEVEL_INDICATOR = -1;
     private static final byte FREE_PAGE_LEVEL_INDICATOR = -2;
-
-    // Arbitrarily chosen magic integer.
-    protected static final int OBSOLETE_MAGIC_VALID_INT = 0x5bd1e995;
     protected static final int MAGIC_VALID_INT = 0x1B16DA7A;
-
-    protected static final int TUPLE_COUNT_OFFSET = 0; //0
-    protected static final int FREE_SPACE_OFFSET = TUPLE_COUNT_OFFSET + 4; //4
-    protected static final int MAX_PAGE_OFFSET = FREE_SPACE_OFFSET + 4; //8
-    protected static final int LEVEL_OFFSET = MAX_PAGE_OFFSET + 12; //20
-    protected static final int NEXT_PAGE_OFFSET = LEVEL_OFFSET + 1; // 21
-    protected static final int VALID_OFFSET = NEXT_PAGE_OFFSET + 4; // 25
-
-    // The ADDITIONAL_FILTERING_PAGE_OFF is used only for LSM indexes.
-    // We store the page id that will be used to store the information of the the filter that is associated with a disk component.
-    // It is only set in the first meta page other meta pages (i.e., with level -2) have junk in the max page field.
-    private static final int ADDITIONAL_FILTERING_PAGE_OFFSET = VALID_OFFSET + 4; // 29
-    public static final int LSN_OFFSET = ADDITIONAL_FILTERING_PAGE_OFFSET + 4; // 33
-    private static final int LAST_MARKER_LSN_OFFSET = LSN_OFFSET + 8; // 41
-    public static final int STORAGE_VERSION_OFFSET = LAST_MARKER_LSN_OFFSET + 4; //45
-    public static final int ROOT_PAGE_NUMBER = STORAGE_VERSION_OFFSET + 4; //49
-    private static final int HEADER_END_OFFSET = ROOT_PAGE_NUMBER + 4; // 53
+    protected static final int MAX_PAGE_OFFSET = ITreeIndexFrame.Constants.RESERVED_HEADER_SIZE;
+    protected static final int NEXT_PAGE_OFFSET = MAX_PAGE_OFFSET + 4;
+    protected static final int VALID_OFFSET = NEXT_PAGE_OFFSET + 4;
+    protected static final int STORAGE_VERSION_OFFSET = VALID_OFFSET + 4;
+    protected static final int ROOT_PAGE_OFFSET = STORAGE_VERSION_OFFSET + 4;
+    protected static final int FREE_PAGE_COUNT_OFFSET = ROOT_PAGE_OFFSET + 4;
+    protected static final int HEADER_END_OFFSET = FREE_PAGE_COUNT_OFFSET + 4;
 
     protected ICachedPage page = null;
     protected ByteBuffer buf = null;
@@ -70,44 +66,35 @@
 
     @Override
     public int getFreePage() {
-        int tupleCount = buf.getInt(TUPLE_COUNT_OFFSET);
-        if (tupleCount > 0) {
-            // return the last page from the linked list of free pages
-            // TODO: this is a dumb policy, but good enough for now
-            int lastPageOff = buf.getInt(FREE_SPACE_OFFSET) - 4;
-            buf.putInt(FREE_SPACE_OFFSET, lastPageOff);
-            buf.putInt(TUPLE_COUNT_OFFSET, tupleCount - 1);
-            return buf.getInt(lastPageOff);
-        } else {
-            return -1;
+        int freePages = buf.getInt(FREE_PAGE_COUNT_OFFSET);
+        if (freePages > 0) {
+            decrement(FREE_PAGE_COUNT_OFFSET);
+            return buf.getInt(buf.array().length - Integer.BYTES * freePages);
         }
+        return -1;
     }
 
-    // must be checked before adding free page
-    // user of this class is responsible for getting a free page as a new meta
-    // page, latching it, etc. if there is no space on this page
     @Override
-    public boolean hasSpace() {
-        return buf.getInt(FREE_SPACE_OFFSET) + 4 < buf.capacity();
+    public int getSpace() {
+        return buf.array().length - buf.getInt(Constants.FREE_SPACE_OFFSET) - (Integer.BYTES * buf.getInt(
+                FREE_PAGE_COUNT_OFFSET));
     }
 
-    // no bounds checking is done, there must be free space
     @Override
     public void addFreePage(int freePage) {
-        int freeSpace = buf.getInt(FREE_SPACE_OFFSET);
-        buf.putInt(freeSpace, freePage);
-        buf.putInt(FREE_SPACE_OFFSET, freeSpace + 4);
-        buf.putInt(TUPLE_COUNT_OFFSET, buf.getInt(TUPLE_COUNT_OFFSET) + 1);
+        increment(FREE_PAGE_COUNT_OFFSET);
+        int numFreePages = buf.getInt(FREE_PAGE_COUNT_OFFSET);
+        buf.putInt(buf.array().length - (Integer.BYTES * numFreePages), freePage);
     }
 
     @Override
     public byte getLevel() {
-        return buf.get(LEVEL_OFFSET);
+        return buf.get(Constants.LEVEL_OFFSET);
     }
 
     @Override
     public void setLevel(byte level) {
-        buf.put(LEVEL_OFFSET, level);
+        buf.put(Constants.LEVEL_OFFSET, level);
     }
 
     @Override
@@ -122,16 +109,15 @@
     }
 
     @Override
-    public void initBuffer() {
-        buf.putInt(TUPLE_COUNT_OFFSET, 0);
-        buf.putInt(FREE_SPACE_OFFSET, HEADER_END_OFFSET);
+    public void init() {
+        buf.putInt(Constants.TUPLE_COUNT_OFFSET, 0);
+        buf.putInt(Constants.FREE_SPACE_OFFSET, HEADER_END_OFFSET);
         buf.putInt(MAX_PAGE_OFFSET, 0);
-        buf.put(LEVEL_OFFSET, META_PAGE_LEVEL_INDICATOR);
+        buf.put(Constants.LEVEL_OFFSET, META_PAGE_LEVEL_INDICATOR);
         buf.putInt(NEXT_PAGE_OFFSET, -1);
-        buf.putInt(ADDITIONAL_FILTERING_PAGE_OFFSET, -1);
-        buf.putLong(LAST_MARKER_LSN_OFFSET, -1L);
-        buf.putInt(ROOT_PAGE_NUMBER, 1);
-        buf.putInt(STORAGE_VERSION_OFFSET, VERSION);
+        buf.putInt(ROOT_PAGE_OFFSET, 1);
+        buf.putInt(FREE_PAGE_COUNT_OFFSET, 0);
+        buf.putInt(STORAGE_VERSION_OFFSET, ITreeIndexFrame.Constants.VERSION);
         setValid(false);
     }
 
@@ -141,71 +127,33 @@
     }
 
     @Override
-    public void setNextPage(int nextPage) {
+    public void setNextMetadataPage(int nextPage) {
         buf.putInt(NEXT_PAGE_OFFSET, nextPage);
     }
 
     @Override
     public boolean isValid() {
-        return buf.getInt(VALID_OFFSET) == MAGIC_VALID_INT || buf.getInt(VALID_OFFSET) == OBSOLETE_MAGIC_VALID_INT;
+        return buf.getInt(VALID_OFFSET) == MAGIC_VALID_INT;
     }
 
     @Override
     public void setValid(boolean isValid) {
-        if (isValid) {
-            buf.putInt(VALID_OFFSET, MAGIC_VALID_INT);
-        } else {
-            buf.putInt(VALID_OFFSET, 0);
-        }
-    }
-
-    @Override
-    public long getLSN() {
-        return buf.getLong(LSN_OFFSET);
-    }
-
-    @Override
-    public void setLSN(long lsn) {
-        buf.putLong(LSN_OFFSET, lsn);
+        buf.putInt(VALID_OFFSET, isValid ? MAGIC_VALID_INT : 0);
     }
 
     @Override
     public int getVersion() {
-        if (buf.getInt(VALID_OFFSET) == OBSOLETE_MAGIC_VALID_INT) {
-            return VERSION * -1;
-        } else {
-            return buf.getInt(STORAGE_VERSION_OFFSET);
-        }
+        return buf.getInt(STORAGE_VERSION_OFFSET);
     }
 
     @Override
-    public long getLastMarkerLSN() {
-        return buf.getLong(LAST_MARKER_LSN_OFFSET);
+    public void setRootPageId(int rootPage) {
+        buf.putInt(ROOT_PAGE_OFFSET, rootPage);
     }
 
     @Override
-    public void setLastMarkerLSN(long lsn) {
-        buf.putLong(LAST_MARKER_LSN_OFFSET, lsn);
-    }
-
-    @Override
-    public int getLSMComponentFilterPageId() {
-        return buf.getInt(ADDITIONAL_FILTERING_PAGE_OFFSET);
-    }
-
-    @Override
-    public void setLSMComponentFilterPageId(int filterPage) {
-        buf.putInt(ADDITIONAL_FILTERING_PAGE_OFFSET, filterPage);
-    }
-
-    @Override
-    public void setRootPageNumber(int rootPage) {
-        buf.putInt(ROOT_PAGE_NUMBER, rootPage);
-    }
-
-    @Override
-    public int getRootPageNumber() {
-        return buf.getInt(ROOT_PAGE_NUMBER);
+    public int getRootPageId() {
+        return buf.getInt(ROOT_PAGE_OFFSET);
     }
 
     @Override
@@ -217,4 +165,153 @@
     public boolean isFreePage() {
         return getLevel() == FREE_PAGE_LEVEL_INDICATOR;
     }
+
+    @Override
+    public void get(IValueReference key, IPointable value) {
+        int tupleCount = getTupleCount();
+        int tupleStart = getTupleStart(0);
+        for (int i = 0; i < tupleCount; i++) {
+            if (isInner(key, tupleStart)) {
+                get(tupleStart + key.getLength() + Integer.BYTES, value);
+                return;
+            }
+            tupleStart = getNextTupleStart(tupleStart);
+        }
+        value.set(null, 0, 0);
+    }
+
+    private int find(IValueReference key) {
+        int tupleCount = getTupleCount();
+        int tupleStart = getTupleStart(0);
+        for (int i = 0; i < tupleCount; i++) {
+            if (isInner(key, tupleStart)) {
+                return i;
+            }
+            tupleStart = getNextTupleStart(tupleStart);
+        }
+        return -1;
+    }
+
+    private void get(int offset, IPointable value) {
+        int valueLength = buf.getInt(offset);
+        value.set(buf.array(), offset + Integer.BYTES, valueLength);
+    }
+
+    private static final int compare(byte[] b1, int s1, byte[] b2, int s2, int l) {
+        for (int i = 0; i < l; i++) {
+            if (b1[s1 + i] != b2[s2 + i]) {
+                return b1[s1 + i] - b2[s2 + i];
+            }
+        }
+        return 0;
+    }
+
+    private boolean isInner(IValueReference key, int tupleOffset) {
+        int keySize = buf.getInt(tupleOffset);
+        if (keySize == key.getLength()) {
+            return LIFOMetaDataFrame.compare(key.getByteArray(), key.getStartOffset(), buf.array(), tupleOffset
+                    + Integer.BYTES, keySize) == 0;
+        }
+        return false;
+    }
+
+    private int getTupleStart(int index) {
+        int offset = HEADER_END_OFFSET;
+        int i = 0;
+        while (i < index) {
+            i++;
+            offset = getNextTupleStart(offset);
+        }
+        return offset;
+    }
+
+    private int getNextTupleStart(int prevTupleOffset) {
+        int keyLength = buf.getInt(prevTupleOffset);
+        int offset = prevTupleOffset + keyLength + Integer.BYTES;
+        return offset + buf.getInt(offset) + Integer.BYTES;
+    }
+
+    private void put(int index, IValueReference value) throws HyracksDataException {
+        int offset = getTupleStart(index);
+        int length = buf.getInt(offset);
+        offset += Integer.BYTES + length;
+        length = buf.getInt(offset);
+        if (length != value.getLength()) {
+            throw new HyracksDataException("This frame doesn't support overwriting dynamically sized values");
+        }
+        offset += Integer.BYTES;
+        System.arraycopy(value.getByteArray(), value.getStartOffset(), buf.array(), offset, value.getLength());
+    }
+
+    @Override
+    public void put(IValueReference key, IValueReference value) throws HyracksDataException {
+        int index = find(key);
+        if (index >= 0) {
+            put(index, value);
+        } else {
+            int offset = buf.getInt(Constants.FREE_SPACE_OFFSET);
+            int available = getSpace();
+            int required = key.getLength() + Integer.BYTES + Integer.BYTES + value.getLength();
+            if (available < required) {
+                throw new HyracksDataException("Available space in the page ("
+                        + available + ") is not enough to store the key value pair(" + required + ")");
+            }
+            buf.putInt(offset, key.getLength());
+            offset += Integer.BYTES;
+            System.arraycopy(key.getByteArray(), key.getStartOffset(), buf.array(), offset, key.getLength());
+            offset += key.getLength();
+            buf.putInt(offset, value.getLength());
+            offset += Integer.BYTES;
+            System.arraycopy(value.getByteArray(), value.getStartOffset(), buf.array(), offset, value.getLength());
+            offset += value.getLength();
+            increment(Constants.TUPLE_COUNT_OFFSET);
+            buf.putInt(Constants.FREE_SPACE_OFFSET, offset);
+        }
+    }
+
+    @Override
+    public int getTupleCount() {
+        return buf.getInt(Constants.TUPLE_COUNT_OFFSET);
+    }
+
+    private void increment(int offset) {
+        buf.putInt(offset, buf.getInt(offset) + 1);
+    }
+
+    private void decrement(int offset) {
+        buf.putInt(offset, buf.getInt(offset) - 1);
+    }
+
+    @Override
+    public int getOffset(IValueReference key) {
+        int index = find(key);
+        if (index >= 0) {
+            int offset = getTupleStart(index);
+            return offset + key.getLength() + 2 * Integer.BYTES;
+        }
+        return -1;
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder aString = new StringBuilder(this.getClass().getSimpleName()).append('\n').
+                append("Tuple Count: " + getTupleCount()).append('\n').
+                append("Free Space offset: " + buf.getInt(Constants.FREE_SPACE_OFFSET)).append('\n').
+                append("Level: " + buf.get(Constants.LEVEL_OFFSET)).append('\n').
+                append("Version: " + buf.getInt(STORAGE_VERSION_OFFSET)).append('\n').
+                append("Max Page: " + buf.getInt(MAX_PAGE_OFFSET)).append('\n').
+                append("Root Page: " + buf.getInt(ROOT_PAGE_OFFSET)).append('\n').
+                append("Number of free pages: " + buf.getInt(FREE_PAGE_COUNT_OFFSET));
+        int tupleCount = getTupleCount();
+        int offset;
+        for (int i = 0; i < tupleCount; i++) {
+            offset = getTupleStart(i);
+            int keyLength = buf.getInt(offset);
+            aString.append('\n').append("Key " + i + " size = " + keyLength);
+            offset += Integer.BYTES + keyLength;
+            int valueLength = buf.getInt(offset);
+            aString.append(", Value " + i + " size = " + valueLength);
+        }
+        return aString.toString();
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/LIFOMetaDataFrameFactory.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/LIFOMetaDataFrameFactory.java
index b0406c2..4dbba02 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/LIFOMetaDataFrameFactory.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/LIFOMetaDataFrameFactory.java
@@ -18,12 +18,12 @@
  */
 package org.apache.hyracks.storage.am.common.frames;
 
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrameFactory;
 
-public class LIFOMetaDataFrameFactory implements ITreeIndexMetaDataFrameFactory {
+public class LIFOMetaDataFrameFactory implements ITreeIndexMetadataFrameFactory {
     @Override
-    public ITreeIndexMetaDataFrame createFrame() {
+    public ITreeIndexMetadataFrame createFrame() {
         return new LIFOMetaDataFrame();
     }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/TreeIndexNSMFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/TreeIndexNSMFrame.java
index 47530a8..0948704 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/TreeIndexNSMFrame.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/TreeIndexNSMFrame.java
@@ -34,15 +34,12 @@
 
 public abstract class TreeIndexNSMFrame implements ITreeIndexFrame {
 
-    protected static final int pageLsnOff = 0; // 0
-    protected static final int tupleCountOff = pageLsnOff + 8; // 8
-    protected static final int freeSpaceOff = tupleCountOff + 4; // 12
-    protected static final int totalFreeSpaceOff = freeSpaceOff + 4; // 16
-    protected static final int levelOff = totalFreeSpaceOff + 4; // 20
-    protected static final int flagOff = levelOff + 1; // 21
-
-    protected static final byte smFlagBit           = 0x1;
-    protected static final byte largeFlagBit        = 0x2;
+    protected static final int PAGE_LSN_OFFSET = Constants.RESERVED_HEADER_SIZE;
+    protected static final int TOTAL_FREE_SPACE_OFFSET = PAGE_LSN_OFFSET + 8;
+    protected static final int FLAG_OFFSET = TOTAL_FREE_SPACE_OFFSET + 4;
+    protected static final int RESERVED_HEADER_SIZE = FLAG_OFFSET + 1;
+    protected static final byte SMALL_FLAG_BIT = 0x1;
+    protected static final byte LARGE_FLAG_BIT = 0x2;
 
     protected ICachedPage page = null;
     protected ByteBuffer buf = null;
@@ -60,12 +57,12 @@
 
     @Override
     public void initBuffer(byte level) {
-        buf.putLong(pageLsnOff, 0); // TODO: might to set to a different lsn
+        buf.putLong(PAGE_LSN_OFFSET, 0); // TODO: might to set to a different lsn
         // during creation
-        buf.putInt(tupleCountOff, 0);
+        buf.putInt(Constants.TUPLE_COUNT_OFFSET, 0);
         resetSpaceParams();
-        buf.put(levelOff, level);
-        buf.put(flagOff, (byte) 0);
+        buf.put(Constants.LEVEL_OFFSET, level);
+        buf.put(FLAG_OFFSET, (byte) 0);
     }
 
     @Override
@@ -75,56 +72,56 @@
 
     @Override
     public boolean isLeaf() {
-        return buf.get(levelOff) == 0;
+        return buf.get(Constants.LEVEL_OFFSET) == 0;
     }
 
     public boolean getSmFlag() {
-        return (buf.get(flagOff) & smFlagBit) != 0;
+        return (buf.get(FLAG_OFFSET) & SMALL_FLAG_BIT) != 0;
     }
 
     public void setSmFlag(boolean smFlag) {
         if (smFlag) {
-            buf.put(flagOff, (byte) (buf.get(flagOff) | smFlagBit));
+            buf.put(FLAG_OFFSET, (byte) (buf.get(FLAG_OFFSET) | SMALL_FLAG_BIT));
         } else {
-            buf.put(flagOff, (byte) (buf.get(flagOff) & ~smFlagBit));
+            buf.put(FLAG_OFFSET, (byte) (buf.get(FLAG_OFFSET) & ~SMALL_FLAG_BIT));
         }
     }
 
     public void setLargeFlag(boolean largeFlag) {
         if (largeFlag) {
-            buf.put(flagOff, (byte) (buf.get(flagOff) | largeFlagBit));
+            buf.put(FLAG_OFFSET, (byte) (buf.get(FLAG_OFFSET) | LARGE_FLAG_BIT));
         } else {
-            buf.put(flagOff, (byte) (buf.get(flagOff) & ~largeFlagBit));
+            buf.put(FLAG_OFFSET, (byte) (buf.get(FLAG_OFFSET) & ~LARGE_FLAG_BIT));
         }
     }
 
     public boolean getLargeFlag() {
-        return (buf.get(flagOff) & largeFlagBit) != 0;
+        return (buf.get(FLAG_OFFSET) & LARGE_FLAG_BIT) != 0;
     }
 
     @Override
     public boolean isInterior() {
-        return buf.get(levelOff) > 0;
+        return buf.get(Constants.LEVEL_OFFSET) > 0;
     }
 
     @Override
     public byte getLevel() {
-        return buf.get(levelOff);
+        return buf.get(Constants.LEVEL_OFFSET);
     }
 
     @Override
     public void setLevel(byte level) {
-        buf.put(levelOff, level);
+        buf.put(Constants.LEVEL_OFFSET, level);
     }
 
     @Override
     public int getFreeSpaceOff() {
-        return buf.getInt(freeSpaceOff);
+        return buf.getInt(Constants.FREE_SPACE_OFFSET);
     }
 
     @Override
     public void setFreeSpaceOff(int freeSpace) {
-        buf.putInt(freeSpaceOff, freeSpace);
+        buf.putInt(Constants.FREE_SPACE_OFFSET, freeSpace);
     }
 
     @Override
@@ -146,8 +143,8 @@
     @Override
     public boolean compact() {
         resetSpaceParams();
-        int tupleCount = buf.getInt(tupleCountOff);
-        int freeSpace = buf.getInt(freeSpaceOff);
+        int tupleCount = buf.getInt(Constants.TUPLE_COUNT_OFFSET);
+        int freeSpace = buf.getInt(Constants.FREE_SPACE_OFFSET);
         // Sort the slots by the tuple offset they point to.
         ArrayList<SlotOffTupleOff> sortedTupleOffs = new ArrayList<>();
         sortedTupleOffs.ensureCapacity(tupleCount);
@@ -161,7 +158,7 @@
         // the left, reclaiming free space.
         for (int i = 0; i < sortedTupleOffs.size(); i++) {
             int tupleOff = sortedTupleOffs.get(i).tupleOff;
-            frameTuple.resetByTupleOffset(buf, tupleOff);
+            frameTuple.resetByTupleOffset(buf.array(), tupleOff);
             int tupleEndOff = frameTuple.getFieldStart(frameTuple.getFieldCount() - 1)
                     + frameTuple.getFieldLength(frameTuple.getFieldCount() - 1);
             int tupleLength = tupleEndOff - tupleOff;
@@ -170,8 +167,8 @@
             freeSpace += tupleLength;
         }
         // Update contiguous free space pointer and total free space indicator.
-        buf.putInt(freeSpaceOff, freeSpace);
-        buf.putInt(totalFreeSpaceOff, buf.capacity() - freeSpace - tupleCount * slotManager.getSlotSize());
+        buf.putInt(Constants.FREE_SPACE_OFFSET, freeSpace);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.capacity() - freeSpace - tupleCount * slotManager.getSlotSize());
         return false;
     }
 
@@ -179,7 +176,7 @@
     public void delete(ITupleReference tuple, int tupleIndex) {
         int slotOff = slotManager.getSlotOff(tupleIndex);
         int tupleOff = slotManager.getTupleOff(slotOff);
-        frameTuple.resetByTupleOffset(buf, tupleOff);
+        frameTuple.resetByTupleOffset(buf.array(), tupleOff);
         int tupleSize = tupleWriter.bytesRequired(frameTuple);
 
         // perform deletion (we just do a memcpy to overwrite the slot)
@@ -188,20 +185,21 @@
         System.arraycopy(buf.array(), slotStartOff, buf.array(), slotStartOff + slotManager.getSlotSize(), length);
 
         // maintain space information
-        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) - 1);
-        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + tupleSize + slotManager.getSlotSize());
+        buf.putInt(Constants.TUPLE_COUNT_OFFSET, buf.getInt(Constants.TUPLE_COUNT_OFFSET) - 1);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.getInt(TOTAL_FREE_SPACE_OFFSET) + tupleSize + slotManager
+                .getSlotSize());
     }
 
     @Override
     public FrameOpSpaceStatus hasSpaceInsert(ITupleReference tuple) throws HyracksDataException {
         int bytesRequired = tupleWriter.bytesRequired(tuple);
         // Enough space in the contiguous space region?
-        if (bytesRequired + slotManager.getSlotSize() <= buf.capacity() - buf.getInt(freeSpaceOff)
-                - (buf.getInt(tupleCountOff) * slotManager.getSlotSize())) {
+        if (bytesRequired + slotManager.getSlotSize() <= buf.capacity() - buf.getInt(Constants.FREE_SPACE_OFFSET)
+                - (buf.getInt(Constants.TUPLE_COUNT_OFFSET) * slotManager.getSlotSize())) {
             return FrameOpSpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
         }
         // Enough space after compaction?
-        if (bytesRequired + slotManager.getSlotSize() <= buf.getInt(totalFreeSpaceOff)) {
+        if (bytesRequired + slotManager.getSlotSize() <= buf.getInt(TOTAL_FREE_SPACE_OFFSET)) {
             return FrameOpSpaceStatus.SUFFICIENT_SPACE;
         }
         return FrameOpSpaceStatus.INSUFFICIENT_SPACE;
@@ -223,29 +221,30 @@
         }
         // Enough space if we delete the old tuple and insert the new one
         // without compaction?
-        if (newTupleBytes <= buf.capacity() - buf.getInt(freeSpaceOff)
-                - (buf.getInt(tupleCountOff) * slotManager.getSlotSize())) {
+        if (newTupleBytes <= buf.capacity() - buf.getInt(Constants.FREE_SPACE_OFFSET)
+                - (buf.getInt(Constants.TUPLE_COUNT_OFFSET) * slotManager.getSlotSize())) {
             return FrameOpSpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
         }
         // Enough space if we delete the old tuple and compact?
-        if (additionalBytesRequired <= buf.getInt(totalFreeSpaceOff)) {
+        if (additionalBytesRequired <= buf.getInt(TOTAL_FREE_SPACE_OFFSET)) {
             return FrameOpSpaceStatus.SUFFICIENT_SPACE;
         }
         return FrameOpSpaceStatus.INSUFFICIENT_SPACE;
     }
 
     protected void resetSpaceParams() {
-        buf.putInt(freeSpaceOff, getPageHeaderSize());
-        buf.putInt(totalFreeSpaceOff, buf.capacity() - getPageHeaderSize());
+        buf.putInt(Constants.FREE_SPACE_OFFSET, getPageHeaderSize());
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.capacity() - getPageHeaderSize());
     }
 
     @Override
     public void insert(ITupleReference tuple, int tupleIndex) {
-        slotManager.insertSlot(tupleIndex, buf.getInt(freeSpaceOff));
-        int bytesWritten = tupleWriter.writeTuple(tuple, buf.array(), buf.getInt(freeSpaceOff));
-        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
-        buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + bytesWritten);
-        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());
+        slotManager.insertSlot(tupleIndex, buf.getInt(Constants.FREE_SPACE_OFFSET));
+        int bytesWritten = tupleWriter.writeTuple(tuple, buf.array(), buf.getInt(Constants.FREE_SPACE_OFFSET));
+        buf.putInt(Constants.TUPLE_COUNT_OFFSET, buf.getInt(Constants.TUPLE_COUNT_OFFSET) + 1);
+        buf.putInt(Constants.FREE_SPACE_OFFSET, buf.getInt(Constants.FREE_SPACE_OFFSET) + bytesWritten);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.getInt(TOTAL_FREE_SPACE_OFFSET) - bytesWritten - slotManager
+                .getSlotSize());
     }
 
     @Override
@@ -260,31 +259,31 @@
         } else {
             // Insert the new tuple at the end of the free space, and change the
             // slot value (effectively "deleting" the old tuple).
-            int newTupleOff = buf.getInt(freeSpaceOff);
+            int newTupleOff = buf.getInt(Constants.FREE_SPACE_OFFSET);
             bytesWritten = tupleWriter.writeTuple(newTuple, buf.array(), newTupleOff);
             // Update slot value.
             buf.putInt(slotOff, newTupleOff);
             // Update contiguous free space pointer.
-            buf.putInt(freeSpaceOff, newTupleOff + bytesWritten);
+            buf.putInt(Constants.FREE_SPACE_OFFSET, newTupleOff + bytesWritten);
         }
-        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + oldTupleBytes - bytesWritten);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.getInt(TOTAL_FREE_SPACE_OFFSET) + oldTupleBytes - bytesWritten);
     }
 
     @Override
     public String printHeader() {
         StringBuilder strBuilder = new StringBuilder();
-        strBuilder.append("pageLsnOff:        " + pageLsnOff + "\n");
-        strBuilder.append("tupleCountOff:     " + tupleCountOff + "\n");
-        strBuilder.append("freeSpaceOff:      " + freeSpaceOff + "\n");
-        strBuilder.append("totalFreeSpaceOff: " + totalFreeSpaceOff + "\n");
-        strBuilder.append("levelOff:          " + levelOff + "\n");
-        strBuilder.append("flagOff:           " + flagOff + "\n");
+        strBuilder.append("pageLsnOff:        " + PAGE_LSN_OFFSET + "\n");
+        strBuilder.append("tupleCountOff:     " + Constants.TUPLE_COUNT_OFFSET + "\n");
+        strBuilder.append("freeSpaceOff:      " + Constants.FREE_SPACE_OFFSET + "\n");
+        strBuilder.append("totalFreeSpaceOff: " + TOTAL_FREE_SPACE_OFFSET + "\n");
+        strBuilder.append("levelOff:          " + Constants.LEVEL_OFFSET + "\n");
+        strBuilder.append("flagOff:           " + FLAG_OFFSET + "\n");
         return strBuilder.toString();
     }
 
     @Override
     public int getTupleCount() {
-        return buf.getInt(tupleCountOff);
+        return buf.getInt(Constants.TUPLE_COUNT_OFFSET);
     }
 
     @Override
@@ -299,17 +298,17 @@
 
     @Override
     public long getPageLsn() {
-        return buf.getLong(pageLsnOff);
+        return buf.getLong(PAGE_LSN_OFFSET);
     }
 
     @Override
     public void setPageLsn(long pageLsn) {
-        buf.putLong(pageLsnOff, pageLsn);
+        buf.putLong(PAGE_LSN_OFFSET, pageLsn);
     }
 
     @Override
     public int getTotalFreeSpace() {
-        return buf.getInt(totalFreeSpaceOff);
+        return buf.getInt(TOTAL_FREE_SPACE_OFFSET);
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/AppendOnlyLinkedMetadataPageManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/AppendOnlyLinkedMetadataPageManager.java
index 4126c19..32d2515 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/AppendOnlyLinkedMetadataPageManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/AppendOnlyLinkedMetadataPageManager.java
@@ -19,12 +19,13 @@
 package org.apache.hyracks.storage.am.common.freepage;
 
 import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.data.std.api.IValueReference;
 import org.apache.hyracks.storage.am.common.api.IMetadataPageManager;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
-import org.apache.hyracks.storage.am.common.frames.LIFOMetaDataFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrameFactory;
 import org.apache.hyracks.storage.am.common.impls.AbstractTreeIndex;
 import org.apache.hyracks.storage.common.buffercache.BufferCache;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
@@ -36,23 +37,23 @@
     private final IBufferCache bufferCache;
     private int metadataPage = IBufferCache.INVALID_PAGEID;
     private int fileId = -1;
-    private final ITreeIndexMetaDataFrameFactory frameFactory;
+    private final ITreeIndexMetadataFrameFactory frameFactory;
     ICachedPage confiscatedPage;
     ICachedPage filterPage;
     boolean ready = false;
 
-    public AppendOnlyLinkedMetadataPageManager(IBufferCache bufferCache, ITreeIndexMetaDataFrameFactory frameFactory) {
+    public AppendOnlyLinkedMetadataPageManager(IBufferCache bufferCache, ITreeIndexMetadataFrameFactory frameFactory) {
         this.bufferCache = bufferCache;
         this.frameFactory = frameFactory;
     }
 
     @Override
-    public void releasePage(ITreeIndexMetaDataFrame metaFrame, int freePageNum) throws HyracksDataException {
+    public void releasePage(ITreeIndexMetadataFrame metaFrame, int freePageNum) throws HyracksDataException {
         ICachedPage metaPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
         metaPage.acquireWriteLatch();
         try {
             metaFrame.setPage(metaPage);
-            if (metaFrame.hasSpace()) {
+            if (metaFrame.getSpace() > Integer.BYTES) {
                 metaFrame.addFreePage(freePageNum);
             } else {
                 int newPageNum = metaFrame.getFreePage();
@@ -66,8 +67,8 @@
                     int metaMaxPage = metaFrame.getMaxPage();
                     System.arraycopy(metaPage.getBuffer().array(), 0, newNode.getBuffer().array(), 0,
                             metaPage.getBuffer().capacity());
-                    metaFrame.initBuffer();
-                    metaFrame.setNextPage(newPageNum);
+                    metaFrame.init();
+                    metaFrame.setNextMetadataPage(newPageNum);
                     metaFrame.setMaxPage(metaMaxPage);
                     metaFrame.addFreePage(freePageNum);
                 } finally {
@@ -82,7 +83,7 @@
     }
 
     @Override
-    public void releaseBlock(ITreeIndexMetaDataFrame metaFrame, int startingPage, int count)
+    public void releaseBlock(ITreeIndexMetadataFrame metaFrame, int startingPage, int count)
             throws HyracksDataException {
         for (int i = 0; i < count; i++) {
             releasePage(metaFrame, startingPage + i);
@@ -90,7 +91,7 @@
     }
 
     @Override
-    public int takePage(ITreeIndexMetaDataFrame metaFrame) throws HyracksDataException {
+    public int takePage(ITreeIndexMetadataFrame metaFrame) throws HyracksDataException {
         confiscatedPage.acquireWriteLatch();
         int freePage = IBufferCache.INVALID_PAGEID;
         try {
@@ -140,14 +141,14 @@
     }
 
     @Override
-    public int takeBlock(ITreeIndexMetaDataFrame metaFrame, int count) throws HyracksDataException {
+    public int takeBlock(ITreeIndexMetadataFrame metaFrame, int count) throws HyracksDataException {
         int maxPage = metaFrame.getMaxPage();
         metaFrame.setMaxPage(maxPage + count);
         return maxPage + 1;
     }
 
     @Override
-    public int getMaxPageId(ITreeIndexMetaDataFrame metaFrame) throws HyracksDataException {
+    public int getMaxPageId(ITreeIndexMetadataFrame metaFrame) throws HyracksDataException {
         ICachedPage metaNode;
         if (confiscatedPage == null) {
             int mdPage = getMetadataPageId();
@@ -173,42 +174,13 @@
     }
 
     @Override
-    public void setFilterPageId(int filterPageId) throws HyracksDataException {
-        ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
-        confiscatedPage.acquireWriteLatch();
-        try {
-            metaFrame.setPage(confiscatedPage);
-            metaFrame.setLSMComponentFilterPageId(filterPageId);
-        } finally {
-            confiscatedPage.releaseWriteLatch(false);
-        }
-    }
-
-    @Override
-    public int getFilterPageId() throws HyracksDataException {
-        ICachedPage metaNode = (confiscatedPage == null) ? bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId,
-                getMetadataPageId()), false) : confiscatedPage;
-        ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
-        metaNode.acquireReadLatch();
-        try {
-            metaFrame.setPage(metaNode);
-            return metaFrame.getLSMComponentFilterPageId();
-        } finally {
-            metaNode.releaseReadLatch();
-            if (confiscatedPage == null) {
-                bufferCache.unpin(metaNode);
-            }
-        }
-    }
-
-    @Override
     public void init(ITreeIndexFrameFactory interiorFrameFactory, ITreeIndexFrameFactory leafFrameFactory)
             throws HyracksDataException {
         // an initialized append only tree is always completely empty with size = 0, hence, this operation is a No Op
     }
 
     @Override
-    public ITreeIndexMetaDataFrame createMetadataFrame() {
+    public ITreeIndexMetadataFrame createMetadataFrame() {
         return frameFactory.createFrame();
     }
 
@@ -222,11 +194,11 @@
             if (confiscatedPage != null) {
                 throw new HyracksDataException("Metadata Page Manager is already initialized");
             }
-            ITreeIndexMetaDataFrame metaFrame = createMetadataFrame();
+            ITreeIndexMetadataFrame metaFrame = createMetadataFrame();
             ICachedPage metaNode = bufferCache.confiscatePage(BufferCache.INVALID_DPID);
             try {
                 metaFrame.setPage(metaNode);
-                metaFrame.initBuffer();
+                metaFrame.init();
                 metaFrame.setMaxPage(-1);
             } finally {
                 confiscatedPage = metaNode;
@@ -238,8 +210,7 @@
     public void close() throws HyracksDataException {
         if (ready) {
             IFIFOPageQueue queue = bufferCache.createFIFOQueue();
-            writeFilterPage(queue);
-            ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
+            ITreeIndexMetadataFrame metaFrame = frameFactory.createFrame();
             confiscatedPage.acquireWriteLatch();
             try {
                 metaFrame.setPage(confiscatedPage);
@@ -259,17 +230,6 @@
         confiscatedPage = null;
     }
 
-    private void writeFilterPage(IFIFOPageQueue queue) throws HyracksDataException {
-        if (filterPage != null) {
-            ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
-            metaFrame.setPage(confiscatedPage);
-            int finalFilterPage = takePage(metaFrame);
-            setFilterPageId(finalFilterPage);
-            bufferCache.setPageDiskId(filterPage, BufferedFileHandle.getDiskPageId(fileId, finalFilterPage));
-            queue.put(filterPage);
-        }
-    }
-
     /**
      * For storage on append-only media (such as HDFS), the meta data page has to be written last.
      * However, some implementations still write the meta data to the front. To deal with this as well
@@ -281,13 +241,10 @@
      */
     @Override
     public int getMetadataPageId() throws HyracksDataException {
-        // if found already, just return it
         if (metadataPage != IBufferCache.INVALID_PAGEID) {
             return metadataPage;
         }
-        // get the number of pages of the file
         int pages = bufferCache.getNumPagesOfFile(fileId);
-        //if there are no pages in the file yet, we're just initializing
         if (pages == 0) {
             return 0;
         }
@@ -296,93 +253,17 @@
     }
 
     @Override
-    public long getLSN() throws HyracksDataException {
-        ICachedPage metaNode;
-        if (confiscatedPage == null) {
-            metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
-        } else {
-            metaNode = confiscatedPage;
-        }
-        ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
-        metaNode.acquireReadLatch();
-        try {
-            metaFrame.setPage(metaNode);
-            return metaFrame.getLSN();
-        } finally {
-            metaNode.releaseReadLatch();
-            if (confiscatedPage == null) {
-                bufferCache.unpin(metaNode);
-            }
-        }
-    }
-
-    @Override
-    public void setLSN(long lsn) throws HyracksDataException {
-        ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
-        confiscatedPage.acquireWriteLatch();
-        try {
-            metaFrame.setPage(confiscatedPage);
-            metaFrame.setLSN(lsn);
-        } finally {
-            confiscatedPage.releaseWriteLatch(false);
-        }
-    }
-
-    @Override
-    public void setFilterPage(ICachedPage filterPage) {
-        filterPage.releaseWriteLatch(false);
-        this.filterPage = filterPage;
-    }
-
-    @Override
-    public ICachedPage getFilterPage() throws HyracksDataException {
-        filterPage = bufferCache.confiscatePage(IBufferCache.INVALID_DPID);
-        filterPage.acquireWriteLatch();
-        return filterPage;
-    }
-
-    @Override
     public boolean isEmpty(ITreeIndexFrame frame, int rootPage) throws HyracksDataException {
         return bufferCache.getNumPagesOfFile(fileId) <= AbstractTreeIndex.MINIMAL_TREE_PAGE_COUNT;
     }
 
     @Override
-    public long getLSNOffset() throws HyracksDataException {
-        int metadataPageNum = getMetadataPageId();
-        if (metadataPageNum != IBufferCache.INVALID_PAGEID) {
-            return ((long) metadataPageNum * bufferCache.getPageSizeWithHeader()) + LIFOMetaDataFrame.LSN_OFFSET;
-        }
-        return IMetadataPageManager.Constants.INVALID_LSN_OFFSET;
-    }
-
-    @Override
-    public long getLastMarkerLSN() throws HyracksDataException {
-        ICachedPage metaNode;
-        if (confiscatedPage == null) {
-            metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
-        } else {
-            metaNode = confiscatedPage;
-        }
-        ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
-        metaNode.acquireReadLatch();
-        try {
-            metaFrame.setPage(metaNode);
-            return metaFrame.getLastMarkerLSN();
-        } finally {
-            metaNode.releaseReadLatch();
-            if (confiscatedPage == null) {
-                bufferCache.unpin(metaNode);
-            }
-        }
-    }
-
-    @Override
     public void setRootPageId(int rootPage) throws HyracksDataException {
-        ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
+        ITreeIndexMetadataFrame metaFrame = frameFactory.createFrame();
         confiscatedPage.acquireWriteLatch();
         try {
             metaFrame.setPage(confiscatedPage);
-            metaFrame.setRootPageNumber(rootPage);
+            metaFrame.setRootPageId(rootPage);
         } finally {
             confiscatedPage.releaseWriteLatch(false);
         }
@@ -397,11 +278,11 @@
         } else {
             metaNode = confiscatedPage;
         }
-        ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
+        ITreeIndexMetadataFrame metaFrame = frameFactory.createFrame();
         metaNode.acquireReadLatch();
         try {
             metaFrame.setPage(metaNode);
-            return metaFrame.getRootPageNumber();
+            return metaFrame.getRootPageId();
         } finally {
             metaNode.releaseReadLatch();
             if (confiscatedPage == null) {
@@ -414,4 +295,60 @@
     public int getBulkLoadLeaf() throws HyracksDataException {
         return 0;
     }
+
+    @Override
+    public void put(ITreeIndexMetadataFrame frame, IValueReference key, IValueReference value)
+            throws HyracksDataException {
+        confiscatedPage.acquireWriteLatch();
+        try {
+            frame.setPage(confiscatedPage);
+            frame.put(key, value);
+        } finally {
+            confiscatedPage.releaseWriteLatch(false);
+        }
+    }
+
+    private ICachedPage pinPage() throws HyracksDataException {
+        return confiscatedPage == null ? bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()),
+                false) : confiscatedPage;
+    }
+
+    private void unpinPage(ICachedPage page) throws HyracksDataException {
+        if (confiscatedPage == null) {
+            bufferCache.unpin(page);
+        }
+    }
+
+    @Override
+    public void get(ITreeIndexMetadataFrame frame, IValueReference key, IPointable value) throws HyracksDataException {
+        ICachedPage page = pinPage();
+        page.acquireReadLatch();
+        try {
+            frame.setPage(page);
+            frame.get(key, value);
+        } finally {
+            page.releaseReadLatch();
+            unpinPage(page);
+        }
+    }
+
+    @Override
+    public long getFileOffset(ITreeIndexMetadataFrame frame, IValueReference key) throws HyracksDataException {
+        int pageId = getMetadataPageId();
+        if (pageId != IBufferCache.INVALID_PAGEID) {
+            ICachedPage page = pinPage();
+            page.acquireReadLatch();
+            try {
+                frame.setPage(page);
+                int inPageOffset = frame.getOffset(key);
+                return inPageOffset >= 0 ? ((long) pageId * bufferCache.getPageSizeWithHeader()) + frame.getOffset(key)
+                        + IBufferCache.RESERVED_HEADER_BYTES
+                        : -1L;
+            } finally {
+                page.releaseReadLatch();
+                unpinPage(page);
+            }
+        }
+        return -1L;
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/LinkedMetaDataPageManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/LinkedMetaDataPageManager.java
index 6cedb4d..686fe78 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/LinkedMetaDataPageManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/LinkedMetaDataPageManager.java
@@ -19,12 +19,13 @@
 package org.apache.hyracks.storage.am.common.freepage;
 
 import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.data.std.api.IValueReference;
 import org.apache.hyracks.storage.am.common.api.IMetadataPageManager;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
-import org.apache.hyracks.storage.am.common.frames.LIFOMetaDataFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrameFactory;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
 import org.apache.hyracks.storage.common.buffercache.ICachedPage;
 import org.apache.hyracks.storage.common.file.BufferedFileHandle;
@@ -32,23 +33,23 @@
 public class LinkedMetaDataPageManager implements IMetadataPageManager {
     private final IBufferCache bufferCache;
     private int fileId = -1;
-    private final ITreeIndexMetaDataFrameFactory frameFactory;
+    private final ITreeIndexMetadataFrameFactory frameFactory;
     private boolean ready = false;
 
-    public LinkedMetaDataPageManager(IBufferCache bufferCache, ITreeIndexMetaDataFrameFactory frameFactory) {
+    public LinkedMetaDataPageManager(IBufferCache bufferCache, ITreeIndexMetadataFrameFactory frameFactory) {
         this.bufferCache = bufferCache;
         this.frameFactory = frameFactory;
     }
 
     @Override
-    public void releasePage(ITreeIndexMetaDataFrame metaFrame, int freePageNum) throws HyracksDataException {
+    public void releasePage(ITreeIndexMetadataFrame metaFrame, int freePageNum) throws HyracksDataException {
         // Get the metadata node
         ICachedPage metaPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
         metaPage.acquireWriteLatch();
         try {
             metaFrame.setPage(metaPage);
 
-            if (metaFrame.hasSpace()) {
+            if (metaFrame.getSpace() > Integer.BYTES) {
                 metaFrame.addFreePage(freePageNum);
             } else {
                 // allocate a new page in the chain of meta pages
@@ -68,8 +69,8 @@
                     System.arraycopy(metaPage.getBuffer().array(), 0, newNode.getBuffer().array(), 0,
                             metaPage.getBuffer().capacity());
 
-                    metaFrame.initBuffer();
-                    metaFrame.setNextPage(newPageNum);
+                    metaFrame.init();
+                    metaFrame.setNextMetadataPage(newPageNum);
                     metaFrame.setMaxPage(metaMaxPage);
                     metaFrame.addFreePage(freePageNum);
                 } finally {
@@ -84,7 +85,7 @@
     }
 
     @Override
-    public void releaseBlock(ITreeIndexMetaDataFrame metaFrame, int startingPage, int count)
+    public void releaseBlock(ITreeIndexMetadataFrame metaFrame, int startingPage, int count)
             throws HyracksDataException {
         for (int i = 0; i < count; i++) {
             releasePage(metaFrame, startingPage + i);
@@ -92,11 +93,9 @@
     }
 
     @Override
-    public int takePage(ITreeIndexMetaDataFrame metaFrame) throws HyracksDataException {
+    public int takePage(ITreeIndexMetadataFrame metaFrame) throws HyracksDataException {
         ICachedPage metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
-
         metaNode.acquireWriteLatch();
-
         int freePage = IBufferCache.INVALID_PAGEID;
         try {
             metaFrame.setPage(metaNode);
@@ -151,14 +150,14 @@
     }
 
     @Override
-    public int takeBlock(ITreeIndexMetaDataFrame metaFrame, int count) throws HyracksDataException {
+    public int takeBlock(ITreeIndexMetadataFrame metaFrame, int count) throws HyracksDataException {
         int maxPage = metaFrame.getMaxPage();
         metaFrame.setMaxPage(maxPage + count);
         return maxPage + 1;
     }
 
     @Override
-    public int getMaxPageId(ITreeIndexMetaDataFrame metaFrame) throws HyracksDataException {
+    public int getMaxPageId(ITreeIndexMetadataFrame metaFrame) throws HyracksDataException {
         ICachedPage metaNode;
         int mdPage = getMetadataPageId();
         if (mdPage < 0) {
@@ -179,43 +178,10 @@
     }
 
     @Override
-    public void setFilterPageId(int filterPageId) throws HyracksDataException {
-        ICachedPage metaNode;
-        int mdPage = getMetadataPageId();
-        if (mdPage < 0) {
-            return;
-        }
-        metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, mdPage), false);
-        ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
-        metaNode.acquireWriteLatch();
-        try {
-            metaFrame.setPage(metaNode);
-            metaFrame.setLSMComponentFilterPageId(filterPageId);
-        } finally {
-            metaNode.releaseWriteLatch(true);
-            bufferCache.unpin(metaNode);
-        }
-    }
-
-    @Override
-    public int getFilterPageId() throws HyracksDataException {
-        ICachedPage metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
-        ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
-        metaNode.acquireReadLatch();
-        try {
-            metaFrame.setPage(metaNode);
-            return metaFrame.getLSMComponentFilterPageId();
-        } finally {
-            metaNode.releaseReadLatch();
-            bufferCache.unpin(metaNode);
-        }
-    }
-
-    @Override
     public void init(ITreeIndexFrameFactory interiorFrameFactory, ITreeIndexFrameFactory leafFrameFactory)
             throws HyracksDataException {
         // initialize meta data page
-        ITreeIndexMetaDataFrame metaFrame = createMetadataFrame();
+        ITreeIndexMetadataFrame metaFrame = createMetadataFrame();
         int metaPage = getMetadataPageId();
         if (metaPage == IBufferCache.INVALID_PAGEID) {
             throw new HyracksDataException("No valid metadata found in this file.");
@@ -224,8 +190,8 @@
         metaNode.acquireWriteLatch();
         try {
             metaFrame.setPage(metaNode);
-            metaFrame.initBuffer();
-            metaFrame.setRootPageNumber(1);
+            metaFrame.init();
+            metaFrame.setRootPageId(1);
             metaFrame.setMaxPage(1);
         } finally {
             metaNode.releaseWriteLatch(true);
@@ -247,7 +213,7 @@
     }
 
     @Override
-    public ITreeIndexMetaDataFrame createMetadataFrame() {
+    public ITreeIndexMetadataFrame createMetadataFrame() {
         return frameFactory.createFrame();
     }
 
@@ -259,11 +225,11 @@
     @Override
     public void setRootPageId(int rootPage) throws HyracksDataException {
         ICachedPage metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
-        ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
+        ITreeIndexMetadataFrame metaFrame = frameFactory.createFrame();
         metaNode.acquireWriteLatch();
         try {
             metaFrame.setPage(metaNode);
-            metaFrame.setRootPageNumber(rootPage);
+            metaFrame.setRootPageId(rootPage);
         } finally {
             metaNode.releaseWriteLatch(true);
             bufferCache.unpin(metaNode);
@@ -276,7 +242,7 @@
         if (ready) {
             ICachedPage metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()),
                     false);
-            ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
+            ITreeIndexMetadataFrame metaFrame = frameFactory.createFrame();
             metaNode.acquireWriteLatch();
             try {
                 metaFrame.setPage(metaNode);
@@ -306,95 +272,14 @@
     }
 
     @Override
-    public long getLSN() throws HyracksDataException {
-        ICachedPage metaNode;
-        metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
-        ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
-        metaNode.acquireReadLatch();
-        try {
-            metaFrame.setPage(metaNode);
-            return metaFrame.getLSN();
-        } finally {
-            metaNode.releaseReadLatch();
-            bufferCache.unpin(metaNode);
-        }
-    }
-
-    @Override
-    public void setLSN(long lsn) throws HyracksDataException {
-        ICachedPage metaNode;
-        metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
-        ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
-        metaNode.acquireWriteLatch();
-        try {
-            metaFrame.setPage(metaNode);
-            metaFrame.setLSN(lsn);
-        } finally {
-            metaNode.releaseWriteLatch(true);
-            bufferCache.unpin(metaNode);
-        }
-    }
-
-    @Override
-    public void setFilterPage(ICachedPage filterPage) throws HyracksDataException {
-        filterPage.releaseWriteLatch(true);
-        bufferCache.unpin(filterPage);
-        bufferCache.flushDirtyPage(filterPage);
-    }
-
-    @Override
-    public ICachedPage getFilterPage() throws HyracksDataException {
-        ITreeIndexMetaDataFrame metadataFrame = frameFactory.createFrame();
-        int metaPageId = getMetadataPageId();
-        ICachedPage metadataPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, metaPageId), false);
-        metadataPage.acquireWriteLatch();
-        int filterPageId;
-        try {
-            metadataFrame.setPage(metadataPage);
-            filterPageId = takePage(metadataFrame);
-            metadataFrame.setLSMComponentFilterPageId(filterPageId);
-        } finally {
-            metadataPage.releaseWriteLatch(true);
-            bufferCache.unpin(metadataPage);
-        }
-        ICachedPage filterPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, filterPageId), true);
-        filterPage.acquireWriteLatch();
-        return filterPage;
-    }
-
-    @Override
-    public long getLSNOffset() throws HyracksDataException {
-        int metadataPageNum = getMetadataPageId();
-        if (metadataPageNum != IBufferCache.INVALID_PAGEID) {
-            return ((long) metadataPageNum * bufferCache.getPageSizeWithHeader()) + LIFOMetaDataFrame.LSN_OFFSET;
-        }
-        return IMetadataPageManager.Constants.INVALID_LSN_OFFSET;
-    }
-
-    @Override
-    public long getLastMarkerLSN() throws HyracksDataException {
-        ICachedPage metaNode;
-        metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
-        ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
-        metaNode.acquireReadLatch();
-        try {
-            metaFrame.setPage(metaNode);
-            return metaFrame.getLastMarkerLSN();
-        } finally {
-            metaNode.releaseReadLatch();
-            bufferCache.unpin(metaNode);
-        }
-    }
-
-    @Override
     public int getRootPageId() throws HyracksDataException {
         ICachedPage metaNode;
         metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
-        ITreeIndexMetaDataFrame metaFrame = frameFactory.createFrame();
+        ITreeIndexMetadataFrame metaFrame = frameFactory.createFrame();
         metaNode.acquireReadLatch();
         try {
             metaFrame.setPage(metaNode);
-            return metaFrame.getRootPageNumber();
+            return metaFrame.getRootPageId();
         } finally {
             metaNode.releaseReadLatch();
             bufferCache.unpin(metaNode);
@@ -412,14 +297,39 @@
         rootNode.acquireReadLatch();
         try {
             frame.setPage(rootNode);
-            if (frame.getLevel() == 0 && frame.getTupleCount() == 0) {
-                return true;
-            } else {
-                return false;
-            }
+            return frame.getLevel() == 0 && frame.getTupleCount() == 0;
         } finally {
             rootNode.releaseReadLatch();
             bufferCache.unpin(rootNode);
         }
     }
+
+    @Override
+    public void put(ITreeIndexMetadataFrame frame, IValueReference key, IValueReference value)
+            throws HyracksDataException {
+        throw new HyracksDataException("Unsupported Operation");
+    }
+
+    @Override
+    public void get(ITreeIndexMetadataFrame frame, IValueReference key, IPointable value) throws HyracksDataException {
+        throw new HyracksDataException("Unsupported Operation");
+    }
+
+    @Override
+    public long getFileOffset(ITreeIndexMetadataFrame frame, IValueReference key) throws HyracksDataException {
+        int metadataPageNum = getMetadataPageId();
+        if (metadataPageNum != IBufferCache.INVALID_PAGEID) {
+            ICachedPage metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()),
+                    false);
+            metaNode.acquireReadLatch();
+            try {
+                frame.setPage(metaNode);
+                return ((long) metadataPageNum * bufferCache.getPageSizeWithHeader()) + frame.getOffset(key);
+            } finally {
+                metaNode.releaseReadLatch();
+                bufferCache.unpin(metaNode);
+            }
+        }
+        return -1;
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/MutableArrayValueReference.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/MutableArrayValueReference.java
new file mode 100644
index 0000000..627994c
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/MutableArrayValueReference.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hyracks.storage.am.common.freepage;
+
+import org.apache.hyracks.data.std.api.IValueReference;
+
+public class MutableArrayValueReference implements IValueReference {
+    private byte[] array;
+
+    public MutableArrayValueReference() {
+        //mutable array. user doesn't need to specify the array in advance
+    }
+
+    public MutableArrayValueReference(byte[] array) {
+        this.array = array;
+    }
+
+    public void set(byte[] array) {
+        this.array = array;
+    }
+
+    @Override
+    public byte[] getByteArray() {
+        return array;
+    }
+
+    @Override
+    public int getStartOffset() {
+        return 0;
+    }
+
+    @Override
+    public int getLength() {
+        return array == null ? 0 : array.length;
+    }
+
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/AbstractTreeIndex.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/AbstractTreeIndex.java
index f8539cb..0af7939 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/AbstractTreeIndex.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/AbstractTreeIndex.java
@@ -32,7 +32,7 @@
 import org.apache.hyracks.storage.am.common.api.ITreeIndexAccessor;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrame;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexTupleWriter;
 import org.apache.hyracks.storage.am.common.api.IndexException;
 import org.apache.hyracks.storage.am.common.api.TreeIndexException;
@@ -261,7 +261,7 @@
         protected final int leafMaxBytes;
         protected final int interiorMaxBytes;
         protected final ArrayList<NodeFrontier> nodeFrontiers = new ArrayList<>();
-        protected final ITreeIndexMetaDataFrame metaFrame;
+        protected final ITreeIndexMetadataFrame metaFrame;
         protected final ITreeIndexTupleWriter tupleWriter;
         protected ITreeIndexFrame leafFrame;
         protected ITreeIndexFrame interiorFrame;
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/tuples/SimpleTupleReference.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/tuples/SimpleTupleReference.java
index 9b16514..0c2081a4 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/tuples/SimpleTupleReference.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/tuples/SimpleTupleReference.java
@@ -19,14 +19,13 @@
 
 package org.apache.hyracks.storage.am.common.tuples;
 
-import java.nio.ByteBuffer;
-
+import org.apache.hyracks.data.std.primitive.ShortPointable;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexTupleReference;
 
 public class SimpleTupleReference implements ITreeIndexTupleReference {
 
-    protected ByteBuffer buf;
+    protected byte[] buf;
     protected int fieldStartIndex;
     protected int fieldCount;
     protected int tupleStartOff;
@@ -34,14 +33,14 @@
     protected int fieldSlotsBytes;
 
     @Override
-    public void resetByTupleOffset(ByteBuffer buf, int tupleStartOff) {
+    public void resetByTupleOffset(byte[] buf, int tupleStartOff) {
         this.buf = buf;
         this.tupleStartOff = tupleStartOff;
     }
 
     @Override
     public void resetByTupleIndex(ITreeIndexFrame frame, int tupleIndex) {
-        resetByTupleOffset(frame.getBuffer(), frame.getTupleOffset(tupleIndex));
+        resetByTupleOffset(frame.getBuffer().array(), frame.getTupleOffset(tupleIndex));
     }
 
     @Override
@@ -65,16 +64,16 @@
 
     @Override
     public byte[] getFieldData(int fIdx) {
-        return buf.array();
+        return buf;
     }
 
     @Override
     public int getFieldLength(int fIdx) {
         if (fIdx == 0) {
-            return buf.getShort(tupleStartOff + nullFlagsBytes);
+            return ShortPointable.getShort(buf, tupleStartOff + nullFlagsBytes);
         } else {
-            return buf.getShort(tupleStartOff + nullFlagsBytes + fIdx * 2)
-                    - buf.getShort(tupleStartOff + nullFlagsBytes + ((fIdx - 1) * 2));
+            return ShortPointable.getShort(buf, tupleStartOff + nullFlagsBytes + fIdx * 2)
+                    - ShortPointable.getShort(buf, tupleStartOff + nullFlagsBytes + ((fIdx - 1) * 2));
         }
     }
 
@@ -84,7 +83,7 @@
             return tupleStartOff + nullFlagsBytes + fieldSlotsBytes;
         } else {
             return tupleStartOff + nullFlagsBytes + fieldSlotsBytes
-                    + buf.getShort(tupleStartOff + nullFlagsBytes + ((fIdx - 1) * 2));
+                    + ShortPointable.getShort(buf, tupleStartOff + nullFlagsBytes + ((fIdx - 1) * 2));
         }
     }
 
@@ -98,6 +97,7 @@
 
     @Override
     public int getTupleSize() {
-        return nullFlagsBytes + fieldSlotsBytes + buf.getShort(tupleStartOff + nullFlagsBytes + (fieldCount-1) * 2);
+        return nullFlagsBytes + fieldSlotsBytes + ShortPointable.getShort(buf, tupleStartOff + nullFlagsBytes
+                + (fieldCount - 1) * 2);
     }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/tuples/TypeAwareTupleReference.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/tuples/TypeAwareTupleReference.java
index 27a767f..e278f6c 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/tuples/TypeAwareTupleReference.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/tuples/TypeAwareTupleReference.java
@@ -19,8 +19,6 @@
 
 package org.apache.hyracks.storage.am.common.tuples;
 
-import java.nio.ByteBuffer;
-
 import org.apache.hyracks.api.dataflow.value.ITypeTraits;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexTupleReference;
@@ -28,7 +26,7 @@
 import org.apache.hyracks.util.encoding.VarLenIntEncoderDecoder.VarLenIntDecoder;
 
 public class TypeAwareTupleReference implements ITreeIndexTupleReference {
-    protected ByteBuffer buf;
+    protected byte[] buf;
     protected int fieldStartIndex;
     protected int fieldCount;
     protected int tupleStartOff;
@@ -46,7 +44,7 @@
     }
 
     @Override
-    public void resetByTupleOffset(ByteBuffer buf, int tupleStartOff) {
+    public void resetByTupleOffset(byte[] buf, int tupleStartOff) {
         this.buf = buf;
         this.tupleStartOff = tupleStartOff;
 
@@ -54,7 +52,7 @@
         int field = 0;
         int cumul = 0;
         int end = fieldStartIndex + fieldCount;
-        encDec.reset(buf.array(), tupleStartOff + nullFlagsBytes);
+        encDec.reset(buf, tupleStartOff + nullFlagsBytes);
         for (int i = fieldStartIndex; i < end; i++) {
             if (!typeTraits[i].isFixedLength()) {
                 cumul += encDec.decode();
@@ -69,7 +67,7 @@
 
     @Override
     public void resetByTupleIndex(ITreeIndexFrame frame, int tupleIndex) {
-        resetByTupleOffset(frame.getBuffer(), frame.getTupleOffset(tupleIndex));
+        resetByTupleOffset(frame.getBuffer().array(), frame.getTupleOffset(tupleIndex));
     }
 
     @Override
@@ -99,7 +97,7 @@
 
     @Override
     public byte[] getFieldData(int fIdx) {
-        return buf.array();
+        return buf;
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/util/TreeIndexBufferCacheWarmup.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/util/TreeIndexBufferCacheWarmup.java
index 3a346c2..2b47238 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/util/TreeIndexBufferCacheWarmup.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/util/TreeIndexBufferCacheWarmup.java
@@ -25,7 +25,7 @@
 import org.apache.hyracks.util.MathUtil;
 import org.apache.hyracks.storage.am.common.api.IMetadataPageManager;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrame;
 import org.apache.hyracks.storage.common.arraylist.IntArrayList;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
 import org.apache.hyracks.storage.common.buffercache.ICachedPage;
@@ -46,7 +46,7 @@
     }
 
     public void warmup(ITreeIndexFrame frame,
-            ITreeIndexMetaDataFrame metaFrame, int[] warmupTreeLevels,
+            ITreeIndexMetadataFrame metaFrame, int[] warmupTreeLevels,
             int[] warmupRepeats) throws HyracksDataException {
         bufferCache.openFile(fileId);
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/util/TreeIndexStats.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/util/TreeIndexStats.java
index cd4854f..4ee9f53 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/util/TreeIndexStats.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/util/TreeIndexStats.java
@@ -21,7 +21,7 @@
 import java.text.DecimalFormat;
 
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrame;
 
 public class TreeIndexStats {
 
@@ -55,7 +55,7 @@
         }
     }
 
-    public void add(ITreeIndexMetaDataFrame metaFrame) {
+    public void add(ITreeIndexMetadataFrame metaFrame) {
         if (metaFrame.isFreePage()) {
             freePages++;
         } else if (metaFrame.isMetadataPage()) {
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/util/TreeIndexStatsGatherer.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/util/TreeIndexStatsGatherer.java
index 0dfac84..7e14b4a 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/util/TreeIndexStatsGatherer.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/util/TreeIndexStatsGatherer.java
@@ -21,7 +21,7 @@
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.storage.am.common.api.IPageManager;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrame;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
 import org.apache.hyracks.storage.common.buffercache.ICachedPage;
 import org.apache.hyracks.storage.common.file.BufferedFileHandle;
@@ -43,7 +43,7 @@
     }
 
     public TreeIndexStats gatherStats(ITreeIndexFrame leafFrame,
-            ITreeIndexFrame interiorFrame, ITreeIndexMetaDataFrame metaFrame)
+            ITreeIndexFrame interiorFrame, ITreeIndexMetadataFrame metaFrame)
             throws HyracksDataException {
 
         bufferCache.openFile(fileId);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/test/java/org/apache/hyracks/storage/am/common/frames/LIFOMetadataFrameTest.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/test/java/org/apache/hyracks/storage/am/common/frames/LIFOMetadataFrameTest.java
new file mode 100644
index 0000000..fbb930d
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/test/java/org/apache/hyracks/storage/am/common/frames/LIFOMetadataFrameTest.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hyracks.storage.am.common.frames;
+
+import java.nio.ByteBuffer;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.primitive.LongPointable;
+import org.apache.hyracks.storage.am.common.freepage.MutableArrayValueReference;
+import org.apache.hyracks.storage.common.buffercache.VirtualPage;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class LIFOMetadataFrameTest {
+
+    @Test
+    public void test() throws HyracksDataException {
+        LIFOMetaDataFrame frame = new LIFOMetaDataFrame();
+        VirtualPage page = new VirtualPage(ByteBuffer.allocate(512), 512);
+        MutableArrayValueReference testKey = new MutableArrayValueReference("TestLSNKey".getBytes());
+        frame.setPage(page);
+        frame.init();
+        LongPointable longPointable = (LongPointable) LongPointable.FACTORY.createPointable();
+        frame.get(testKey, longPointable);
+        Assert.assertNull(longPointable.getByteArray());
+        byte[] longBytes = new byte[Long.BYTES];
+        MutableArrayValueReference value = new MutableArrayValueReference(longBytes);
+        int space = frame.getSpace() - (value.getLength() + Integer.BYTES * 2
+                + testKey.getLength());
+        for (long l = 1L; l < 52L; l++) {
+            LongPointable.setLong(longBytes, 0, l);
+            frame.put(testKey, value);
+            Assert.assertEquals(space, frame.getSpace());
+            frame.get(testKey, longPointable);
+            Assert.assertEquals(l, longPointable.longValue());
+        }
+        Assert.assertEquals(frame.getFreePage(), -1);
+        // add 10 pages and monitor space
+        for (int i = 0; i < 10; i++) {
+            frame.addFreePage(i);
+            space -= Integer.BYTES;
+            Assert.assertEquals(space, frame.getSpace());
+        }
+        for (int i = 9; i >= 0; i--) {
+            int freePage = frame.getFreePage();
+            Assert.assertEquals(freePage, i);
+            space += Integer.BYTES;
+            Assert.assertEquals(space, frame.getSpace());
+        }
+        Assert.assertTrue(frame.getFreePage() < 0);
+    }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
index 66540ab..b77cc15 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
@@ -71,6 +71,7 @@
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
 import org.apache.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
+import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType;
 import org.apache.hyracks.storage.am.lsm.common.freepage.VirtualFreePageManager;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.impls.BlockingIOOperationCallbackWrapper;
@@ -97,7 +98,7 @@
 
     private final boolean needKeyDupCheck;
     private final int[] btreeFields;
-    // Primary LSMBTree has a Bloomfilter, but Secondary one doesn't have. 
+    // Primary LSMBTree has a Bloomfilter, but Secondary one doesn't have.
     private final boolean hasBloomFilter;
 
     public LSMBTree(IIOManager ioManager, List<IVirtualBufferCache> virtualBufferCaches,
@@ -494,7 +495,6 @@
         }
         component.setMostRecentMarkerLSN(flushOp.getPrevMarkerLSN());
         bulkLoader.end();
-
         return component;
     }
 
@@ -744,6 +744,10 @@
                 if (isEmptyComponent) {
                     cleanupArtifacts();
                 } else {
+                    //TODO(amoudi): Ensure Bulk load follow the same lifecycle Other Operations (Flush, Merge, etc).
+                    //then after operation should be called from harness as well
+                    //https://issues.apache.org/jira/browse/ASTERIXDB-1764
+                    ioOpCallback.afterOperation(LSMOperationType.FLUSH, null, component);
                     lsmHarness.addBulkLoadedComponent(component);
                 }
             }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponent.java
index 9768aa3..244f7a9 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponent.java
@@ -66,6 +66,7 @@
 
     public void readMostRecentMarkerLSN(BTree treeIndex) throws HyracksDataException {
         IMetadataPageManager treeMetaManager = (IMetadataPageManager) treeIndex.getPageManager();
-        mostRecentMarkerLSN = treeMetaManager.getLastMarkerLSN();
+        treeMetaManager.get(treeMetaManager.createMetadataFrame(), MARKER_LSN_KEY, pointable);
+        mostRecentMarkerLSN = pointable.getByteArray() == null ? -1L : pointable.longValue();
     }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/tuples/LSMBTreeTupleReference.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/tuples/LSMBTreeTupleReference.java
index 4d6fa61..73aa321 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/tuples/LSMBTreeTupleReference.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/tuples/LSMBTreeTupleReference.java
@@ -19,8 +19,6 @@
 
 package org.apache.hyracks.storage.am.lsm.btree.tuples;
 
-import java.nio.ByteBuffer;
-
 import org.apache.hyracks.api.dataflow.value.ITypeTraits;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
 import org.apache.hyracks.storage.am.common.tuples.TypeAwareTupleReference;
@@ -39,6 +37,7 @@
         this.numKeyFields = numKeyFields;
     }
 
+    @Override
     public void setFieldCount(int fieldCount) {
         super.setFieldCount(fieldCount);
         // Don't change the fieldCount in reset calls.
@@ -53,7 +52,7 @@
     }
 
     @Override
-    public void resetByTupleOffset(ByteBuffer buf, int tupleStartOff) {
+    public void resetByTupleOffset(byte[] buf, int tupleStartOff) {
         this.buf = buf;
         this.tupleStartOff = tupleStartOff;
         if (numKeyFields != typeTraits.length) {
@@ -72,7 +71,7 @@
 
     @Override
     public void resetByTupleIndex(ITreeIndexFrame frame, int tupleIndex) {
-        resetByTupleOffset(frame.getBuffer(), frame.getTupleOffset(tupleIndex));
+        resetByTupleOffset(frame.getBuffer().array(), frame.getTupleOffset(tupleIndex));
     }
 
     @Override
@@ -85,7 +84,7 @@
     public boolean isAntimatter() {
           // Check if the leftmost bit is 0 or 1.
         final byte mask = (byte) (1 << 7);
-        if ((buf.array()[tupleStartOff] & mask) != 0) {
+        if ((buf[tupleStartOff] & mask) != 0) {
             return true;
         }
         return false;
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/util/LSMBTreeUtils.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/util/LSMBTreeUtils.java
index 69c6666..9ca96e4 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/util/LSMBTreeUtils.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/util/LSMBTreeUtils.java
@@ -90,9 +90,8 @@
         if (filterCmpFactories != null) {
             TypeAwareTupleWriterFactory filterTupleWriterFactory = new TypeAwareTupleWriterFactory(filterTypeTraits);
             filterFactory = new LSMComponentFilterFactory(filterTupleWriterFactory, filterCmpFactories);
-            filterFrameFactory = new LSMComponentFilterFrameFactory(filterTupleWriterFactory,
-                    diskBufferCache.getPageSize());
-            filterManager = new LSMComponentFilterManager(diskBufferCache, filterFrameFactory);
+            filterFrameFactory = new LSMComponentFilterFrameFactory(filterTupleWriterFactory);
+            filterManager = new LSMComponentFilterManager(filterFrameFactory);
         }
 
         //Primary LSMBTree index has a BloomFilter.
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/pom.xml b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/pom.xml
index da7c397..43e1ff3 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/pom.xml
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/pom.xml
@@ -16,17 +16,14 @@
  ! specific language governing permissions and limitations
  ! under the License.
  !-->
-
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
   <modelVersion>4.0.0</modelVersion>
   <artifactId>hyracks-storage-am-lsm-common</artifactId>
-
   <parent>
     <groupId>org.apache.hyracks</groupId>
     <artifactId>hyracks</artifactId>
     <version>0.2.18-SNAPSHOT</version>
   </parent>
-
   <licenses>
     <license>
       <name>Apache License, Version 2.0</name>
@@ -35,11 +32,9 @@
       <comments>A business-friendly OSS license</comments>
     </license>
   </licenses>
-
   <properties>
     <root.dir>${basedir}/../..</root.dir>
   </properties>
-
   <dependencies>
     <dependency>
       <groupId>org.apache.hyracks</groupId>
@@ -76,5 +71,10 @@
       <artifactId>hyracks-dataflow-std</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.hyracks</groupId>
+      <artifactId>hyracks-data-std</artifactId>
+      <version>${project.version}</version>
+    </dependency>
   </dependencies>
-</project>
+</project>
\ No newline at end of file
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterFrameFactory.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterFrameFactory.java
index cf77f30..1c853c8 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterFrameFactory.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterFrameFactory.java
@@ -19,5 +19,5 @@
 package org.apache.hyracks.storage.am.lsm.common.api;
 
 public interface ILSMComponentFilterFrameFactory {
-    public ILSMComponentFilterFrame createFrame();
+    public ILSMComponentFilterReference createFrame();
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterManager.java
index 20598ca..12cafef 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterManager.java
@@ -22,9 +22,7 @@
 
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
-import org.apache.hyracks.storage.am.common.api.IIndexBulkLoader;
 import org.apache.hyracks.storage.am.common.api.ITreeIndex;
-import org.apache.hyracks.storage.am.common.impls.AbstractTreeIndex.AbstractTreeIndexBulkLoader;
 
 public interface ILSMComponentFilterManager {
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterReference.java
similarity index 66%
rename from hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterFrame.java
rename to hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterReference.java
index 4c159b7..5e6f4eb 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterFrame.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterReference.java
@@ -18,27 +18,23 @@
  */
 package org.apache.hyracks.storage.am.lsm.common.api;
 
+import org.apache.hyracks.data.std.api.IPointable;
+import org.apache.hyracks.data.std.api.IValueReference;
 import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
-import org.apache.hyracks.storage.common.buffercache.ICachedPage;
 
-public interface ILSMComponentFilterFrame {
+public interface ILSMComponentFilterReference extends IValueReference, IPointable {
 
-    public ICachedPage getPage();
+    void writeMinTuple(ITupleReference tuple);
 
-    public void setPage(ICachedPage page);
+    void writeMaxTuple(ITupleReference tuple);
 
-    public void writeMinTuple(ITupleReference tuple);
+    boolean isMinTupleSet();
 
-    public void writeMaxTuple(ITupleReference tuple);
+    boolean isMaxTupleSet();
 
-    public void initBuffer();
+    ITupleReference getMinTuple();
 
-    public boolean isMinTupleSet();
+    ITupleReference getMaxTuple();
 
-    public boolean isMaxTupleSet();
-
-    public ITupleReference getMinTuple();
-
-    public ITupleReference getMaxTuple();
-
+    void reset();
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/frames/LSMComponentFilterFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/frames/LSMComponentFilterFrame.java
deleted file mode 100644
index 5d74da1..0000000
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/frames/LSMComponentFilterFrame.java
+++ /dev/null
@@ -1,108 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.hyracks.storage.am.lsm.common.frames;
-
-import java.nio.ByteBuffer;
-
-import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexTupleReference;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexTupleWriter;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilterFrame;
-import org.apache.hyracks.storage.common.buffercache.ICachedPage;
-
-public class LSMComponentFilterFrame implements ILSMComponentFilterFrame {
-
-    // This page consists of two tuples that represents the minimum and maximum tuples in an LSM component.
-
-    // A-one byte to indicate whether the filter tuples were set yet.
-    private static final int minTupleIsSetIndicatorOff = 0;
-    private static final int maxTupleIsSetIndicatorOff = 1;
-
-    private final int minTupleOff;
-    private final int maxTupleOff;
-
-    private final ITreeIndexTupleWriter tupleWriter;
-
-    protected ICachedPage page = null;
-    protected ByteBuffer buf = null;
-
-    private ITreeIndexTupleReference minTuple;
-    private ITreeIndexTupleReference maxTuple;
-
-    public LSMComponentFilterFrame(ITreeIndexTupleWriter tupleWriter, int pageSize) {
-        this.tupleWriter = tupleWriter;
-        this.minTupleOff = maxTupleIsSetIndicatorOff + 1;
-        this.maxTupleOff = maxTupleIsSetIndicatorOff + 1 + (pageSize / 2);
-
-        this.minTuple = tupleWriter.createTupleReference();
-        this.maxTuple = tupleWriter.createTupleReference();
-    }
-
-    @Override
-    public void initBuffer() {
-        buf.put(minTupleIsSetIndicatorOff, (byte) 0);
-        buf.put(maxTupleIsSetIndicatorOff, (byte) 0);
-    }
-
-    @Override
-    public ICachedPage getPage() {
-        return page;
-    }
-
-    @Override
-    public void setPage(ICachedPage page) {
-        this.page = page;
-        this.buf = page.getBuffer();
-    }
-
-    @Override
-    public void writeMinTuple(ITupleReference tuple) {
-        tupleWriter.writeTuple(tuple, buf.array(), minTupleOff);
-        buf.put(minTupleIsSetIndicatorOff, (byte) 1);
-    }
-
-    @Override
-    public void writeMaxTuple(ITupleReference tuple) {
-        tupleWriter.writeTuple(tuple, buf.array(), maxTupleOff);
-        buf.put(maxTupleIsSetIndicatorOff, (byte) 1);
-    }
-
-    @Override
-    public boolean isMinTupleSet() {
-        return buf.get(minTupleIsSetIndicatorOff) == (byte) 1 ? true : false;
-    }
-
-    @Override
-    public boolean isMaxTupleSet() {
-        return buf.get(maxTupleIsSetIndicatorOff) == (byte) 1 ? true : false;
-    }
-
-    @Override
-    public ITupleReference getMinTuple() {
-        minTuple.resetByTupleOffset(buf, minTupleOff);
-        return minTuple;
-    }
-
-    @Override
-    public ITupleReference getMaxTuple() {
-        maxTuple.resetByTupleOffset(buf, maxTupleOff);
-        return maxTuple;
-    }
-}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/frames/LSMComponentFilterFrameFactory.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/frames/LSMComponentFilterFrameFactory.java
index 7a75534..999f72b0 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/frames/LSMComponentFilterFrameFactory.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/frames/LSMComponentFilterFrameFactory.java
@@ -20,21 +20,19 @@
 package org.apache.hyracks.storage.am.lsm.common.frames;
 
 import org.apache.hyracks.storage.am.common.api.ITreeIndexTupleWriterFactory;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilterFrame;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilterFrameFactory;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilterReference;
 
 public class LSMComponentFilterFrameFactory implements ILSMComponentFilterFrameFactory {
 
     private final ITreeIndexTupleWriterFactory tupleWriterFactory;
-    private final int pageSize;
 
-    public LSMComponentFilterFrameFactory(ITreeIndexTupleWriterFactory tupleWriterFactory, int pageSize) {
+    public LSMComponentFilterFrameFactory(ITreeIndexTupleWriterFactory tupleWriterFactory) {
         this.tupleWriterFactory = tupleWriterFactory;
-        this.pageSize = pageSize;
     }
 
     @Override
-    public ILSMComponentFilterFrame createFrame() {
-        return new LSMComponentFilterFrame(tupleWriterFactory.createTupleWriter(), pageSize);
+    public ILSMComponentFilterReference createFrame() {
+        return new LSMComponentFilterReference(tupleWriterFactory.createTupleWriter());
     }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/frames/LSMComponentFilterReference.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/frames/LSMComponentFilterReference.java
new file mode 100644
index 0000000..f04cbb0
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/frames/LSMComponentFilterReference.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hyracks.storage.am.lsm.common.frames;
+
+import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.hyracks.data.std.primitive.BooleanPointable;
+import org.apache.hyracks.data.std.primitive.IntegerPointable;
+import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexTupleReference;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexTupleWriter;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilterReference;
+
+/**
+ * When written to disk:
+ * [min set?][max set?][min length][min][max length][max]
+ */
+public class LSMComponentFilterReference implements ILSMComponentFilterReference {
+
+    private static final int MIN_SET_INDICATOR_OFFSET = 0;
+    private static final int MAX_SET_INDICATOR_OFFSET = 1;
+
+    private final ArrayBackedValueStorage min;
+    private final ArrayBackedValueStorage max;
+    private ArrayBackedValueStorage binaryFilter;
+    private final ITreeIndexTupleWriter tupleWriter;
+    private ITreeIndexTupleReference minTuple;
+    private ITreeIndexTupleReference maxTuple;
+
+    public LSMComponentFilterReference(ITreeIndexTupleWriter tupleWriter) {
+        this.tupleWriter = tupleWriter;
+        min = new ArrayBackedValueStorage();
+        max = new ArrayBackedValueStorage();
+        binaryFilter = new ArrayBackedValueStorage();
+        minTuple = tupleWriter.createTupleReference();
+        maxTuple = tupleWriter.createTupleReference();
+    }
+
+    @Override
+    public void writeMinTuple(ITupleReference tuple) {
+        binaryFilter.reset();
+        min.reset();
+        min.setSize(tupleWriter.bytesRequired(tuple));
+        tupleWriter.writeTuple(tuple, min.getByteArray(), 0);
+    }
+
+    @Override
+    public void writeMaxTuple(ITupleReference tuple) {
+        binaryFilter.reset();
+        max.reset();
+        max.setSize(tupleWriter.bytesRequired(tuple));
+        tupleWriter.writeTuple(tuple, max.getByteArray(), 0);
+    }
+
+    @Override
+    public boolean isMinTupleSet() {
+        return min.getLength() > 0;
+    }
+
+    @Override
+    public boolean isMaxTupleSet() {
+        return max.getLength() > 0;
+    }
+
+    @Override
+    public ITupleReference getMinTuple() {
+        minTuple.resetByTupleOffset(min.getByteArray(), 0);
+        return minTuple;
+    }
+
+    @Override
+    public ITupleReference getMaxTuple() {
+        maxTuple.resetByTupleOffset(max.getByteArray(), 0);
+        return maxTuple;
+    }
+
+    @Override
+    public byte[] getByteArray() {
+        if (binaryFilter.getLength() == 0) {
+            int binarySize = 2;
+            if (min.getLength() > 0) {
+                binarySize += Integer.BYTES + min.getLength();
+            }
+            if (max.getLength() > 0) {
+                binarySize += Integer.BYTES + max.getLength();
+            }
+            binaryFilter.setSize(binarySize);
+            byte[] buf = binaryFilter.getByteArray();
+            BooleanPointable.setBoolean(buf, MIN_SET_INDICATOR_OFFSET, min.getLength() == 0);
+            BooleanPointable.setBoolean(buf, MAX_SET_INDICATOR_OFFSET, max.getLength() == 0);
+            int offset = 2;
+            if (min.getLength() > 0) {
+                IntegerPointable.setInteger(buf, offset, min.getLength());
+                offset += Integer.BYTES;
+                System.arraycopy(min.getByteArray(), 0, buf, offset, min.getLength());
+            }
+            if (max.getLength() > 0) {
+                IntegerPointable.setInteger(buf, offset, max.getLength());
+                offset += Integer.BYTES;
+                System.arraycopy(max.getByteArray(), 0, buf, offset, max.getLength());
+            }
+        }
+        return binaryFilter.getByteArray();
+    }
+
+    @Override
+    public int getStartOffset() {
+        return 0;
+    }
+
+    @Override
+    public int getLength() {
+        return binaryFilter.getLength();
+    }
+
+    @Override
+    public void reset() {
+        min.reset();
+        max.reset();
+        binaryFilter.reset();
+    }
+
+    @Override
+    public void set(byte[] bytes, int start, int length) {
+        boolean isMinSet = BooleanPointable.getBoolean(bytes, MIN_SET_INDICATOR_OFFSET + start);
+        boolean isMaxSet = BooleanPointable.getBoolean(bytes, MAX_SET_INDICATOR_OFFSET + start);
+        int srcOffset = start + 2;
+        if (isMinSet) {
+            min.setSize(IntegerPointable.getInteger(bytes, srcOffset));
+            srcOffset += Integer.BYTES;
+            System.arraycopy(bytes, srcOffset, min.getByteArray(), 0, min.getLength());
+            srcOffset += min.getLength();
+        }
+        if (isMaxSet) {
+            max.setSize(IntegerPointable.getInteger(bytes, srcOffset));
+            srcOffset += Integer.BYTES;
+            System.arraycopy(bytes, srcOffset, max.getByteArray(), 0, max.getLength());
+        }
+        binaryFilter.reset();
+    }
+
+    @Override
+    public void set(IValueReference pointer) {
+        set(pointer.getByteArray(), pointer.getStartOffset(), pointer.getLength());
+    }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/freepage/VirtualFreePageManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/freepage/VirtualFreePageManager.java
index 5771728..08c75dc 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/freepage/VirtualFreePageManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/freepage/VirtualFreePageManager.java
@@ -25,7 +25,7 @@
 import org.apache.hyracks.storage.am.common.api.IPageManager;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrame;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
 import org.apache.hyracks.storage.common.buffercache.ICachedPage;
 import org.apache.hyracks.storage.common.file.BufferedFileHandle;
@@ -44,34 +44,34 @@
     }
 
     @Override
-    public int takePage(ITreeIndexMetaDataFrame metaFrame) throws HyracksDataException {
+    public int takePage(ITreeIndexMetadataFrame metaFrame) throws HyracksDataException {
         // The very first call returns page id 2 because the BTree uses
         // the first page as metadata page, and the second page as root page.
         return currentPageId.incrementAndGet();
     }
 
     @Override
-    public int takeBlock(ITreeIndexMetaDataFrame metaFrame, int count) throws HyracksDataException {
+    public int takeBlock(ITreeIndexMetadataFrame metaFrame, int count) throws HyracksDataException {
         return currentPageId.getAndUpdate(operand -> operand + count) + 1;
     }
 
     @Override
-    public int getMaxPageId(ITreeIndexMetaDataFrame metaFrame) throws HyracksDataException {
+    public int getMaxPageId(ITreeIndexMetadataFrame metaFrame) throws HyracksDataException {
         return currentPageId.get();
     }
 
     @Override
-    public ITreeIndexMetaDataFrame createMetadataFrame() {
+    public ITreeIndexMetadataFrame createMetadataFrame() {
         return null;
     }
 
     @Override
-    public void releasePage(ITreeIndexMetaDataFrame metaFrame, int freePage) throws HyracksDataException {
+    public void releasePage(ITreeIndexMetadataFrame metaFrame, int freePage) throws HyracksDataException {
         throw new HyracksDataException("Pages of an in memory index are released through the virtual buffer cache");
     }
 
     @Override
-    public void releaseBlock(ITreeIndexMetaDataFrame metaFrame, int startingPage, int count)
+    public void releaseBlock(ITreeIndexMetadataFrame metaFrame, int startingPage, int count)
             throws HyracksDataException {
         throw new HyracksDataException("Pages of an in memory index are released through the virtual buffer cache");
     }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMComponent.java
index c678878..cf50c00 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMComponent.java
@@ -18,15 +18,21 @@
  */
 package org.apache.hyracks.storage.am.lsm.common.impls;
 
+import org.apache.hyracks.data.std.primitive.LongPointable;
+import org.apache.hyracks.storage.am.common.freepage.MutableArrayValueReference;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilter;
 
 public abstract class AbstractLSMComponent implements ILSMComponent {
 
+    public static final MutableArrayValueReference MARKER_LSN_KEY = new MutableArrayValueReference("Marker"
+            .getBytes());
+
     protected ComponentState state;
     protected int readerCount;
     protected final ILSMComponentFilter filter;
     protected long mostRecentMarkerLSN;
+    protected final LongPointable pointable = (LongPointable) LongPointable.FACTORY.createPointable();
 
     public AbstractLSMComponent(ILSMComponentFilter filter, long mostRecentMarkerLSN) {
         this.filter = filter;
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexFileManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexFileManager.java
index 43b6f92..731d312 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexFileManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexFileManager.java
@@ -37,7 +37,8 @@
 import org.apache.hyracks.api.io.FileReference;
 import org.apache.hyracks.api.io.IIOManager;
 import org.apache.hyracks.storage.am.common.api.ITreeIndex;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrame;
 import org.apache.hyracks.storage.am.common.api.IndexException;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexFileManager;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
@@ -94,7 +95,7 @@
             if (metadataPage < 0) {
                 return TreeIndexState.INVALID;
             }
-            ITreeIndexMetaDataFrame metadataFrame = treeIndex.getPageManager().createMetadataFrame();
+            ITreeIndexMetadataFrame metadataFrame = treeIndex.getPageManager().createMetadataFrame();
             ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(treeIndex.getFileId(), metadataPage),
                     false);
             page.acquireReadLatch();
@@ -102,7 +103,7 @@
                 metadataFrame.setPage(page);
                 if (!metadataFrame.isValid()) {
                     return TreeIndexState.INVALID;
-                } else if (metadataFrame.getVersion() != ITreeIndexMetaDataFrame.VERSION) {
+                } else if (metadataFrame.getVersion() != ITreeIndexFrame.Constants.VERSION) {
                     return TreeIndexState.VERSION_MISMATCH;
                 } else {
                     return TreeIndexState.VALID;
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentFilter.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentFilter.java
index f7bc2ad..32937a9 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentFilter.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentFilter.java
@@ -70,7 +70,7 @@
             tupleWriter.writeTuple(tuple, minTupleBytes, 0);
             minTupleBuf = ByteBuffer.wrap(minTupleBytes);
             minTuple = tupleWriter.createTupleReference();
-            ((ITreeIndexTupleReference) minTuple).resetByTupleOffset(minTupleBuf, 0);
+            ((ITreeIndexTupleReference) minTuple).resetByTupleOffset(minTupleBuf.array(), 0);
         } else {
             int c = cmp.compare(tuple, minTuple);
             if (c < 0) {
@@ -82,7 +82,7 @@
                 } else {
                     tupleWriter.writeTuple(tuple, minTupleBytes, 0);
                 }
-                ((ITreeIndexTupleReference) minTuple).resetByTupleOffset(minTupleBuf, 0);
+                ((ITreeIndexTupleReference) minTuple).resetByTupleOffset(minTupleBuf.array(), 0);
             }
         }
         if (maxTuple == null) {
@@ -91,7 +91,7 @@
             tupleWriter.writeTuple(tuple, maxTupleBytes, 0);
             maxTupleBuf = ByteBuffer.wrap(maxTupleBytes);
             maxTuple = tupleWriter.createTupleReference();
-            ((ITreeIndexTupleReference) maxTuple).resetByTupleOffset(maxTupleBuf, 0);
+            ((ITreeIndexTupleReference) maxTuple).resetByTupleOffset(maxTupleBuf.array(), 0);
         } else {
             int c = cmp.compare(tuple, maxTuple);
             if (c > 0) {
@@ -103,7 +103,7 @@
                 } else {
                     tupleWriter.writeTuple(tuple, maxTupleBytes, 0);
                 }
-                ((ITreeIndexTupleReference) maxTuple).resetByTupleOffset(maxTupleBuf, 0);
+                ((ITreeIndexTupleReference) maxTuple).resetByTupleOffset(maxTupleBuf.array(), 0);
             }
         }
     }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentFilterManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentFilterManager.java
index 0fbbbdf..a8b4c70 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentFilterManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentFilterManager.java
@@ -25,22 +25,19 @@
 import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
 import org.apache.hyracks.storage.am.common.api.IMetadataPageManager;
 import org.apache.hyracks.storage.am.common.api.ITreeIndex;
+import org.apache.hyracks.storage.am.common.freepage.MutableArrayValueReference;
 import org.apache.hyracks.storage.am.common.ophelpers.MultiComparator;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilter;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilterFrame;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilterFrameFactory;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilterManager;
-import org.apache.hyracks.storage.common.buffercache.IBufferCache;
-import org.apache.hyracks.storage.common.buffercache.ICachedPage;
-import org.apache.hyracks.storage.common.file.BufferedFileHandle;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilterReference;
 
 public class LSMComponentFilterManager implements ILSMComponentFilterManager {
 
-    private final IBufferCache bufferCache;
+    public static final MutableArrayValueReference FILTER_KEY = new MutableArrayValueReference("Filter".getBytes());
     private final ILSMComponentFilterFrameFactory filterFrameFactory;
 
-    public LSMComponentFilterManager(IBufferCache bufferCache, ILSMComponentFilterFrameFactory filterFrameFactory) {
-        this.bufferCache = bufferCache;
+    public LSMComponentFilterManager(ILSMComponentFilterFrameFactory filterFrameFactory) {
         this.filterFrameFactory = filterFrameFactory;
     }
 
@@ -56,11 +53,8 @@
     @Override
     public void writeFilterInfo(ILSMComponentFilter filter, ITreeIndex treeIndex) throws HyracksDataException {
         IMetadataPageManager treeMetaManager = (IMetadataPageManager) treeIndex.getPageManager();
-        ICachedPage filterPage = treeMetaManager.getFilterPage();
+        ILSMComponentFilterReference filterFrame = filterFrameFactory.createFrame();
         try {
-            ILSMComponentFilterFrame filterFrame = filterFrameFactory.createFrame();
-            filterFrame.setPage(filterPage);
-            filterFrame.initBuffer();
             if (filter.getMinTuple() != null) {
                 filterFrame.writeMinTuple(filter.getMinTuple());
             }
@@ -68,39 +62,23 @@
                 filterFrame.writeMaxTuple(filter.getMaxTuple());
             }
         } finally {
-            treeMetaManager.setFilterPage(filterPage);
+            treeMetaManager.put(treeMetaManager.createMetadataFrame(), FILTER_KEY, filterFrame);
         }
     }
 
     @Override
     public boolean readFilterInfo(ILSMComponentFilter filter, ITreeIndex treeIndex) throws HyracksDataException {
-        int fileId = treeIndex.getFileId();
         IMetadataPageManager treeMetaManager = (IMetadataPageManager) treeIndex.getPageManager();
-        int componentFilterPageId = treeMetaManager.getFilterPageId();
-        if (componentFilterPageId < 0) {
+        ILSMComponentFilterReference filterFrame = filterFrameFactory.createFrame();
+        treeMetaManager.get(treeMetaManager.createMetadataFrame(), FILTER_KEY, filterFrame);
+        // TODO: Filters never have one of min/max set and the other not
+        if (!filterFrame.isMinTupleSet() || !filterFrame.isMaxTupleSet()) {
             return false;
         }
-
-        ICachedPage filterPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, componentFilterPageId),
-                false);
-
-        filterPage.acquireReadLatch();
-        try {
-            ILSMComponentFilterFrame filterFrame = filterFrameFactory.createFrame();
-            filterFrame.setPage(filterPage);
-
-            if (!filterFrame.isMinTupleSet() || !filterFrame.isMaxTupleSet()) {
-                return false;
-            }
-            List<ITupleReference> filterTuples = new ArrayList<>();
-            filterTuples.add(filterFrame.getMinTuple());
-            filterTuples.add(filterFrame.getMaxTuple());
-            updateFilterInfo(filter, filterTuples);
-
-        } finally {
-            filterPage.releaseReadLatch();
-            bufferCache.unpin(filterPage);
-        }
+        List<ITupleReference> filterTuples = new ArrayList<>();
+        filterTuples.add(filterFrame.getMinTuple());
+        filterTuples.add(filterFrame.getMaxTuple());
+        updateFilterInfo(filter, filterTuples);
         return true;
     }
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
index 896d513..a6868f0 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
@@ -66,7 +66,7 @@
         //only durable indexes are replicated
         this.replicationEnabled = replicationEnabled && lsmIndex.isDurable();
         if (replicationEnabled) {
-            this.componentsToBeReplicated = new ArrayList<ILSMComponent>();
+            this.componentsToBeReplicated = new ArrayList<>();
         }
     }
 
@@ -314,7 +314,7 @@
                         for (ILSMComponent inactiveComp : inactiveDiskComponents) {
                             if (((AbstractDiskLSMComponent) inactiveComp).getFileReferenceCount() == 1) {
                                 if (inactiveDiskComponentsToBeDeleted == null) {
-                                    inactiveDiskComponentsToBeDeleted = new LinkedList<ILSMComponent>();
+                                    inactiveDiskComponentsToBeDeleted = new LinkedList<>();
                                 }
                                 inactiveDiskComponentsToBeDeleted.add(inactiveComp);
                             }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/VirtualBufferCache.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/VirtualBufferCache.java
index 416f859..49f9060 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/VirtualBufferCache.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/VirtualBufferCache.java
@@ -22,9 +22,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -36,7 +34,7 @@
 import org.apache.hyracks.storage.common.buffercache.ICachedPage;
 import org.apache.hyracks.storage.common.buffercache.IExtraPageBlockHelper;
 import org.apache.hyracks.storage.common.buffercache.IFIFOPageQueue;
-import org.apache.hyracks.storage.common.buffercache.IQueueInfo;
+import org.apache.hyracks.storage.common.buffercache.VirtualPage;
 import org.apache.hyracks.storage.common.file.BufferedFileHandle;
 import org.apache.hyracks.storage.common.file.IFileMapManager;
 import org.apache.hyracks.storage.common.file.TransientFileMapManager;
@@ -106,19 +104,19 @@
                 VirtualPage prev = null;
                 VirtualPage curr = bucket.cachedPage;
                 while (curr != null) {
-                    if (BufferedFileHandle.getFileId(curr.dpid) == fileId) {
+                    if (BufferedFileHandle.getFileId(curr.dpid()) == fileId) {
                         if (prev == null) {
-                            bucket.cachedPage = curr.next;
+                            bucket.cachedPage = curr.next();
                             curr.reset();
                             curr = bucket.cachedPage;
                         } else {
-                            prev.next = curr.next;
+                            prev.next(curr.next());
                             curr.reset();
-                            curr = prev.next;
+                            curr = prev.next();
                         }
                     } else {
                         prev = curr;
-                        curr = curr.next;
+                        curr = curr.next();
                     }
                 }
             } finally {
@@ -134,18 +132,18 @@
             int end = nextFree - 1;
             while (start < end) {
                 VirtualPage lastUsed = pages.get(end);
-                while (end > 0 && lastUsed.dpid == -1) {
+                while (end > 0 && lastUsed.dpid() == -1) {
                     --end;
                     lastUsed = pages.get(end);
                 }
 
                 if (end == 0) {
-                    nextFree = lastUsed.dpid == -1 ? 0 : 1;
+                    nextFree = lastUsed.dpid() == -1 ? 0 : 1;
                     break;
                 }
 
                 VirtualPage firstUnused = pages.get(start);
-                while (start < end && firstUnused.dpid != -1) {
+                while (start < end && firstUnused.dpid() != -1) {
                     ++start;
                     firstUnused = pages.get(start);
                 }
@@ -177,10 +175,10 @@
         try {
             page = bucket.cachedPage;
             while (page != null) {
-                if (page.dpid == dpid) {
+                if (page.dpid() == dpid) {
                     return page;
                 }
-                page = page.next;
+                page = page.next();
             }
 
             if (!newPage) {
@@ -189,7 +187,7 @@
             }
 
             page = getOrAllocPage(dpid);
-            page.next = bucket.cachedPage;
+            page.next(bucket.cachedPage);
             bucket.cachedPage = page;
         } finally {
             bucket.bucketLock.unlock();
@@ -207,14 +205,14 @@
         VirtualPage page;
         synchronized (pages) {
             if (nextFree >= pages.size()) {
-                page = new VirtualPage(allocator.allocate(pageSize, 1)[0]);
-                page.multiplier = 1;
+                page = new VirtualPage(allocator.allocate(pageSize, 1)[0], pageSize);
+                page.multiplier(1);
                 pages.add(page);
             } else {
                 page = pages.get(nextFree);
             }
             ++nextFree;
-            page.dpid = dpid;
+            page.dpid(dpid);
         }
         return page;
     }
@@ -248,8 +246,8 @@
         } else {
             largePages.getAndAdd(multiplier - origMultiplier);
         }
-        ((VirtualPage) cPage).buffer = newBuffer;
-        ((VirtualPage) cPage).multiplier = multiplier;
+        ((VirtualPage) cPage).buffer(newBuffer);
+        ((VirtualPage) cPage).multiplier(multiplier);
     }
 
     @Override
@@ -352,76 +350,6 @@
         }
     }
 
-    private class VirtualPage implements ICachedPage {
-        ByteBuffer buffer;
-        final ReadWriteLock latch;
-        volatile long dpid;
-        int multiplier;
-        VirtualPage next;
-
-        public VirtualPage(ByteBuffer buffer) {
-            this.buffer = buffer;
-            latch = new ReentrantReadWriteLock(true);
-            dpid = -1;
-            next = null;
-        }
-
-        public void reset() {
-            dpid = -1;
-            next = null;
-        }
-
-        @Override
-        public ByteBuffer getBuffer() {
-            return buffer;
-        }
-
-        @Override
-        public void acquireReadLatch() {
-            latch.readLock().lock();
-        }
-
-        @Override
-        public void releaseReadLatch() {
-            latch.readLock().unlock();
-        }
-
-        @Override
-        public void acquireWriteLatch() {
-            latch.writeLock().lock();
-        }
-
-        @Override
-        public void releaseWriteLatch(boolean markDirty) {
-            latch.writeLock().unlock();
-        }
-
-        @Override
-        public boolean confiscated() {
-            return false;
-        }
-
-        @Override
-        public IQueueInfo getQueueInfo() {
-            return null;
-        }
-
-        @Override
-        public void setQueueInfo(IQueueInfo queueInfo) {
-            throw new UnsupportedOperationException();
-        }
-
-        @Override
-        public int getPageSize() {
-            return pageSize;
-        }
-
-        @Override
-        public int getFrameSizeMultiplier() {
-            return multiplier;
-        }
-    }
-
     //These 4 methods aren't applicable here.
     @Override
     public int createMemFile() throws HyracksDataException {
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
index cb4b67a..403fa09 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
@@ -68,6 +68,7 @@
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
 import org.apache.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
+import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType;
 import org.apache.hyracks.storage.am.lsm.common.freepage.VirtualFreePageManager;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.impls.BTreeFactory;
@@ -772,6 +773,7 @@
                 if (isEmptyComponent) {
                     cleanupArtifacts();
                 } else {
+                    ioOpCallback.afterOperation(LSMOperationType.FLUSH, null, component);
                     lsmHarness.addBulkLoadedComponent(component);
                 }
             }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/util/InvertedIndexUtils.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/util/InvertedIndexUtils.java
index 702e653..b8dd258 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/util/InvertedIndexUtils.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/util/InvertedIndexUtils.java
@@ -164,9 +164,8 @@
         if (filterCmpFactories != null) {
             TypeAwareTupleWriterFactory filterTupleWriterFactory = new TypeAwareTupleWriterFactory(filterTypeTraits);
             filterFactory = new LSMComponentFilterFactory(filterTupleWriterFactory, filterCmpFactories);
-            filterFrameFactory = new LSMComponentFilterFrameFactory(filterTupleWriterFactory,
-                    diskBufferCache.getPageSize());
-            filterManager = new LSMComponentFilterManager(diskBufferCache, filterFrameFactory);
+            filterFrameFactory = new LSMComponentFilterFrameFactory(filterTupleWriterFactory);
+            filterManager = new LSMComponentFilterManager(filterFrameFactory);
         }
         LSMInvertedIndex invIndex = new LSMInvertedIndex(ioManager, virtualBufferCaches, invIndexFactory,
                 deletedKeysBTreeFactory,
@@ -216,9 +215,8 @@
         if (filterCmpFactories != null) {
             TypeAwareTupleWriterFactory filterTupleWriterFactory = new TypeAwareTupleWriterFactory(filterTypeTraits);
             filterFactory = new LSMComponentFilterFactory(filterTupleWriterFactory, filterCmpFactories);
-            filterFrameFactory = new LSMComponentFilterFrameFactory(filterTupleWriterFactory,
-                    diskBufferCache.getPageSize());
-            filterManager = new LSMComponentFilterManager(diskBufferCache, filterFrameFactory);
+            filterFrameFactory = new LSMComponentFilterFrameFactory(filterTupleWriterFactory);
+            filterManager = new LSMComponentFilterManager(filterFrameFactory);
         }
         PartitionedLSMInvertedIndex invIndex = new PartitionedLSMInvertedIndex(ioManager, virtualBufferCaches,
                 invIndexFactory,
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
index 0ce35d4..bfaef38 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
@@ -67,6 +67,7 @@
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
 import org.apache.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
+import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentFilterManager;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMTreeIndexAccessor;
@@ -599,6 +600,7 @@
                 if (isEmptyComponent) {
                     cleanupArtifacts();
                 } else {
+                    ioOpCallback.afterOperation(LSMOperationType.FLUSH, null, component);
                     lsmHarness.addBulkLoadedComponent(component);
                 }
             }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
index 5565872..6b67494 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
@@ -58,6 +58,7 @@
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
 import org.apache.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
+import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentFilterManager;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMIndexSearchCursor;
@@ -433,6 +434,7 @@
                 if (isEmptyComponent) {
                     cleanupArtifacts();
                 } else {
+                    ioOpCallback.afterOperation(LSMOperationType.FLUSH, null, component);
                     lsmHarness.addBulkLoadedComponent(component);
                 }
             }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/TreeTupleSorter.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/TreeTupleSorter.java
index 429c65d..70b1141 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/TreeTupleSorter.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/TreeTupleSorter.java
@@ -61,11 +61,13 @@
         cmp = MultiComparator.create(comparatorFactories);
     }
 
+    @Override
     public void reset() {
         numTuples = 0;
         currentTupleIndex = 0;
     }
 
+    @Override
     public boolean hasNext() throws HyracksDataException {
         if (numTuples <= currentTupleIndex) {
             return false;
@@ -77,17 +79,19 @@
                 false);
         try {
             leafFrame1.setPage(node1);
-            frameTuple1.resetByTupleOffset(leafFrame1.getBuffer(), tPointers[currentTupleIndex * 2 + 1]);
+            frameTuple1.resetByTupleOffset(leafFrame1.getBuffer().array(), tPointers[currentTupleIndex * 2 + 1]);
         } finally {
             bufferCache.unpin(node1);
         }
         return true;
     }
 
+    @Override
     public void next() {
         currentTupleIndex++;
     }
 
+    @Override
     public ITupleReference getTuple() {
         return frameTuple1;
     }
@@ -138,8 +142,9 @@
                 }
                 --c;
             }
-            if (b > c)
+            if (b > c) {
                 break;
+            }
             swap(tPointers, b++, c--);
         }
 
@@ -185,8 +190,8 @@
         leafFrame2.setPage(node2);
 
         try {
-            frameTuple1.resetByTupleOffset(leafFrame1.getBuffer(), j1);
-            frameTuple2.resetByTupleOffset(leafFrame2.getBuffer(), j2);
+            frameTuple1.resetByTupleOffset(leafFrame1.getBuffer().array(), j1);
+            frameTuple2.resetByTupleOffset(leafFrame2.getBuffer().array(), j2);
 
             return cmp.selectiveFieldCompare(frameTuple1, frameTuple2, comparatorFields);
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/tuples/LSMRTreeTupleReference.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/tuples/LSMRTreeTupleReference.java
index 4c5c9a4..f49ecb0 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/tuples/LSMRTreeTupleReference.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/tuples/LSMRTreeTupleReference.java
@@ -39,7 +39,7 @@
     public boolean isAntimatter() {
         // Check if the leftmost bit is 0 or 1.
         final byte mask = (byte) (1 << 7);
-        if ((buf.array()[tupleStartOff] & mask) != 0) {
+        if ((buf[tupleStartOff] & mask) != 0) {
             return true;
         }
         return false;
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/tuples/LSMRTreeTupleReferenceForPointMBR.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/tuples/LSMRTreeTupleReferenceForPointMBR.java
index 48b22ec..4ea10a3 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/tuples/LSMRTreeTupleReferenceForPointMBR.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/tuples/LSMRTreeTupleReferenceForPointMBR.java
@@ -19,8 +19,6 @@
 
 package org.apache.hyracks.storage.am.lsm.rtree.tuples;
 
-import java.nio.ByteBuffer;
-
 import org.apache.hyracks.api.dataflow.value.ITypeTraits;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMTreeTupleReference;
@@ -36,7 +34,7 @@
     private final int nullFlagsBytes;
     private final int[] decodedFieldSlots;
 
-    private ByteBuffer buf;
+    private byte[] buf;
     private int tupleStartOff;
     private int dataStartOff;
     private final boolean antimatterAware;
@@ -55,7 +53,7 @@
     }
 
     @Override
-    public void resetByTupleOffset(ByteBuffer buf, int tupleStartOff) {
+    public void resetByTupleOffset(byte[] buf, int tupleStartOff) {
         this.buf = buf;
         this.tupleStartOff = tupleStartOff;
 
@@ -74,7 +72,7 @@
             decodedFieldSlots[field++] = decodedFieldSlots[i];
         }
         //step3. decode field slots for value field
-        encDec.reset(buf.array(), tupleStartOff + nullFlagsBytes);
+        encDec.reset(buf, tupleStartOff + nullFlagsBytes);
         for (int i = inputKeyFieldCount; i < inputTotalFieldCount; i++) {
             if (!typeTraits[i].isFixedLength()) {
                 //value fields
@@ -92,7 +90,7 @@
 
     @Override
     public void resetByTupleIndex(ITreeIndexFrame frame, int tupleIndex) {
-        resetByTupleOffset(frame.getBuffer(), frame.getTupleOffset(tupleIndex));
+        resetByTupleOffset(frame.getBuffer().array(), frame.getTupleOffset(tupleIndex));
     }
 
     @Override
@@ -112,7 +110,7 @@
 
     @Override
     public byte[] getFieldData(int fIdx) {
-        return buf.array();
+        return buf;
     }
 
     @Override
@@ -154,7 +152,7 @@
     public boolean isAntimatter() {
         // Check if the leftmost bit is 0 or 1.
         final byte mask = (byte) (1 << 7);
-        if ((buf.array()[tupleStartOff] & mask) != 0) {
+        if ((buf[tupleStartOff] & mask) != 0) {
             return true;
         }
         return false;
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java
index 158b095..624c7de 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java
@@ -128,9 +128,8 @@
         if (filterCmpFactories != null) {
             TypeAwareTupleWriterFactory filterTupleWriterFactory = new TypeAwareTupleWriterFactory(filterTypeTraits);
             filterFactory = new LSMComponentFilterFactory(filterTupleWriterFactory, filterCmpFactories);
-            filterFrameFactory = new LSMComponentFilterFrameFactory(filterTupleWriterFactory,
-                    diskBufferCache.getPageSize());
-            filterManager = new LSMComponentFilterManager(diskBufferCache, filterFrameFactory);
+            filterFrameFactory = new LSMComponentFilterFrameFactory(filterTupleWriterFactory);
+            filterManager = new LSMComponentFilterManager(filterFrameFactory);
         }
         ILSMIndexFileManager fileNameManager = new LSMRTreeFileManager(ioManager, diskFileMapProvider, file,
                 diskRTreeFactory,
@@ -214,9 +213,8 @@
         if (filterCmpFactories != null) {
             TypeAwareTupleWriterFactory filterTupleWriterFactory = new TypeAwareTupleWriterFactory(filterTypeTraits);
             filterFactory = new LSMComponentFilterFactory(filterTupleWriterFactory, filterCmpFactories);
-            filterFrameFactory = new LSMComponentFilterFrameFactory(filterTupleWriterFactory,
-                    diskBufferCache.getPageSize());
-            filterManager = new LSMComponentFilterManager(diskBufferCache, filterFrameFactory);
+            filterFrameFactory = new LSMComponentFilterFrameFactory(filterTupleWriterFactory);
+            filterManager = new LSMComponentFilterManager(filterFrameFactory);
         }
         ILSMIndexFileManager fileNameManager = new LSMRTreeWithAntiMatterTuplesFileManager(ioManager,
                 diskFileMapProvider, file,
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RStarTreePolicy.java b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RStarTreePolicy.java
index a46f755..bf37eab 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RStarTreePolicy.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RStarTreePolicy.java
@@ -220,19 +220,19 @@
         }
 
         int tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
-        frameTuple.resetByTupleOffset(buf, tupleOff);
+        frameTuple.resetByTupleOffset(buf.array(), tupleOff);
         int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, keyValueProviders.length);
 
         splitKey.initData(splitKeySize);
         leftRTreeFrame.adjustMBR();
         rTreeTupleWriterleftRTreeFrame.writeTupleFields(leftRTreeFrame.getMBRTuples(), 0,
                 rTreeSplitKey.getLeftPageBuffer(), 0);
-        rTreeSplitKey.getLeftTuple().resetByTupleOffset(rTreeSplitKey.getLeftPageBuffer(), 0);
+        rTreeSplitKey.getLeftTuple().resetByTupleOffset(rTreeSplitKey.getLeftPageBuffer().array(), 0);
 
         ((IRTreeFrame) rightFrame).adjustMBR();
         rTreeTupleWriterRightFrame.writeTupleFields(((RTreeNSMFrame) rightFrame).getMBRTuples(), 0,
                 rTreeSplitKey.getRightPageBuffer(), 0);
-        rTreeSplitKey.getRightTuple().resetByTupleOffset(rTreeSplitKey.getRightPageBuffer(), 0);
+        rTreeSplitKey.getRightTuple().resetByTupleOffset(rTreeSplitKey.getRightPageBuffer().array(), 0);
 
         tupleEntries1.clear();
         tupleEntries2.clear();
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreeNSMFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreeNSMFrame.java
index 84cb82e..9acfd39 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreeNSMFrame.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreeNSMFrame.java
@@ -35,20 +35,20 @@
 import org.apache.hyracks.storage.common.buffercache.IExtraPageBlockHelper;
 
 public abstract class RTreeNSMFrame extends TreeIndexNSMFrame implements IRTreeFrame {
-    protected static final int pageNsnOff = flagOff + 1; // 22
-    protected static final int rightPageOff = pageNsnOff + 8; // 30
+    protected static final int PAGE_NSN_OFFSET = TreeIndexNSMFrame.RESERVED_HEADER_SIZE;
+    protected static final int RIGHT_PAGE_OFFSET = PAGE_NSN_OFFSET + 8;
 
     protected ITreeIndexTupleReference[] mbrTuples;
     protected ITreeIndexTupleReference cmpFrameTuple;
 
-    private static final double doubleEpsilon = computeDoubleEpsilon();
+    private static final double DOUBLE_EPSILON = computeDoubleEpsilon();
     protected final IPrimitiveValueProvider[] keyValueProviders;
 
     protected IRTreePolicy rtreePolicy;
     protected final boolean isPointMBR;
 
     public RTreeNSMFrame(ITreeIndexTupleWriter tupleWriter, IPrimitiveValueProvider[] keyValueProviders,
-                         RTreePolicyType rtreePolicyType, boolean isPointMBR) {
+            RTreePolicyType rtreePolicyType, boolean isPointMBR) {
         super(tupleWriter, new UnorderedSlotManager());
         this.mbrTuples = new ITreeIndexTupleReference[keyValueProviders.length];
         for (int i = 0; i < keyValueProviders.length; i++) {
@@ -58,9 +58,9 @@
         this.keyValueProviders = keyValueProviders;
 
         if (rtreePolicyType == RTreePolicyType.RTREE) {
-            rtreePolicy = new RTreePolicy(tupleWriter, keyValueProviders, cmpFrameTuple, totalFreeSpaceOff);
+            rtreePolicy = new RTreePolicy(tupleWriter, keyValueProviders, cmpFrameTuple, TOTAL_FREE_SPACE_OFFSET);
         } else {
-            rtreePolicy = new RStarTreePolicy(tupleWriter, keyValueProviders, cmpFrameTuple, totalFreeSpaceOff);
+            rtreePolicy = new RStarTreePolicy(tupleWriter, keyValueProviders, cmpFrameTuple, TOTAL_FREE_SPACE_OFFSET);
         }
         this.isPointMBR = isPointMBR;
     }
@@ -75,38 +75,38 @@
     }
 
     public static double doubleEpsilon() {
-        return doubleEpsilon;
+        return DOUBLE_EPSILON;
     }
 
     @Override
     public void initBuffer(byte level) {
         super.initBuffer(level);
-        buf.putLong(pageNsnOff, 0);
-        buf.putInt(rightPageOff, -1);
+        buf.putLong(PAGE_NSN_OFFSET, 0);
+        buf.putInt(RIGHT_PAGE_OFFSET, -1);
     }
 
     public void setTupleCount(int tupleCount) {
-        buf.putInt(tupleCountOff, tupleCount);
+        buf.putInt(Constants.TUPLE_COUNT_OFFSET, tupleCount);
     }
 
     @Override
     public void setPageNsn(long pageNsn) {
-        buf.putLong(pageNsnOff, pageNsn);
+        buf.putLong(PAGE_NSN_OFFSET, pageNsn);
     }
 
     @Override
     public long getPageNsn() {
-        return buf.getLong(pageNsnOff);
+        return buf.getLong(PAGE_NSN_OFFSET);
     }
 
     @Override
     public int getRightPage() {
-        return buf.getInt(rightPageOff);
+        return buf.getInt(RIGHT_PAGE_OFFSET);
     }
 
     @Override
     public void setRightPage(int rightPage) {
-        buf.putInt(rightPageOff, rightPage);
+        buf.putInt(RIGHT_PAGE_OFFSET, rightPage);
     }
 
     public ITreeIndexTupleReference[] getMBRTuples() {
@@ -115,7 +115,7 @@
 
     @Override
     public void split(ITreeIndexFrame rightFrame, ITupleReference tuple, ISplitKey splitKey,
-                      IExtraPageBlockHelper extraPageBlockHelper, IBufferCache bufferCache)
+            IExtraPageBlockHelper extraPageBlockHelper, IBufferCache bufferCache)
             throws HyracksDataException {
         rtreePolicy.split(this, buf, rightFrame, slotManager, frameTuple, tuple, splitKey);
     }
@@ -156,11 +156,25 @@
 
     @Override
     public int getPageHeaderSize() {
-        return rightPageOff + 4;
+        return RIGHT_PAGE_OFFSET + 4;
     }
 
     @Override
     public void setMultiComparator(MultiComparator cmp) {
         // currently, R-Tree Frames are unsorted
     }
+
+    @Override
+    public String toString() {
+        return new StringBuilder(this.getClass().getSimpleName()).append('\n').append(
+                "Tuple Count: " + getTupleCount()).append('\n').append("Free Space offset: " + buf
+                        .getInt(Constants.FREE_SPACE_OFFSET)).append('\n').append("Level: " + buf
+                                .get(Constants.LEVEL_OFFSET)).append('\n').append("LSN: "
+                                        + buf.getLong(PAGE_LSN_OFFSET)).append('\n').append(
+                                                "Total Free Space: " + buf.getInt(TOTAL_FREE_SPACE_OFFSET)).append(
+                                                        '\n').append("Flag: " + buf.get(
+                                                                FLAG_OFFSET)).append('\n')
+                .append("NSN: " + buf.getLong(PAGE_NSN_OFFSET)).append('\n').append("Right Page:")
+                .append(buf.getInt(RIGHT_PAGE_OFFSET)).toString();
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreeNSMInteriorFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreeNSMInteriorFrame.java
index a368c9a..efdb67f 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreeNSMInteriorFrame.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreeNSMInteriorFrame.java
@@ -141,10 +141,10 @@
     public boolean compact() {
         resetSpaceParams();
 
-        int tupleCount = buf.getInt(tupleCountOff);
-        int freeSpace = buf.getInt(freeSpaceOff);
+        int tupleCount = buf.getInt(Constants.TUPLE_COUNT_OFFSET);
+        int freeSpace = buf.getInt(Constants.FREE_SPACE_OFFSET);
 
-        ArrayList<SlotOffTupleOff> sortedTupleOffs = new ArrayList<SlotOffTupleOff>();
+        ArrayList<SlotOffTupleOff> sortedTupleOffs = new ArrayList<>();
         sortedTupleOffs.ensureCapacity(tupleCount);
         for (int i = 0; i < tupleCount; i++) {
             int slotOff = slotManager.getSlotOff(i);
@@ -155,7 +155,7 @@
 
         for (int i = 0; i < sortedTupleOffs.size(); i++) {
             int tupleOff = sortedTupleOffs.get(i).tupleOff;
-            frameTuple.resetByTupleOffset(buf, tupleOff);
+            frameTuple.resetByTupleOffset(buf.array(), tupleOff);
 
             int tupleEndOff = frameTuple.getFieldStart(frameTuple.getFieldCount() - 1)
                     + frameTuple.getFieldLength(frameTuple.getFieldCount() - 1);
@@ -166,8 +166,8 @@
             freeSpace += tupleLength;
         }
 
-        buf.putInt(freeSpaceOff, freeSpace);
-        buf.putInt(totalFreeSpaceOff, buf.capacity() - freeSpace - tupleCount * slotManager.getSlotSize());
+        buf.putInt(Constants.FREE_SPACE_OFFSET, freeSpace);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.capacity() - freeSpace - tupleCount * slotManager.getSlotSize());
 
         return false;
     }
@@ -175,23 +175,25 @@
     @Override
     public FrameOpSpaceStatus hasSpaceInsert(ITupleReference tuple) throws HyracksDataException {
         int bytesRequired = tupleWriter.bytesRequired(tuple) + childPtrSize;
-        if (bytesRequired + slotManager.getSlotSize() <= buf.capacity() - buf.getInt(freeSpaceOff)
-                - (buf.getInt(tupleCountOff) * slotManager.getSlotSize()))
+        if (bytesRequired + slotManager.getSlotSize() <= buf.capacity() - buf.getInt(Constants.FREE_SPACE_OFFSET)
+                - (buf.getInt(Constants.TUPLE_COUNT_OFFSET) * slotManager.getSlotSize())) {
             return FrameOpSpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
-        else if (bytesRequired + slotManager.getSlotSize() <= buf.getInt(totalFreeSpaceOff))
+        } else if (bytesRequired + slotManager.getSlotSize() <= buf.getInt(TOTAL_FREE_SPACE_OFFSET)) {
             return FrameOpSpaceStatus.SUFFICIENT_SPACE;
-        else
+        } else {
             return FrameOpSpaceStatus.INSUFFICIENT_SPACE;
+        }
     }
 
     public FrameOpSpaceStatus hasSpaceInsert(int bytesRequired) {
-        if (bytesRequired + slotManager.getSlotSize() <= buf.capacity() - buf.getInt(freeSpaceOff)
-                - (buf.getInt(tupleCountOff) * slotManager.getSlotSize()))
+        if (bytesRequired + slotManager.getSlotSize() <= buf.capacity() - buf.getInt(Constants.FREE_SPACE_OFFSET)
+                - (buf.getInt(Constants.TUPLE_COUNT_OFFSET) * slotManager.getSlotSize())) {
             return FrameOpSpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
-        else if (bytesRequired + slotManager.getSlotSize() <= buf.getInt(totalFreeSpaceOff))
+        } else if (bytesRequired + slotManager.getSlotSize() <= buf.getInt(TOTAL_FREE_SPACE_OFFSET)) {
             return FrameOpSpaceStatus.SUFFICIENT_SPACE;
-        else
+        } else {
             return FrameOpSpaceStatus.INSUFFICIENT_SPACE;
+        }
     }
 
     @Override
@@ -220,6 +222,7 @@
                         tupleB.getFieldData(cmp.getKeyFieldCount() - 1), getChildPointerOff(tupleB), childPtrSize);
     }
 
+    @Override
     public int getTupleSize(ITupleReference tuple) {
         return tupleWriter.bytesRequired(tuple) + childPtrSize;
     }
@@ -231,16 +234,17 @@
     @Override
     public void insert(ITupleReference tuple, int tupleIndex) {
         frameTuple.setFieldCount(tuple.getFieldCount());
-        slotManager.insertSlot(AbstractSlotManager.GREATEST_KEY_INDICATOR, buf.getInt(freeSpaceOff));
-        int freeSpace = buf.getInt(freeSpaceOff);
+        slotManager.insertSlot(AbstractSlotManager.GREATEST_KEY_INDICATOR, buf.getInt(Constants.FREE_SPACE_OFFSET));
+        int freeSpace = buf.getInt(Constants.FREE_SPACE_OFFSET);
         int bytesWritten = tupleWriter.writeTupleFields(tuple, 0, tuple.getFieldCount(), buf.array(), freeSpace);
         System.arraycopy(tuple.getFieldData(tuple.getFieldCount() - 1), getChildPointerOff(tuple), buf.array(),
                 freeSpace + bytesWritten, childPtrSize);
         int tupleSize = bytesWritten + childPtrSize;
 
-        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
-        buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + tupleSize);
-        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - tupleSize - slotManager.getSlotSize());
+        buf.putInt(Constants.TUPLE_COUNT_OFFSET, buf.getInt(Constants.TUPLE_COUNT_OFFSET) + 1);
+        buf.putInt(Constants.FREE_SPACE_OFFSET, buf.getInt(Constants.FREE_SPACE_OFFSET) + tupleSize);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.getInt(TOTAL_FREE_SPACE_OFFSET) - tupleSize - slotManager
+                .getSlotSize());
 
     }
 
@@ -250,7 +254,7 @@
         int slotOff = slotManager.getSlotOff(tupleIndex);
 
         int tupleOff = slotManager.getTupleOff(slotOff);
-        frameTuple.resetByTupleOffset(buf, tupleOff);
+        frameTuple.resetByTupleOffset(buf.array(), tupleOff);
         int tupleSize = tupleWriter.bytesRequired(frameTuple);
 
         // perform deletion (we just do a memcpy to overwrite the slot)
@@ -259,9 +263,9 @@
         System.arraycopy(buf.array(), slotStartOff, buf.array(), slotStartOff + slotManager.getSlotSize(), length);
 
         // maintain space information
-        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) - 1);
-        buf.putInt(totalFreeSpaceOff,
-                buf.getInt(totalFreeSpaceOff) + tupleSize + childPtrSize + slotManager.getSlotSize());
+        buf.putInt(Constants.TUPLE_COUNT_OFFSET, buf.getInt(Constants.TUPLE_COUNT_OFFSET) - 1);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET,
+                buf.getInt(TOTAL_FREE_SPACE_OFFSET) + tupleSize + childPtrSize + slotManager.getSlotSize());
     }
 
     @Override
@@ -288,12 +292,12 @@
 
     // For debugging.
     public ArrayList<Integer> getChildren(MultiComparator cmp) {
-        ArrayList<Integer> ret = new ArrayList<Integer>();
+        ArrayList<Integer> ret = new ArrayList<>();
         frameTuple.setFieldCount(cmp.getKeyFieldCount());
-        int tupleCount = buf.getInt(tupleCountOff);
+        int tupleCount = buf.getInt(Constants.TUPLE_COUNT_OFFSET);
         for (int i = 0; i < tupleCount; i++) {
             int tupleOff = slotManager.getTupleOff(slotManager.getSlotOff(i));
-            frameTuple.resetByTupleOffset(buf, tupleOff);
+            frameTuple.resetByTupleOffset(buf.array(), tupleOff);
             int intVal = IntegerPointable.getInteger(buf.array(), frameTuple.getFieldStart(frameTuple.getFieldCount() - 1)
             + frameTuple.getFieldLength(frameTuple.getFieldCount() - 1));
             ret.add(intVal);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreeNSMLeafFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreeNSMLeafFrame.java
index ed8f14c..b8b8877 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreeNSMLeafFrame.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreeNSMLeafFrame.java
@@ -71,18 +71,19 @@
         return true;
     }
 
+    @Override
     public int getTupleSize(ITupleReference tuple) {
         return tupleWriter.bytesRequired(tuple);
     }
 
     @Override
     public void insert(ITupleReference tuple, int tupleIndex) {
-        slotManager.insertSlot(-1, buf.getInt(freeSpaceOff));
-        int bytesWritten = tupleWriter.writeTuple(tuple, buf.array(), buf.getInt(freeSpaceOff));
-
-        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
-        buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + bytesWritten);
-        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - bytesWritten - slotManager.getSlotSize());
+        slotManager.insertSlot(-1, buf.getInt(Constants.FREE_SPACE_OFFSET));
+        int bytesWritten = tupleWriter.writeTuple(tuple, buf.array(), buf.getInt(Constants.FREE_SPACE_OFFSET));
+        buf.putInt(Constants.TUPLE_COUNT_OFFSET, buf.getInt(Constants.TUPLE_COUNT_OFFSET) + 1);
+        buf.putInt(Constants.FREE_SPACE_OFFSET, buf.getInt(Constants.FREE_SPACE_OFFSET) + bytesWritten);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.getInt(TOTAL_FREE_SPACE_OFFSET) - bytesWritten - slotManager
+                .getSlotSize());
     }
 
     @Override
@@ -90,7 +91,7 @@
         int slotOff = slotManager.getSlotOff(tupleIndex);
 
         int tupleOff = slotManager.getTupleOff(slotOff);
-        frameTuple.resetByTupleOffset(buf, tupleOff);
+        frameTuple.resetByTupleOffset(buf.array(), tupleOff);
         int tupleSize = tupleWriter.bytesRequired(frameTuple);
 
         // perform deletion (we just do a memcpy to overwrite the slot)
@@ -99,8 +100,9 @@
         System.arraycopy(buf.array(), slotStartOff, buf.array(), slotStartOff + slotManager.getSlotSize(), length);
 
         // maintain space information
-        buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) - 1);
-        buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + tupleSize + slotManager.getSlotSize());
+        buf.putInt(Constants.TUPLE_COUNT_OFFSET, buf.getInt(Constants.TUPLE_COUNT_OFFSET) - 1);
+        buf.putInt(TOTAL_FREE_SPACE_OFFSET, buf.getInt(TOTAL_FREE_SPACE_OFFSET) + tupleSize + slotManager
+                .getSlotSize());
     }
 
     @Override
@@ -108,6 +110,7 @@
         return frameTuple.getFieldCount();
     }
 
+    @Override
     public ITupleReference getBeforeTuple(ITupleReference tuple, int targetTupleIndex, MultiComparator cmp)
             throws HyracksDataException {
         // Examine the tuple index to determine whether it is valid or not.
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreePolicy.java b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreePolicy.java
index 2ca9548..31466ee 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreePolicy.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreePolicy.java
@@ -191,18 +191,18 @@
         }
 
         int tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
-        frameTuple.resetByTupleOffset(buf, tupleOff);
+        frameTuple.resetByTupleOffset(buf.array(), tupleOff);
         int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, keyValueProviders.length);
 
         splitKey.initData(splitKeySize);
         leftRTreeFrame.adjustMBR();
         rTreeTupleWriterLeftFrame.writeTupleFields(leftRTreeFrame.getMBRTuples(), 0, rTreeSplitKey.getLeftPageBuffer(), 0);
-        rTreeSplitKey.getLeftTuple().resetByTupleOffset(rTreeSplitKey.getLeftPageBuffer(), 0);
+        rTreeSplitKey.getLeftTuple().resetByTupleOffset(rTreeSplitKey.getLeftPageBuffer().array(), 0);
 
         ((IRTreeFrame) rightFrame).adjustMBR();
         rTreeTupleWriterRightFrame.writeTupleFields(((RTreeNSMFrame) rightFrame).getMBRTuples(), 0,
                 rTreeSplitKey.getRightPageBuffer(), 0);
-        rTreeSplitKey.getRightTuple().resetByTupleOffset(rTreeSplitKey.getRightPageBuffer(), 0);
+        rTreeSplitKey.getRightTuple().resetByTupleOffset(rTreeSplitKey.getRightPageBuffer().array(), 0);
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTree.java
index 0dbdaba..9cd16c4 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTree.java
@@ -988,7 +988,7 @@
                 //not a leaf
                 if (nodeFrontiers.indexOf(n) != 0) {
                     interiorFrame.setPage(n.page);
-                    mbrTuple.resetByTupleOffset(mbr, 0);
+                    mbrTuple.resetByTupleOffset(mbr.array(), 0);
                     interiorFrame.insert(mbrTuple, -1);
                     interiorFrame.getBuffer().putInt(
                             interiorFrame.getTupleOffset(interiorFrame.getTupleCount() - 1) + mbrTuple.getTupleSize(),
@@ -1041,7 +1041,7 @@
                 mbr = ByteBuffer.allocate(bytesRequired);
             }
             interiorFrameTupleWriter.writeTupleFields(((RTreeNSMFrame) lowerFrame).getMBRTuples(), 0, mbr, 0);
-            mbrTuple.resetByTupleOffset(mbr, 0);
+            mbrTuple.resetByTupleOffset(mbr.array(), 0);
 
             NodeFrontier frontier = nodeFrontiers.get(level);
             interiorFrame.setPage(frontier.page);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTreeOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTreeOpContext.java
index e25c644..4bc4d61 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTreeOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTreeOpContext.java
@@ -28,7 +28,7 @@
 import org.apache.hyracks.storage.am.common.api.IIndexOperationContext;
 import org.apache.hyracks.storage.am.common.api.IModificationOperationCallback;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexCursor;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrame;
 import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
 import org.apache.hyracks.storage.am.common.ophelpers.MultiComparator;
 import org.apache.hyracks.storage.am.rtree.api.IRTreeInteriorFrame;
@@ -46,7 +46,7 @@
     public ITreeIndexCursor cursor;
     public RTreeCursorInitialState cursorInitialState;
     public final IPageManager freePageManager;
-    public final ITreeIndexMetaDataFrame metaFrame;
+    public final ITreeIndexMetadataFrame metaFrame;
     public RTreeSplitKey splitKey;
     public ITupleReference tuple;
     // Used to record the pageIds and pageLsns of the visited pages.
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTreeSplitKey.java b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTreeSplitKey.java
index d89c12b..f6bdcfb 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTreeSplitKey.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTreeSplitKey.java
@@ -41,6 +41,7 @@
         this.rightTuple = rightTuple;
     }
 
+    @Override
     public void initData(int keySize) {
         // try to reuse existing memory from a lower-level split if possible
         this.keySize = keySize;
@@ -63,8 +64,8 @@
             rightPageBuf = ByteBuffer.wrap(rightPageData);
         }
 
-        leftTuple.resetByTupleOffset(leftPageBuf, 0);
-        rightTuple.resetByTupleOffset(rightPageBuf, 0);
+        leftTuple.resetByTupleOffset(leftPageBuf.array(), 0);
+        rightTuple.resetByTupleOffset(rightPageBuf.array(), 0);
     }
 
     public void resetLeftPage() {
@@ -93,18 +94,22 @@
         return rightTuple;
     }
 
+    @Override
     public int getLeftPage() {
         return leftPageBuf.getInt(keySize);
     }
 
+    @Override
     public int getRightPage() {
         return rightPageBuf.getInt(keySize);
     }
 
+    @Override
     public void setLeftPage(int page) {
         leftPageBuf.putInt(keySize, page);
     }
 
+    @Override
     public void setRightPage(int page) {
         rightPageBuf.putInt(keySize, page);
     }
@@ -115,12 +120,12 @@
         copy.leftPageData = leftPageData.clone();
         copy.leftPageBuf = ByteBuffer.wrap(copy.leftPageData);
         copy.leftTuple.setFieldCount(leftTuple.getFieldCount());
-        copy.leftTuple.resetByTupleOffset(copy.leftPageBuf, 0);
+        copy.leftTuple.resetByTupleOffset(copy.leftPageBuf.array(), 0);
 
         copy.rightPageData = rightPageData.clone();
         copy.rightPageBuf = ByteBuffer.wrap(copy.rightPageData);
         copy.rightTuple.setFieldCount(rightTuple.getFieldCount());
-        copy.rightTuple.resetByTupleOffset(copy.rightPageBuf, 0);
+        copy.rightTuple.resetByTupleOffset(copy.rightPageBuf.array(), 0);
         return copy;
     }
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/VirtualPage.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/VirtualPage.java
new file mode 100644
index 0000000..cfca77a
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/VirtualPage.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hyracks.storage.common.buffercache;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class VirtualPage implements ICachedPage {
+    private final ReadWriteLock latch;
+    private final int pageSize;
+    private ByteBuffer buffer;
+    private volatile long dpid;
+    private int multiplier;
+    private VirtualPage next;
+
+    public VirtualPage(ByteBuffer buffer, int pageSize) {
+        this.buffer = buffer;
+        this.pageSize = pageSize;
+        latch = new ReentrantReadWriteLock(true);
+        dpid = -1;
+        next = null;
+    }
+
+    public void reset() {
+        dpid = -1;
+        next = null;
+    }
+
+    @Override
+    public ByteBuffer getBuffer() {
+        return buffer;
+    }
+
+    @Override
+    public void acquireReadLatch() {
+        latch.readLock().lock();
+    }
+
+    @Override
+    public void releaseReadLatch() {
+        latch.readLock().unlock();
+    }
+
+    @Override
+    public void acquireWriteLatch() {
+        latch.writeLock().lock();
+    }
+
+    @Override
+    public void releaseWriteLatch(boolean markDirty) {
+        latch.writeLock().unlock();
+    }
+
+    @Override
+    public boolean confiscated() {
+        return false;
+    }
+
+    @Override
+    public IQueueInfo getQueueInfo() {
+        return null;
+    }
+
+    @Override
+    public void setQueueInfo(IQueueInfo queueInfo) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public int getPageSize() {
+        return pageSize;
+    }
+
+    @Override
+    public int getFrameSizeMultiplier() {
+        return multiplier;
+    }
+
+    public void multiplier(int multiplier) {
+        this.multiplier = multiplier;
+    }
+
+    public long dpid() {
+        return dpid;
+    }
+
+    public void dpid(long dpid) {
+        this.dpid = dpid;
+    }
+
+    public VirtualPage next() {
+        return next;
+    }
+
+    public void next(VirtualPage next) {
+        this.next = next;
+    }
+
+    public ByteBuffer buffer() {
+        return buffer;
+    }
+
+    public void buffer(ByteBuffer buffer) {
+        this.buffer = buffer;
+    }
+
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeExamplesTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeExamplesTest.java
index 70e92a3..dd4dbbb 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeExamplesTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeExamplesTest.java
@@ -26,7 +26,7 @@
 import org.apache.hyracks.storage.am.btree.util.BTreeTestHarness;
 import org.apache.hyracks.storage.am.btree.util.BTreeUtils;
 import org.apache.hyracks.storage.am.common.api.ITreeIndex;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrameFactory;
 import org.apache.hyracks.storage.am.common.api.TreeIndexException;
 import org.apache.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
 import org.apache.hyracks.storage.am.common.freepage.LinkedMetaDataPageManager;
@@ -50,7 +50,7 @@
     protected ITreeIndex createTreeIndex(ITypeTraits[] typeTraits, IBinaryComparatorFactory[] cmpFactories,
             int[] bloomFilterKeyFields, ITypeTraits[] filterTypeTraits, IBinaryComparatorFactory[] filterCmpFactories,
             int[] btreeFields, int[] filterFields) throws TreeIndexException {
-        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+        ITreeIndexMetadataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
         LinkedMetaDataPageManager freePageManager = new LinkedMetaDataPageManager(harness.getBufferCache(),
                 metaFrameFactory);
         return BTreeUtils.createBTree(harness.getBufferCache(), harness.getFileMapProvider(), typeTraits, cmpFactories,
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeSearchCursorTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeSearchCursorTest.java
index 6dfeffd..b880a01 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeSearchCursorTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeSearchCursorTest.java
@@ -65,7 +65,7 @@
     private final int fieldCount = 2;
     private final ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
     private final TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
-    private final ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+    private final ITreeIndexMetadataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
     private final Random rnd = new Random(50);
 
     @Before
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeSearchOperationCallbackTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeSearchOperationCallbackTest.java
index 2c82fab..797ee49 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeSearchOperationCallbackTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeSearchOperationCallbackTest.java
@@ -22,7 +22,7 @@
 import org.apache.hyracks.storage.am.btree.frames.BTreeLeafFrameType;
 import org.apache.hyracks.storage.am.btree.util.BTreeTestHarness;
 import org.apache.hyracks.storage.am.btree.util.BTreeUtils;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrameFactory;
 import org.apache.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
 import org.apache.hyracks.storage.am.common.freepage.LinkedMetaDataPageManager;
 
@@ -35,7 +35,7 @@
 
     @Override
     protected void createIndexInstance() throws Exception {
-        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+        ITreeIndexMetadataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
         LinkedMetaDataPageManager freePageManager = new LinkedMetaDataPageManager(harness.getBufferCache(),
                 metaFrameFactory);
         index = BTreeUtils.createBTree(harness.getBufferCache(), harness.getFileMapProvider(),
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeStatsTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeStatsTest.java
index 489c2c7..e4c1449 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeStatsTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeStatsTest.java
@@ -89,11 +89,11 @@
         TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
         ITreeIndexFrameFactory leafFrameFactory = new BTreeNSMLeafFrameFactory(tupleWriterFactory);
         ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(tupleWriterFactory);
-        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+        ITreeIndexMetadataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
 
         IBTreeLeafFrame leafFrame = (IBTreeLeafFrame) leafFrameFactory.createFrame();
         IBTreeInteriorFrame interiorFrame = (IBTreeInteriorFrame) interiorFrameFactory.createFrame();
-        ITreeIndexMetaDataFrame metaFrame = metaFrameFactory.createFrame();
+        ITreeIndexMetadataFrame metaFrame = metaFrameFactory.createFrame();
 
         IMetadataPageManager freePageManager = new LinkedMetaDataPageManager(bufferCache, metaFrameFactory);
 
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeUpdateSearchTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeUpdateSearchTest.java
index 8f373c1..493a7ab 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeUpdateSearchTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeUpdateSearchTest.java
@@ -73,7 +73,7 @@
         TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
         ITreeIndexFrameFactory leafFrameFactory = new BTreeNSMLeafFrameFactory(tupleWriterFactory);
         ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(tupleWriterFactory);
-        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+        ITreeIndexMetadataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
 
         IBTreeLeafFrame leafFrame = (IBTreeLeafFrame) leafFrameFactory.createFrame();
 
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/perf/BTreeRunner.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/perf/BTreeRunner.java
index 005b09a..5b84652 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/perf/BTreeRunner.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/perf/BTreeRunner.java
@@ -26,7 +26,7 @@
 import org.apache.hyracks.storage.am.btree.exceptions.BTreeException;
 import org.apache.hyracks.storage.am.btree.frames.BTreeLeafFrameType;
 import org.apache.hyracks.storage.am.btree.util.BTreeUtils;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrameFactory;
 import org.apache.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
 import org.apache.hyracks.storage.am.common.freepage.LinkedMetaDataPageManager;
 import org.apache.hyracks.storage.common.file.IFileMapProvider;
@@ -49,7 +49,7 @@
         TestStorageManagerComponentHolder.init(pageSize, numPages, MAX_OPEN_FILES);
         bufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
         IFileMapProvider fmp = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
-        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+        ITreeIndexMetadataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
         LinkedMetaDataPageManager freePageManager = new LinkedMetaDataPageManager(bufferCache, metaFrameFactory);
         btree = BTreeUtils.createBTree(bufferCache, fmp, typeTraits, cmpFactories, BTreeLeafFrameType.REGULAR_NSM,
                 file, freePageManager);
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/perf/ConcurrentSkipListRunner.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/perf/ConcurrentSkipListRunner.java
index 8748f44..8f1d02a 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/perf/ConcurrentSkipListRunner.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/perf/ConcurrentSkipListRunner.java
@@ -135,7 +135,7 @@
                     for (int j = 0; j < batch.size(); j++) {
                         // Copy the tuple to the buffer and set the pre-created tuple ref.
                         tupleWriter.writeTuple(batch.get(j), tupleBuf.array(), tupleIndex * tupleSize);
-                        tuples[tupleIndex].resetByTupleOffset(tupleBuf, tupleIndex * tupleSize);
+                        tuples[tupleIndex].resetByTupleOffset(tupleBuf.array(), tupleIndex * tupleSize);
                         skipList.add(tuples[tupleIndex]);
                         tupleIndex++;
                     }
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/perf/InMemorySortRunner.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/perf/InMemorySortRunner.java
index ac8e836..1e960e0 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/perf/InMemorySortRunner.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/perf/InMemorySortRunner.java
@@ -95,7 +95,7 @@
             for (int j = 0; j < batch.size(); j++) {
                 // Copy the tuple to the buffer and set the pre-created tuple ref.
                 tupleWriter.writeTuple(batch.get(j), tupleBuf.array(), tupleIndex * tupleSize);
-                tuples.get(tupleIndex).resetByTupleOffset(tupleBuf, tupleIndex * tupleSize);
+                tuples.get(tupleIndex).resetByTupleOffset(tupleBuf.array(), tupleIndex * tupleSize);
                 tupleIndex++;
             }
         }
@@ -150,7 +150,7 @@
                     for (int j = 0; j < batch.size(); j++) {
                         // Copy the tuple to the buffer and set the pre-created tuple ref.
                         tupleWriter.writeTuple(batch.get(j), tupleBuf.array(), tupleIndex * tupleSize);
-                        tuples[tupleIndex].resetByTupleOffset(tupleBuf, tupleIndex * tupleSize);
+                        tuples[tupleIndex].resetByTupleOffset(tupleBuf.array(), tupleIndex * tupleSize);
                         skipList.add(tuples[tupleIndex]);
                         tupleIndex++;
                     }
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/tuples/LSMBTreeTuplesTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/tuples/LSMBTreeTuplesTest.java
index 91467de..4743dee 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/tuples/LSMBTreeTuplesTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/tuples/LSMBTreeTuplesTest.java
@@ -103,22 +103,22 @@
                 for (int r = 0; r < 4; r++) {
 
                     // Check matter tuple with lsmBTreeTuple.
-                    lsmBTreeTuple.resetByTupleOffset(matterBuf, 0);
+                    lsmBTreeTuple.resetByTupleOffset(matterBuf.array(), 0);
                     checkTuple(lsmBTreeTuple, numFields, false, fieldSerdes, fields);
 
                     // Create a copy using copyTupleWriter, and verify again.
                     ByteBuffer copyMatterBuf = writeTuple(lsmBTreeTuple, copyTupleWriter);
-                    lsmBTreeTuple.resetByTupleOffset(copyMatterBuf, 0);
+                    lsmBTreeTuple.resetByTupleOffset(copyMatterBuf.array(), 0);
                     checkTuple(lsmBTreeTuple, numFields, false, fieldSerdes, fields);
 
                     // Check antimatter tuple with lsmBTreeTuple.
-                    lsmBTreeTuple.resetByTupleOffset(antimatterBuf, 0);
+                    lsmBTreeTuple.resetByTupleOffset(antimatterBuf.array(), 0);
                     // Should only contain keys.
                     checkTuple(lsmBTreeTuple, numKeyFields, true, fieldSerdes, fields);
 
                     // Create a copy using copyTupleWriter, and verify again.
                     ByteBuffer copyAntimatterBuf = writeTuple(lsmBTreeTuple, copyTupleWriter);
-                    lsmBTreeTuple.resetByTupleOffset(copyAntimatterBuf, 0);
+                    lsmBTreeTuple.resetByTupleOffset(copyAntimatterBuf.array(), 0);
                     // Should only contain keys.
                     checkTuple(lsmBTreeTuple, numKeyFields, true, fieldSerdes, fields);
 
@@ -126,17 +126,17 @@
                     // We should be able to manually set a prefix of the fields
                     // (the passed type traits in the tuple factory's constructor).
                     maxLsmBTreeTuple.setFieldCount(numFields);
-                    maxLsmBTreeTuple.resetByTupleOffset(matterBuf, 0);
+                    maxLsmBTreeTuple.resetByTupleOffset(matterBuf.array(), 0);
                     checkTuple(maxLsmBTreeTuple, numFields, false, fieldSerdes, fields);
 
                     // Check antimatter tuple with maxLsmBTreeTuple.
-                    maxLsmBTreeTuple.resetByTupleOffset(antimatterBuf, 0);
+                    maxLsmBTreeTuple.resetByTupleOffset(antimatterBuf.array(), 0);
                     // Should only contain keys.
                     checkTuple(maxLsmBTreeTuple, numKeyFields, true, fieldSerdes, fields);
 
                     // Resetting maxLsmBTreeTuple should set its field count to
                     // maxFieldSerdes.length, based on the its type traits.
-                    maxLsmBTreeTuple.resetByTupleOffset(maxMatterBuf, 0);
+                    maxLsmBTreeTuple.resetByTupleOffset(maxMatterBuf.array(), 0);
                     checkTuple(maxLsmBTreeTuple, maxFieldSerdes.length, false, maxFieldSerdes, maxFields);
                 }
             }
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-rtree-test/src/test/java/org/apache/hyracks/storage/am/rtree/RTreeSearchCursorTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-rtree-test/src/test/java/org/apache/hyracks/storage/am/rtree/RTreeSearchCursorTest.java
index 96366b5..6064c1b 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-rtree-test/src/test/java/org/apache/hyracks/storage/am/rtree/RTreeSearchCursorTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-rtree-test/src/test/java/org/apache/hyracks/storage/am/rtree/RTreeSearchCursorTest.java
@@ -38,7 +38,7 @@
 import org.apache.hyracks.storage.am.common.api.ITreeIndexAccessor;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexCursor;
 import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
-import org.apache.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrameFactory;
 import org.apache.hyracks.storage.am.common.api.TreeIndexException;
 import org.apache.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
 import org.apache.hyracks.storage.am.common.freepage.LinkedMetaDataPageManager;
@@ -110,7 +110,7 @@
                 .createPrimitiveValueProviderFactories(cmpFactories.length, IntegerPointable.FACTORY);
 
         RTreeTypeAwareTupleWriterFactory tupleWriterFactory = new RTreeTypeAwareTupleWriterFactory(typeTraits);
-        ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
+        ITreeIndexMetadataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
 
         ITreeIndexFrameFactory interiorFrameFactory = new RTreeNSMInteriorFrameFactory(tupleWriterFactory,
                 valueProviderFactories, RTreePolicyType.RTREE, false);