diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
index d528925..caf862b 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
@@ -24,6 +24,8 @@
 import java.rmi.RemoteException;
 import java.rmi.server.UnicastRemoteObject;
 import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -116,6 +118,9 @@
 import org.apache.hyracks.storage.common.buffercache.ICacheMemoryAllocator;
 import org.apache.hyracks.storage.common.buffercache.IPageCleanerPolicy;
 import org.apache.hyracks.storage.common.buffercache.IPageReplacementStrategy;
+import org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider;
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheReadContext;
+import org.apache.hyracks.storage.common.file.BufferedFileHandle;
 import org.apache.hyracks.storage.common.file.FileMapManager;
 import org.apache.hyracks.storage.common.file.ILocalResourceRepositoryFactory;
 import org.apache.hyracks.storage.common.file.IResourceIdFactory;
@@ -204,6 +209,7 @@
             IConfigValidatorFactory configValidatorFactory, IReplicationStrategyFactory replicationStrategyFactory,
             boolean initialRun) throws IOException {
         ioManager = getServiceContext().getIoManager();
+        IBufferCacheReadContext defaultContext = DefaultBufferCachePageOperationContextProvider.DEFAULT;
         if (isCloudDeployment()) {
             persistenceIOManager =
                     CloudManagerProvider.createIOManager(cloudProperties, ioManager, namespacePathResolver);
@@ -265,6 +271,7 @@
                 this.ncServiceContext);
         receptionist = receptionistFactory.create();
 
+        Map<Integer, BufferedFileHandle> fileInfoMap = new HashMap<>();
         if (replicationProperties.isReplicationEnabled()) {
             if (LOGGER.isInfoEnabled()) {
                 LOGGER.info("Replication is enabled");
@@ -283,10 +290,11 @@
 
             bufferCache = new BufferCache(persistenceIOManager, prs, pcp, new FileMapManager(),
                     storageProperties.getBufferCacheMaxOpenFiles(), ioQueueLen, getServiceContext().getThreadFactory(),
-                    replicationManager);
+                    replicationManager, fileInfoMap);
         } else {
             bufferCache = new BufferCache(persistenceIOManager, prs, pcp, new FileMapManager(),
-                    storageProperties.getBufferCacheMaxOpenFiles(), ioQueueLen, getServiceContext().getThreadFactory());
+                    storageProperties.getBufferCacheMaxOpenFiles(), ioQueueLen, getServiceContext().getThreadFactory(),
+                    fileInfoMap, defaultContext);
         }
 
         NodeControllerService ncs = (NodeControllerService) getServiceContext().getControllerService();
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/GlobalVirtualBufferCache.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/GlobalVirtualBufferCache.java
index fb88c65..20a1260 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/GlobalVirtualBufferCache.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/GlobalVirtualBufferCache.java
@@ -18,6 +18,8 @@
  */
 package org.apache.asterix.common.context;
 
+import static org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider.DEFAULT;
+
 import java.io.IOException;
 import java.io.OutputStream;
 import java.util.ArrayList;
@@ -49,6 +51,8 @@
 import org.apache.hyracks.storage.common.buffercache.IPageWriteCallback;
 import org.apache.hyracks.storage.common.buffercache.IPageWriteFailureCallback;
 import org.apache.hyracks.storage.common.buffercache.VirtualPage;
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheReadContext;
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheWriteContext;
 import org.apache.hyracks.storage.common.file.BufferedFileHandle;
 import org.apache.hyracks.storage.common.file.IFileMapManager;
 import org.apache.hyracks.util.ExitUtil;
@@ -264,20 +268,20 @@
     }
 
     @Override
-    public ICachedPage pin(long dpid, boolean newPage) throws HyracksDataException {
-        ICachedPage page = vbc.pin(dpid, newPage);
-        if (newPage) {
+    public ICachedPage pin(long dpid) throws HyracksDataException {
+        return pin(dpid, DEFAULT);
+    }
+
+    @Override
+    public ICachedPage pin(long dpid, IBufferCacheReadContext context) throws HyracksDataException {
+        ICachedPage page = vbc.pin(dpid, context);
+        if (context.isNewPage()) {
             incrementFilteredMemoryComponentUsage(dpid, 1);
             checkAndNotifyFlushThread();
         }
         return page;
     }
 
-    @Override
-    public ICachedPage pin(long dpid, boolean newPage, boolean incrementStats) throws HyracksDataException {
-        return pin(dpid, newPage);
-    }
-
     private void incrementFilteredMemoryComponentUsage(long dpid, int pages) {
         if (filteredMemoryComponentMaxNumPages > 0) {
             // update memory usage of filtered index
@@ -316,7 +320,12 @@
 
     @Override
     public void unpin(ICachedPage page) throws HyracksDataException {
-        vbc.unpin(page);
+        unpin(page, DEFAULT);
+    }
+
+    @Override
+    public void unpin(ICachedPage page, IBufferCacheReadContext context) throws HyracksDataException {
+        vbc.unpin(page, context);
     }
 
     @Override
@@ -391,8 +400,9 @@
     }
 
     @Override
-    public IFIFOPageWriter createFIFOWriter(IPageWriteCallback callback, IPageWriteFailureCallback failureCallback) {
-        return vbc.createFIFOWriter(callback, failureCallback);
+    public IFIFOPageWriter createFIFOWriter(IPageWriteCallback callback, IPageWriteFailureCallback failureCallback,
+            IBufferCacheWriteContext context) {
+        return vbc.createFIFOWriter(callback, failureCallback, context);
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/RuntimeContext.java b/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/RuntimeContext.java
index 4632f2d..69fc6d1 100644
--- a/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/RuntimeContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/RuntimeContext.java
@@ -19,6 +19,7 @@
 
 package org.apache.hyracks.examples.btree.helper;
 
+import java.util.HashMap;
 import java.util.concurrent.ThreadFactory;
 
 import org.apache.hyracks.api.application.INCServiceContext;
@@ -35,6 +36,7 @@
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
 import org.apache.hyracks.storage.common.buffercache.ICacheMemoryAllocator;
 import org.apache.hyracks.storage.common.buffercache.IPageReplacementStrategy;
+import org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider;
 import org.apache.hyracks.storage.common.file.FileMapManager;
 import org.apache.hyracks.storage.common.file.IFileMapManager;
 import org.apache.hyracks.storage.common.file.IFileMapProvider;
@@ -58,7 +60,7 @@
         ThreadFactory threadFactory = Thread::new;
         this.ioManager = appCtx.getIoManager();
         bufferCache = new BufferCache(ioManager, prs, new DelayPageCleanerPolicy(1000), fileMapManager, 100, 10,
-                threadFactory);
+                threadFactory, new HashMap<>(), DefaultBufferCachePageOperationContextProvider.DEFAULT);
         ILocalResourceRepositoryFactory localResourceRepositoryFactory = new TransientLocalResourceRepositoryFactory();
         localResourceRepository = localResourceRepositoryFactory.createRepository();
         resourceIdFactory = (new ResourceIdFactoryProvider(localResourceRepository)).createResourceIdFactory();
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-bloomfilter/src/main/java/org/apache/hyracks/storage/am/bloomfilter/impls/BloomFilter.java b/hyracks-fullstack/hyracks/hyracks-storage-am-bloomfilter/src/main/java/org/apache/hyracks/storage/am/bloomfilter/impls/BloomFilter.java
index 6c0badc..d94435d 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-bloomfilter/src/main/java/org/apache/hyracks/storage/am/bloomfilter/impls/BloomFilter.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-bloomfilter/src/main/java/org/apache/hyracks/storage/am/bloomfilter/impls/BloomFilter.java
@@ -31,6 +31,7 @@
 import org.apache.hyracks.storage.common.buffercache.IFIFOPageWriter;
 import org.apache.hyracks.storage.common.buffercache.IPageWriteCallback;
 import org.apache.hyracks.storage.common.buffercache.PageWriteFailureCallback;
+import org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCacheWriteContext;
 import org.apache.hyracks.storage.common.file.BufferedFileHandle;
 
 public class BloomFilter {
@@ -92,7 +93,7 @@
                 pages = new ICachedPage[numPages];
             }
             for (int i = 0; i < numPages; i++) {
-                pages[i] = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, i + 1), false);
+                pages[i] = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, i + 1));
             }
             pagesPinned = true;
         }
@@ -155,7 +156,7 @@
         if (pagesPinned) {
             page = pages[pageId];
         } else {
-            page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId + 1), false);
+            page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId + 1));
             unpinWhenExit = true;
         }
         ByteBuffer buffer = page.getBuffer();
@@ -186,7 +187,7 @@
 
             // we increment the page id by one, since the metadata page id of the filter is 0.
             ICachedPage page =
-                    bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, (int) (hash / numBitsPerPage) + 1), false);
+                    bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, (int) (hash / numBitsPerPage) + 1));
             page.acquireReadLatch();
             try {
                 ByteBuffer buffer = page.getBuffer();
@@ -235,7 +236,7 @@
             version = DEFAULT_BLOOM_FILTER_VERSION;
             return;
         }
-        ICachedPage metaPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, METADATA_PAGE_ID), false);
+        ICachedPage metaPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, METADATA_PAGE_ID));
         metaPage.acquireReadLatch();
         try {
             numPages = metaPage.getBuffer().getInt(NUM_PAGES_OFFSET);
@@ -302,7 +303,7 @@
             if (!isActivated) {
                 throw HyracksDataException.create(ErrorCode.CANNOT_CREATE_BLOOM_FILTER_BUILDER_FOR_INACTIVE_FILTER);
             }
-            pageWriter = bufferCache.createFIFOWriter(callback, this);
+            pageWriter = bufferCache.createFIFOWriter(callback, this, DefaultBufferCacheWriteContext.INSTANCE);
             this.estimatedNumElements = estimatedNumElemenets;
             this.numHashes = numHashes;
             numBits = this.estimatedNumElements * numBitsPerElement;
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 5b1dd8c..89aa769 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
@@ -19,6 +19,8 @@
 
 package org.apache.hyracks.storage.am.btree.impls;
 
+import static org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider.NEW;
+
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -99,7 +101,7 @@
         RangePredicate diskOrderScanPred = new RangePredicate(null, null, true, true, ctx.getCmp(), ctx.getCmp());
         int maxPageId = freePageManager.getMaxPageId(ctx.getMetaFrame());
         int currentPageId = bulkloadLeafStart;
-        final ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), currentPageId), false);
+        final ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), currentPageId));
         page.acquireReadLatch();
         try {
             cursor.setBufferCache(bufferCache);
@@ -131,7 +133,7 @@
     }
 
     private void validate(BTreeOpContext ctx, int pageId) throws HyracksDataException {
-        ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId), false);
+        ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId));
         ctx.getInteriorFrame().setPage(page);
         PageValidationInfo currentPvi = ctx.getValidationInfos().peekFirst();
 
@@ -214,7 +216,7 @@
         ICachedPage originalPage = ctx.getInteriorFrame().getPage();
         for (int i = 0; i < ctx.getSmPages().size(); i++) {
             int pageId = ctx.getSmPages().get(i);
-            ICachedPage smPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId), false);
+            ICachedPage smPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId));
             smPage.acquireWriteLatch();
             try {
                 ctx.getInteriorFrame().setPage(smPage);
@@ -235,11 +237,11 @@
     private void createNewRoot(BTreeOpContext ctx) throws HyracksDataException {
         // Make sure the root is always in the same page.
         ICachedPage leftNode =
-                bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), ctx.getSplitKey().getLeftPage()), false);
+                bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), ctx.getSplitKey().getLeftPage()));
         leftNode.acquireWriteLatch();
         try {
             int newLeftId = freePageManager.takePage(ctx.getMetaFrame());
-            ICachedPage newLeftNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), newLeftId), true);
+            ICachedPage newLeftNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), newLeftId), NEW);
             newLeftNode.acquireWriteLatch();
             try {
                 boolean largePage = false;
@@ -347,7 +349,7 @@
             }
         }
         int rightPageId = freePageManager.takePage(ctx.getMetaFrame());
-        ICachedPage rightNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), rightPageId), true);
+        ICachedPage rightNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), rightPageId), NEW);
         rightNode.acquireWriteLatch();
         try {
             IBTreeLeafFrame rightFrame = ctx.createLeafFrame();
@@ -467,7 +469,7 @@
             case INSUFFICIENT_SPACE: {
                 int rightPageId = freePageManager.takePage(ctx.getMetaFrame());
                 ICachedPage rightNode =
-                        bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), rightPageId), true);
+                        bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), rightPageId), NEW);
                 rightNode.acquireWriteLatch();
                 try {
                     IBTreeFrame rightFrame = ctx.createInteriorFrame();
@@ -546,7 +548,7 @@
     }
 
     private ICachedPage isConsistent(int pageId, BTreeOpContext ctx) throws Exception {
-        ICachedPage node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId), false);
+        ICachedPage node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId));
         node.acquireReadLatch();
         ctx.getInteriorFrame().setPage(node);
         boolean isConsistent = ctx.getPageLsns().getLast() == ctx.getInteriorFrame().getPageLsn();
