asterixdb-notifications mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "abdullah alamoudi (Code Review)" <do-not-re...@asterixdb.incubator.apache.org>
Subject Change in asterixdb[master]: Change the API for writing and reading metadata pages
Date Mon, 19 Dec 2016 04:19:56 GMT
abdullah alamoudi has uploaded a new change for review.

  https://asterix-gerrit.ics.uci.edu/1396

Change subject: Change the API for writing and reading metadata pages
......................................................................

Change the API for writing and reading metadata pages

Change-Id: Iadad522ab5568677aa816c74fc1d63acad505380
---
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeIOOperationCallback.java
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallback.java
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMInvertedIndexIOOperationCallback.java
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMRTreeIOOperationCallback.java
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogRecord.java
M asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/MetadataBootstrap.java
M asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/valueextractors/TupleCopyValueExtractor.java
M asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
M asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationManager.java
M asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMIndexFileProperties.java
M asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
M hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/IntegerPointable.java
M hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/primitive/LongPointable.java
M hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/util/ArrayBackedValueStorage.java
M hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/util/ByteArrayAccessibleOutputStream.java
M hyracks-fullstack/hyracks/hyracks-data/hyracks-data-std/src/main/java/org/apache/hyracks/data/std/util/GrowableArray.java
M hyracks-fullstack/hyracks/hyracks-hdfs/hyracks-hdfs-core/src/main/java/org/apache/hyracks/hdfs/lib/RawBinaryComparatorFactory.java
M hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeFieldPrefixNSMLeafFrame.java
M hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeNSMInteriorFrame.java
M hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/frames/BTreeNSMLeafFrame.java
M hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTree.java
M hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeOpContext.java
M hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeSplitKey.java
M hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/FieldPrefixPrefixTupleReference.java
M hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/FieldPrefixTupleReference.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/pom.xml
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IMetadataPageManager.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IPageManager.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexFrame.java
D hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetaDataFrame.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetaDataFrameFactory.java
A hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetadataFrame.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexTupleReference.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexDataflowHelper.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/LIFOMetaDataFrame.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/LIFOMetaDataFrameFactory.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/TreeIndexNSMFrame.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/AppendOnlyLinkedMetadataPageManager.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/LinkedMetaDataPageManager.java
A hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/MutableArrayValueReference.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/AbstractTreeIndex.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/tuples/SimpleTupleReference.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/tuples/TypeAwareTupleReference.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/util/TreeIndexBufferCacheWarmup.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/util/TreeIndexStats.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/util/TreeIndexStatsGatherer.java
A hyracks-fullstack/hyracks/hyracks-storage-am-common/src/test/java/org/apache/hyracks/storage/am/common/frames/LIFOMetadataFrameTest.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponent.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/tuples/LSMBTreeTupleReference.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/pom.xml
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterFrameFactory.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterManager.java
R hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterReference.java
D hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/frames/LSMComponentFilterFrame.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/frames/LSMComponentFilterFrameFactory.java
A hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/frames/LSMComponentFilterReference.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/freepage/VirtualFreePageManager.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMComponent.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexFileManager.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentFilter.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentFilterManager.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/VirtualBufferCache.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/TreeTupleSorter.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/tuples/LSMRTreeTupleReference.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/tuples/LSMRTreeTupleReferenceForPointMBR.java
M hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RStarTreePolicy.java
M hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreeNSMFrame.java
M hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreeNSMInteriorFrame.java
M hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreeNSMLeafFrame.java
M hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreePolicy.java
M hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTree.java
M hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTreeOpContext.java
M hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTreeSplitKey.java
A hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/VirtualPage.java
M hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/BTreeStatsTest.java
M hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/perf/ConcurrentSkipListRunner.java
M hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/perf/InMemorySortRunner.java
M hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/tuples/LSMBTreeTuplesTest.java
81 files changed, 1,446 insertions(+), 1,112 deletions(-)


  git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/96/1396/1

diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java
index ad4d9e1..bf6e6a0 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java
@@ -22,14 +22,18 @@
 import java.util.List;
 
 import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.primitive.LongPointable;
 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.lsm.common.api.ILSMComponent;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
 import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType;
 
 // A single LSMIOOperationCallback per LSM index used to perform actions around Flush and Merge operations
 public abstract class AbstractLSMIOOperationCallback implements ILSMIOOperationCallback {
+    public static final MutableArrayValueReference LSN_KEY = new MutableArrayValueReference("LSN"
+            .getBytes());
 
     // First LSN per mutable component
     protected long[] firstLSNs;
@@ -90,14 +94,17 @@
 
     public abstract long getComponentLSN(List<ILSMComponent> oldComponents) throws HyracksDataException;
 
-    protected void putLSNIntoMetadata(ITreeIndex treeIndex, List<ILSMComponent> oldComponents)
+    public void putLSNIntoMetadata(ITreeIndex treeIndex, List<ILSMComponent> oldComponents)
             throws HyracksDataException {
-        long componentLSN = getComponentLSN(oldComponents);
-        ((IMetadataPageManager) treeIndex.getPageManager()).setLSN(componentLSN);
+        byte[] lsn = new byte[Long.BYTES];
+        LongPointable.setLong(lsn, 0, getComponentLSN(oldComponents));
+        ((IMetadataPageManager) treeIndex.getPageManager()).put(LSN_KEY, new MutableArrayValueReference(lsn));
     }
 
     public static long getTreeIndexLSN(ITreeIndex treeIndex) throws HyracksDataException {
-        return ((IMetadataPageManager) treeIndex.getPageManager()).getLSN();
+        LongPointable pointable = new LongPointable();
+        ((IMetadataPageManager) treeIndex.getPageManager()).get(LSN_KEY, pointable);
+        return pointable.getLength() == 0 ? -1L : pointable.longValue();
     }
 
     public void updateLastLSN(long lastLSN) {
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeIOOperationCallback.java
index f0296ea..ceb175d 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeIOOperationCallback.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeIOOperationCallback.java
@@ -68,8 +68,8 @@
             throws HyracksDataException {
         if (diskComponentFilePath.endsWith(LSMBTreeFileManager.BTREE_STRING)) {
             LSMBTreeDiskComponent btreeComponent = (LSMBTreeDiskComponent) diskComponent;
-            return ((IMetadataPageManager) btreeComponent.getBTree().getPageManager()).getLSNOffset();
+            return ((IMetadataPageManager) btreeComponent.getBTree().getPageManager()).getFileOffset(LSN_KEY);
         }
-        return IMetadataPageManager.Constants.INVALID_LSN_OFFSET;
+        return -1L;
     }
 }
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallback.java
index 2b384fa..963eab8 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallback.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallback.java
@@ -61,8 +61,8 @@
             throws HyracksDataException {
         if (diskComponentFilePath.endsWith(LSMBTreeWithBuddyFileManager.BTREE_STRING)) {
             LSMBTreeWithBuddyDiskComponent btreeComponent = (LSMBTreeWithBuddyDiskComponent) diskComponent;
-            return ((IMetadataPageManager) btreeComponent.getBTree().getPageManager()).getLSNOffset();
+            return ((IMetadataPageManager) btreeComponent.getBTree().getPageManager()).getFileOffset(LSN_KEY);
         }
-        return IMetadataPageManager.Constants.INVALID_LSN_OFFSET;
+        return -1L;
     }
 }
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMInvertedIndexIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMInvertedIndexIOOperationCallback.java
index 3be31d6..ef1e001 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMInvertedIndexIOOperationCallback.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMInvertedIndexIOOperationCallback.java
@@ -66,8 +66,9 @@
             throws HyracksDataException {
         if (diskComponentFilePath.endsWith(LSMInvertedIndexFileManager.DELETED_KEYS_BTREE_SUFFIX)) {
             LSMInvertedIndexDiskComponent invIndexComponent = (LSMInvertedIndexDiskComponent) diskComponent;
-            return ((IMetadataPageManager) invIndexComponent.getDeletedKeysBTree().getPageManager()).getLSNOffset();
+            return ((IMetadataPageManager) invIndexComponent.getDeletedKeysBTree().getPageManager()).getFileOffset(
+                    LSN_KEY);
         }
-        return IMetadataPageManager.Constants.INVALID_LSN_OFFSET;
+        return -1L;
     }
 }
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMRTreeIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMRTreeIOOperationCallback.java
index 7b04b74..612c388 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMRTreeIOOperationCallback.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/LSMRTreeIOOperationCallback.java
@@ -66,8 +66,8 @@
             throws HyracksDataException {
         if (diskComponentFilePath.endsWith(LSMRTreeFileManager.RTREE_STRING)) {
             LSMRTreeDiskComponent rtreeComponent = (LSMRTreeDiskComponent) diskComponent;
-            return ((IMetadataPageManager) rtreeComponent.getRTree().getPageManager()).getLSNOffset();
+            return ((IMetadataPageManager) rtreeComponent.getRTree().getPageManager()).getFileOffset(LSN_KEY);
         }
-        return IMetadataPageManager.Constants.INVALID_LSN_OFFSET;
+        return -1L;
     }
 }
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogRecord.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogRecord.java
index 361ac7c..d691b18 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogRecord.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogRecord.java
@@ -392,7 +392,7 @@
             throw new BufferUnderflowException();
         }
         destTuple.setFieldCount(fieldCnt);