@@ -560,7 +562,7 @@
 
     private void performOp(int pageId, ICachedPage parent, boolean parentIsReadLatched, BTreeOpContext ctx)
             throws HyracksDataException {
-        ICachedPage node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId), false);
+        ICachedPage node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId));
         ctx.getInteriorFrame().setPage(node);
         // this check performs an unprotected read in the page
         // the following could happen: TODO fill out
@@ -624,8 +626,8 @@
                             case UPDATE: {
                                 // Is there a propagated split key?
                                 if (ctx.getSplitKey().getBuffer() != null) {
-                                    ICachedPage interiorNode = bufferCache
-                                            .pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId), false);
+                                    ICachedPage interiorNode =
+                                            bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId));
                                     interiorNode.acquireWriteLatch();
                                     try {
                                         // Insert or update op. Both can cause split keys to propagate upwards.
@@ -760,7 +762,7 @@
     public void printTree(int pageId, ICachedPage parent, boolean unpin, IBTreeLeafFrame leafFrame,
             IBTreeInteriorFrame interiorFrame, byte treeHeight, ISerializerDeserializer[] keySerdes,
             StringBuilder strBuilder, MultiComparator cmp) throws Exception {
-        ICachedPage node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId), false);
+        ICachedPage node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId));
         node.acquireReadLatch();
         try {
             if (parent != null && unpin) {
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeCountingSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeCountingSearchCursor.java
index 71df593..a010fde 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeCountingSearchCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeCountingSearchCursor.java
@@ -121,7 +121,7 @@
         do {
             final ICachedPage nextLeaf;
             try {
-                nextLeaf = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, nextLeafPage), false);
+                nextLeaf = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, nextLeafPage));
                 if (exclusiveLatchNodes) {
                     nextLeaf.acquireWriteLatch();
                 } else {
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeRangeSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeRangeSearchCursor.java
index cf3727a..034a714 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeRangeSearchCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeRangeSearchCursor.java
@@ -308,7 +308,7 @@
     }
 
     protected ICachedPage acquirePage(int pageId) throws HyracksDataException {
-        ICachedPage nextPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId), false);
+        ICachedPage nextPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId));
         if (exclusiveLatchNodes) {
             nextPage.acquireWriteLatch();
         } else {
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTree.java
index f58d43f..9bbd056 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTree.java
@@ -56,7 +56,7 @@
         RangePredicate diskOrderScanPred = new RangePredicate(null, null, true, true, ctx.getCmp(), ctx.getCmp());
         int maxPageId = freePageManager.getMaxPageId(ctx.getMetaFrame());
         int currentPageId = bulkloadLeafStart;
-        ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), currentPageId), false);
+        ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), currentPageId));
         try {
             cursor.setBufferCache(bufferCache);
             cursor.setFileId(getFileId());
@@ -101,7 +101,7 @@
                 }
             }
         }
-        ICachedPage rootNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), rootPage), false);
+        ICachedPage rootNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), rootPage));
         searchDown(rootNode, rootPage, ctx, cursor);
     }
 
@@ -125,7 +125,7 @@
                 bufferCache.unpin(currentPage);
                 pageId = childPageId;
 
-                currentPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), childPageId), false);
+                currentPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), childPageId));
                 ctx.getInteriorFrame().setPage(currentPage);
             }
 
@@ -223,7 +223,7 @@
 
         @Override
         protected ICachedPage acquireNextPage() throws HyracksDataException {
-            ICachedPage nextPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, currentPageId), false);
+            ICachedPage nextPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, currentPageId));
             return nextPage;
         }
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTreeRangeSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTreeRangeSearchCursor.java
index d788398..3dc4df7 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTreeRangeSearchCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTreeRangeSearchCursor.java
@@ -73,7 +73,7 @@
     @Override
     protected ICachedPage acquirePage(int pageId) throws HyracksDataException {
         stats.getPageCounter().update(1);
-        return bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId), false);
+        return bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId));
     }
 
 }
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 45fbd1b..501f8a6 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
@@ -34,6 +34,7 @@
 import org.apache.hyracks.storage.common.buffercache.IFIFOPageWriter;
 import org.apache.hyracks.storage.common.buffercache.IPageWriteFailureCallback;
 import org.apache.hyracks.storage.common.buffercache.NoOpPageWriteCallback;
+import org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCacheWriteContext;
 import org.apache.hyracks.storage.common.compression.file.ICompressedPageWriter;
 import org.apache.hyracks.storage.common.file.BufferedFileHandle;
 
@@ -52,7 +53,7 @@
 
     @Override
     public void releasePage(ITreeIndexMetadataFrame metaFrame, int freePageNum) throws HyracksDataException {
-        ICachedPage metaPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
+        ICachedPage metaPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()));
         metaPage.acquireWriteLatch();
         try {
             metaFrame.setPage(metaPage);
@@ -64,7 +65,7 @@
                     throw new HyracksDataException(
                             "Inconsistent Meta Page State. It has no space, but it also has no entries.");
                 }
-                ICachedPage newNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, newPageNum), false);
+                ICachedPage newNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, newPageNum));
                 newNode.acquireWriteLatch();
                 try {
                     int metaMaxPage = metaFrame.getMaxPage();
@@ -103,7 +104,7 @@
             if (freePage < 0) { // no free page entry on this page
                 int nextPage = metaFrame.getNextMetadataPage();
                 if (nextPage > 0) { // sibling may have free pages
-                    ICachedPage nextNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, nextPage), false);
+                    ICachedPage nextNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, nextPage));
                     nextNode.acquireWriteLatch();
                     // we copy over the free space entries of nextpage into the
                     // first meta page (metaDataPage)
@@ -158,7 +159,7 @@
             if (mdPage < 0) {
                 return IBufferCache.INVALID_PAGEID;
             }
-            metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, mdPage), false);
+            metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, mdPage));
         } else {
             metaNode = confiscatedPage;
         }
@@ -212,7 +213,8 @@
     @Override
     public void close(IPageWriteFailureCallback failureCallback) throws HyracksDataException {
         if (ready) {
-            IFIFOPageWriter pageWriter = bufferCache.createFIFOWriter(NoOpPageWriteCallback.INSTANCE, failureCallback);
+            IFIFOPageWriter pageWriter = bufferCache.createFIFOWriter(NoOpPageWriteCallback.INSTANCE, failureCallback,
+                    DefaultBufferCacheWriteContext.INSTANCE);
             ITreeIndexMetadataFrame metaFrame = frameFactory.createFrame();
             confiscatedPage.acquireWriteLatch();
             try {
@@ -282,7 +284,7 @@
     public int getRootPageId() throws HyracksDataException {
         ICachedPage metaNode;
         if (confiscatedPage == null) {
-            metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
+            metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()));
         } else {
             metaNode = confiscatedPage;
         }
@@ -320,8 +322,7 @@
     }
 
     private ICachedPage pinPage() throws HyracksDataException {
-        return confiscatedPage == null
-                ? bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false)
+        return confiscatedPage == null ? bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()))
                 : confiscatedPage;
     }
 
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 e348e24..932ad6c 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
@@ -18,6 +18,8 @@
  */
 package org.apache.hyracks.storage.am.common.freepage;
 
+import static org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider.NEW;
+
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.data.std.api.IPointable;
 import org.apache.hyracks.data.std.api.IValueReference;
@@ -32,8 +34,7 @@
 import org.apache.hyracks.storage.common.file.BufferedFileHandle;
 
 /**
- * @deprecated
- *             This class must not be used. Instead, use {@link AppendOnlyLinkedMetadataPageManager}
+ * @deprecated This class must not be used. Instead, use {@link AppendOnlyLinkedMetadataPageManager}
  */
 @Deprecated
 public class LinkedMetaDataPageManager implements IMetadataPageManager {
@@ -50,7 +51,7 @@
     @Override
     public void releasePage(ITreeIndexMetadataFrame metaFrame, int freePageNum) throws HyracksDataException {
         // Get the metadata node
-        ICachedPage metaPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
+        ICachedPage metaPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()));
         metaPage.acquireWriteLatch();
         try {
             metaFrame.setPage(metaPage);
@@ -65,7 +66,7 @@
                             "Inconsistent Meta Page State. It has no space, but it also has no entries.");
                 }
 
-                ICachedPage newNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, newPageNum), false);
+                ICachedPage newNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, newPageNum));
                 newNode.acquireWriteLatch();
 
                 try {
@@ -100,7 +101,7 @@
 
     @Override
     public int takePage(ITreeIndexMetadataFrame metaFrame) throws HyracksDataException {
-        ICachedPage metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
+        ICachedPage metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()));
         metaNode.acquireWriteLatch();
         int freePage = IBufferCache.INVALID_PAGEID;
         try {
@@ -109,7 +110,7 @@
             if (freePage < 0) { // no free page entry on this page
                 int nextPage = metaFrame.getNextMetadataPage();
                 if (nextPage > 0) { // sibling may have free pages
-                    ICachedPage nextNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, nextPage), false);
+                    ICachedPage nextNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, nextPage));
 
                     nextNode.acquireWriteLatch();
                     // we copy over the free space entries of nextpage into the
@@ -169,7 +170,7 @@
         if (mdPage < 0) {
             return IBufferCache.INVALID_PAGEID;
         }
-        metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, mdPage), false);
+        metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, mdPage));
 
         metaNode.acquireReadLatch();
         int maxPage = -1;
@@ -192,7 +193,7 @@
         if (metaPage == IBufferCache.INVALID_PAGEID) {
             throw new HyracksDataException("No valid metadata found in this file.");
         }
-        ICachedPage metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), true);
+        ICachedPage metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), NEW);
         metaNode.acquireWriteLatch();
         try {
             metaFrame.setPage(metaNode);
@@ -205,7 +206,7 @@
             bufferCache.unpin(metaNode);
         }
         int rootPage = getRootPageId();