-        destTuple.resetByTupleOffset(srcBuffer, srcBuffer.position());
+        destTuple.resetByTupleOffset(srcBuffer.array(), srcBuffer.position());
         srcBuffer.position(srcBuffer.position() + size);
         return destTuple;
     }
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/MetadataBootstrap.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/MetadataBootstrap.java
index d8dfc9e..712975d 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/MetadataBootstrap.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/MetadataBootstrap.java
@@ -78,7 +78,7 @@
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.io.FileReference;
 import org.apache.hyracks.api.io.IIOManager;
-import org.apache.hyracks.storage.am.common.frames.LIFOMetaDataFrame;
+import org.apache.hyracks.storage.am.common.api.ITreeIndexFrame;
 import org.apache.hyracks.storage.am.lsm.btree.impls.LSMBTree;
 import org.apache.hyracks.storage.am.lsm.btree.util.LSMBTreeUtils;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
@@ -378,7 +378,7 @@
                             LocalResource.LSMBTreeResource);
             ILocalResourceFactory localResourceFactory = localResourceFactoryProvider.getLocalResourceFactory();
             localResourceRepository.insert(localResourceFactory.createLocalResource(resourceID, resourceName,
-                    LIFOMetaDataFrame.VERSION, metadataPartition.getPartitionId()));
+                    ITreeIndexFrame.Constants.VERSION, metadataPartition.getPartitionId()));
             dataLifecycleManager.register(file.getRelativePath(), lsmBtree);
         } else {
             final LocalResource resource = localResourceRepository.get(file.getRelativePath());
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/valueextractors/TupleCopyValueExtractor.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/valueextractors/TupleCopyValueExtractor.java
index 33f8969..4c0f48f 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/valueextractors/TupleCopyValueExtractor.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/valueextractors/TupleCopyValueExtractor.java
@@ -53,7 +53,7 @@
         tupleBytes = new byte[numBytes];
         tupleWriter.writeTuple(tuple, tupleBytes, 0);
         buf = ByteBuffer.wrap(tupleBytes);
-        tupleReference.resetByTupleOffset(buf, 0);
+        tupleReference.resetByTupleOffset(buf.array(), 0);
         return tupleReference;
     }
 }
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
index e978ad2..d6f2633 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
@@ -71,7 +71,6 @@
 import org.apache.asterix.replication.storage.ReplicaResourcesManager;
 import org.apache.asterix.transaction.management.resource.PersistentLocalResourceRepository;
 import org.apache.hyracks.api.application.INCApplicationContext;
-import org.apache.hyracks.storage.am.common.api.IMetadataPageManager;
 import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
 import org.apache.hyracks.util.StorageUtil;