-        ICachedPage rootNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, rootPage), true);
+        ICachedPage rootNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, rootPage), NEW);
         rootNode.acquireWriteLatch();
         try {
             ITreeIndexFrame leafFrame = leafFrameFactory.createFrame();
@@ -230,7 +231,7 @@
 
     @Override
     public void setRootPageId(int rootPage) throws HyracksDataException {
-        ICachedPage metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
+        ICachedPage metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()));
         ITreeIndexMetadataFrame metaFrame = frameFactory.createFrame();
         metaNode.acquireWriteLatch();
         try {
@@ -246,8 +247,7 @@
     @Override
     public void close(IPageWriteFailureCallback callback) throws HyracksDataException {
         if (ready) {
-            ICachedPage metaNode =
-                    bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
+            ICachedPage metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()));
             ITreeIndexMetadataFrame metaFrame = frameFactory.createFrame();
             metaNode.acquireWriteLatch();
             try {
@@ -280,7 +280,7 @@
     @Override
     public int getRootPageId() throws HyracksDataException {
         ICachedPage metaNode;
-        metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
+        metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()));
         ITreeIndexMetadataFrame metaFrame = frameFactory.createFrame();
         metaNode.acquireReadLatch();
         try {
@@ -299,7 +299,7 @@
 
     @Override
     public boolean isEmpty(ITreeIndexFrame frame, int rootPage) throws HyracksDataException {
-        ICachedPage rootNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, rootPage), false);
+        ICachedPage rootNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, rootPage));
         rootNode.acquireReadLatch();
         try {
             frame.setPage(rootNode);
@@ -325,8 +325,7 @@
     public long getFileOffset(ITreeIndexMetadataFrame frame, IValueReference key) throws HyracksDataException {
         int metadataPageNum = getMetadataPageId();
         if (metadataPageNum != IBufferCache.INVALID_PAGEID) {
-            ICachedPage metaNode =
-                    bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()), false);
+            ICachedPage metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getMetadataPageId()));
             metaNode.acquireReadLatch();
             try {
                 frame.setPage(metaNode);
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 a319c53..ff1011a 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
@@ -167,7 +167,7 @@
     }
 
     public byte getTreeHeight(ITreeIndexFrame frame) throws HyracksDataException {
-        ICachedPage rootNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, rootPage), false);
+        ICachedPage rootNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, rootPage));
         rootNode.acquireReadLatch();
         try {
             frame.setPage(rootNode);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/AbstractTreeIndexBulkLoader.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/AbstractTreeIndexBulkLoader.java
index 45a88a7..1ed1264 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/AbstractTreeIndexBulkLoader.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/AbstractTreeIndexBulkLoader.java
@@ -36,6 +36,7 @@
 import org.apache.hyracks.storage.common.buffercache.IFIFOPageWriter;
 import org.apache.hyracks.storage.common.buffercache.IPageWriteCallback;
 import org.apache.hyracks.storage.common.buffercache.PageWriteFailureCallback;
+import org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCacheWriteContext;
 import org.apache.hyracks.storage.common.compression.file.ICompressedPageWriter;
 import org.apache.hyracks.storage.common.file.BufferedFileHandle;
 
@@ -76,7 +77,7 @@
         interiorFrame = treeIndex.getInteriorFrameFactory().createFrame();
         metaFrame = freePageManager.createMetadataFrame();
 
-        pageWriter = bufferCache.createFIFOWriter(callback, this);
+        pageWriter = bufferCache.createFIFOWriter(callback, this, DefaultBufferCacheWriteContext.INSTANCE);
 
         if (!treeIndex.isEmptyTree(leafFrame)) {
             throw HyracksDataException.create(ErrorCode.CANNOT_BULK_LOAD_NON_EMPTY_TREE);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/TreeIndexDiskOrderScanCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/TreeIndexDiskOrderScanCursor.java
index 36fba76..14decb4 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/TreeIndexDiskOrderScanCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/TreeIndexDiskOrderScanCursor.java
@@ -151,7 +151,7 @@
     }
 
     protected ICachedPage acquireNextPage() throws HyracksDataException {
-        ICachedPage nextPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, currentPageId), false);
+        ICachedPage nextPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, currentPageId));
         nextPage.acquireReadLatch();
         return nextPage;
     }
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 959f27a..4b8337b 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
@@ -53,7 +53,7 @@
         // scan entire file to determine pages in each level
         int maxPageId = freePageManager.getMaxPageId(metaFrame);
         for (int pageId = 0; pageId <= maxPageId; pageId++) {
-            ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId), false);
+            ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId));
             page.acquireReadLatch();
             try {
                 frame.setPage(page);
@@ -87,7 +87,7 @@
                         int pageId = remainingPageIds[index];
 
                         // pin & latch then immediately unlatch & unpin
-                        ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId), false);
+                        ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId));
                         page.acquireReadLatch();
                         page.releaseReadLatch();
                         bufferCache.unpin(page);
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 5410282..b6f239b 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
@@ -49,7 +49,7 @@
         treeIndexStats.begin();
         int maxPageId = freePageManager.getMaxPageId(metaFrame);
         for (int pageId = 0; pageId <= maxPageId; pageId++) {
-            ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId), false);
+            ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId));
             page.acquireReadLatch();
             try {
                 metaFrame.setPage(page);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree-column/src/main/java/org/apache/hyracks/storage/am/lsm/btree/column/impls/btree/ColumnBTreeRangeSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree-column/src/main/java/org/apache/hyracks/storage/am/lsm/btree/column/impls/btree/ColumnBTreeRangeSearchCursor.java
index 39952df..abac582 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree-column/src/main/java/org/apache/hyracks/storage/am/lsm/btree/column/impls/btree/ColumnBTreeRangeSearchCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree-column/src/main/java/org/apache/hyracks/storage/am/lsm/btree/column/impls/btree/ColumnBTreeRangeSearchCursor.java
@@ -86,7 +86,7 @@
     private void fetchNextLeafPage(int leafPage) throws HyracksDataException {
         int nextLeafPage = leafPage;
         do {
-            ICachedPage nextLeaf = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, nextLeafPage), false);
+            ICachedPage nextLeaf = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, nextLeafPage));
             stats.getPageCounter().update(1);
             bufferCache.unpin(page0);
             page0 = nextLeaf;
@@ -285,7 +285,7 @@
     @Override
     public ICachedPage pin(int pageId) throws HyracksDataException {
         stats.getPageCounter().update(1);
-        return bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId), false);
+        return bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId));
     }
 
     @Override
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 0e4f835..bccce90 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
@@ -19,6 +19,8 @@
 
 package org.apache.hyracks.storage.am.lsm.common.freepage;
 
+import static org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider.NEW;
+
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -97,11 +99,11 @@
     public void init(ITreeIndexFrameFactory interiorFrameFactory, ITreeIndexFrameFactory leafFrameFactory)
             throws HyracksDataException {
         currentPageId.set(1);
-        ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, 0), true);
+        ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, 0), NEW);
         page.acquireWriteLatch();
         page.releaseWriteLatch(false);
         bufferCache.unpin(page);
-        page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, currentPageId.get()), true);
+        page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, currentPageId.get()), NEW);
         if (leafFrameFactory != null) {
             page.acquireWriteLatch();
             try {
@@ -133,7 +135,7 @@
 
     @Override
     public boolean isEmpty(ITreeIndexFrame frame, int rootPage) throws HyracksDataException {
-        ICachedPage rootNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, rootPage), false);
+        ICachedPage rootNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, rootPage));
         rootNode.acquireReadLatch();
         try {
             frame.setPage(rootNode);
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 cff54a0..0c442a7 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
@@ -115,8 +115,7 @@
                 return TreeIndexState.INVALID;
             }
             ITreeIndexMetadataFrame metadataFrame = treeIndex.getPageManager().createMetadataFrame();
-            ICachedPage page =
-                    bufferCache.pin(BufferedFileHandle.getDiskPageId(treeIndex.getFileId(), metadataPage), false);
+            ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(treeIndex.getFileId(), metadataPage));
             page.acquireReadLatch();
             try {
                 metadataFrame.setPage(page);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/MultitenantVirtualBufferCache.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/MultitenantVirtualBufferCache.java
index 959bb50..d145558 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/MultitenantVirtualBufferCache.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/MultitenantVirtualBufferCache.java
@@ -18,6 +18,8 @@
  */
 package org.apache.hyracks.storage.am.lsm.common.impls;
 
+import static org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider.DEFAULT;
+
 import java.util.HashMap;
 import java.util.Map;
 
@@ -31,6 +33,8 @@
 import org.apache.hyracks.storage.common.buffercache.IFIFOPageWriter;
 import org.apache.hyracks.storage.common.buffercache.IPageWriteCallback;
 import org.apache.hyracks.storage.common.buffercache.IPageWriteFailureCallback;
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheReadContext;
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheWriteContext;
 import org.apache.hyracks.storage.common.file.IFileMapManager;
 import org.apache.hyracks.util.JSONUtil;
 
@@ -65,18 +69,23 @@
     }
 
     @Override