@@ -344,7 +343,7 @@
                 }
                 if (afp.isLSMComponentFile()) {
                     String componentId = LSMComponentProperties.getLSMComponentID(afp.getFilePath());
-                    if (afp.getLSNByteOffset() != IMetadataPageManager.Constants.INVALID_LSN_OFFSET) {
+                    if (afp.getLSNByteOffset() >= 0L) {
                         LSMComponentLSNSyncTask syncTask = new LSMComponentLSNSyncTask(componentId,
                                 destFile.getAbsolutePath(), afp.getLSNByteOffset());
                         lsmComponentRemoteLSN2LocalLSNMappingTaskQ.offer(syncTask);
@@ -391,8 +390,7 @@
                             try (RandomAccessFile fromFile = new RandomAccessFile(filePath, "r");
                                     FileChannel fileChannel = fromFile.getChannel();) {
                                 long fileSize = fileChannel.size();
-                                fileProperties.initialize(filePath, fileSize, replicaId, false,
-                                        IMetadataPageManager.Constants.INVALID_LSN_OFFSET, false);
+                                fileProperties.initialize(filePath, fileSize, replicaId, false, -1L, false);
                                 outBuffer = ReplicationProtocol.writeFileReplicationRequest(outBuffer, fileProperties,
                                         ReplicationRequestType.REPLICATE_FILE);
 
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationManager.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationManager.java
index ce7ab5e..28e955a 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationManager.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationManager.java
@@ -83,7 +83,6 @@
 import org.apache.hyracks.api.replication.IReplicationJob.ReplicationExecutionType;
 import org.apache.hyracks.api.replication.IReplicationJob.ReplicationJobType;
 import org.apache.hyracks.api.replication.IReplicationJob.ReplicationOperation;
-import org.apache.hyracks.storage.am.common.api.IMetadataPageManager;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexReplicationJob;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
@@ -333,8 +332,8 @@
                                 asterixFileProperties.initialize(filePath, fileSize, nodeId, isLSMComponentFile,
                                         LSNByteOffset, remainingFiles == 0);
                             } else {
-                                asterixFileProperties.initialize(filePath, fileSize, nodeId, isLSMComponentFile,
-                                        IMetadataPageManager.Constants.INVALID_LSN_OFFSET, remainingFiles == 0);
+                                asterixFileProperties.initialize(filePath, fileSize, nodeId, isLSMComponentFile, -1L,
+                                        remainingFiles == 0);
                             }
 
                             requestBuffer = ReplicationProtocol.writeFileReplicationRequest(requestBuffer,
@@ -372,8 +371,8 @@
                 } else if (job.getOperation() == ReplicationOperation.DELETE) {
                     for (String filePath : job.getJobFiles()) {
                         remainingFiles--;
-                        asterixFileProperties.initialize(filePath, -1, nodeId, isLSMComponentFile,
-                                IMetadataPageManager.Constants.INVALID_LSN_OFFSET, remainingFiles == 0);
+                        asterixFileProperties.initialize(filePath, -1, nodeId, isLSMComponentFile, -1L,
+                                remainingFiles == 0);
                         ReplicationProtocol.writeFileReplicationRequest(requestBuffer, asterixFileProperties,
                                 ReplicationRequestType.DELETE_FILE);
 
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMIndexFileProperties.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMIndexFileProperties.java
index 2a219c1..eb9e82d 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMIndexFileProperties.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMIndexFileProperties.java
@@ -25,7 +25,6 @@
 import java.io.OutputStream;
 
 import org.apache.asterix.common.utils.StoragePathUtil;
-import org.apache.hyracks.storage.am.common.api.IMetadataPageManager;
 
 public class LSMIndexFileProperties {
 
@@ -49,8 +48,7 @@
     }
 
     public LSMIndexFileProperties(LSMComponentProperties lsmComponentProperties) {
-        initialize(lsmComponentProperties.getComponentId(), -1, lsmComponentProperties.getNodeId(), false,
-                IMetadataPageManager.Constants.INVALID_LSN_OFFSET, false);
+        initialize(lsmComponentProperties.getComponentId(), -1, lsmComponentProperties.getNodeId(), false, -1L, false);
     }
 
     public void initialize(String filePath, long fileSize, String nodeId, boolean lsmComponentFile, long LSNByteOffset,
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
index 6646defb..ebf79cf 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
@@ -55,7 +55,7 @@
 import org.apache.hyracks.api.replication.IReplicationJob.ReplicationExecutionType;
 import org.apache.hyracks.api.replication.IReplicationJob.ReplicationJobType;
 import org.apache.hyracks.api.replication.IReplicationJob.ReplicationOperation;
-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.ILocalResourceRepository;
 import org.apache.hyracks.storage.common.file.LocalResource;
 
@@ -164,7 +164,8 @@
             }
 
             LocalResource rootLocalResource = new LocalResource(STORAGE_LOCAL_RESOURCE_ID,
-                    storageMetadataFile.getRelativePath(), 0, LIFOMetaDataFrame.VERSION, storageRootDirPath);
+                    storageMetadataFile.getRelativePath(), 0, ITreeIndexFrame.Constants.VERSION,
+                    storageRootDirPath);
             insert(rootLocalResource);
             LOGGER.log(Level.INFO, "created the root-metadata-file: " + storageMetadataFile.getAbsolutePath());
         }
@@ -343,7 +344,7 @@
         try (FileInputStream fis = new FileInputStream(file);
                 ObjectInputStream oisFromFis = new ObjectInputStream(fis)) {
             LocalResource resource = (LocalResource) oisFromFis.readObject();
-            if (resource.getVersion() == LIFOMetaDataFrame.VERSION) {
+            if (resource.getVersion() == ITreeIndexFrame.Constants.VERSION) {
                 return resource;
             } else {
                 throw new AsterixException("Storage version mismatch.");
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..db8d0c6 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();
+    public static final IBinaryComparatorFactory INSTANCE = new RawBinaryComparatorFactory();
 
     private RawBinaryComparatorFactory() {
     }
 
     @Override
     public IBinaryComparator createBinaryComparator() {
-        return new IBinaryComparator() {
-
-            @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);
-            }
-
-        };
+        return RawBinaryComparatorFactory::compare;
     }
-}
+
+    private 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..39327c6 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
@@ -721,7 +721,7 @@
         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
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..d7f4335 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/pom.xml
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/pom.xml
@@ -80,5 +80,11 @@
       <artifactId>hyracks-data-std</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <version>4.12</version>
+      <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..961e288 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,61 @@
 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
+     * @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(IValueReference key, IValueReference value) throws HyracksDataException;
 
     /**
-     * Set the cached page to manage for filter data
-     *
-     * @param page
-     *            The page to manage
+     * put the key value pair in the metadata page using the passed frame
+     * @param frame
+     * @param key
+     * @param value
      * @throws HyracksDataException
      */
-    void setFilterPage(ICachedPage page) throws HyracksDataException;
+    void put(ITreeIndexMetadataFrame frame, IValueReference key, IValueReference value) throws HyracksDataException;
 
     /**
-     * Get filter page if exists, create and return a new one if it doesn't
-     * @return
+     * get the value of the key from the metadata page
+     * @param key
+     * @param value
      * @throws HyracksDataException
      */
-    ICachedPage getFilterPage() throws HyracksDataException;
+    void get(IValueReference key, IPointable value) throws HyracksDataException;
+    
+    /**
+     * get the value of the key from the metadata page using the passed frame
+     * @param frame
+     * @param key
+     * @param value
+     * @throws HyracksDataException
+     */
+    void get(ITreeIndexMetadataFrame frame, IValueReference key, IPointable value) throws HyracksDataException;
 
     /**
-     * @return The LSN byte offset in the LSM disk component if the index is valid,
-     *         otherwise {@link #INVALID_LSN_OFFSET}.
+     * @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
      * @throws HyracksDataException
      */
-    long getLSNOffset() throws HyracksDataException;
-    long getLastMarkerLSN() throws HyracksDataException;
+    long getFileOffset(IValueReference key) throws HyracksDataException;
+
+    /**
+     * @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
+     */
+    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/ITreeIndexMetaDataFrameFactory.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetaDataFrameFactory.java
index ff7deaf..6f37d75 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
@@ -20,5 +20,5 @@
 package org.apache.hyracks.storage.am.common.api;
 
 public interface ITreeIndexMetaDataFrameFactory {
-    public ITreeIndexMetaDataFrame createFrame();
+    public ITreeIndexMetadataFrame createFrame();
 }
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..98c3bbf
--- /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,161 @@
+/*
+ * 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);
+
+    @Deprecated
+    boolean isMetadataPage();
+
+    @Deprecated
+    boolean isFreePage();
+}
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..5da1a13 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,30 @@
  * 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 {
+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 +56,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);
+        int numFreePages = buf.getInt(FREE_PAGE_COUNT_OFFSET);
+        buf.putInt(buf.array().length - (Integer.BYTES * numFreePages), freePage);
+        increment(FREE_PAGE_COUNT_OFFSET);
     }
 
     @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 +99,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 +117,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 +155,147 @@
     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);
+            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..4a3b831 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.ITreeIndexMetadataFrame;
 import org.apache.hyracks.storage.am.common.api.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..74cd0cf 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.impls.AbstractTreeIndex;
 import org.apache.hyracks.storage.common.buffercache.BufferCache;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
@@ -47,12 +48,12 @@
     }
 
     @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,22 @@
     }
 
     @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);
-            }
-        }
+    public long getFileOffset(IValueReference key) throws HyracksDataException {
+        return getFileOffset(createMetadataFrame(), key);
     }
 
     @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 +283,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 +300,67 @@
     public int getBulkLoadLeaf() throws HyracksDataException {
         return 0;
     }
+
+    @Override
+    public void put(IValueReference key, IValueReference value) throws HyracksDataException {
+        put(frameFactory.createFrame(), key, value);
+    }
+
+    @Override
+    public void get(IValueReference key, IPointable value) throws HyracksDataException {
+        get(createMetadataFrame(), key, value);
+    }
+
+    @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);
+                return ((long) pageId * bufferCache.getPageSizeWithHeader()) + frame.getOffset(key);
+            } finally {
+                page.releaseReadLatch();
+                unpinPage(page);
+            }
+        }
+        return -1;
+    }
 }
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..962708b 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.common.buffercache.IBufferCache;
 import org.apache.hyracks.storage.common.buffercache.ICachedPage;
 import org.apache.hyracks.storage.common.file.BufferedFileHandle;
@@ -41,14 +42,14 @@
     }
 
     @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,33 @@
     }
 
     @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 {
+    public long getFileOffset(IValueReference key) throws HyracksDataException {
         int metadataPageNum = getMetadataPageId();
         if (metadataPageNum != IBufferCache.INVALID_PAGEID) {
-            return ((long) metadataPageNum * bufferCache.getPageSizeWithHeader()) + LIFOMetaDataFrame.LSN_OFFSET;
+            ITreeIndexMetadataFrame metadataFrame = createMetadataFrame();
+            ICachedPage metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()),
+                    false);
+            metaNode.acquireReadLatch();
+            try {
+                metadataFrame.setPage(metaNode);
+                return ((long) metadataPageNum * bufferCache.getPageSizeWithHeader()) + metadataFrame.getOffset(key);
+            } finally {
+                metaNode.releaseReadLatch();
+                bufferCache.unpin(metaNode);
+            }
         }
-        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);
-        }
+        return -1;
     }
 
     @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 +316,36 @@
         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(IValueReference key, IValueReference value) throws HyracksDataException {
+        throw new HyracksDataException("Unsupported Operation");
+    }
+
+    @Override
+    public void get(IValueReference key, IPointable value) throws HyracksDataException {
+        throw new HyracksDataException("Unsupported Operation");
+    }
+
+    @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 {
+        throw new HyracksDataException("Unsupported Operation");
+    }
 }
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..3151345 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
@@ -494,7 +494,6 @@
         }
         component.setMostRecentMarkerLSN(flushOp.getPrevMarkerLSN());
         bulkLoader.end();
-
         return 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..ab7d61b 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(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-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..295a32f 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,7 +20,7 @@
 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.ILSMComponentFilterReference;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilterFrameFactory;
 
 public class LSMComponentFilterFrameFactory implements ILSMComponentFilterFrameFactory {
@@ -34,7 +34,7 @@
     }
 
     @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..5fc63fc
--- /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..2a6af64 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,17 +25,17 @@
 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.am.lsm.common.api.ILSMComponentFilterReference;
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
-import org.apache.hyracks.storage.common.buffercache.ICachedPage;
-import org.apache.hyracks.storage.common.file.BufferedFileHandle;
 
 public class LSMComponentFilterManager implements ILSMComponentFilterManager {
 
+    public static final MutableArrayValueReference FILTER_KEY = new MutableArrayValueReference("Filter".getBytes());
     private final IBufferCache bufferCache;
     private final ILSMComponentFilterFrameFactory filterFrameFactory;
 
@@ -56,11 +56,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 +65,23 @@
                 filterFrame.writeMaxTuple(filter.getMaxTuple());
             }
         } finally {
-            treeMetaManager.setFilterPage(filterPage);
+            treeMetaManager.put(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(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
@@ -349,76 +347,6 @@
 
         public CacheBucket() {
             this.bucketLock = new ReentrantLock();
-        }
-    }
-
-    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;
         }
     }
 
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-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..ca98dc3 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 DOUEBLE_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 DOUEBLE_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/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..cb01237 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
@@ -93,7 +93,7 @@
 
         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-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);
                 }
             }

-- 
To view, visit https://asterix-gerrit.ics.uci.edu/1396
To unsubscribe, visit https://asterix-gerrit.ics.uci.edu/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: Iadad522ab5568677aa816c74fc1d63acad505380
Gerrit-PatchSet: 1
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Owner: abdullah alamoudi <bamousaa@gmail.com>

Mime
View raw message