-    public ICachedPage pin(long dpid, boolean newPage) throws HyracksDataException {
-        return vbc.pin(dpid, newPage);
+    public ICachedPage pin(long dpid) throws HyracksDataException {
+        return pin(dpid, DEFAULT);
     }
 
     @Override
-    public ICachedPage pin(long dpid, boolean newPage, boolean incrementStats) throws HyracksDataException {
-        return vbc.pin(dpid, newPage);
+    public ICachedPage pin(long dpid, IBufferCacheReadContext context) throws HyracksDataException {
+        return vbc.pin(dpid, context);
     }
 
     @Override
     public void unpin(ICachedPage page) throws HyracksDataException {
-        vbc.unpin(page);
+        unpin(page, DEFAULT);
+    }
+
+    @Override
+    public void unpin(ICachedPage page, IBufferCacheReadContext context) throws HyracksDataException {
+        vbc.unpin(page, context);
     }
 
     @Override
@@ -160,8 +169,9 @@
     }
 
     @Override
-    public IFIFOPageWriter createFIFOWriter(IPageWriteCallback callback, IPageWriteFailureCallback failureCallback) {
-        return vbc.createFIFOWriter(callback, failureCallback);
+    public IFIFOPageWriter createFIFOWriter(IPageWriteCallback callback, IPageWriteFailureCallback failureCallback,
+            IBufferCacheWriteContext context) {
+        return vbc.createFIFOWriter(callback, failureCallback, context);
     }
 
     @Override
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 e222461..72f1e4a 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
@@ -18,6 +18,8 @@
  */
 package org.apache.hyracks.storage.am.lsm.common.impls;
 
+import static org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider.DEFAULT;
+
 import java.nio.ByteBuffer;
 import java.util.HashMap;
 import java.util.Map;
@@ -39,6 +41,8 @@
 import org.apache.hyracks.storage.common.buffercache.IPageWriteCallback;
 import org.apache.hyracks.storage.common.buffercache.IPageWriteFailureCallback;
 import org.apache.hyracks.storage.common.buffercache.VirtualPage;
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheReadContext;
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheWriteContext;
 import org.apache.hyracks.storage.common.file.BufferedFileHandle;
 import org.apache.hyracks.storage.common.file.FileMapManager;
 import org.apache.hyracks.storage.common.file.IFileMapManager;
@@ -228,7 +232,13 @@
     }
 
     @Override
-    public ICachedPage pin(long dpid, boolean newPage) throws HyracksDataException {
+    public ICachedPage pin(long dpid) throws HyracksDataException {
+        return pin(dpid, DEFAULT);
+    }
+
+    @Override
+    public ICachedPage pin(long dpid, IBufferCacheReadContext context) throws HyracksDataException {
+        boolean newPage = context.isNewPage();
         VirtualPage page;
         int hash = hash(dpid);
         CacheBucket bucket = buckets[hash];
@@ -260,11 +270,6 @@
         return page;
     }
 
-    @Override
-    public ICachedPage pin(long dpid, boolean newPage, boolean incrementStats) throws HyracksDataException {
-        return pin(dpid, newPage);
-    }
-
     private int hash(long dpid) {
         int hashValue = (int) dpid ^ (Integer.reverse((int) (dpid >>> 32)) >>> 1);
         return hashValue % buckets.length;
@@ -326,6 +331,10 @@
     }
 
     @Override
+    public void unpin(ICachedPage page, IBufferCacheReadContext context) throws HyracksDataException {
+    }
+
+    @Override
     public void flush(ICachedPage page) throws HyracksDataException {
         throw new UnsupportedOperationException();
     }
@@ -400,7 +409,8 @@
     }
 
     @Override
-    public IFIFOPageWriter createFIFOWriter(IPageWriteCallback callback, IPageWriteFailureCallback failureCallback) {
+    public IFIFOPageWriter createFIFOWriter(IPageWriteCallback callback, IPageWriteFailureCallback failureCallback,
+            IBufferCacheWriteContext context) {
         throw new UnsupportedOperationException("Virtual buffer caches don't have FIFO writers");
     }
 
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/AbstractOnDiskInvertedListCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/AbstractOnDiskInvertedListCursor.java
index 9886da4..c5ad6eb 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/AbstractOnDiskInvertedListCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/AbstractOnDiskInvertedListCursor.java
@@ -130,7 +130,8 @@
 
     /**
      * Tries to allocate enough buffers to read the inverted list at once. If memory budget is not enough, this method
-     * stops allocating buffers. */
+     * stops allocating buffers.
+     */
     protected void allocateBuffers() throws HyracksDataException {
         do {
             ByteBuffer tmpBuffer = bufferManagerForSearch.acquireFrame(bufferCache.getPageSize());
@@ -205,7 +206,7 @@
         int currentBufferIdx = 0;
         ByteBuffer tmpBuffer;
         for (int i = bufferStartPageId; i <= endPageId; i++) {
-            page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, i), false);
+            page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, i));
             stats.getPageCounter().update(1);
             // Copies the content to the buffer (working memory).
             // Assumption: processing inverted list takes time; so, we don't want to keep them on the buffer cache.
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndex.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndex.java
index 97a541a..8ef078a 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndex.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndex.java
@@ -71,6 +71,7 @@
 import org.apache.hyracks.storage.common.buffercache.IPageWriteCallback;
 import org.apache.hyracks.storage.common.buffercache.NoOpPageWriteCallback;
 import org.apache.hyracks.storage.common.buffercache.PageWriteFailureCallback;
+import org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCacheWriteContext;
 import org.apache.hyracks.storage.common.file.BufferedFileHandle;
 
 /**
@@ -89,6 +90,7 @@
 
     // Type traits to be appended to the token type trait which finally form the BTree field type traits.
     protected static final ITypeTraits[] btreeValueTypeTraits = new ITypeTraits[4];
+
     static {
         // startPageId
         btreeValueTypeTraits[0] = IntegerPointable.TYPE_TRAITS;
@@ -298,7 +300,7 @@
             currentPageId = startPageId;
             currentPage = bufferCache.confiscatePage(BufferedFileHandle.getDiskPageId(fileId, currentPageId));
             invListBuilder.setTargetBuffer(currentPage.getBuffer().array(), 0);
-            queue = bufferCache.createFIFOWriter(callback, this);
+            queue = bufferCache.createFIFOWriter(callback, this, DefaultBufferCacheWriteContext.INSTANCE);
         }
 
         protected void pinNextPage() throws HyracksDataException {
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/fixedsize/FixedSizeElementInvertedListScanCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/fixedsize/FixedSizeElementInvertedListScanCursor.java
index 458eb6b..f237865 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/fixedsize/FixedSizeElementInvertedListScanCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/fixedsize/FixedSizeElementInvertedListScanCursor.java
@@ -119,7 +119,7 @@
             return;
         }
         unloadPages();
-        page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, currentPageId), false);
+        page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, currentPageId));
         pinnedPageId = currentPageId;
         pinned = true;
         stats.getPageCounter().update(1);
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 9ef305c..3d3a069 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
@@ -76,8 +76,7 @@
         // We don't latch pages since this code is only used by flush () before
         // bulk-loading the r-tree to disk and flush is not concurrent.
         //
-        ICachedPage node1 =
-                bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, tPointers[currentTupleIndex * 2]), false);
+        ICachedPage node1 = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, tPointers[currentTupleIndex * 2]));
         try {
             leafFrame1.setPage(node1);
             frameTuple1.resetByTupleOffset(leafFrame1.getBuffer().array(), tPointers[currentTupleIndex * 2 + 1]);
@@ -183,10 +182,10 @@
         int j1 = tPointers[tp1 * 2 + 1];
         int i2 = tp2i;
         int j2 = tp2j;
-        ICachedPage node1 = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, i1), false);
+        ICachedPage node1 = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, i1));
         try {
             leafFrame1.setPage(node1);
-            ICachedPage node2 = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, i2), false);
+            ICachedPage node2 = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, i2));
             try {
                 leafFrame2.setPage(node2);
                 frameTuple1.resetByTupleOffset(leafFrame1.getBuffer().array(), j1);
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 d85200f..356cc70 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
@@ -19,6 +19,8 @@
 
 package org.apache.hyracks.storage.am.rtree.impls;
 
+import static org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider.NEW;
+
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
@@ -104,7 +106,7 @@
     public void printTree(int pageId, ICachedPage parent, boolean unpin, IRTreeLeafFrame leafFrame,
             IRTreeInteriorFrame interiorFrame, byte treeHeight, ISerializerDeserializer[] keySerdes,
             StringBuilder strBuilder, MultiComparator cmp) throws Exception {
-        ICachedPage node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId), false);
+        ICachedPage node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId));
         node.acquireReadLatch();
         try {
             if (parent != null && unpin == true) {
@@ -174,7 +176,7 @@
 
             while (true) {
                 if (!writeLatched) {
-                    node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId), false);
+                    node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId));
                     ctx.getInteriorFrame().setPage(node);
                     isLeaf = ctx.getInteriorFrame().isLeaf();
                     if (isLeaf) {
@@ -232,7 +234,7 @@
                             readLatched = false;
                             bufferCache.unpin(node);
 
-                            node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId), false);
+                            node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId));
                             node.acquireWriteLatch();
                             writeLatched = true;
                             ctx.getInteriorFrame().setPage(node);
@@ -349,7 +351,7 @@
             case INSUFFICIENT_SPACE: {
                 int rightPageId = freePageManager.takePage(ctx.getMetaFrame());
                 ICachedPage rightNode =
-                        bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), rightPageId), true);
+                        bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), rightPageId), NEW);
                 rightNode.acquireWriteLatch();
 
                 try {
@@ -394,7 +396,7 @@
                 if (pageId == rootPage) {
                     int newLeftId = freePageManager.takePage(ctx.getMetaFrame());
                     ICachedPage newLeftNode =
-                            bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), newLeftId), true);
+                            bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), newLeftId), NEW);
                     newLeftNode.acquireWriteLatch();
                     succeeded = false;
                     try {
@@ -452,7 +454,7 @@
         boolean succeeded = false;
         boolean writeLatched = false;
         int parentId = ctx.getPathList().getLastPageId();
-        ICachedPage parentNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), parentId), false);
+        ICachedPage parentNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), parentId));
         parentNode.acquireWriteLatch();
         writeLatched = true;
         ctx.getInteriorFrame().setPage(parentNode);
@@ -478,7 +480,7 @@
                     }
 
                     parentId = rightPage;
-                    parentNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), parentId), false);
+                    parentNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), parentId));
                     parentNode.acquireWriteLatch();
                     writeLatched = true;
                     ctx.getInteriorFrame().setPage(parentNode);
@@ -532,7 +534,7 @@
                 pageId = ctx.getTraverseList().getFirstPageId();
                 parentIndex = ctx.getTraverseList().getFirstPageIndex();
 
-                node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId), false);
+                node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId));
                 node.acquireReadLatch();
                 readLatched = true;
                 ctx.getInteriorFrame().setPage(node);
@@ -615,7 +617,7 @@
                 int pageId = ctx.getPathList().getLastPageId();
                 long parentLsn = ctx.getPathList().getLastPageLsn();
                 ctx.getPathList().moveLast();
-                node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId), false);
+                node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId));
                 node.acquireReadLatch();
                 readLatched = true;
                 ctx.getInteriorFrame().setPage(node);
@@ -648,7 +650,7 @@
                         readLatched = false;
                         bufferCache.unpin(node);
 
-                        node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId), false);
+                        node = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), pageId));
                         node.acquireWriteLatch();
                         writeLatched = true;
                         ctx.getLeafFrame().setPage(node);
@@ -737,7 +739,7 @@
         int currentPageId = bulkloadLeafStart;
         int maxPageId = freePageManager.getMaxPageId(ctx.getMetaFrame());
 
-        ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), currentPageId), false);
+        ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(getFileId(), currentPageId));
         page.acquireReadLatch();
         try {
             cursor.setBufferCache(bufferCache);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTreeSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTreeSearchCursor.java
index 80fc5bd..f78f010 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTreeSearchCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/impls/RTreeSearchCursor.java
@@ -115,7 +115,7 @@
             if (fileId < 0) {
                 throw new IllegalStateException();
             }
-            ICachedPage node = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId), false);
+            ICachedPage node = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId));
             node.acquireReadLatch();
             stats.getPageCounter().update(1);
             readLatched = true;
@@ -154,7 +154,7 @@
                 } else {
                     page = node;
                     this.pageId = pageId; // This is only needed for the
-                                          // LSMRTree flush operation
+                    // LSMRTree flush operation
                     leafFrame.setPage(page);
                     tupleIndex = 0;
                     succeeded = true;
@@ -191,15 +191,15 @@
                     if (leafFrame.intersect(searchKey, i, cmp)) {
                         frameTuple.resetByTupleIndex(leafFrame, i);
                         currentTupleIndex = i; // This is only needed for the
-                                               // LSMRTree flush operation
+                        // LSMRTree flush operation
                         tupleIndexInc = i + 1;
                         return true;
                     }
                 } else {
                     frameTuple.resetByTupleIndex(leafFrame, i);
                     currentTupleIndex = i; // This is only needed for the
-                                           // LSMRTree
-                                           // flush operation
+                    // LSMRTree
+                    // flush operation
                     tupleIndexInc = i + 1;
                     return true;
                 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java
index fed916d..1660b7e 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java
@@ -43,6 +43,10 @@
 import org.apache.hyracks.api.lifecycle.ILifeCycleComponent;
 import org.apache.hyracks.api.replication.IIOReplicationManager;
 import org.apache.hyracks.api.util.ExceptionUtils;
+import org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider;
+import org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCacheWriteContext;
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheReadContext;
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheWriteContext;
 import org.apache.hyracks.storage.common.compression.file.ICompressedPageWriter;
 import org.apache.hyracks.storage.common.file.BufferedFileHandle;
 import org.apache.hyracks.storage.common.file.IFileMapManager;
@@ -52,6 +56,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+// TODO should be trimmed from all useless artifacts that are required by tests
 public class BufferCache implements IBufferCacheInternal, ILifeCycleComponent, IThreadStatsCollector {
 
     private static final Logger LOGGER = LogManager.getLogger();
@@ -67,6 +72,7 @@
 
     private final int pageSize;
     private final int maxOpenFiles;
+    private final IBufferCacheReadContext defaultContext;
     final IIOManager ioManager;
     private final CacheBucket[] pageMap;
     private final IPageReplacementStrategy pageReplacementStrategy;
@@ -92,11 +98,13 @@
 
     public BufferCache(IIOManager ioManager, IPageReplacementStrategy pageReplacementStrategy,
             IPageCleanerPolicy pageCleanerPolicy, IFileMapManager fileMapManager, int maxOpenFiles, int ioQueuelen,
-            ThreadFactory threadFactory) {
+            ThreadFactory threadFactory, Map<Integer, BufferedFileHandle> fileInfoMap,
+            IBufferCacheReadContext defaultContext) {
         this.headerPageCache = new ArrayBlockingQueue<>(ioQueuelen);
         this.ioManager = ioManager;
         this.pageSize = pageReplacementStrategy.getPageSize();
         this.maxOpenFiles = maxOpenFiles;
+        this.defaultContext = defaultContext;
         pageReplacementStrategy.setBufferCache(this);
         pageMap = new CacheBucket[pageReplacementStrategy.getMaxAllowedNumPages() * MAP_FACTOR + 1];
         for (int i = 0; i < pageMap.length; ++i) {
@@ -107,7 +115,7 @@
         this.fileMapManager = fileMapManager;
 
         Executor executor = Executors.newCachedThreadPool(threadFactory);
-        fileInfoMap = new HashMap<>();
+        this.fileInfoMap = fileInfoMap;
         cleanerThread = new CleanerThread();
         executor.execute(cleanerThread);
         closed = false;
@@ -123,10 +131,10 @@
     //this constructor is used when replication is enabled to pass the IIOReplicationManager
     public BufferCache(IIOManager ioManager, IPageReplacementStrategy pageReplacementStrategy,
             IPageCleanerPolicy pageCleanerPolicy, IFileMapManager fileMapManager, int maxOpenFiles, int ioQueueLen,
-            ThreadFactory threadFactory, IIOReplicationManager ioReplicationManager) {
-
+            ThreadFactory threadFactory, IIOReplicationManager ioReplicationManager,
+            Map<Integer, BufferedFileHandle> fileInfoMap) {
         this(ioManager, pageReplacementStrategy, pageCleanerPolicy, fileMapManager, maxOpenFiles, ioQueueLen,
-                threadFactory);
+                threadFactory, fileInfoMap, DefaultBufferCachePageOperationContextProvider.DEFAULT);
         this.ioReplicationManager = ioReplicationManager;
     }
 
@@ -164,19 +172,20 @@
     }
 
     @Override
-    public ICachedPage pin(long dpid, boolean newPage) throws HyracksDataException {
-        return pin(dpid, newPage, true);
+    public ICachedPage pin(long dpid) throws HyracksDataException {
+        return pin(dpid, defaultContext);
     }
 
     @Override
-    public ICachedPage pin(long dpid, boolean newPage, boolean incrementStats) throws HyracksDataException {
+    public ICachedPage pin(long dpid, IBufferCacheReadContext context) throws HyracksDataException {
+        boolean newPage = context.isNewPage();
         // Calling the pinSanityCheck should be used only for debugging, since
         // the synchronized block over the fileInfoMap is a hot spot.
         if (DEBUG) {
             pinSanityCheck(dpid);
         }
         final IThreadStats threadStats = statsSubscribers.get(Thread.currentThread());
-        if (threadStats != null && incrementStats) {
+        if (threadStats != null && context.incrementStats()) {
             threadStats.pagePinned();
         }
         CachedPage cPage = findPage(dpid);
@@ -193,12 +202,16 @@
                     confiscateLock.unlock();
                 }
             }
+
+            // Notify context page is going to be pinned
+            context.onPin(cPage);
+
             // Resolve race of multiple threads trying to read the page from
             // disk.
             synchronized (cPage) {
                 if (!cPage.valid) {
                     try {
-                        tryRead(cPage, incrementStats);
+                        tryRead(cPage, context);
                         cPage.valid = true;
                     } catch (Exception e) {
                         LOGGER.log(ExceptionUtils.causedByInterrupt(e) ? Level.DEBUG : Level.WARN,
@@ -525,10 +538,10 @@
         return false;
     }
 
-    private void tryRead(CachedPage cPage, boolean incrementStats) throws HyracksDataException {
+    private void tryRead(CachedPage cPage, IBufferCacheReadContext context) throws HyracksDataException {
         for (int i = 1; i <= MAX_PAGE_READ_ATTEMPTS; i++) {
             try {
-                read(cPage, incrementStats);
+                read(cPage, context);
                 return;
             } catch (HyracksDataException readException) {
                 if (readException.matches(ErrorCode.CANNOT_READ_CLOSED_FILE) && i != MAX_PAGE_READ_ATTEMPTS) {
@@ -552,12 +565,12 @@
         }
     }
 
-    private void read(CachedPage cPage, boolean incrementStats) throws HyracksDataException {
+    private void read(CachedPage cPage, IBufferCacheReadContext context) throws HyracksDataException {
         BufferedFileHandle fInfo = getFileHandle(cPage);
         cPage.buffer.clear();
         fInfo.read(cPage);
         final IThreadStats threadStats = statsSubscribers.get(Thread.currentThread());
-        if (threadStats != null && incrementStats) {
+        if (threadStats != null && context.incrementStats()) {
             threadStats.coldRead();
         }
     }
@@ -568,7 +581,7 @@
         pageReplacementStrategy.resizePage((ICachedPageInternal) cPage, totalPages, extraPageBlockHelper);
     }
 
-    void write(CachedPage cPage) throws HyracksDataException {
+    void write(CachedPage cPage, IBufferCacheWriteContext context) throws HyracksDataException {
         BufferedFileHandle fInfo = getFileHandle(cPage);
         // synchronize on fInfo to prevent the file handle from being deleted until the page is written.
         synchronized (fInfo) {
@@ -582,9 +595,16 @@
 
     @Override
     public void unpin(ICachedPage page) throws HyracksDataException {
+        unpin(page, defaultContext);
+    }
+
+    @Override
+    public void unpin(ICachedPage page, IBufferCacheReadContext context) throws HyracksDataException {
         if (closed) {
             throw new HyracksDataException("unpin called on a closed cache");
         }
+
+        context.onUnpin(page);
         int pinCount = ((CachedPage) page).pinCount.decrementAndGet();
         if (DEBUG && pinCount == 0) {
             pinnedPageOwner.remove(page);
@@ -622,6 +642,9 @@
         }
     }
 
+    /**
+     * This class is useless in an actual deployment as on-disk pages are immutable. Only tests need those.
+     */
     private class CleanerThread implements Runnable {
         private volatile boolean shutdownStart = false;
         private volatile boolean shutdownComplete = false;
@@ -665,7 +688,7 @@
             }
             boolean cleaned = true;
             try {
-                write(cPage);
+                write(cPage, DefaultBufferCacheWriteContext.INSTANCE);
             } catch (HyracksDataException e) {
                 LOGGER.log(Level.WARN, "Unable to write dirty page", e);
                 cleaned = false;
@@ -894,7 +917,7 @@
             int pinCount;
             if (cPage.dirty.get()) {
                 if (flushDirtyPages) {
-                    write(cPage);
+                    write(cPage, DefaultBufferCacheWriteContext.INSTANCE);
                 }
                 cPage.dirty.set(false);
                 pinCount = cPage.pinCount.decrementAndGet();
@@ -1384,8 +1407,9 @@
     }
 
     @Override
-    public IFIFOPageWriter createFIFOWriter(IPageWriteCallback callback, IPageWriteFailureCallback failureCallback) {
-        return new FIFOLocalWriter(this, callback, failureCallback);
+    public IFIFOPageWriter createFIFOWriter(IPageWriteCallback callback, IPageWriteFailureCallback failureCallback,
+            IBufferCacheWriteContext context) {
+        return new FIFOLocalWriter(this, callback, failureCallback, context);
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCacheHeaderHelper.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCacheHeaderHelper.java
index a913513..a105e60 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCacheHeaderHelper.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCacheHeaderHelper.java
@@ -25,8 +25,8 @@
 import org.apache.hyracks.api.compression.ICompressorDecompressor;
 
 public class BufferCacheHeaderHelper {
-    private static final int FRAME_MULTIPLIER_OFF = 0;
-    private static final int EXTRA_BLOCK_PAGE_ID_OFF = FRAME_MULTIPLIER_OFF + 4; // 4
+    public static final int FRAME_MULTIPLIER_OFF = 0;
+    public static final int EXTRA_BLOCK_PAGE_ID_OFF = FRAME_MULTIPLIER_OFF + 4; // 4
 
     private final ByteBuffer[] array;
     private final int pageSizeWithHeader;
@@ -67,6 +67,10 @@
         return buf;
     }
 
+    public ByteBuffer getBuffer() {
+        return buf;
+    }
+
     private void setPageInfo(CachedPage cPage) {
         buf.putInt(FRAME_MULTIPLIER_OFF, cPage.getFrameSizeMultiplier());
         buf.putInt(EXTRA_BLOCK_PAGE_ID_OFF, cPage.getExtraBlockPageId());
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/DebugBufferCache.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/DebugBufferCache.java
index c76c781..43e5940 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/DebugBufferCache.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/DebugBufferCache.java
@@ -19,11 +19,15 @@
 
 package org.apache.hyracks.storage.common.buffercache;
 
+import static org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider.DEFAULT;
+
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.io.FileReference;
 import org.apache.hyracks.api.replication.IIOReplicationManager;
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheReadContext;
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheWriteContext;
 
 /**
  * Implementation of an IBufferCache that counts the number of pins/unpins,
@@ -77,20 +81,25 @@
     }
 
     @Override
-    public ICachedPage pin(long dpid, boolean newPage, boolean incrementStats) throws HyracksDataException {
-        ICachedPage page = bufferCache.pin(dpid, newPage, incrementStats);
+    public ICachedPage pin(long dpid) throws HyracksDataException {
+        return pin(dpid, DEFAULT);
+    }
+
+    @Override
+    public ICachedPage pin(long dpid, IBufferCacheReadContext context) throws HyracksDataException {
+        ICachedPage page = bufferCache.pin(dpid, context);
         pinCount.addAndGet(1);
         return page;
     }
 
     @Override
-    public ICachedPage pin(long dpid, boolean newPage) throws HyracksDataException {
-        return pin(dpid, newPage, true);
+    public void unpin(ICachedPage page) throws HyracksDataException {
+        unpin(page, DEFAULT);
     }
 
     @Override
-    public void unpin(ICachedPage page) throws HyracksDataException {
-        bufferCache.unpin(page);
+    public void unpin(ICachedPage page, IBufferCacheReadContext context) throws HyracksDataException {
+        bufferCache.unpin(page, context);
         unpinCount.addAndGet(1);
     }
 
@@ -199,8 +208,9 @@
     }
 
     @Override
-    public IFIFOPageWriter createFIFOWriter(IPageWriteCallback callback, IPageWriteFailureCallback failureCallback) {
-        return bufferCache.createFIFOWriter(callback, failureCallback);
+    public IFIFOPageWriter createFIFOWriter(IPageWriteCallback callback, IPageWriteFailureCallback failureCallback,
+            IBufferCacheWriteContext context) {
+        return bufferCache.createFIFOWriter(callback, failureCallback, context);
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/DefaultDiskCachedPageAllocator.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/DefaultDiskCachedPageAllocator.java
new file mode 100644
index 0000000..8888cc8
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/DefaultDiskCachedPageAllocator.java
@@ -0,0 +1,33 @@
+/*
+ * 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;
+
+public final class DefaultDiskCachedPageAllocator implements IDiskCachedPageAllocator {
+    public static final IDiskCachedPageAllocator INSTANCE = new DefaultDiskCachedPageAllocator();
+
+    private DefaultDiskCachedPageAllocator() {
+    }
+
+    @Override
+    public CachedPage allocate(int cpid, ByteBuffer buffer, IPageReplacementStrategy replacementStrategy) {
+        return new CachedPage(cpid, buffer, replacementStrategy);
+    }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/FIFOLocalWriter.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/FIFOLocalWriter.java
index e74fe59..ce7fb5b 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/FIFOLocalWriter.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/FIFOLocalWriter.java
@@ -15,6 +15,7 @@
 
 package org.apache.hyracks.storage.common.buffercache;
 
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheWriteContext;
 import org.apache.hyracks.util.ExitUtil;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -27,12 +28,14 @@
 
     private final IPageWriteCallback callback;
     private final IPageWriteFailureCallback failureCallback;
+    private final IBufferCacheWriteContext context;
 
     public FIFOLocalWriter(BufferCache bufferCache, IPageWriteCallback callback,
-            IPageWriteFailureCallback failureCallback) {
+            IPageWriteFailureCallback failureCallback, IBufferCacheWriteContext context) {
         this.bufferCache = bufferCache;
         this.callback = callback;
         this.failureCallback = failureCallback;
+        this.context = context;
     }
 
     @SuppressWarnings("squid:S1181") // System must halt on all IO errors
@@ -41,7 +44,7 @@
         CachedPage cPage = (CachedPage) page;
         try {
             callback.beforeWrite(cPage);
-            bufferCache.write(cPage);
+            bufferCache.write(cPage, context);
             callback.afterWrite(cPage);
         } catch (Exception e) {
             handleWriteFailure(page, e);
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IBufferCache.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IBufferCache.java
index 3e977bd..d11a065 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IBufferCache.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IBufferCache.java
@@ -21,6 +21,8 @@
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.io.FileReference;
 import org.apache.hyracks.api.replication.IIOReplicationManager;
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheReadContext;
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheWriteContext;
 import org.apache.hyracks.storage.common.compression.file.ICompressedPageWriter;
 
 public interface IBufferCache {
@@ -32,52 +34,42 @@
     /**
      * Create file on disk
      *
-     * @param fileRef
-     *            the file to create
+     * @param fileRef the file to create
      * @return the file id
-     * @throws HyracksDataException
-     *             if the file already exists or attempt to create the file failed
+     * @throws HyracksDataException if the file already exists or attempt to create the file failed
      */
     int createFile(FileReference fileRef) throws HyracksDataException;
 
     /**
      * Open the file and register it (if not registered) with the file map manager
      *
-     * @param fileRef
-     *            the file to open
+     * @param fileRef the file to open
      * @return the file id
-     * @throws HyracksDataException
-     *             if the file doesn't exist or buffer cache failed to open the file
+     * @throws HyracksDataException if the file doesn't exist or buffer cache failed to open the file
      */
     int openFile(FileReference fileRef) throws HyracksDataException;
 
     /**
      * Open the mapped file with the passed file id
      *
-     * @param fileId
-     *            the file id
-     * @throws HyracksDataException
-     *             if the file doesn't exist or buffer cache fails to open the file
+     * @param fileId the file id
+     * @throws HyracksDataException if the file doesn't exist or buffer cache fails to open the file
      */
     void openFile(int fileId) throws HyracksDataException;
 
     /**
      * close the file
      *
-     * @param fileId
-     *            the file id
-     * @throws HyracksDataException
-     *             if file doesn't exist or is not open
+     * @param fileId the file id
+     * @throws HyracksDataException if file doesn't exist or is not open
      */
     void closeFile(int fileId) throws HyracksDataException;
 
     /**
      * delete the file from memory and disk
      *
-     * @param fileId
-     *            the file id
-     * @throws HyracksDataException
-     *             if the file doesn't exist or if a failure to delete takes place
+     * @param fileId the file id
+     * @throws HyracksDataException if the file doesn't exist or if a failure to delete takes place
      */
     void deleteFile(int fileId) throws HyracksDataException;
 
@@ -90,56 +82,44 @@
     void deleteFile(FileReference file) throws HyracksDataException;
 
     /**
-     * Pin the page so it can't be evicted from the buffer cache...
+     * Pin a page to indicate it is not evictable
+     * Note: this will use the default {@link IBufferCacheReadContext}
      *
-     * @param dpid
-     *            page id is a unique id that is a combination of file id and page id
-     * @param newPage
-     *            whether this page is expected to be new.
-     *            NOTE: undefined:
-     *            -- what if the flag is true but the page exists?
-     *            -- what if the flag is false but the page doesn't exist
+     * @param dpid page id is a unique id that is a combination of file id and page id
      * @return the pinned page
-     * @throws HyracksDataException
      */
-    ICachedPage pin(long dpid, boolean newPage) throws HyracksDataException;
+    ICachedPage pin(long dpid) throws HyracksDataException;
 
     /**
-     * Pin the page so it can't be evicted from the buffer cache...
+     * Pin a page to indicate it is not evictable
      *
-     * @param dpid
-     *            page id is a unique id that is a combination of file id and page id
-     * @param newPage
-     *            whether this page is expected to be new.
-     *            NOTE: undefined:
-     *            -- what if the flag is true but the page exists?
-     *            -- what if the flag is false but the page doesn't exist
-     *
-     * @param incrementStats
-     *            whether to increment the coldRead and pinCount counters when
-     *            the page is pinned. this is to not bias the count when using
-     *            accessors that cause nested pins due to wrapping file handles,
-     *            like compression
-     *
+     * @param dpid    page id is a unique id that is a combination of file id and page id
+     * @param context read context
      * @return the pinned page
-     * @throws HyracksDataException
      */
-    ICachedPage pin(long dpid, boolean newPage, boolean incrementStats) throws HyracksDataException;
+    ICachedPage pin(long dpid, IBufferCacheReadContext context) throws HyracksDataException;
 
     /**
-     * Unpin a pinned page so its buffer can be recycled
+     * Unpin a page to indicate can be evicted
+     * Note: this will use the default {@link IBufferCacheReadContext}
      *
-     * @param page
-     *            the page
-     * @throws HyracksDataException
+     * @param page the page
      */
     void unpin(ICachedPage page) throws HyracksDataException;
 
     /**
+     * Unpin a page to indicate can be evicted
+     * Note: this will use the default {@link IBufferCacheReadContext}
+     *
+     * @param page    the page
+     * @param context read context
+     */
+    void unpin(ICachedPage page, IBufferCacheReadContext context) throws HyracksDataException;
+
+    /**
      * Flush the page if it is dirty
      *
-     * @param page
-     *            the page to flush
+     * @param page the page to flush
      * @throws HyracksDataException
      */
     void flush(ICachedPage page) throws HyracksDataException;
@@ -148,10 +128,8 @@
      * Force bits that have been already flushed to disk
      * This method doesn't flush all dirty pages to disk but simply calls the sync method on the filesystem api
      *
-     * @param fileId
-     *            the file id
-     * @param metadata
-     *            whether metadata should be synced as well
+     * @param fileId   the file id
+     * @param metadata whether metadata should be synced as well
      * @throws HyracksDataException
      */
     void force(int fileId, boolean metadata) throws HyracksDataException;
@@ -159,8 +137,7 @@
     /**
      * Take a page such that no one else has access to it
      *
-     * @param dpid
-     *            the unique (fileId,pageId)
+     * @param dpid the unique (fileId,pageId)
      * @return the confiscated page or null if no page is available
      * @throws HyracksDataException
      */
@@ -174,14 +151,10 @@
     /**
      * Take a large page such that no one else has access to it
      *
-     * @param dpid
-     *            the unique (fileId,pageId)
-     * @param multiplier
-     *            how many multiples of the original page size
-     * @param extraBlockPageId
-     *            the page id where the large block comes from
-     * @return
-     *         the confiscated page or null if a large page couldn't be found
+     * @param dpid             the unique (fileId,pageId)
+     * @param multiplier       how many multiples of the original page size
+     * @param extraBlockPageId the page id where the large block comes from
+     * @return the confiscated page or null if a large page couldn't be found
      * @throws HyracksDataException
      */
     ICachedPage confiscateLargePage(long dpid, int multiplier, int extraBlockPageId) throws HyracksDataException;
@@ -189,18 +162,15 @@
     /**
      * Return and re-insert a confiscated page
      *
-     * @param page
-     *            the confiscated page
+     * @param page the confiscated page
      */
     void returnPage(ICachedPage page);
 
     /**
      * Return a confiscated page
      *
-     * @param page
-     *            the confiscated page
-     * @param reinsert
-     *            if true, return the page to the cache, otherwise, destroy
+     * @param page     the confiscated page
+     * @param reinsert if true, return the page to the cache, otherwise, destroy
      */
     void returnPage(ICachedPage page, boolean reinsert);
 
@@ -226,8 +196,7 @@
     /**
      * Get the number of pages used for a file
      *
-     * @param fileId
-     *            the file id
+     * @param fileId the file id
      * @return the number of pages used for the file
      * @throws HyracksDataException
      */
@@ -236,8 +205,7 @@
     /**
      * Get the reference count for a file (num of open - num of close)
      *
-     * @param fileId
-     *            the file
+     * @param fileId the file
      * @return the reference count
      */
     int getFileReferenceCount(int fileId);
@@ -251,11 +219,18 @@
     void close() throws HyracksDataException;
 
     /**
+     * A sequential page writer
+     *
+     * @param callback        call back when a page is successfully written
+     * @param failureCallback call back when a page is failed to be written
+     * @param context         write context
      * @return an instance of {@link IFIFOPageWriter} that can be used to write pages to the file
      */
-    IFIFOPageWriter createFIFOWriter(IPageWriteCallback callback, IPageWriteFailureCallback failureCallback);
+    IFIFOPageWriter createFIFOWriter(IPageWriteCallback callback, IPageWriteFailureCallback failureCallback,
+            IBufferCacheWriteContext context);
 
     // TODO: remove the replication out of the buffer cache interface
+
     /**
      * @return true if replication is enabled, false otherwise
      */
@@ -268,7 +243,7 @@
 
     /**
      * Deletes the file and recycle all of its pages without flushing them.
-     *
+     * <p>
      * ONLY call this if you absolutely, positively know this file has no dirty pages in the cache!
      * Bypasses the normal lifecycle of a file handle and evicts all references to it immediately.
      */
@@ -277,12 +252,9 @@
     /**
      * Resize the page
      *
-     * @param page
-     *            the page to resize
-     * @param multiplier
-     *            how many multiples of the original page size
-     * @param extraPageBlockHelper
-     *            helper to determine the location of the resize block
+     * @param page                 the page to resize
+     * @param multiplier           how many multiples of the original page size
+     * @param extraPageBlockHelper helper to determine the location of the resize block
      * @throws HyracksDataException
      */
     void resizePage(ICachedPage page, int multiplier, IExtraPageBlockHelper extraPageBlockHelper)
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IDiskCachedPageAllocator.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IDiskCachedPageAllocator.java
new file mode 100644
index 0000000..cb0e970
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IDiskCachedPageAllocator.java
@@ -0,0 +1,26 @@
+/*
+ * 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;
+
+@FunctionalInterface
+public interface IDiskCachedPageAllocator {
+    CachedPage allocate(int cpid, ByteBuffer buffer, IPageReplacementStrategy replacementStrategy);
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/AbstractBufferCacheReadContext.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/AbstractBufferCacheReadContext.java
new file mode 100644
index 0000000..13b8ee9
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/AbstractBufferCacheReadContext.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.common.buffercache.context.page;
+
+import static org.apache.hyracks.storage.common.buffercache.BufferCacheHeaderHelper.EXTRA_BLOCK_PAGE_ID_OFF;
+import static org.apache.hyracks.storage.common.buffercache.BufferCacheHeaderHelper.FRAME_MULTIPLIER_OFF;
+import static org.apache.hyracks.storage.common.buffercache.IBufferCache.RESERVED_HEADER_BYTES;
+
+import java.nio.ByteBuffer;
+
+import org.apache.hyracks.control.nc.io.IOManager;
+import org.apache.hyracks.storage.common.buffercache.BufferCacheHeaderHelper;
+import org.apache.hyracks.storage.common.buffercache.CachedPage;
+import org.apache.hyracks.storage.common.buffercache.ICachedPage;
+import org.apache.hyracks.storage.common.file.BufferedFileHandle;
+
+abstract class AbstractBufferCacheReadContext implements IBufferCacheReadContext {
+    @Override
+    public final void onPin(ICachedPage page) {
+        // NoOp
+    }
+
+    @Override
+    public final void onUnpin(ICachedPage page) {
+        // NoOp
+    }
+
+    @Override
+    public final ByteBuffer processHeader(IOManager ioManager, BufferedFileHandle fileHandle,
+            BufferCacheHeaderHelper header, CachedPage cPage) {
+        ByteBuffer buf = header.getBuffer();
+        cPage.setFrameSizeMultiplier(buf.getInt(FRAME_MULTIPLIER_OFF));
+        cPage.setExtraBlockPageId(buf.getInt(EXTRA_BLOCK_PAGE_ID_OFF));
+        buf.position(RESERVED_HEADER_BYTES);
+        return buf;
+    }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/DefaultBufferCachePageOperationContextProvider.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/DefaultBufferCachePageOperationContextProvider.java
new file mode 100644
index 0000000..a187b9b
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/DefaultBufferCachePageOperationContextProvider.java
@@ -0,0 +1,39 @@
+/*
+ * 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.context.page;
+
+public class DefaultBufferCachePageOperationContextProvider {
+    private DefaultBufferCachePageOperationContextProvider() {
+    }
+
+    /**
+     * Pin a page and increment stats
+     */
+    public static final IBufferCacheReadContext DEFAULT = new DefaultBufferCacheReadContext();
+
+    /**
+     * Pin a new page
+     */
+    public static final IBufferCacheReadContext NEW = new PinNewBufferCacheReadContext();
+
+    /**
+     * Pin a page and do not increment the stats of the pinned page
+     */
+    public static final IBufferCacheReadContext NO_STATS = new NoStatsBufferCacheReadContext();
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/DefaultBufferCacheReadContext.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/DefaultBufferCacheReadContext.java
new file mode 100644
index 0000000..2a67a06
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/DefaultBufferCacheReadContext.java
@@ -0,0 +1,31 @@
+/*
+ * 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.context.page;
+
+public final class DefaultBufferCacheReadContext extends AbstractBufferCacheReadContext {
+    @Override
+    public boolean isNewPage() {
+        return false;
+    }
+
+    @Override
+    public boolean incrementStats() {
+        return true;
+    }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/DefaultBufferCacheWriteContext.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/DefaultBufferCacheWriteContext.java
new file mode 100644
index 0000000..a5ef722
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/DefaultBufferCacheWriteContext.java
@@ -0,0 +1,44 @@
+/*
+ * 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.context.page;
+
+import java.nio.ByteBuffer;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.io.IFileHandle;
+import org.apache.hyracks.control.nc.io.IOManager;
+
+public final class DefaultBufferCacheWriteContext implements IBufferCacheWriteContext {
+    public static final IBufferCacheWriteContext INSTANCE = new DefaultBufferCacheWriteContext();
+
+    private DefaultBufferCacheWriteContext() {
+    }
+
+    @Override
+    public int write(IOManager ioManager, IFileHandle handle, long offset, ByteBuffer data)
+            throws HyracksDataException {
+        return ioManager.doSyncWrite(handle, offset, data);
+    }
+
+    @Override
+    public long write(IOManager ioManager, IFileHandle handle, long offset, ByteBuffer[] data)
+            throws HyracksDataException {
+        return ioManager.doSyncWrite(handle, offset, data);
+    }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/IBufferCacheReadContext.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/IBufferCacheReadContext.java
new file mode 100644
index 0000000..216ebfb
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/IBufferCacheReadContext.java
@@ -0,0 +1,72 @@
+/*
+ * 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.context.page;
+
+import java.nio.ByteBuffer;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.control.nc.io.IOManager;
+import org.apache.hyracks.storage.common.buffercache.BufferCacheHeaderHelper;
+import org.apache.hyracks.storage.common.buffercache.CachedPage;
+import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+import org.apache.hyracks.storage.common.buffercache.ICachedPage;
+import org.apache.hyracks.storage.common.file.BufferedFileHandle;
+
+/**
+ * Provide a context to {@link IBufferCache} pin/unpin operations as well as processing the header of the first
+ * pages when the {@link IBufferCache} requires to read from disk
+ */
+public interface IBufferCacheReadContext {
+    /**
+     * Signals a page was found and {@link IBufferCache} is about to pin the requested page
+     *
+     * @param page that will be pinned
+     */
+    void onPin(ICachedPage page);
+
+    /**
+     * Signals that a page will be unpinned
+     *
+     * @param page that will be unpinned
+     */
+    void onUnpin(ICachedPage page);
+
+    /**
+     * @return true if pinning a new page, false otherwise
+     */
+    boolean isNewPage();
+
+    /**
+     * @return true to increment {@link IBufferCache} stats, false otherwise
+     */
+    boolean incrementStats();
+
+    /**
+     * Processing the header of a read page during a pin operation
+     * Note: This operation modifies the position and limit of the header
+     *
+     * @param ioManager  I/O manager
+     * @param fileHandle file the page was read from
+     * @param header     header of the page
+     * @param cPage      the pinned page
+     * @return the byte buffer of the header after processing it
+     */
+    ByteBuffer processHeader(IOManager ioManager, BufferedFileHandle fileHandle, BufferCacheHeaderHelper header,
+            CachedPage cPage) throws HyracksDataException;
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/IBufferCacheWriteContext.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/IBufferCacheWriteContext.java
new file mode 100644
index 0000000..7afb4ce
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/IBufferCacheWriteContext.java
@@ -0,0 +1,31 @@
+/*
+ * 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.context.page;
+
+import java.nio.ByteBuffer;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.io.IFileHandle;
+import org.apache.hyracks.control.nc.io.IOManager;
+
+public interface IBufferCacheWriteContext {
+    int write(IOManager ioManager, IFileHandle handle, long offset, ByteBuffer data) throws HyracksDataException;
+
+    long write(IOManager ioManager, IFileHandle handle, long offset, ByteBuffer[] data) throws HyracksDataException;
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/NoStatsBufferCacheReadContext.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/NoStatsBufferCacheReadContext.java
new file mode 100644
index 0000000..fa5ac65
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/NoStatsBufferCacheReadContext.java
@@ -0,0 +1,31 @@
+/*
+ * 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.context.page;
+
+final class NoStatsBufferCacheReadContext extends AbstractBufferCacheReadContext {
+    @Override
+    public boolean isNewPage() {
+        return false;
+    }
+
+    @Override
+    public boolean incrementStats() {
+        return false;
+    }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/PinNewBufferCacheReadContext.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/PinNewBufferCacheReadContext.java
new file mode 100644
index 0000000..4eda54e
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/context/page/PinNewBufferCacheReadContext.java
@@ -0,0 +1,32 @@
+/*
+ * 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.context.page;
+
+final class PinNewBufferCacheReadContext extends AbstractBufferCacheReadContext {
+
+    @Override
+    public boolean isNewPage() {
+        return true;
+    }
+
+    @Override
+    public boolean incrementStats() {
+        return true;
+    }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/compression/file/CompressedFileManager.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/compression/file/CompressedFileManager.java
index 8db171a..0a7bcf0 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/compression/file/CompressedFileManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/compression/file/CompressedFileManager.java
@@ -18,6 +18,8 @@
  */
 package org.apache.hyracks.storage.common.compression.file;
 
+import static org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider.NO_STATS;
+
 import java.nio.ByteBuffer;
 import java.util.EnumSet;
 
@@ -32,10 +34,10 @@
 /**
  * CompressedFileManager is responsible to manage the Look Aside File (LAF file), which contains
  * the compression information. LAF file format is as follow:
- *
+ * <p>
  * [<offset0, size0>, <offset1, size1> .... <offsetN, sizeN>]
  * Each entry <offsetM, sizeM> is an entry of 16-bytes for page M (8 bytes for offset and 8 for size).
- *
+ * <p>
  * The file is responsible to store the beginning and the size of each page after compression.
  */
 public class CompressedFileManager {
@@ -92,11 +94,9 @@
      * If the file is empty (i.e. the number of pages is zero)
      * Then the state will be WRITABLE.
      *
-     * @return
-     *         true if the file exists and was closed
-     *         false otherwise
-     * @throws IllegalStateException
-     *             if the the file is not in CLOSED state
+     * @return true if the file exists and was closed
+     * false otherwise
+     * @throws IllegalStateException if the the file is not in CLOSED state
      */
     public boolean open() throws HyracksDataException {
         ensureState(CLOSED);
@@ -115,11 +115,9 @@
     /**
      * Close LAF file
      *
-     * @return
-     *         true if the LAF file exists and was OPEN
-     *         false otherwise
-     * @throws IllegalStateException
-     *             if the the file is not in OPEN state
+     * @return true if the LAF file exists and was OPEN
+     * false otherwise
+     * @throws IllegalStateException if the the file is not in OPEN state
      */
     public boolean close() throws HyracksDataException {
         ensureState(OPEN_OR_INVALID);
@@ -136,8 +134,7 @@
     /**
      * Close and purge LAF file
      *
-     * @throws IllegalStateException
-     *             if the the file is not in OPEN state
+     * @throws IllegalStateException if the the file is not in OPEN state
      */
     public void purge() throws HyracksDataException {
         if (close()) {
@@ -148,8 +145,7 @@
     /**
      * Delete LAF file
      *
-     * @throws IllegalStateException
-     *             if the the file is not in CLOSED state
+     * @throws IllegalStateException if the the file is not in CLOSED state
      */
     public void delete() throws HyracksDataException {
         ensureState(CLOSED_OR_INVALID);
@@ -161,8 +157,7 @@
     /**
      * Force LAF file content to disk
      *
-     * @throws IllegalStateException
-     *             if the the file is not in CLOSED state
+     * @throws IllegalStateException if the the file is not in CLOSED state
      */
     public void force(boolean metadata) throws HyracksDataException {
         ensureState(OPEN);
@@ -203,15 +198,11 @@
     /**
      * Add extra page information (offset, size) after compression.
      *
-     * @param extraPageId
-     *            extra page ID
-     * @param size
-     *            size of the extra page
-     * @param extraPageIndex
-     *            the index of the extra page (starting from 0)
+     * @param extraPageId    extra page ID
+     * @param size           size of the extra page
+     * @param extraPageIndex the index of the extra page (starting from 0)
      * @return offset for the compressed page.
-     * @throws IllegalStateException
-     *             If the file is not in WRITABLE state
+     * @throws IllegalStateException If the file is not in WRITABLE state
      */
     public long writeExtraPageInfo(int extraPageId, long size, int extraPageIndex) throws HyracksDataException {
         ensureState(WRITABLE);
@@ -231,10 +222,8 @@
      * This methods is used by {@link LAFWriter#endWriting()} to signal the end of writing.
      * After calling this methods, LAF file will be READ-ONLY.
      *
-     * @param totalNumOfPages
-     *            The total number of pages of the index
-     * @throws IllegalStateException
-     *             If the file is not in WRITABLE state
+     * @param totalNumOfPages The total number of pages of the index
+     * @throws IllegalStateException If the file is not in WRITABLE state
      */
     void endWriting(int totalNumOfPages) {
         ensureState(WRITABLE);
@@ -268,6 +257,29 @@
         setCompressedPageInfo(compressedPage.getExtraBlockPageId() + extraPageIndex, compressedPage);
     }
 
+    public long getTotalCompressedSize(int startPageId, int numberOfPages) throws HyracksDataException {
+        int lafPageId = -1;
+        ICachedPage lafPage = null;
+        long totalSize = 0;
+        try {
+            for (int i = startPageId; i < startPageId + numberOfPages; i++) {
+                int nextLafId = getLAFPageId(i);
+                if (nextLafId != lafPageId) {
+                    lafPageId = nextLafId;
+                    lafPage = pinNext(lafPage, nextLafId);
+                }
+
+                int entryOffset = i * ENTRY_LENGTH % bufferCache.getPageSize();
+                totalSize += lafPage.getBuffer().getLong(entryOffset + SIZE_ENTRY_OFFSET);
+            }
+        } finally {
+            // It will unpin only as nextPageId is -1
+            pinNext(lafPage, -1);
+        }
+
+        return totalSize;
+    }
+
     /* ************************
      * LAF general methods
      * ************************
@@ -284,6 +296,17 @@
         return compressorDecompressor;
     }
 
+    public long getPageOffset(int pageId) throws HyracksDataException {
+        ICachedPage page = pinAndGetPage(pageId);
+        try {
+            final ByteBuffer buf = page.getBuffer();
+            int entryOffset = pageId * ENTRY_LENGTH % bufferCache.getPageSize();
+            return buf.getLong(entryOffset);
+        } finally {
+            bufferCache.unpin(page);
+        }
+    }
+
     /* ************************
      * Private methods
      * ************************
@@ -316,7 +339,7 @@
         final long dpid = getDiskPageId(numOfPages - 1);
         //exclude the LAF from the buffer cache pin/read stats, because it is hot in any
         //sane scenario. otherwise it inflates the query's cache ratio greatly. 
-        final ICachedPage page = bufferCache.pin(dpid, false, false);
+        final ICachedPage page = bufferCache.pin(dpid, NO_STATS);
         try {
             final ByteBuffer buf = page.getBuffer();
 
@@ -334,11 +357,15 @@
     }
 
     private ICachedPage pinAndGetPage(int compressedPageId) throws HyracksDataException {
-        final int pageId = compressedPageId * ENTRY_LENGTH / bufferCache.getPageSize();
+        final int pageId = getLAFPageId(compressedPageId);
         //don't increment the stats with this pin. this is because we are pinning on behalf
         //of a caller, so our pins will be an interfering access pattern that conflates itself
         //with the caller's cache pattern.
-        return bufferCache.pin(getDiskPageId(pageId), false, false);
+        return bufferCache.pin(getDiskPageId(pageId), NO_STATS);
+    }
+
+    private int getLAFPageId(int compressedPageId) {
+        return compressedPageId * ENTRY_LENGTH / bufferCache.getPageSize();
     }
 
     private long getDiskPageId(int pageId) {
@@ -368,4 +395,11 @@
             bufferCache.unpin(page);
         }
     }
+
+    private ICachedPage pinNext(ICachedPage page, int nextPageId) throws HyracksDataException {
+        if (page != null) {
+            bufferCache.unpin(page);
+        }
+        return nextPageId >= 0 ? bufferCache.pin(getDiskPageId(nextPageId)) : null;
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/compression/file/LAFWriter.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/compression/file/LAFWriter.java
index 3226786..8598401 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/compression/file/LAFWriter.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/compression/file/LAFWriter.java
@@ -35,6 +35,7 @@
 import org.apache.hyracks.storage.common.buffercache.IFIFOPageWriter;
 import org.apache.hyracks.storage.common.buffercache.NoOpPageWriteCallback;
 import org.apache.hyracks.storage.common.buffercache.PageWriteFailureCallback;
+import org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCacheWriteContext;
 import org.apache.hyracks.storage.common.file.BufferedFileHandle;
 import org.apache.hyracks.util.annotations.NotThreadSafe;
 
@@ -65,7 +66,8 @@
         recycledFrames = new ArrayDeque<>();
         this.fileId = compressedFileManager.getFileId();
         failureCallback = new PageWriteFailureCallback();
-        pageWriter = bufferCache.createFIFOWriter(NoOpPageWriteCallback.INSTANCE, failureCallback);
+        pageWriter = bufferCache.createFIFOWriter(NoOpPageWriteCallback.INSTANCE, failureCallback,
+                DefaultBufferCacheWriteContext.INSTANCE);
         maxNumOfEntries = bufferCache.getPageSize() / ENTRY_LENGTH;
         lastOffset = 0;
         totalNumOfPages = 0;
diff --git a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestStorageManagerComponentHolder.java b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestStorageManagerComponentHolder.java
index ee10176..00a041f 100644
--- a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestStorageManagerComponentHolder.java
+++ b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestStorageManagerComponentHolder.java
@@ -20,6 +20,7 @@
 
 import java.io.File;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.concurrent.ThreadFactory;
 
@@ -43,6 +44,7 @@
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
 import org.apache.hyracks.storage.common.buffercache.ICacheMemoryAllocator;
 import org.apache.hyracks.storage.common.buffercache.IPageReplacementStrategy;
+import org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider;
 import org.apache.hyracks.storage.common.file.FileMapManager;
 import org.apache.hyracks.storage.common.file.IFileMapManager;
 import org.apache.hyracks.storage.common.file.IFileMapProvider;
@@ -147,7 +149,8 @@
         IPageReplacementStrategy prs = new ClockPageReplacementStrategy(allocator, pageSize, numPages);
         IFileMapProvider fileMapProvider = getFileMapProvider();
         bufferCache = new BufferCache(ioManager, prs, new DelayPageCleanerPolicy(1000),
-                (IFileMapManager) fileMapProvider, maxOpenFiles, 10, threadFactory);
+                (IFileMapManager) fileMapProvider, maxOpenFiles, 10, threadFactory, new HashMap<>(),
+                DefaultBufferCachePageOperationContextProvider.DEFAULT);
         return bufferCache;
     }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/FieldPrefixNSMTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/FieldPrefixNSMTest.java
index e9b9e7f..8b5fe68 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/FieldPrefixNSMTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/FieldPrefixNSMTest.java
@@ -19,6 +19,8 @@
 
 package org.apache.hyracks.storage.am.btree;
 
+import static org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider.NEW;
+
 import java.io.DataOutput;
 import java.util.Random;
 
@@ -128,7 +130,7 @@
         bufferCache.createFile(harness.getFileReference());
         int btreeFileId = bufferCache.openFile(harness.getFileReference());
         IHyracksTaskContext ctx = harness.getHyracksTaskContext();
-        ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(btreeFileId, 0), true);
+        ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(btreeFileId, 0), NEW);
         try {
 
             BTreeTypeAwareTupleWriter tupleWriter = new BTreeTypeAwareTupleWriter(typeTraits, false, null, null);
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/StorageFileAccessTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/StorageFileAccessTest.java
index 576c6bb..bdcb206 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/StorageFileAccessTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-btree-test/src/test/java/org/apache/hyracks/storage/am/btree/StorageFileAccessTest.java
@@ -87,7 +87,7 @@
             }
 
             try {
-                ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId), false);
+                ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId));
                 LatchType latch = null;
 
                 switch (fta) {
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestVirtualBufferCache.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestVirtualBufferCache.java
index e697b54..154e34a 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestVirtualBufferCache.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestVirtualBufferCache.java
@@ -18,6 +18,8 @@
  */
 package org.apache.hyracks.storage.am.lsm.btree.impl;
 
+import static org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider.DEFAULT;
+
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
@@ -33,6 +35,9 @@
 import org.apache.hyracks.storage.common.buffercache.IFIFOPageWriter;
 import org.apache.hyracks.storage.common.buffercache.IPageWriteCallback;
 import org.apache.hyracks.storage.common.buffercache.IPageWriteFailureCallback;
+import org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCacheWriteContext;
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheReadContext;
+import org.apache.hyracks.storage.common.buffercache.context.page.IBufferCacheWriteContext;
 import org.apache.hyracks.storage.common.file.IFileMapManager;
 
 public class TestVirtualBufferCache implements IVirtualBufferCache {
@@ -89,8 +94,13 @@
     }
 
     @Override
-    public ICachedPage pin(long dpid, boolean newPage) throws HyracksDataException {
-        ICachedPage page = vbc.pin(dpid, newPage);
+    public ICachedPage pin(long dpid) throws HyracksDataException {
+        return pin(dpid, DEFAULT);
+    }
+
+    @Override
+    public ICachedPage pin(long dpid, IBufferCacheReadContext context) throws HyracksDataException {
+        ICachedPage page = vbc.pin(dpid, context);
         // the memory component can be full after each but, but isFull may not be called by the memory component
         // for correctness, we call isFull here after each pin
         for (ILSMMemoryComponent component : isFullMap.keySet()) {
@@ -101,13 +111,13 @@
     }
 
     @Override
-    public ICachedPage pin(long dpid, boolean newPage, boolean incrementStats) throws HyracksDataException {
-        return pin(dpid, newPage);
+    public void unpin(ICachedPage page) throws HyracksDataException {
+        unpin(page, DEFAULT);
     }
 
     @Override
-    public void unpin(ICachedPage page) throws HyracksDataException {
-        vbc.unpin(page);
+    public void unpin(ICachedPage page, IBufferCacheReadContext context) throws HyracksDataException {
+        vbc.unpin(page, context);
     }
 
     @Override
@@ -172,8 +182,9 @@
     }
 
     @Override
-    public IFIFOPageWriter createFIFOWriter(IPageWriteCallback callback, IPageWriteFailureCallback failureCallback) {
-        return vbc.createFIFOWriter(callback, failureCallback);
+    public IFIFOPageWriter createFIFOWriter(IPageWriteCallback callback, IPageWriteFailureCallback failureCallback,
+            IBufferCacheWriteContext context) {
+        return vbc.createFIFOWriter(callback, failureCallback, DefaultBufferCacheWriteContext.INSTANCE);
     }
 
     @Override
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/org/apache/hyracks/storage/am/lsm/common/test/VirtualBufferCacheTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/org/apache/hyracks/storage/am/lsm/common/test/VirtualBufferCacheTest.java
index 0e749fc..e7d258c 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/org/apache/hyracks/storage/am/lsm/common/test/VirtualBufferCacheTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/org/apache/hyracks/storage/am/lsm/common/test/VirtualBufferCacheTest.java
@@ -18,6 +18,7 @@
  */
 package org.apache.hyracks.storage.am.lsm.common.test;
 
+import static org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider.NEW;
 import static org.junit.Assert.assertTrue;
 
 import java.util.Collections;
@@ -71,7 +72,7 @@
         }
 
         public void pin(VirtualBufferCache vbc, int multiplier) throws HyracksDataException {
-            ICachedPage p = vbc.pin(BufferedFileHandle.getDiskPageId(fileId, pinCount), true);
+            ICachedPage p = vbc.pin(BufferedFileHandle.getDiskPageId(fileId, pinCount), NEW);
             pinnedPages.add(p);
             pinCount++;
             totalNumPages++;
@@ -148,7 +149,7 @@
                 switch (req.getType()) {
                     case PIN_PAGE:
                         ICachedPage p = vbc.pin(
-                                BufferedFileHandle.getDiskPageId(fileState.fileId, fileState.helper.pinCount), true);
+                                BufferedFileHandle.getDiskPageId(fileState.fileId, fileState.helper.pinCount), NEW);
                         fileState.helper.pinnedPages.add(p);
                         ++fileState.helper.pinCount;
                         break;
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheRegressionTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheRegressionTest.java
index 3a857cc..4dc165f 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheRegressionTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheRegressionTest.java
@@ -18,6 +18,7 @@
  */
 package org.apache.hyracks.storage.common;
 
+import static org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider.NEW;
 import static org.junit.Assert.fail;
 
 import java.io.File;
@@ -95,7 +96,7 @@
 
         // Fill the first page with known data and make it dirty by write
         // latching it.
-        ICachedPage writePage = bufferCache.pin(BufferedFileHandle.getDiskPageId(firstFileId, 0), true);
+        ICachedPage writePage = bufferCache.pin(BufferedFileHandle.getDiskPageId(firstFileId, 0), NEW);
         writePage.acquireWriteLatch();
         try {
             ByteBuffer buf = writePage.getBuffer();
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheTest.java
index 6ed4a09..e797967 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheTest.java
@@ -18,6 +18,8 @@
  */
 package org.apache.hyracks.storage.common;
 
+import static org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCachePageOperationContextProvider.NEW;
+
 import java.io.File;
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
@@ -44,6 +46,7 @@
 import org.apache.hyracks.storage.common.buffercache.IBufferCache;
 import org.apache.hyracks.storage.common.buffercache.ICachedPage;
 import org.apache.hyracks.storage.common.buffercache.NoOpPageWriteCallback;
+import org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCacheWriteContext;
 import org.apache.hyracks.storage.common.file.BufferedFileHandle;
 import org.apache.hyracks.test.support.TestStorageManagerComponentHolder;
 import org.apache.hyracks.test.support.TestUtils;
@@ -97,7 +100,8 @@
             long dpid = BufferedFileHandle.getDiskPageId(fileId, i);
             ICachedPage page = bufferCache.confiscatePage(dpid);
             page.getBuffer().putInt(0, i);
-            bufferCache.createFIFOWriter(NoOpPageWriteCallback.INSTANCE, HaltOnFailureCallback.INSTANCE).write(page);
+            bufferCache.createFIFOWriter(NoOpPageWriteCallback.INSTANCE, HaltOnFailureCallback.INSTANCE,
+                    DefaultBufferCacheWriteContext.INSTANCE).write(page);
         }
         bufferCache.closeFile(fileId);
         ExecutorService executor = Executors.newFixedThreadPool(bufferCacheNumPages);
@@ -126,7 +130,7 @@
                         pageNumber = (pageNumber + 1) % numPages;
                         try {
                             long dpid = BufferedFileHandle.getDiskPageId(fileId, pageNumber);
-                            ICachedPage page = bufferCache.pin(dpid, false);
+                            ICachedPage page = bufferCache.pin(dpid);
                             successfulReads++;
                             bufferCache.unpin(page);
                         } catch (HyracksDataException hde) {
@@ -186,7 +190,7 @@
         ICachedPage page = null;
 
         // pin page should succeed
-        page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId), true);
+        page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId), NEW);
         page.acquireWriteLatch();
         try {
             for (int i = 0; i < num; i++) {
@@ -203,7 +207,7 @@
         bufferCache.openFile(fileId);
 
         // tryPin should succeed because page should still be cached
-        page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId), false);
+        page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId));
         Assert.assertNotNull(page);
         page.acquireReadLatch();
         try {
@@ -298,7 +302,7 @@
             fileIds.add(fileId);
 
             ICachedPage page = null;
-            page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId), true);
+            page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId), NEW);
             page.acquireWriteLatch();
             try {
                 ArrayList<Integer> values = new ArrayList<>();
@@ -373,7 +377,7 @@
 
             // pin first page and verify contents
             ICachedPage page = null;
-            page = bufferCache.pin(BufferedFileHandle.getDiskPageId(closedFileId, testPageId), false);
+            page = bufferCache.pin(BufferedFileHandle.getDiskPageId(closedFileId, testPageId));
             page.acquireReadLatch();
             try {
                 ArrayList<Integer> values = pageContents.get(closedFileId);
@@ -411,7 +415,7 @@
             Thread interruptedReader = null;
             try {
                 for (int i = 0; i < expectedPinCount; i++) {
-                    ICachedPage aPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId), false);
+                    ICachedPage aPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId));
                     bufferCache.unpin(aPage);
                     ((CachedPage) aPage).invalidate();
                     actualPinCount.incrementAndGet();
@@ -420,7 +424,7 @@
                         interruptedReader = new Thread(() -> {
                             try {
                                 Thread.currentThread().interrupt();
-                                bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId + 1), false);
+                                bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId + 1));
                             } catch (Exception e) {
                                 e.printStackTrace();
                             }
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheWithCompressionTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheWithCompressionTest.java
index 963b1d8..ad3c3b4 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheWithCompressionTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheWithCompressionTest.java
@@ -46,6 +46,7 @@
 import org.apache.hyracks.storage.common.buffercache.ICachedPage;
 import org.apache.hyracks.storage.common.buffercache.IFIFOPageWriter;
 import org.apache.hyracks.storage.common.buffercache.NoOpPageWriteCallback;
+import org.apache.hyracks.storage.common.buffercache.context.page.DefaultBufferCacheWriteContext;
 import org.apache.hyracks.storage.common.compression.SnappyCompressorDecompressorFactory;
 import org.apache.hyracks.storage.common.compression.file.CompressedFileReference;
 import org.apache.hyracks.storage.common.compression.file.ICompressedPageWriter;
@@ -100,8 +101,8 @@
         final int numPages = 16;
         bufferCache.openFile(fileId);
         final ICompressedPageWriter writer = bufferCache.getCompressedPageWriter(fileId);
-        final IFIFOPageWriter pageWriter =
-                bufferCache.createFIFOWriter(NoOpPageWriteCallback.INSTANCE, HaltOnFailureCallback.INSTANCE);
+        final IFIFOPageWriter pageWriter = bufferCache.createFIFOWriter(NoOpPageWriteCallback.INSTANCE,
+                HaltOnFailureCallback.INSTANCE, DefaultBufferCacheWriteContext.INSTANCE);
         for (int i = 0; i < numPages; i++) {
             long dpid = BufferedFileHandle.getDiskPageId(fileId, i);
             ICachedPage page = bufferCache.confiscatePage(dpid);
@@ -137,7 +138,7 @@
                         pageNumber = (pageNumber + 1) % numPages;
                         try {
                             long dpid = BufferedFileHandle.getDiskPageId(fileId, pageNumber);
-                            ICachedPage page = bufferCache.pin(dpid, false);
+                            ICachedPage page = bufferCache.pin(dpid);
                             successfulReads++;
                             bufferCache.unpin(page);
                         } catch (HyracksDataException hde) {
@@ -206,8 +207,8 @@
         for (int i = 0; i < num; i++) {
             page.getBuffer().putInt(i * 4, i);
         }
-        final IFIFOPageWriter pageWriter =
-                bufferCache.createFIFOWriter(NoOpPageWriteCallback.INSTANCE, HaltOnFailureCallback.INSTANCE);
+        final IFIFOPageWriter pageWriter = bufferCache.createFIFOWriter(NoOpPageWriteCallback.INSTANCE,
+                HaltOnFailureCallback.INSTANCE, DefaultBufferCacheWriteContext.INSTANCE);
         pageWriter.write(page);
         writer.endWriting();
         bufferCache.closeFile(fileId);
@@ -216,7 +217,7 @@
         bufferCache.openFile(fileId);
 
         // tryPin should succeed because page should still be cached
-        page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId), false);
+        page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId));
         Assert.assertNotNull(page);
         try {
             // verify contents of page
@@ -260,8 +261,8 @@
                 values.add(x);
             }
             pageContents.put(fileId, values);
-            final IFIFOPageWriter pageWriter =
-                    bufferCache.createFIFOWriter(NoOpPageWriteCallback.INSTANCE, HaltOnFailureCallback.INSTANCE);
+            final IFIFOPageWriter pageWriter = bufferCache.createFIFOWriter(NoOpPageWriteCallback.INSTANCE,
+                    HaltOnFailureCallback.INSTANCE, DefaultBufferCacheWriteContext.INSTANCE);
             pageWriter.write(page);
             writer.endWriting();
         }
@@ -323,7 +324,7 @@
 
             // pin first page and verify contents
             ICachedPage page = null;
-            page = bufferCache.pin(BufferedFileHandle.getDiskPageId(closedFileId, testPageId), false);
+            page = bufferCache.pin(BufferedFileHandle.getDiskPageId(closedFileId, testPageId));
             try {
                 ArrayList<Integer> values = pageContents.get(closedFileId);
                 for (int j = 0; j < values.size(); j++) {
@@ -359,7 +360,7 @@
             Thread interruptedReader = null;
             try {
                 for (int i = 0; i < expectedPinCount; i++) {
-                    ICachedPage aPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId), false);
+                    ICachedPage aPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId));
                     bufferCache.unpin(aPage);
                     ((CachedPage) aPage).invalidate();
                     actualPinCount.incrementAndGet();
@@ -368,7 +369,7 @@
                         interruptedReader = new Thread(() -> {
                             try {
                                 Thread.currentThread().interrupt();
-                                bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId + 1), false);
+                                bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId + 1));
                             } catch (Exception e) {
                                 e.printStackTrace();
                             }