IGNITE-5849 Introduced meta store
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java
index 64c5927..6802a3f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java
@@ -52,6 +52,11 @@
         throws IgniteCheckedException;
 
     /**
+     * Initializes disk cache store structures.
+     */
+    public void initializeForMetastorage() throws IgniteCheckedException;
+
+    /**
      * Callback called when a cache is stopping. After this callback is invoked, no data associated with
      * the given cache will be stored on disk.
      *
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageInsertFragmentRecord.java b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageInsertFragmentRecord.java
index e07c388..5324d56 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageInsertFragmentRecord.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageInsertFragmentRecord.java
@@ -19,7 +19,8 @@
 
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.internal.pagemem.PageMemory;
-import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.AbstractDataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
 import org.apache.ignite.internal.util.tostring.GridToStringExclude;
 import org.apache.ignite.internal.util.typedef.internal.S;
 
@@ -54,7 +55,7 @@
 
     /** {@inheritDoc} */
     @Override public void applyDelta(PageMemory pageMem, long pageAddr) throws IgniteCheckedException {
-        DataPageIO io = DataPageIO.VERSIONS.forPage(pageAddr);
+        AbstractDataPageIO io = PageIO.getPageIO(pageAddr);
 
         io.addRowFragment(pageAddr, payload, lastLink, pageMem.pageSize());
     }
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageInsertRecord.java b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageInsertRecord.java
index f315058..2c9a8e7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageInsertRecord.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageInsertRecord.java
@@ -19,7 +19,8 @@
 
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.internal.pagemem.PageMemory;
-import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.AbstractDataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
 import org.apache.ignite.internal.util.typedef.internal.S;
 
 /**
@@ -55,7 +56,7 @@
     @Override public void applyDelta(PageMemory pageMem, long pageAddr) throws IgniteCheckedException {
         assert payload != null;
 
-        DataPageIO io = DataPageIO.VERSIONS.forPage(pageAddr);
+        AbstractDataPageIO io = PageIO.getPageIO(pageAddr);
 
         io.addRow(pageAddr, payload, pageMem.pageSize());
     }
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageRemoveRecord.java b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageRemoveRecord.java
index 484ec87..f7776be 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageRemoveRecord.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageRemoveRecord.java
@@ -19,7 +19,8 @@
 
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.internal.pagemem.PageMemory;
-import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.AbstractDataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
 import org.apache.ignite.internal.util.typedef.internal.S;
 
 /**
@@ -50,7 +51,7 @@
     /** {@inheritDoc} */
     @Override public void applyDelta(PageMemory pageMem, long pageAddr)
         throws IgniteCheckedException {
-        DataPageIO io = DataPageIO.VERSIONS.forPage(pageAddr);
+        AbstractDataPageIO io = PageIO.getPageIO(pageAddr);
 
         io.removeRow(pageAddr, itemId, pageMem.pageSize());
     }
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageSetFreeListPageRecord.java b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageSetFreeListPageRecord.java
index 0ade484..e679611 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageSetFreeListPageRecord.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageSetFreeListPageRecord.java
@@ -19,7 +19,8 @@
 
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.internal.pagemem.PageMemory;
-import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.AbstractDataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
 import org.apache.ignite.internal.util.typedef.internal.S;
 
 import static org.apache.ignite.internal.pagemem.wal.record.WALRecord.RecordType.DATA_PAGE_SET_FREE_LIST_PAGE;
@@ -51,7 +52,7 @@
 
     /** {@inheritDoc} */
     @Override public void applyDelta(PageMemory pageMem, long pageAddr) throws IgniteCheckedException {
-        DataPageIO io = DataPageIO.VERSIONS.forPage(pageAddr);
+        AbstractDataPageIO io = PageIO.getPageIO(pageAddr);
 
         io.setFreeListPageId(pageAddr, freeListPage);
     }
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageUpdateRecord.java b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageUpdateRecord.java
index 8ea2981..ed469a4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageUpdateRecord.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/DataPageUpdateRecord.java
@@ -19,7 +19,8 @@
 
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.internal.pagemem.PageMemory;
-import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.AbstractDataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
 import org.apache.ignite.internal.util.typedef.internal.S;
 
 /**
@@ -68,7 +69,7 @@
     @Override public void applyDelta(PageMemory pageMem, long pageAddr) throws IgniteCheckedException {
         assert payload != null;
 
-        DataPageIO io = DataPageIO.VERSIONS.forPage(pageAddr);
+        AbstractDataPageIO io = PageIO.getPageIO(pageAddr);
 
         io.updateRow(pageAddr, itemId, pageMem.pageSize(), payload, null, 0);
     }
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index f3759e0..9dec24b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -91,6 +91,7 @@
 import org.apache.ignite.internal.processors.cache.persistence.MemoryPolicy;
 import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager;
 import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeList;
+import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteCacheSnapshotManager;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotDiscoveryMessage;
 import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
@@ -3762,7 +3763,7 @@
         if (ccfg != null) {
             cloneCheckSerializable(ccfg);
 
-            if (desc != null) {
+            if (desc != null || MetaStorage.METASTORAGE_CACHE_NAME.equals(cacheName)) {
                 if (failIfExists) {
                     throw new CacheExistsException("Failed to start cache " +
                         "(a cache with the same name is already started): " + cacheName);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
index ba6f7d0..b9404ef 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
@@ -41,8 +41,8 @@
 import org.apache.ignite.internal.processors.cache.persistence.CacheSearchRow;
 import org.apache.ignite.internal.processors.cache.persistence.RootPage;
 import org.apache.ignite.internal.processors.cache.persistence.RowStore;
-import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeListImpl;
 import org.apache.ignite.internal.processors.cache.persistence.tree.BPlusTree;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO;
 import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
 import org.apache.ignite.internal.processors.cache.query.GridCacheQueryManager;
 import org.apache.ignite.internal.processors.cache.tree.CacheDataRowStore;
@@ -1179,12 +1179,12 @@
             // Use grp.sharedGroup() flag since it is possible cacheId is not yet set here.
             boolean sizeWithCacheId = grp.sharedGroup();
 
-            int oldLen = FreeListImpl.getRowSize(oldRow, sizeWithCacheId);
+            int oldLen = DataPageIO.getRowSize(oldRow, sizeWithCacheId);
 
             if (oldLen > updateValSizeThreshold)
                 return false;
 
-            int newLen = FreeListImpl.getRowSize(dataRow, sizeWithCacheId);
+            int newLen = DataPageIO.getRowSize(dataRow, sizeWithCacheId);
 
             return oldLen == newLen;
         }
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRow.java
index 57aeaef..4604826 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRow.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRow.java
@@ -24,7 +24,7 @@
 /**
  * Cache data row.
  */
-public interface CacheDataRow extends CacheSearchRow {
+public interface CacheDataRow extends CacheSearchRow, Storable {
     /**
      * @return Cache value.
      */
@@ -43,12 +43,12 @@
     /**
      * @return Partition for this key.
      */
-    public int partition();
+    @Override public int partition();
 
     /**
      * @param link Link for this row.
      */
-    public void link(long link);
+    @Override public void link(long link);
 
     /**
      * @param key Key.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
index 9a2e028..ae27f51 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
@@ -77,6 +77,8 @@
 import org.apache.ignite.internal.NodeStoppingException;
 import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
 import org.apache.ignite.internal.mem.DirectMemoryProvider;
+import org.apache.ignite.internal.mem.file.MappedFileMemoryProvider;
+import org.apache.ignite.internal.mem.unsafe.UnsafeMemoryProvider;
 import org.apache.ignite.internal.pagemem.FullPageId;
 import org.apache.ignite.internal.pagemem.PageIdUtils;
 import org.apache.ignite.internal.pagemem.PageMemory;
@@ -108,6 +110,7 @@
 import org.apache.ignite.internal.processors.cache.persistence.file.FileIO;
 import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStore;
 import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager;
+import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage;
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.CheckpointMetricsTracker;
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryEx;
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryImpl;
@@ -148,6 +151,7 @@
 import static java.nio.file.StandardOpenOption.READ;
 import static org.apache.ignite.IgniteSystemProperties.IGNITE_PDS_SKIP_CRC;
 import static org.apache.ignite.IgniteSystemProperties.IGNITE_PDS_WAL_REBALANCE_THRESHOLD;
+import static org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage.METASTORAGE_CACHE_ID;
 
 /**
  *
@@ -157,6 +161,9 @@
     /** */
     public static final String IGNITE_PDS_CHECKPOINT_TEST_SKIP_SYNC = "IGNITE_PDS_CHECKPOINT_TEST_SKIP_SYNC";
 
+    /** MemoryPolicyConfiguration name reserved for meta store. */
+    private static final String METASTORE_MEMORY_POLICY_NAME = "metastoreMemPlc";
+
     /** Default checkpointing page buffer size (may be adjusted by Ignite). */
     public static final Long DFLT_CHECKPOINTING_PAGE_BUFFER_SIZE = 256L * 1024 * 1024;
 
@@ -312,6 +319,9 @@
     /** Number of pages in current checkpoint. */
     private volatile int currCheckpointPagesCnt;
 
+    /** */
+    private MetaStorage metaStorage;
+
     /**
      * @param ctx Kernal context.
      */
@@ -356,6 +366,31 @@
     }
 
     /** {@inheritDoc} */
+    @Override protected void initPageMemoryPolicies(MemoryConfiguration memCfg) throws IgniteCheckedException {
+        super.initPageMemoryPolicies(memCfg);
+
+        addMemoryPolicy(
+            memCfg,
+            createStoreMemoryPolicy(memCfg),
+            METASTORE_MEMORY_POLICY_NAME
+        );
+    }
+
+    /**
+     * @param memCfg Memory configuration.
+     * @return Memoty polict configuration.
+     */
+    private MemoryPolicyConfiguration createStoreMemoryPolicy(MemoryConfiguration memCfg) {
+        MemoryPolicyConfiguration cfg = new MemoryPolicyConfiguration();
+
+        cfg.setName(METASTORE_MEMORY_POLICY_NAME);
+        cfg.setInitialSize(memCfg.getSystemCacheInitialSize());
+        cfg.setMaxSize(memCfg.getSystemCacheMaxSize());
+
+        return cfg;
+    }
+
+    /** {@inheritDoc} */
     @Override protected void start0() throws IgniteCheckedException {
         super.start0();
 
@@ -393,6 +428,8 @@
                 fileLockHolder = new FileLockHolder(storeMgr.workDir().getPath(), kernalCtx, log);
 
             persStoreMetrics.wal(cctx.wal());
+
+            // Here we can get data from metastorage
         }
     }
 
@@ -484,6 +521,13 @@
         }
 
         super.onActivate(ctx);
+
+        if (!cctx.localNode().isClient()) {
+            cctx.pageStore().initializeForMetastorage();
+
+            metaStorage = new MetaStorage(cctx.wal(), memPlcMap.get(METASTORE_MEMORY_POLICY_NAME),
+                (MemoryMetricsImpl)memMetricsMap.get(METASTORE_MEMORY_POLICY_NAME));
+        }
     }
 
     /** {@inheritDoc} */
@@ -594,6 +638,8 @@
             // This method should return a pointer to the last valid record in the WAL.
             WALPointer restore = restoreMemory(status);
 
+            metaStorage.init(this);
+
             cctx.wal().resumeLogging(restore);
 
             cctx.wal().log(new MemoryRecoveryRecord(U.currentTimeMillis()));
@@ -606,6 +652,50 @@
         }
     }
 
+    /**
+     * @throws IgniteCheckedException If failed.
+     */
+    private void getMetastoreData() throws IgniteCheckedException {
+        try {
+            MemoryConfiguration memCfg = cctx.kernalContext().config().getMemoryConfiguration();
+
+            MemoryPolicyConfiguration plcCfg = createStoreMemoryPolicy(memCfg);
+
+            File allocPath = buildAllocPath(plcCfg);
+
+            DirectMemoryProvider memProvider = allocPath == null ?
+                new UnsafeMemoryProvider(log) :
+                new MappedFileMemoryProvider(
+                    log,
+                    allocPath);
+
+            MemoryMetricsImpl memMetrics = new MemoryMetricsImpl(plcCfg);
+
+            PageMemoryEx storePageMem = (PageMemoryEx)createPageMemory(memProvider, memCfg, plcCfg, memMetrics);
+
+            MemoryPolicy storeMemPlc = new MemoryPolicy(storePageMem, plcCfg, memMetrics, createPageEvictionTracker(plcCfg, storePageMem));
+
+            CheckpointStatus status = readCheckpointStatus();
+
+            cctx.pageStore().initializeForMetastorage();
+
+            restoreMemory(status, true, storePageMem);
+
+            metaStorage = new MetaStorage(cctx.wal(), storeMemPlc, memMetrics, true);
+
+            metaStorage.init(this);
+
+            // here get some data
+
+            metaStorage = null;
+
+            storePageMem.stop();
+        }
+        catch (StorageException e) {
+            throw new IgniteCheckedException(e);
+        }
+    }
+
     /** {@inheritDoc} */
     @Override public void lock() throws IgniteCheckedException {
         if (fileLockHolder != null) {
@@ -1447,6 +1537,17 @@
      * @param status Checkpoint status.
      */
     private WALPointer restoreMemory(CheckpointStatus status) throws IgniteCheckedException {
+        return restoreMemory(status, false, (PageMemoryEx)metaStorage.pageMemory());
+    }
+
+    /**
+     * @param status Checkpoint status.
+     * @param storeOnly If {@code True} restores Metastorage only.
+     */
+    private WALPointer restoreMemory(CheckpointStatus status, boolean storeOnly,
+        PageMemoryEx storePageMem) throws IgniteCheckedException {
+        assert !storeOnly || storePageMem != null;
+
         if (log.isInfoEnabled())
             log.info("Checking memory state [lastValidPos=" + status.endPtr + ", lastMarked="
                 + status.startPtr + ", lastCheckpointId=" + status.cpStartId + ']');
@@ -1496,9 +1597,13 @@
                             // Here we do not require tag check because we may be applying memory changes after
                             // several repetitive restarts and the same pages may have changed several times.
                             int grpId = pageRec.fullPageId().groupId();
+
+                            if (storeOnly && grpId != METASTORAGE_CACHE_ID)
+                                continue;
+
                             long pageId = pageRec.fullPageId().pageId();
 
-                            PageMemoryEx pageMem = getPageMemoryForCacheGroup(grpId);
+                            PageMemoryEx pageMem = grpId == METASTORAGE_CACHE_ID ? storePageMem : getPageMemoryForCacheGroup(grpId);
 
                             long page = pageMem.acquirePage(grpId, pageId, true);
 
@@ -1526,9 +1631,13 @@
                             PartitionDestroyRecord destroyRec = (PartitionDestroyRecord)rec;
 
                             final int gId = destroyRec.groupId();
+
+                            if (storeOnly && gId != METASTORAGE_CACHE_ID)
+                                continue;
+
                             final int pId = destroyRec.partitionId();
 
-                            PageMemoryEx pageMem = getPageMemoryForCacheGroup(gId);
+                            PageMemoryEx pageMem = gId == METASTORAGE_CACHE_ID ? storePageMem : getPageMemoryForCacheGroup(gId);
 
                             pageMem.clearAsync(new P3<Integer, Long, Integer>() {
                                 @Override public boolean apply(Integer cacheId, Long pageId, Integer tag) {
@@ -1544,9 +1653,13 @@
                             PageDeltaRecord r = (PageDeltaRecord)rec;
 
                             int grpId = r.groupId();
+
+                            if (storeOnly && grpId != METASTORAGE_CACHE_ID)
+                                continue;
+
                             long pageId = r.pageId();
 
-                            PageMemoryEx pageMem = getPageMemoryForCacheGroup(grpId);
+                            PageMemoryEx pageMem = grpId == METASTORAGE_CACHE_ID ? storePageMem : getPageMemoryForCacheGroup(grpId);
 
                             // Here we do not require tag check because we may be applying memory changes after
                             // several repetitive restarts and the same pages may have changed several times.
@@ -1572,6 +1685,9 @@
             }
         }
 
+        if (storeOnly)
+            return null;
+
         if (status.needRestoreMemory()) {
             if (apply)
                 throw new IgniteCheckedException("Failed to restore memory state (checkpoint marker is present " +
@@ -2601,12 +2717,19 @@
 
                     int grpId = fullId.groupId();
 
-                    CacheGroupContext grp = context().cache().cacheGroup(grpId);
+                    PageMemoryEx pageMem;
 
-                    if (grp == null)
-                        continue;
+                    if (grpId != MetaStorage.METASTORAGE_CACHE_ID) {
+                        CacheGroupContext grp = context().cache().cacheGroup(grpId);
 
-                    PageMemoryEx pageMem = (PageMemoryEx)grp.memoryPolicy().pageMemory();
+                        if (grp == null)
+                            continue;
+
+                        pageMem = (PageMemoryEx)grp.memoryPolicy().pageMemory();
+                    }
+                    else
+                        pageMem = (PageMemoryEx)metaStorage.pageMemory();
+
 
                     Integer tag = pageMem.getForCheckpoint(
                         fullId, tmpWriteBuf, persStoreMetrics.metricsEnabled() ? tracker : null);
@@ -3323,4 +3446,9 @@
     public PersistenceMetricsImpl persistentStoreMetricsImpl() {
         return persStoreMetrics;
     }
+
+    /** {@inheritDoc} */
+    @Override public MetaStorage metaStorage() {
+        return metaStorage;
+    }
 }
\ No newline at end of file
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
index 5c91a4f..2fe40cc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
@@ -50,7 +50,7 @@
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap;
-import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeListImpl;
+import org.apache.ignite.internal.processors.cache.persistence.freelist.CacheFreeListImpl;
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryEx;
 import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId;
 import org.apache.ignite.internal.processors.cache.persistence.partstate.PagesAllocationRange;
@@ -80,7 +80,7 @@
  */
 public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl implements DbCheckpointListener {
     /** */
-    private MetaStore metaStore;
+    private IndexStorage indexStorage;
 
     /** */
     private ReuseListImpl reuseList;
@@ -100,7 +100,7 @@
 
         RootPage metastoreRoot = metas.treeRoot;
 
-        metaStore = new MetadataStorage(grp.memoryPolicy().pageMemory(),
+        indexStorage = new IndexStorageImpl(grp.memoryPolicy().pageMemory(),
             ctx.wal(),
             globalRemoveId(),
             grp.groupId(),
@@ -121,7 +121,7 @@
             try {
                 final String name = "PendingEntries";
 
-                RootPage pendingRootPage = metaStore.getOrAllocateForTree(name);
+                RootPage pendingRootPage = indexStorage.getOrAllocateForTree(name);
 
                 pendingEntries = new PendingEntriesTree(
                     grp,
@@ -177,7 +177,7 @@
         boolean wasSaveToMeta = false;
 
         if (rowStore0 != null) {
-            FreeListImpl freeList = (FreeListImpl)rowStore0.freeList();
+            CacheFreeListImpl freeList = (CacheFreeListImpl)rowStore0.freeList();
 
             freeList.saveMetadata();
 
@@ -484,7 +484,7 @@
         if (grp.sharedGroup())
             idxName = Integer.toString(cacheId) + "_" + idxName;
 
-        return metaStore.getOrAllocateForTree(idxName);
+        return indexStorage.getOrAllocateForTree(idxName);
     }
 
     /** {@inheritDoc} */
@@ -492,7 +492,7 @@
         if (grp.sharedGroup())
             idxName = Integer.toString(cacheId) + "_" + idxName;
 
-        metaStore.dropRootPage(idxName);
+        indexStorage.dropRootPage(idxName);
     }
 
     /** {@inheritDoc} */
@@ -612,7 +612,7 @@
         for (CacheDataStore store : partDataStores.values()) {
             assert store instanceof GridCacheDataStore;
 
-            FreeListImpl freeList = ((GridCacheDataStore)store).freeList;
+            CacheFreeListImpl freeList = ((GridCacheDataStore)store).freeList;
 
             if (freeList == null)
                 continue;
@@ -862,7 +862,7 @@
         private String name;
 
         /** */
-        private volatile FreeListImpl freeList;
+        private volatile CacheFreeListImpl freeList;
 
         /** */
         private volatile CacheDataStore delegate;
@@ -912,7 +912,7 @@
 
                     RootPage reuseRoot = metas.reuseListRoot;
 
-                    freeList = new FreeListImpl(
+                    freeList = new CacheFreeListImpl(
                         grp.groupId(),
                         grp.cacheOrGroupName() + "-" + partId,
                         grp.memoryPolicy().memoryMetrics(),
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java
index d7682f0..d9f7836 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java
@@ -51,8 +51,9 @@
 import org.apache.ignite.internal.processors.cache.persistence.evict.Random2LruPageEvictionTracker;
 import org.apache.ignite.internal.processors.cache.persistence.evict.RandomLruPageEvictionTracker;
 import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderSettings;
+import org.apache.ignite.internal.processors.cache.persistence.freelist.CacheFreeListImpl;
 import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeList;
-import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeListImpl;
+import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage;
 import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
 import org.apache.ignite.internal.processors.cluster.IgniteChangeGlobalStateSupport;
 import org.apache.ignite.internal.util.typedef.F;
@@ -92,10 +93,10 @@
     protected MemoryPolicy dfltMemPlc;
 
     /** */
-    private Map<String, FreeListImpl> freeListMap;
+    private Map<String, CacheFreeListImpl> freeListMap;
 
     /** */
-    private FreeListImpl dfltFreeList;
+    private CacheFreeListImpl dfltFreeList;
 
     /** Page size from memory configuration, may be set only for fake(standalone) IgniteCacheDataBaseSharedManager */
     private int pageSize;
@@ -170,7 +171,7 @@
 
             MemoryMetricsImpl memMetrics = (MemoryMetricsImpl) memMetricsMap.get(memPlcCfg.getName());
 
-            FreeListImpl freeList = new FreeListImpl(0,
+            CacheFreeListImpl freeList = new CacheFreeListImpl(0,
                     cctx.igniteInstanceName(),
                     memMetrics,
                     memPlc,
@@ -211,9 +212,9 @@
         MemoryPolicyConfiguration[] memPlcsCfgs = memCfg.getMemoryPolicies();
 
         if (memPlcsCfgs == null) {
-            //reserve place for default and system memory policies
-            memPlcMap = U.newHashMap(2);
-            memMetricsMap = U.newHashMap(2);
+            //reserve place for default, system and store memory policies
+            memPlcMap = U.newHashMap(3);
+            memMetricsMap = U.newHashMap(3);
 
             addMemoryPolicy(
                 memCfg,
@@ -228,8 +229,8 @@
 
             if (DFLT_MEM_PLC_DEFAULT_NAME.equals(dfltMemPlcName) && !hasCustomDefaultMemoryPolicy(memPlcsCfgs)) {
                 //reserve additional place for default and system memory policies
-                memPlcMap = U.newHashMap(memPlcsCfgs.length + 2);
-                memMetricsMap = U.newHashMap(memPlcsCfgs.length + 2);
+                memPlcMap = U.newHashMap(memPlcsCfgs.length + 3);
+                memMetricsMap = U.newHashMap(memPlcsCfgs.length + 3);
 
                 addMemoryPolicy(
                     memCfg,
@@ -240,9 +241,9 @@
                 U.warn(log, "No user-defined default MemoryPolicy found; system default of 1GB size will be used.");
             }
             else {
-                //reserve additional space for system memory policy only
-                memPlcMap = U.newHashMap(memPlcsCfgs.length + 1);
-                memMetricsMap = U.newHashMap(memPlcsCfgs.length + 1);
+                //reserve additional space for system and store memory policies
+                memPlcMap = U.newHashMap(memPlcsCfgs.length + 2);
+                memMetricsMap = U.newHashMap(memPlcsCfgs.length + 2);
             }
 
             for (MemoryPolicyConfiguration memPlcCfg : memPlcsCfgs)
@@ -265,7 +266,7 @@
      * @param memPlcName Memory policy name.
      * @throws IgniteCheckedException If failed to initialize swap path.
      */
-    private void addMemoryPolicy(
+    protected void addMemoryPolicy(
         MemoryConfiguration memCfg,
         MemoryPolicyConfiguration memPlcCfg,
         String memPlcName
@@ -298,11 +299,11 @@
      */
     protected IgniteOutClosure<Float> fillFactorProvider(final String memPlcName) {
         return new IgniteOutClosure<Float>() {
-            private FreeListImpl freeList;
+            private CacheFreeListImpl freeList;
 
             @Override public Float apply() {
                 if (freeList == null) {
-                    FreeListImpl freeList0 = freeListMap.get(memPlcName);
+                    CacheFreeListImpl freeList0 = freeListMap.get(memPlcName);
 
                     if (freeList0 == null)
                         return (float) 0;
@@ -583,7 +584,7 @@
      */
     public void dumpStatistics(IgniteLogger log) {
         if (freeListMap != null) {
-            for (FreeListImpl freeList : freeListMap.values())
+            for (CacheFreeListImpl freeList : freeListMap.values())
                 freeList.dumpStatistics(log);
         }
     }
@@ -865,7 +866,7 @@
 
         int sysPageSize = pageMem.systemPageSize();
 
-        FreeListImpl freeListImpl = freeListMap.get(plcCfg.getName());
+        CacheFreeListImpl freeListImpl = freeListMap.get(plcCfg.getName());
 
         for (;;) {
             long allocatedPagesCnt = pageMem.loadedPages();
@@ -912,7 +913,7 @@
      * @param plc Memory Policy Configuration.
      * @param pageMem Page memory.
      */
-    private PageEvictionTracker createPageEvictionTracker(MemoryPolicyConfiguration plc, PageMemory pageMem) {
+    protected PageEvictionTracker createPageEvictionTracker(MemoryPolicyConfiguration plc, PageMemory pageMem) {
         if (plc.getPageEvictionMode() == DataPageEvictionMode.DISABLED || cctx.gridConfig().isPersistentStoreEnabled())
             return new NoOpPageEvictionTracker();
 
@@ -940,7 +941,7 @@
      *
      * @throws IgniteCheckedException If resolving swap directory fails.
      */
-    @Nullable private File buildAllocPath(MemoryPolicyConfiguration plc) throws IgniteCheckedException {
+    @Nullable File buildAllocPath(MemoryPolicyConfiguration plc) throws IgniteCheckedException {
         String path = plc.getSwapFilePath();
 
         if (path == null)
@@ -1038,4 +1039,11 @@
     protected void setPageSize(int pageSize) {
         this.pageSize = pageSize;
     }
+
+    /**
+     * @return MetaStorage
+     */
+    public MetaStorage metaStorage() {
+        return null;
+    }
 }
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/MetaStore.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IndexStorage.java
similarity index 97%
rename from modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/MetaStore.java
rename to modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IndexStorage.java
index c09ce4e..5141b04 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/MetaStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IndexStorage.java
@@ -22,7 +22,7 @@
 /**
  * Meta store.
  */
-public interface MetaStore {
+public interface IndexStorage {
     /**
      * Get or allocate initial page for an index.
      *
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/MetadataStorage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IndexStorageImpl.java
similarity index 98%
rename from modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/MetadataStorage.java
rename to modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IndexStorageImpl.java
index e667807..7daef3c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/MetadataStorage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IndexStorageImpl.java
@@ -37,7 +37,7 @@
 /**
  * Metadata storage.
  */
-public class MetadataStorage implements MetaStore {
+public class IndexStorageImpl implements IndexStorage {
     /** Max index name length (bytes num) */
     public static final int MAX_IDX_NAME_LEN = 255;
 
@@ -69,7 +69,7 @@
      * @param pageMem Page memory.
      * @param wal Write ahead log manager.
      */
-    public MetadataStorage(
+    public IndexStorageImpl(
         final PageMemory pageMem,
         final IgniteWriteAheadLogManager wal,
         final AtomicLong globalRmvId,
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/MemoryMetricsImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/MemoryMetricsImpl.java
index 32618744..1da52ca 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/MemoryMetricsImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/MemoryMetricsImpl.java
@@ -69,7 +69,7 @@
 
     /**
      * @param memPlcCfg MemoryPolicyConfiguration.
-    */
+     */
     public MemoryMetricsImpl(MemoryPolicyConfiguration memPlcCfg) {
         this(memPlcCfg, null);
     }
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/Storable.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/Storable.java
new file mode 100644
index 0000000..ae200df
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/Storable.java
@@ -0,0 +1,38 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence;
+
+/**
+ * Simple interface for data, store in some RowStore.
+ */
+public interface Storable {
+    /**
+     * @param link Link for this row.
+     */
+    public void link(long link);
+
+    /**
+     * @return Link for this row.
+     */
+    public long link();
+
+    /**
+     * @return Partition.
+     */
+    public int partition();
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java
index ed82127..e2df6e3 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java
@@ -48,6 +48,7 @@
 import org.apache.ignite.internal.processors.cache.GridCacheSharedManagerAdapter;
 import org.apache.ignite.internal.processors.cache.StoredCacheData;
 import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderSettings;
+import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteCacheSnapshotManager;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.marshaller.Marshaller;
@@ -195,6 +196,20 @@
     }
 
     /** {@inheritDoc} */
+    @Override public void initializeForMetastorage()
+        throws IgniteCheckedException {
+        int grpId = MetaStorage.METASTORAGE_CACHE_ID;
+
+        if (!idxCacheStores.containsKey(grpId)) {
+            CacheStoreHolder holder = initDir(new File(storeWorkDir, "metastorage"), grpId, 1);
+
+            CacheStoreHolder old = idxCacheStores.put(grpId, holder);
+
+            assert old == null : "Non-null old store holder for metastorage";
+        }
+    }
+
+    /** {@inheritDoc} */
     @Override public void storeCacheData(StoredCacheData cacheData, boolean overwrite) throws IgniteCheckedException {
         File cacheWorkDir = cacheWorkDirectory(cacheData.config());
         File file;
@@ -344,19 +359,30 @@
 
         File cacheWorkDir = cacheWorkDirectory(ccfg);
 
+        return initDir(cacheWorkDir, grpDesc.groupId(), grpDesc.config().getAffinity().partitions());
+    }
+
+    /**
+     * @param cacheWorkDir Work directory.
+     * @param grpId Group ID.
+     * @param partitions Number of partitions.
+     * @return Cache store holder.
+     * @throws IgniteCheckedException If failed.
+     */
+    private CacheStoreHolder initDir(File cacheWorkDir, int grpId, int partitions) throws IgniteCheckedException {
         boolean dirExisted = checkAndInitCacheWorkDir(cacheWorkDir);
 
         File idxFile = new File(cacheWorkDir, INDEX_FILE_NAME);
 
         if (dirExisted && !idxFile.exists())
-            grpsWithoutIdx.add(grpDesc.groupId());
+            grpsWithoutIdx.add(grpId);
 
         FileVersionCheckingFactory pageStoreFactory = new FileVersionCheckingFactory(
             pstCfg.getFileIOFactory(), igniteCfg.getMemoryConfiguration());
 
         FilePageStore idxStore = pageStoreFactory.createPageStore(PageMemory.FLAG_IDX, idxFile);
 
-        FilePageStore[] partStores = new FilePageStore[grpDesc.config().getAffinity().partitions()];
+        FilePageStore[] partStores = new FilePageStore[partitions];
 
         for (int partId = 0; partId < partStores.length; partId++) {
             FilePageStore partStore = pageStoreFactory.createPageStore(
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeListImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java
similarity index 87%
rename from modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeListImpl.java
rename to modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java
index 3eb62ae..88d19e8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeListImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java
@@ -28,15 +28,13 @@
 import org.apache.ignite.internal.pagemem.wal.record.delta.DataPageInsertRecord;
 import org.apache.ignite.internal.pagemem.wal.record.delta.DataPageRemoveRecord;
 import org.apache.ignite.internal.pagemem.wal.record.delta.DataPageUpdateRecord;
-import org.apache.ignite.internal.processors.cache.CacheObject;
-import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
 import org.apache.ignite.internal.processors.cache.persistence.MemoryMetricsImpl;
 import org.apache.ignite.internal.processors.cache.persistence.MemoryPolicy;
+import org.apache.ignite.internal.processors.cache.persistence.Storable;
 import org.apache.ignite.internal.processors.cache.persistence.evict.PageEvictionTracker;
-import org.apache.ignite.internal.processors.cache.persistence.tree.io.CacheVersionIO;
-import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.AbstractDataPageIO;
 import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPagePayload;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.IOVersions;
 import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
 import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseBag;
 import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
@@ -46,7 +44,7 @@
 
 /**
  */
-public class FreeListImpl extends PagesList implements FreeList, ReuseList {
+public abstract class AbstractFreeList<T extends Storable> extends PagesList implements FreeList<T>, ReuseList {
     /** */
     private static final int BUCKETS = 256; // Must be power of 2.
 
@@ -78,7 +76,7 @@
     private final int emptyDataPagesBucket;
 
     /** */
-    private final PageHandler<CacheDataRow, Boolean> updateRow = new UpdateRowHandler();
+    private final PageHandler<T, Boolean> updateRow = new UpdateRowHandler();
 
     /** */
     private final MemoryMetricsImpl memMetrics;
@@ -89,7 +87,7 @@
     /**
      *
      */
-    private final class UpdateRowHandler extends PageHandler<CacheDataRow, Boolean> {
+    private final class UpdateRowHandler extends PageHandler<T, Boolean> {
         @Override public Boolean run(
             int cacheId,
             long pageId,
@@ -97,12 +95,12 @@
             long pageAddr,
             PageIO iox,
             Boolean walPlc,
-            CacheDataRow row,
+            T row,
             int itemId)
             throws IgniteCheckedException {
-            DataPageIO io = (DataPageIO)iox;
+            AbstractDataPageIO<T> io = (AbstractDataPageIO<T>)iox;
 
-            int rowSize = getRowSize(row, row.cacheId() != 0);
+            int rowSize = io.getRowSize(row);
 
             boolean updated = io.updateRow(pageAddr, itemId, pageSize(), null, row, rowSize);
 
@@ -130,12 +128,12 @@
     }
 
     /** */
-    private final PageHandler<CacheDataRow, Integer> writeRow = new WriteRowHandler();
+    private final PageHandler<T, Integer> writeRow = new WriteRowHandler();
 
     /**
      *
      */
-    private final class WriteRowHandler extends PageHandler<CacheDataRow, Integer> {
+    private final class WriteRowHandler extends PageHandler<T, Integer> {
         @Override public Integer run(
             int cacheId,
             long pageId,
@@ -143,18 +141,18 @@
             long pageAddr,
             PageIO iox,
             Boolean walPlc,
-            CacheDataRow row,
+            T row,
             int written)
             throws IgniteCheckedException {
-            DataPageIO io = (DataPageIO)iox;
+            AbstractDataPageIO<T> io = (AbstractDataPageIO<T>)iox;
 
-            int rowSize = getRowSize(row, row.cacheId() != 0);
+            int rowSize = io.getRowSize(row);
             int oldFreeSpace = io.getFreeSpace(pageAddr);
 
             assert oldFreeSpace > 0 : oldFreeSpace;
 
             // If the full row does not fit into this page write only a fragment.
-            written = (written == 0 && oldFreeSpace >= rowSize) ? addRow(pageId, page, pageAddr, io, row, rowSize):
+            written = (written == 0 && oldFreeSpace >= rowSize) ? addRow(pageId, page, pageAddr, io, row, rowSize) :
                 addRowFragment(pageId, page, pageAddr, io, row, written, rowSize);
 
             // Reread free space after update.
@@ -187,8 +185,8 @@
             long pageId,
             long page,
             long pageAddr,
-            DataPageIO io,
-            CacheDataRow row,
+            AbstractDataPageIO<T> io,
+            T row,
             int rowSize
         ) throws IgniteCheckedException {
             io.addRow(pageAddr, row, rowSize, pageSize());
@@ -227,8 +225,8 @@
             long pageId,
             long page,
             long pageAddr,
-            DataPageIO io,
-            CacheDataRow row,
+            AbstractDataPageIO<T> io,
+            T row,
             int written,
             int rowSize
         ) throws IgniteCheckedException {
@@ -254,7 +252,6 @@
         }
     }
 
-
     /** */
     private final PageHandler<Void, Long> rmvRow = new RemoveRowHandler();
 
@@ -272,11 +269,11 @@
             Void ignored,
             int itemId)
             throws IgniteCheckedException {
-            DataPageIO io = (DataPageIO)iox;
+            AbstractDataPageIO<T> io = (AbstractDataPageIO<T>)iox;
 
             int oldFreeSpace = io.getFreeSpace(pageAddr);
 
-            assert oldFreeSpace >= 0: oldFreeSpace;
+            assert oldFreeSpace >= 0 : oldFreeSpace;
 
             long nextLink = io.removeRow(pageAddr, itemId, pageSize());
 
@@ -320,7 +317,7 @@
      * @param initNew {@code True} if new metadata should be initialized.
      * @throws IgniteCheckedException If failed.
      */
-    public FreeListImpl(
+    public AbstractFreeList(
         int cacheId,
         String name,
         MemoryMetricsImpl memMetrics,
@@ -340,7 +337,7 @@
 
         // TODO this constant is used because currently we cannot reuse data pages as index pages
         // TODO and vice-versa. It should be removed when data storage format is finalized.
-        MIN_SIZE_FOR_DATA_PAGE = pageSize - DataPageIO.MIN_DATA_PAGE_OVERHEAD;
+        MIN_SIZE_FOR_DATA_PAGE = pageSize - AbstractDataPageIO.MIN_DATA_PAGE_OVERHEAD;
 
         int shift = 0;
 
@@ -457,8 +454,8 @@
     }
 
     /** {@inheritDoc} */
-    @Override public void insertDataRow(CacheDataRow row) throws IgniteCheckedException {
-        int rowSize = getRowSize(row, row.cacheId() != 0);
+    @Override public void insertDataRow(T row) throws IgniteCheckedException {
+        int rowSize = ioVersions().latest().getRowSize(row);
 
         int written = 0;
 
@@ -471,14 +468,14 @@
             long pageId = 0L;
 
             if (freeSpace == MIN_SIZE_FOR_DATA_PAGE)
-                pageId = takeEmptyPage(emptyDataPagesBucket, DataPageIO.VERSIONS);
+                pageId = takeEmptyPage(emptyDataPagesBucket, ioVersions());
 
             boolean reuseBucket = false;
 
             // TODO: properly handle reuse bucket.
             if (pageId == 0L) {
                 for (int b = bucket(freeSpace, false) + 1; b < BUCKETS - 1; b++) {
-                    pageId = takeEmptyPage(b, DataPageIO.VERSIONS);
+                    pageId = takeEmptyPage(b, ioVersions());
 
                     if (pageId != 0L) {
                         reuseBucket = isReuseBucket(b);
@@ -493,7 +490,7 @@
             if (allocated)
                 pageId = allocateDataPage(row.partition());
 
-            DataPageIO init = reuseBucket || allocated ? DataPageIO.VERSIONS.latest() : null;
+            AbstractDataPageIO<T> init = reuseBucket || allocated ? ioVersions().latest() : null;
 
             written = write(pageId, writeRow, init, row, written, FAIL_I);
 
@@ -503,7 +500,7 @@
     }
 
     /** {@inheritDoc} */
-    @Override public boolean updateDataRow(long link, CacheDataRow row) throws IgniteCheckedException {
+    @Override public boolean updateDataRow(long link, T row) throws IgniteCheckedException {
         assert link != 0;
 
         long pageId = PageIdUtils.pageId(link);
@@ -563,40 +560,29 @@
 
     /** {@inheritDoc} */
     @Override public void addForRecycle(ReuseBag bag) throws IgniteCheckedException {
-        assert reuseList == this: "not allowed to be a reuse list";
+        assert reuseList == this : "not allowed to be a reuse list";
 
         put(bag, 0, 0, 0L, REUSE_BUCKET);
     }
 
     /** {@inheritDoc} */
     @Override public long takeRecycledPage() throws IgniteCheckedException {
-        assert reuseList == this: "not allowed to be a reuse list";
+        assert reuseList == this : "not allowed to be a reuse list";
 
         return takeEmptyPage(REUSE_BUCKET, null);
     }
 
     /** {@inheritDoc} */
     @Override public long recycledPagesCount() throws IgniteCheckedException {
-        assert reuseList == this: "not allowed to be a reuse list";
+        assert reuseList == this : "not allowed to be a reuse list";
 
         return storedPagesCount(REUSE_BUCKET);
     }
 
     /**
-     * @param row Row.
-     * @param withCacheId If {@code true} adds cache ID size.
-     * @return Entry size on page.
-     * @throws IgniteCheckedException If failed.
+     * @return IOVersions.
      */
-    public static int getRowSize(CacheDataRow row, boolean withCacheId) throws IgniteCheckedException {
-        KeyCacheObject key = row.key();
-        CacheObject val = row.value();
-
-        int keyLen = key.valueBytesLength(null);
-        int valLen = val.valueBytesLength(null);
-
-        return keyLen + valLen + CacheVersionIO.size(row.version(), false) + 8 + (withCacheId ? 4 : 0);
-    }
+    public abstract IOVersions<? extends AbstractDataPageIO<T>> ioVersions();
 
     /** {@inheritDoc} */
     @Override public String toString() {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/CacheFreeListImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/CacheFreeListImpl.java
new file mode 100644
index 0000000..d0a48d5
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/CacheFreeListImpl.java
@@ -0,0 +1,50 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.freelist;
+
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
+import org.apache.ignite.internal.processors.cache.persistence.MemoryMetricsImpl;
+import org.apache.ignite.internal.processors.cache.persistence.MemoryPolicy;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.AbstractDataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.IOVersions;
+import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
+
+/**
+ * FreeList implementation for cache.
+ */
+public class CacheFreeListImpl extends AbstractFreeList<CacheDataRow> {
+    /** {@inheritDoc} */
+    public CacheFreeListImpl(int cacheId, String name, MemoryMetricsImpl memMetrics, MemoryPolicy memPlc,
+        ReuseList reuseList,
+        IgniteWriteAheadLogManager wal, long metaPageId, boolean initNew) throws IgniteCheckedException {
+        super(cacheId, name, memMetrics, memPlc, reuseList, wal, metaPageId, initNew);
+    }
+
+    /** {@inheritDoc} */
+    @Override public IOVersions<? extends AbstractDataPageIO<CacheDataRow>> ioVersions() {
+        return DataPageIO.VERSIONS;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return "FreeList [name=" + name + ']';
+    }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeList.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeList.java
index d2f0099..bdca21c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeList.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeList.java
@@ -19,16 +19,16 @@
 
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteLogger;
-import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
+import org.apache.ignite.internal.processors.cache.persistence.Storable;
 
 /**
  */
-public interface FreeList {
+public interface FreeList<T extends Storable> {
     /**
      * @param row Row.
      * @throws IgniteCheckedException If failed.
      */
-    public void insertDataRow(CacheDataRow row) throws IgniteCheckedException;
+    public void insertDataRow(T row) throws IgniteCheckedException;
 
     /**
      * @param link Row link.
@@ -36,7 +36,7 @@
      * @return {@code True} if was able to update row.
      * @throws IgniteCheckedException If failed.
      */
-    public boolean updateDataRow(long link, CacheDataRow row) throws IgniteCheckedException;
+    public boolean updateDataRow(long link, T row) throws IgniteCheckedException;
 
     /**
      * @param link Row link.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/PagesList.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/PagesList.java
index 8a540a0..d79aa81 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/PagesList.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/PagesList.java
@@ -38,7 +38,7 @@
 import org.apache.ignite.internal.processors.cache.persistence.DataStructure;
 import org.apache.ignite.internal.processors.cache.persistence.freelist.io.PagesListMetaIO;
 import org.apache.ignite.internal.processors.cache.persistence.freelist.io.PagesListNodeIO;
-import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.AbstractDataPageIO;
 import org.apache.ignite.internal.processors.cache.persistence.tree.io.IOVersions;
 import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
 import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseBag;
@@ -698,7 +698,7 @@
             if (needWalDeltaRecord(pageId, page, null))
                 wal.log(new PagesListAddPageRecord(grpId, pageId, dataId));
 
-            DataPageIO dataIO = DataPageIO.VERSIONS.forPage(dataAddr);
+            AbstractDataPageIO dataIO = PageIO.getPageIO(dataAddr);
             dataIO.setFreeListPageId(dataAddr, pageId);
 
             if (needWalDeltaRecord(dataId, dataPage, null))
@@ -729,7 +729,7 @@
         final long dataAddr,
         int bucket
     ) throws IgniteCheckedException {
-        DataPageIO dataIO = DataPageIO.VERSIONS.forPage(dataAddr);
+        AbstractDataPageIO dataIO = PageIO.getPageIO(dataAddr);
 
         // Attempt to add page failed: the node page is full.
         if (isReuseBucket(bucket)) {
@@ -1152,7 +1152,7 @@
         final long dataId,
         final long dataPage,
         final long dataAddr,
-        DataPageIO dataIO,
+        AbstractDataPageIO dataIO,
         int bucket)
         throws IgniteCheckedException {
         final long pageId = dataIO.getFreeListPageId(dataAddr);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetaStorage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetaStorage.java
new file mode 100644
index 0000000..cb588f5
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetaStorage.java
@@ -0,0 +1,338 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.metastorage;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.pagemem.FullPageId;
+import org.apache.ignite.internal.pagemem.PageIdUtils;
+import org.apache.ignite.internal.pagemem.PageMemory;
+import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
+import org.apache.ignite.internal.pagemem.wal.record.delta.MetaPageInitRecord;
+import org.apache.ignite.internal.processors.cache.IncompleteObject;
+import org.apache.ignite.internal.processors.cache.persistence.DbCheckpointListener;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.IgniteCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.MemoryMetricsImpl;
+import org.apache.ignite.internal.processors.cache.persistence.MemoryPolicy;
+import org.apache.ignite.internal.processors.cache.persistence.RootPage;
+import org.apache.ignite.internal.processors.cache.persistence.freelist.AbstractFreeList;
+import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryEx;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.AbstractDataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPagePayload;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.IOVersions;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PagePartitionMetaIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.SimpleDataPageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
+import org.apache.ignite.internal.processors.cache.persistence.tree.util.PageHandler;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.internal.util.typedef.internal.U;
+
+import static org.apache.ignite.internal.pagemem.PageIdUtils.itemId;
+import static org.apache.ignite.internal.pagemem.PageIdUtils.pageId;
+
+/**
+ * General purpose key-value local-only storage.
+ */
+public class MetaStorage implements DbCheckpointListener {
+    /** */
+    public static final String METASTORAGE_CACHE_NAME = "MetaStorage";
+    /** */
+    public static final int METASTORAGE_CACHE_ID = CU.cacheId(METASTORAGE_CACHE_NAME);
+
+    /** */
+    private final IgniteWriteAheadLogManager wal;
+    /** */
+    private final MemoryPolicy memPlc;
+    /** */
+    private MetastorageTree tree;
+    /** */
+    private AtomicLong rmvId = new AtomicLong();
+    /** */
+    private MemoryMetricsImpl memMetrics;
+    /** */
+    private boolean readOnly;
+    /** */
+    private RootPage treeRoot;
+    /** */
+    private RootPage reuseListRoot;
+    /** */
+    private FreeListImpl freeList;
+
+    /** */
+    public MetaStorage(IgniteWriteAheadLogManager wal, MemoryPolicy memPlc, MemoryMetricsImpl memMetrics,
+        boolean readOnly) {
+        this.wal = wal;
+        this.memPlc = memPlc;
+        this.memMetrics = memMetrics;
+        this.readOnly = readOnly;
+    }
+
+    /** */
+    public MetaStorage(IgniteWriteAheadLogManager wal, MemoryPolicy memPlc, MemoryMetricsImpl memMetrics) {
+        this(wal, memPlc, memMetrics, false);
+    }
+
+    /** */
+    public void init(IgniteCacheDatabaseSharedManager db) throws IgniteCheckedException {
+        getOrAllocateMetas();
+
+        freeList = new FreeListImpl(METASTORAGE_CACHE_ID, "metastorage",
+            memMetrics, memPlc, null, wal, reuseListRoot.pageId().pageId(),
+            reuseListRoot.isAllocated());
+
+        MetastorageRowStore rowStore = new MetastorageRowStore(freeList, db);
+
+        tree = new MetastorageTree(METASTORAGE_CACHE_ID, memPlc.pageMemory(), wal, rmvId,
+            freeList, rowStore, treeRoot.pageId().pageId(), treeRoot.isAllocated());
+
+        ((GridCacheDatabaseSharedManager)db).addCheckpointListener(this);
+    }
+
+    /** */
+    public void putData(String key, byte[] data) throws IgniteCheckedException {
+        if (!readOnly) {
+            synchronized (this) {
+                MetastorageDataRow oldRow = tree.findOne(new MetastorageDataRow(key, null));
+
+                if (oldRow != null) {
+                    tree.removex(oldRow);
+                    tree.rowStore().removeRow(oldRow.link());
+                }
+
+                MetastorageDataRow row = new MetastorageDataRow(key, data);
+                tree.rowStore().addRow(row);
+                tree.put(row);
+            }
+        }
+    }
+
+    /** */
+    public MetastorageDataRow getData(String key) throws IgniteCheckedException {
+        MetastorageDataRow row = tree.findOne(new MetastorageDataRow(key, null));
+
+        return row;
+    }
+
+    /** */
+    public void removeData(String key) throws IgniteCheckedException {
+        if (!readOnly)
+            synchronized (this) {
+                MetastorageDataRow row = new MetastorageDataRow(key, null);
+                MetastorageDataRow oldRow = tree.findOne(row);
+
+                if (oldRow != null) {
+                    tree.removex(oldRow);
+                    tree.rowStore().removeRow(oldRow.link());
+                }
+            }
+    }
+
+    /** */
+    private void getOrAllocateMetas() throws IgniteCheckedException {
+        PageMemoryEx pageMem = (PageMemoryEx)memPlc.pageMemory();
+
+        int partId = 0;
+
+        long partMetaId = pageMem.partitionMetaPageId(METASTORAGE_CACHE_ID, partId);
+        long partMetaPage = pageMem.acquirePage(METASTORAGE_CACHE_ID, partMetaId);
+        try {
+            boolean allocated = false;
+            long pageAddr = pageMem.writeLock(METASTORAGE_CACHE_ID, partMetaId, partMetaPage);
+
+            try {
+                long treeRoot, reuseListRoot;
+
+                if (PageIO.getType(pageAddr) != PageIO.T_PART_META) {
+                    // Initialize new page.
+                    if (readOnly)
+                        throw new IgniteCheckedException("metastorage is not initialized");
+
+                    PagePartitionMetaIO io = PagePartitionMetaIO.VERSIONS.latest();
+
+                    io.initNewPage(pageAddr, partMetaId, pageMem.pageSize());
+
+                    treeRoot = pageMem.allocatePage(METASTORAGE_CACHE_ID, partId, PageMemory.FLAG_DATA);
+                    reuseListRoot = pageMem.allocatePage(METASTORAGE_CACHE_ID, partId, PageMemory.FLAG_DATA);
+
+                    assert PageIdUtils.flag(treeRoot) == PageMemory.FLAG_DATA;
+                    assert PageIdUtils.flag(reuseListRoot) == PageMemory.FLAG_DATA;
+
+                    io.setTreeRoot(pageAddr, treeRoot);
+                    io.setReuseListRoot(pageAddr, reuseListRoot);
+
+                    if (PageHandler.isWalDeltaRecordNeeded(pageMem, METASTORAGE_CACHE_ID, partMetaId, partMetaPage, wal, null))
+                        wal.log(new MetaPageInitRecord(
+                            METASTORAGE_CACHE_ID,
+                            partMetaId,
+                            io.getType(),
+                            io.getVersion(),
+                            treeRoot,
+                            reuseListRoot
+                        ));
+
+                    allocated = true;
+                }
+                else {
+                    PagePartitionMetaIO io = PageIO.getPageIO(pageAddr);
+
+                    treeRoot = io.getTreeRoot(pageAddr);
+                    reuseListRoot = io.getReuseListRoot(pageAddr);
+
+                    assert PageIdUtils.flag(treeRoot) == PageMemory.FLAG_DATA :
+                        U.hexLong(treeRoot) + ", part=" + partId + ", METASTORAGE_CACHE_ID=" + METASTORAGE_CACHE_ID;
+                    assert PageIdUtils.flag(reuseListRoot) == PageMemory.FLAG_DATA :
+                        U.hexLong(reuseListRoot) + ", part=" + partId + ", METASTORAGE_CACHE_ID=" + METASTORAGE_CACHE_ID;
+                }
+
+                this.treeRoot = new RootPage(new FullPageId(treeRoot, METASTORAGE_CACHE_ID), allocated);
+                this.reuseListRoot = new RootPage(new FullPageId(reuseListRoot, METASTORAGE_CACHE_ID), allocated);
+            }
+            finally {
+                pageMem.writeUnlock(METASTORAGE_CACHE_ID, partMetaId, partMetaPage, null, allocated);
+            }
+        }
+        finally {
+            pageMem.releasePage(METASTORAGE_CACHE_ID, partMetaId, partMetaPage);
+        }
+    }
+
+    /**
+     * @return Page memory.
+     */
+    public PageMemory pageMemory() {
+        return memPlc.pageMemory();
+    }
+
+    /** {@inheritDoc} */
+    @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
+        freeList.saveMetadata();
+
+        MetastorageRowStore rowStore = tree.rowStore();
+
+        saveStoreMetadata(rowStore, ctx);
+    }
+
+    /**
+     * @param rowStore Store to save metadata.
+     * @throws IgniteCheckedException If failed.
+     */
+    private void saveStoreMetadata(MetastorageRowStore rowStore, Context ctx) throws IgniteCheckedException {
+        FreeListImpl freeList = (FreeListImpl)rowStore.freeList();
+
+        freeList.saveMetadata();
+    }
+
+    /** */
+    public static class FreeListImpl extends AbstractFreeList<MetastorageDataRow> {
+        /** {@inheritDoc} */
+        FreeListImpl(int cacheId, String name, MemoryMetricsImpl memMetrics, MemoryPolicy memPlc,
+            ReuseList reuseList,
+            IgniteWriteAheadLogManager wal, long metaPageId, boolean initNew) throws IgniteCheckedException {
+            super(cacheId, name, memMetrics, memPlc, reuseList, wal, metaPageId, initNew);
+        }
+
+        /** {@inheritDoc} */
+        @Override public IOVersions<? extends AbstractDataPageIO<MetastorageDataRow>> ioVersions() {
+            return SimpleDataPageIO.VERSIONS;
+        }
+
+        /**
+         * Read row from data pages.
+         */
+        final MetastorageDataRow readRow(String key, long link)
+            throws IgniteCheckedException {
+            assert link != 0 : "link";
+
+            long nextLink = link;
+            IncompleteObject incomplete = null;
+            int size = 0;
+
+            boolean first = true;
+
+            do {
+                final long pageId = pageId(nextLink);
+
+                final long page = pageMem.acquirePage(grpId, pageId);
+
+                try {
+                    long pageAddr = pageMem.readLock(grpId, pageId, page); // Non-empty data page must not be recycled.
+
+                    assert pageAddr != 0L : nextLink;
+
+                    try {
+                        SimpleDataPageIO io = (SimpleDataPageIO)ioVersions().forPage(pageAddr);
+
+                        DataPagePayload data = io.readPayload(pageAddr, itemId(nextLink), pageMem.pageSize());
+
+                        nextLink = data.nextLink();
+
+                        if (first) {
+                            if (nextLink == 0) {
+                                // Fast path for a single page row.
+                                return new MetastorageDataRow(link, key, SimpleDataPageIO.readPayload(pageAddr + data.offset()));
+                            }
+
+                            first = false;
+                        }
+
+                        ByteBuffer buf = pageMem.pageBuffer(pageAddr);
+
+                        buf.position(data.offset());
+                        buf.limit(data.offset() + data.payloadSize());
+
+                        if (size == 0) {
+                            if (buf.remaining() >= 2 && incomplete == null) {
+                                // Just read size.
+                                size = buf.getShort();
+                                incomplete = new IncompleteObject(new byte[size]);
+                            }
+                            else {
+                                if (incomplete == null)
+                                    incomplete = new IncompleteObject(new byte[2]);
+
+                                incomplete.readData(buf);
+
+                                if (incomplete.isReady()) {
+                                    size = ByteBuffer.wrap(incomplete.data()).order(buf.order()).getShort();
+                                    incomplete = new IncompleteObject(new byte[size]);
+                                }
+                            }
+                        }
+
+                        if (size != 0 && buf.remaining() > 0)
+                            incomplete.readData(buf);
+                    }
+                    finally {
+                        pageMem.readUnlock(grpId, pageId, page);
+                    }
+                }
+                finally {
+                    pageMem.releasePage(grpId, pageId, page);
+                }
+            }
+            while (nextLink != 0);
+
+            assert incomplete.isReady();
+
+            return new MetastorageDataRow(link, key, incomplete.data());
+        }
+    }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetastorageDataRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetastorageDataRow.java
new file mode 100644
index 0000000..dde30d7
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetastorageDataRow.java
@@ -0,0 +1,92 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.metastorage;
+
+import org.apache.ignite.internal.processors.cache.persistence.Storable;
+
+/**
+ *
+ */
+public class MetastorageDataRow implements MetastorageSearchRow, Storable {
+
+    /* **/
+    private long link;
+
+    /* **/
+    private String key;
+
+    /* **/
+    private byte[] value;
+
+    /* **/
+    public MetastorageDataRow(long link, String key, byte[] value) {
+        this.link = link;
+        this.key = key;
+        this.value = value;
+    }
+
+    /* **/
+    public MetastorageDataRow(String key, byte[] value) {
+        this.key = key;
+        this.value = value;
+    }
+
+    /**
+     * @return Key.
+     */
+    public String key() {
+        return key;
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public int hash() {
+        return key.hashCode();
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public int partition() {
+        return 0;
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public void link(long link) {
+        this.link = link;
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public long link() {
+        return link;
+    }
+
+    /**
+     * @return Value.
+     */
+    public byte[] value() {
+        return value;
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public String toString() {
+        return "key=" + key;
+    }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetastorageRowStore.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetastorageRowStore.java
new file mode 100644
index 0000000..0806b30
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetastorageRowStore.java
@@ -0,0 +1,97 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.metastorage;
+
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.processors.cache.persistence.IgniteCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeList;
+
+/**
+ *
+ */
+public class MetastorageRowStore {
+
+    /** */
+    private final FreeList freeList;
+
+    /** */
+    protected final IgniteCacheDatabaseSharedManager db;
+
+    /** */
+    public MetastorageRowStore(FreeList freeList, IgniteCacheDatabaseSharedManager db) {
+        this.freeList = freeList;
+        this.db = db;
+    }
+
+    /**
+     * @param link Row link.
+     * @return Data row.
+     */
+    public MetastorageDataRow dataRow(String key, long link) throws IgniteCheckedException {
+        return ((MetaStorage.FreeListImpl)freeList).readRow(key, link);
+    }
+
+    /**
+     * @param link Row link.
+     * @throws IgniteCheckedException If failed.
+     */
+    public void removeRow(long link) throws IgniteCheckedException {
+        assert link != 0;
+        db.checkpointReadLock();
+
+        try {
+            freeList.removeDataRowByLink(link);
+        }
+        finally {
+            db.checkpointReadUnlock();
+        }
+    }
+
+    /**
+     * @param row Row.
+     * @throws IgniteCheckedException If failed.
+     */
+    public void addRow(MetastorageDataRow row) throws IgniteCheckedException {
+        db.checkpointReadLock();
+
+        try {
+            freeList.insertDataRow(row);
+        }
+        finally {
+            db.checkpointReadUnlock();
+        }
+    }
+
+    /**
+     * @param link Row link.
+     * @param row New row data.
+     * @return {@code True} if was able to update row.
+     * @throws IgniteCheckedException If failed.
+     */
+    public boolean updateRow(long link, MetastorageDataRow row) throws IgniteCheckedException {
+        return freeList.updateDataRow(link, row);
+    }
+
+    /**
+     * @return Free list.
+     */
+    public FreeList freeList() {
+        return freeList;
+    }
+
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetastorageSearchRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetastorageSearchRow.java
new file mode 100644
index 0000000..601fbc1
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetastorageSearchRow.java
@@ -0,0 +1,38 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.metastorage;
+
+/**
+ *
+ */
+public interface MetastorageSearchRow {
+    /**
+     * @return Key.
+     */
+    public String key();
+
+    /**
+     * @return Link for this row.
+     */
+    public long link();
+
+    /**
+     * @return Key hash code.
+     */
+    public int hash();
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetastorageTree.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetastorageTree.java
new file mode 100644
index 0000000..445522b
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetastorageTree.java
@@ -0,0 +1,266 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.metastorage;
+
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.pagemem.PageMemory;
+import org.apache.ignite.internal.pagemem.PageUtils;
+import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
+import org.apache.ignite.internal.processors.cache.persistence.tree.BPlusTree;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.BPlusIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.BPlusInnerIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.BPlusLeafIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.IOVersions;
+import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
+
+/**
+ *
+ */
+public class MetastorageTree extends BPlusTree<MetastorageSearchRow, MetastorageDataRow> {
+    /** Max key length (bytes num) */
+    public static final int MAX_KEY_LEN = 64;
+
+    /** */
+    private MetastorageRowStore rowStore;
+
+    /**
+     * @param pageMem
+     * @param wal
+     * @param globalRmvId
+     * @param metaPageId
+     * @param reuseList
+     * @throws IgniteCheckedException
+     */
+    public MetastorageTree(int cacheId,
+        PageMemory pageMem,
+        IgniteWriteAheadLogManager wal,
+        AtomicLong globalRmvId,
+        ReuseList reuseList,
+        MetastorageRowStore rowStore,
+        long metaPageId,
+        boolean initNew) throws IgniteCheckedException {
+        super("Metastorage", cacheId, pageMem, wal,
+            globalRmvId, metaPageId, reuseList, MetastorageInnerIO.VERSIONS, MetastoreLeafIO.VERSIONS);
+
+        this.rowStore = rowStore;
+
+        initTree(initNew);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected int compare(BPlusIO<MetastorageSearchRow> io, long pageAddr, int idx,
+        MetastorageSearchRow row) throws IgniteCheckedException {
+
+        String key = ((DataLinkIO)io).getKey(pageAddr, idx);
+
+        return key.compareTo(row.key());
+    }
+
+    /** {@inheritDoc} */
+    @Override protected MetastorageDataRow getRow(BPlusIO<MetastorageSearchRow> io, long pageAddr, int idx,
+        Object x) throws IgniteCheckedException {
+        long link = ((DataLinkIO)io).getLink(pageAddr, idx);
+        String key = ((DataLinkIO)io).getKey(pageAddr, idx);
+
+        return rowStore.dataRow(key, link);
+    }
+
+    /**
+     * @return RowStore.
+     */
+    public MetastorageRowStore rowStore() {
+        return rowStore;
+    }
+
+    /**
+     *
+     */
+    private interface DataLinkIO {
+        /**
+         * @param pageAddr Page address.
+         * @param idx Index.
+         * @return Row link.
+         */
+        public long getLink(long pageAddr, int idx);
+
+        /**
+         * @param pageAddr Page address.
+         * @param idx Index.
+         * @return Key size in bytes.
+         */
+        public short getKeySize(long pageAddr, int idx);
+
+        /**
+         * @param pageAddr Page address.
+         * @param idx Index.
+         * @return Key.
+         */
+        public String getKey(long pageAddr, int idx);
+    }
+
+    /**
+     *
+     */
+    public static class MetastorageInnerIO extends BPlusInnerIO<MetastorageSearchRow> implements DataLinkIO {
+        /** */
+        public static final IOVersions<MetastorageInnerIO> VERSIONS = new IOVersions<>(
+            new MetastorageInnerIO(1)
+        );
+
+        /**
+         * @param ver Page format version.
+         */
+        MetastorageInnerIO(int ver) {
+            super(T_DATA_REF_METASTORAGE_INNER, ver, true, 10 + MAX_KEY_LEN);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void storeByOffset(long pageAddr, int off,
+            MetastorageSearchRow row) throws IgniteCheckedException {
+            assert row.link() != 0;
+
+            PageUtils.putLong(pageAddr, off, row.link());
+
+            byte[] bytes = row.key().getBytes();
+            assert bytes.length <= MAX_KEY_LEN;
+
+            PageUtils.putShort(pageAddr, off + 8, (short)bytes.length);
+            PageUtils.putBytes(pageAddr, off + 10, bytes);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void store(long dstPageAddr, int dstIdx, BPlusIO<MetastorageSearchRow> srcIo, long srcPageAddr,
+            int srcIdx) throws IgniteCheckedException {
+            int srcOff = srcIo.offset(srcIdx);
+            int dstOff = offset(dstIdx);
+
+            long link = ((DataLinkIO)srcIo).getLink(srcPageAddr, srcIdx);
+            short len = ((DataLinkIO)srcIo).getKeySize(srcPageAddr, srcIdx);
+
+            byte[] payload = PageUtils.getBytes(srcPageAddr, srcOff + 10, len);
+
+            PageUtils.putLong(dstPageAddr, dstOff, link);
+            PageUtils.putShort(dstPageAddr, dstOff + 8, len);
+            PageUtils.putBytes(dstPageAddr, dstOff + 10, payload);
+        }
+
+        /** {@inheritDoc} */
+        @Override public MetastorageSearchRow getLookupRow(BPlusTree<MetastorageSearchRow, ?> tree, long pageAddr,
+            int idx) throws IgniteCheckedException {
+            long link = getLink(pageAddr, idx);
+            String key = getKey(pageAddr, idx);
+
+            return new MetsatorageSearchRowImpl(key, link);
+        }
+
+        /** {@inheritDoc} */
+        @Override public long getLink(long pageAddr, int idx) {
+            assert idx < getCount(pageAddr) : idx;
+
+            return PageUtils.getLong(pageAddr, offset(idx));
+        }
+
+        /** {@inheritDoc} */
+        @Override public short getKeySize(long pageAddr, int idx) {
+            return PageUtils.getShort(pageAddr, offset(idx) + 8);
+        }
+
+        /** {@inheritDoc} */
+        @Override public String getKey(long pageAddr, int idx) {
+            int len = PageUtils.getShort(pageAddr, offset(idx) + 8);
+            byte[] bytes = PageUtils.getBytes(pageAddr, offset(idx) + 10, len);
+            return new String(bytes);
+        }
+    }
+
+    /**
+     *
+     */
+    public static class MetastoreLeafIO extends BPlusLeafIO<MetastorageSearchRow> implements DataLinkIO {
+        /** */
+        public static final IOVersions<MetastoreLeafIO> VERSIONS = new IOVersions<>(
+            new MetastoreLeafIO(1)
+        );
+
+        /**
+         * @param ver Page format version.
+         */
+        MetastoreLeafIO(int ver) {
+            super(T_DATA_REF_METASTORAGE_LEAF, ver, 10 + MAX_KEY_LEN);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void storeByOffset(long pageAddr, int off,
+            MetastorageSearchRow row) throws IgniteCheckedException {
+            assert row.link() != 0;
+
+            PageUtils.putLong(pageAddr, off, row.link());
+            byte[] bytes = row.key().getBytes();
+
+            assert bytes.length <= MAX_KEY_LEN;
+
+            PageUtils.putShort(pageAddr, off + 8, (short)bytes.length);
+            PageUtils.putBytes(pageAddr, off + 10, bytes);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void store(long dstPageAddr, int dstIdx, BPlusIO<MetastorageSearchRow> srcIo, long srcPageAddr,
+            int srcIdx) throws IgniteCheckedException {
+            int srcOff = srcIo.offset(srcIdx);
+            int dstOff = offset(dstIdx);
+
+            long link = ((DataLinkIO)srcIo).getLink(srcPageAddr, srcIdx);
+            short len = ((DataLinkIO)srcIo).getKeySize(srcPageAddr, srcIdx);
+
+            byte[] payload = PageUtils.getBytes(srcPageAddr, srcOff + 10, len);
+
+            PageUtils.putLong(dstPageAddr, dstOff, link);
+            PageUtils.putShort(dstPageAddr, dstOff + 8, len);
+            PageUtils.putBytes(dstPageAddr, dstOff + 10, payload);
+        }
+
+        /** {@inheritDoc} */
+        @Override public MetastorageSearchRow getLookupRow(BPlusTree<MetastorageSearchRow, ?> tree, long pageAddr,
+            int idx) throws IgniteCheckedException {
+            long link = getLink(pageAddr, idx);
+            String key = getKey(pageAddr, idx);
+
+            return new MetsatorageSearchRowImpl(key, link);
+        }
+
+        /** {@inheritDoc} */
+        @Override public long getLink(long pageAddr, int idx) {
+            assert idx < getCount(pageAddr) : idx;
+
+            return PageUtils.getLong(pageAddr, offset(idx));
+        }
+
+        /** {@inheritDoc} */
+        @Override public short getKeySize(long pageAddr, int idx) {
+            return PageUtils.getShort(pageAddr, offset(idx) + 8);
+        }
+
+        /** {@inheritDoc} */
+        @Override public String getKey(long pageAddr, int idx) {
+            int len = PageUtils.getShort(pageAddr, offset(idx) + 8);
+            byte[] bytes = PageUtils.getBytes(pageAddr, offset(idx) + 10, len);
+            return new String(bytes);
+        }
+    }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetsatorageSearchRowImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetsatorageSearchRowImpl.java
new file mode 100644
index 0000000..363c14d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetsatorageSearchRowImpl.java
@@ -0,0 +1,55 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.metastorage;
+
+/**
+ *
+ */
+public class MetsatorageSearchRowImpl implements MetastorageSearchRow {
+    /** */
+    private final String key;
+    /** */
+    private final long link;
+
+    /**
+     * @param key Key.
+     * @param link Link.
+     */
+    public MetsatorageSearchRowImpl(String key, long link) {
+        this.key = key;
+        this.link = link;
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public String key() {
+        return key;
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public long link() {
+        return link;
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public int hash() {
+        return key.hashCode();
+    }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/AbstractDataPageIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/AbstractDataPageIO.java
new file mode 100644
index 0000000..8666f75
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/AbstractDataPageIO.java
@@ -0,0 +1,1244 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.tree.io;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.pagemem.PageIdUtils;
+import org.apache.ignite.internal.pagemem.PageMemory;
+import org.apache.ignite.internal.pagemem.PageUtils;
+import org.apache.ignite.internal.processors.cache.persistence.Storable;
+import org.apache.ignite.internal.processors.cache.persistence.tree.util.PageHandler;
+import org.apache.ignite.internal.util.GridStringBuilder;
+import org.apache.ignite.internal.util.typedef.internal.SB;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Data pages IO.
+ */
+public abstract class AbstractDataPageIO<T extends Storable> extends PageIO {
+
+    /** */
+    private static final int SHOW_ITEM = 0b0001;
+
+    /** */
+    private static final int SHOW_PAYLOAD_LEN = 0b0010;
+
+    /** */
+    private static final int SHOW_LINK = 0b0100;
+
+    /** */
+    private static final int FREE_LIST_PAGE_ID_OFF = COMMON_HEADER_END;
+
+    /** */
+    private static final int FREE_SPACE_OFF = FREE_LIST_PAGE_ID_OFF + 8;
+
+    /** */
+    private static final int DIRECT_CNT_OFF = FREE_SPACE_OFF + 2;
+
+    /** */
+    private static final int INDIRECT_CNT_OFF = DIRECT_CNT_OFF + 1;
+
+    /** */
+    private static final int FIRST_ENTRY_OFF = INDIRECT_CNT_OFF + 1;
+
+    /** */
+    public static final int ITEMS_OFF = FIRST_ENTRY_OFF + 2;
+
+    /** */
+    private static final int ITEM_SIZE = 2;
+
+    /** */
+    private static final int PAYLOAD_LEN_SIZE = 2;
+
+    /** */
+    private static final int LINK_SIZE = 8;
+
+    /** */
+    private static final int FRAGMENTED_FLAG = 0b10000000_00000000;
+
+    /** */
+    public static final int MIN_DATA_PAGE_OVERHEAD = ITEMS_OFF + ITEM_SIZE + PAYLOAD_LEN_SIZE + LINK_SIZE;
+
+    /**
+     * @param ver Page format version.
+     */
+    protected AbstractDataPageIO(int type, int ver) {
+        super(type, ver);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void initNewPage(long pageAddr, long pageId, int pageSize) {
+        super.initNewPage(pageAddr, pageId, pageSize);
+
+        setEmptyPage(pageAddr, pageSize);
+        setFreeListPageId(pageAddr, 0L);
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param pageSize Page size.
+     */
+    private void setEmptyPage(long pageAddr, int pageSize) {
+        setDirectCount(pageAddr, 0);
+        setIndirectCount(pageAddr, 0);
+        setFirstEntryOffset(pageAddr, pageSize, pageSize);
+        setRealFreeSpace(pageAddr, pageSize - ITEMS_OFF, pageSize);
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param freeListPageId Free list page ID.
+     */
+    public void setFreeListPageId(long pageAddr, long freeListPageId) {
+        PageUtils.putLong(pageAddr, FREE_LIST_PAGE_ID_OFF, freeListPageId);
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @return Free list page ID.
+     */
+    public long getFreeListPageId(long pageAddr) {
+        return PageUtils.getLong(pageAddr, FREE_LIST_PAGE_ID_OFF);
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param dataOff Data offset.
+     * @param show What elements of data page entry to show in the result.
+     * @return Data page entry size.
+     */
+    private int getPageEntrySize(long pageAddr, int dataOff, int show) {
+        int payloadLen = PageUtils.getShort(pageAddr, dataOff) & 0xFFFF;
+
+        if ((payloadLen & FRAGMENTED_FLAG) != 0)
+            payloadLen &= ~FRAGMENTED_FLAG; // We are fragmented and have a link.
+        else
+            show &= ~SHOW_LINK; // We are not fragmented, never have a link.
+
+        return getPageEntrySize(payloadLen, show);
+    }
+
+    /**
+     * @param payloadLen Length of the payload, may be a full data row or a row fragment length.
+     * @param show What elements of data page entry to show in the result.
+     * @return Data page entry size.
+     */
+    private int getPageEntrySize(int payloadLen, int show) {
+        assert payloadLen > 0 : payloadLen;
+
+        int res = payloadLen;
+
+        if ((show & SHOW_LINK) != 0)
+            res += LINK_SIZE;
+
+        if ((show & SHOW_ITEM) != 0)
+            res += ITEM_SIZE;
+
+        if ((show & SHOW_PAYLOAD_LEN) != 0)
+            res += PAYLOAD_LEN_SIZE;
+
+        return res;
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param dataOff Entry data offset.
+     * @param pageSize Page size.
+     */
+    private void setFirstEntryOffset(long pageAddr, int dataOff, int pageSize) {
+        assert dataOff >= ITEMS_OFF + ITEM_SIZE && dataOff <= pageSize : dataOff;
+
+        PageUtils.putShort(pageAddr, FIRST_ENTRY_OFF, (short)dataOff);
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @return Entry data offset.
+     */
+    private int getFirstEntryOffset(long pageAddr) {
+        return PageUtils.getShort(pageAddr, FIRST_ENTRY_OFF) & 0xFFFF;
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param freeSpace Free space.
+     * @param pageSize Page size.
+     */
+    private void setRealFreeSpace(long pageAddr, int freeSpace, int pageSize) {
+        assert freeSpace == actualFreeSpace(pageAddr, pageSize) : freeSpace + " != " + actualFreeSpace(pageAddr, pageSize);
+
+        PageUtils.putShort(pageAddr, FREE_SPACE_OFF, (short)freeSpace);
+    }
+
+    /**
+     * Free space refers to a "max row size (without any data page specific overhead) which is guaranteed to fit into
+     * this data page".
+     *
+     * @param pageAddr Page address.
+     * @return Free space.
+     */
+    public int getFreeSpace(long pageAddr) {
+        if (getFreeItemSlots(pageAddr) == 0)
+            return 0;
+
+        int freeSpace = getRealFreeSpace(pageAddr);
+
+        // We reserve size here because of getFreeSpace() method semantics (see method javadoc).
+        // It means that we must be able to accommodate a row of size which is equal to getFreeSpace(),
+        // plus we will have data page overhead: header of the page as well as item, payload length and
+        // possibly a link to the next row fragment.
+        freeSpace -= ITEM_SIZE + PAYLOAD_LEN_SIZE + LINK_SIZE;
+
+        return freeSpace < 0 ? 0 : freeSpace;
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @return {@code true} If there is no useful data in this page.
+     */
+    public boolean isEmpty(long pageAddr) {
+        return getDirectCount(pageAddr) == 0;
+    }
+
+    /**
+     * Equivalent for {@link #actualFreeSpace(long, int)} but reads saved value.
+     *
+     * @param pageAddr Page address.
+     * @return Free space.
+     */
+    private int getRealFreeSpace(long pageAddr) {
+        return PageUtils.getShort(pageAddr, FREE_SPACE_OFF);
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param cnt Direct count.
+     */
+    private void setDirectCount(long pageAddr, int cnt) {
+        assert checkCount(cnt) : cnt;
+
+        PageUtils.putByte(pageAddr, DIRECT_CNT_OFF, (byte)cnt);
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @return Direct count.
+     */
+    private int getDirectCount(long pageAddr) {
+        return PageUtils.getByte(pageAddr, DIRECT_CNT_OFF) & 0xFF;
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param c Closure.
+     * @param <T> Closure return type.
+     * @return Collection of closure results for all items in page.
+     * @throws IgniteCheckedException In case of error in closure body.
+     */
+    public <T> List<T> forAllItems(long pageAddr, CC<T> c) throws IgniteCheckedException {
+        long pageId = getPageId(pageAddr);
+
+        int cnt = getDirectCount(pageAddr);
+
+        List<T> res = new ArrayList<>(cnt);
+
+        for (int i = 0; i < cnt; i++) {
+            long link = PageIdUtils.link(pageId, i);
+
+            res.add(c.apply(link));
+        }
+
+        return res;
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param cnt Indirect count.
+     */
+    private void setIndirectCount(long pageAddr, int cnt) {
+        assert checkCount(cnt) : cnt;
+
+        PageUtils.putByte(pageAddr, INDIRECT_CNT_OFF, (byte)cnt);
+    }
+
+    /**
+     * @param idx Index.
+     * @return {@code true} If the index is valid.
+     */
+    protected boolean checkIndex(int idx) {
+        return idx >= 0 && idx < 0xFF;
+    }
+
+    /**
+     * @param cnt Counter value.
+     * @return {@code true} If the counter fits 1 byte.
+     */
+    private boolean checkCount(int cnt) {
+        return cnt >= 0 && cnt <= 0xFF;
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @return Indirect count.
+     */
+    private int getIndirectCount(long pageAddr) {
+        return PageUtils.getByte(pageAddr, INDIRECT_CNT_OFF) & 0xFF;
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @return Number of free entry slots.
+     */
+    private int getFreeItemSlots(long pageAddr) {
+        return 0xFF - getDirectCount(pageAddr);
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param itemId Fixed item ID (the index used for referencing an entry from the outside).
+     * @param directCnt Direct items count.
+     * @param indirectCnt Indirect items count.
+     * @return Found index of indirect item.
+     */
+    private int findIndirectItemIndex(long pageAddr, int itemId, int directCnt, int indirectCnt) {
+        int low = directCnt;
+        int high = directCnt + indirectCnt - 1;
+
+        while (low <= high) {
+            int mid = (low + high) >>> 1;
+
+            int cmp = Integer.compare(itemId(getItem(pageAddr, mid)), itemId);
+
+            if (cmp < 0)
+                low = mid + 1;
+            else if (cmp > 0)
+                high = mid - 1;
+            else
+                return mid; // found
+        }
+
+        throw new IllegalStateException("Item not found: " + itemId);
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param pageSize Page size.
+     * @return String representation.
+     */
+    private String printPageLayout(long pageAddr, int pageSize) {
+        SB b = new SB();
+
+        printPageLayout(pageAddr, pageSize, b);
+
+        return b.toString();
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param pageSize Page size.
+     * @param b B.
+     */
+    protected void printPageLayout(long pageAddr, int pageSize, GridStringBuilder b) {
+        int directCnt = getDirectCount(pageAddr);
+        int indirectCnt = getIndirectCount(pageAddr);
+        int free = getRealFreeSpace(pageAddr);
+
+        boolean valid = directCnt >= indirectCnt;
+
+        b.appendHex(PageIO.getPageId(pageAddr)).a(" [");
+
+        int entriesSize = 0;
+
+        for (int i = 0; i < directCnt; i++) {
+            if (i != 0)
+                b.a(", ");
+
+            short item = getItem(pageAddr, i);
+
+            if (item < ITEMS_OFF || item >= pageSize)
+                valid = false;
+
+            entriesSize += getPageEntrySize(pageAddr, item, SHOW_PAYLOAD_LEN | SHOW_LINK);
+
+            b.a(item);
+        }
+
+        b.a("][");
+
+        Collection<Integer> set = new HashSet<>();
+
+        for (int i = directCnt; i < directCnt + indirectCnt; i++) {
+            if (i != directCnt)
+                b.a(", ");
+
+            short item = getItem(pageAddr, i);
+
+            int itemId = itemId(item);
+            int directIdx = directItemIndex(item);
+
+            if (!set.add(directIdx) || !set.add(itemId))
+                valid = false;
+
+            assert indirectItem(itemId, directIdx) == item;
+
+            if (itemId < directCnt || directIdx < 0 || directIdx >= directCnt)
+                valid = false;
+
+            if (i > directCnt && itemId(getItem(pageAddr, i - 1)) >= itemId)
+                valid = false;
+
+
+            b.a(itemId).a('^').a(directIdx);
+        }
+
+        b.a("][free=").a(free);
+
+        int actualFree = pageSize - ITEMS_OFF - (entriesSize + (directCnt + indirectCnt) * ITEM_SIZE);
+
+        if (free != actualFree) {
+            b.a(", actualFree=").a(actualFree);
+
+            valid = false;
+        }
+        else
+            b.a("]");
+
+        assert valid : b.toString();
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param itemId Fixed item ID (the index used for referencing an entry from the outside).
+     * @param pageSize Page size.
+     * @return Data entry offset in bytes.
+     */
+    protected int getDataOffset(long pageAddr, int itemId, int pageSize) {
+        assert checkIndex(itemId) : itemId;
+
+        int directCnt = getDirectCount(pageAddr);
+
+        assert directCnt > 0 : "itemId=" + itemId + ", directCnt=" + directCnt + ", page=" + printPageLayout(pageAddr, pageSize);
+
+        if (itemId >= directCnt) { // Need to do indirect lookup.
+            int indirectCnt = getIndirectCount(pageAddr);
+
+            // Must have indirect items here.
+            assert indirectCnt > 0 : "itemId=" + itemId + ", directCnt=" + directCnt + ", indirectCnt=" + indirectCnt +
+                ", page=" + printPageLayout(pageAddr, pageSize);
+
+            int indirectItemIdx = findIndirectItemIndex(pageAddr, itemId, directCnt, indirectCnt);
+
+            assert indirectItemIdx >= directCnt : indirectItemIdx + " " + directCnt;
+            assert indirectItemIdx < directCnt + indirectCnt : indirectItemIdx + " " + directCnt + " " + indirectCnt;
+
+            itemId = directItemIndex(getItem(pageAddr, indirectItemIdx));
+
+            assert itemId >= 0 && itemId < directCnt : itemId + " " + directCnt + " " + indirectCnt; // Direct item.
+        }
+
+        return directItemToOffset(getItem(pageAddr, itemId));
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param dataOff Points to the entry start.
+     * @return Link to the next entry fragment or 0 if no fragments left or if entry is not fragmented.
+     */
+    private long getNextFragmentLink(long pageAddr, int dataOff) {
+        assert isFragmented(pageAddr, dataOff);
+
+        return PageUtils.getLong(pageAddr, dataOff + PAYLOAD_LEN_SIZE);
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param dataOff Data offset.
+     * @return {@code true} If the data row is fragmented across multiple pages.
+     */
+    protected boolean isFragmented(long pageAddr, int dataOff) {
+        return (PageUtils.getShort(pageAddr, dataOff) & FRAGMENTED_FLAG) != 0;
+    }
+
+    /**
+     * Sets position to start of actual fragment data and limit to it's end.
+     *
+     * @param pageAddr Page address.
+     * @param itemId Item to position on.
+     * @param pageSize Page size.
+     * @return {@link DataPagePayload} object.
+     */
+    public DataPagePayload readPayload(final long pageAddr, final int itemId, final int pageSize) {
+        int dataOff = getDataOffset(pageAddr, itemId, pageSize);
+
+        boolean fragmented = isFragmented(pageAddr, dataOff);
+        long nextLink = fragmented ? getNextFragmentLink(pageAddr, dataOff) : 0;
+        int payloadSize = getPageEntrySize(pageAddr, dataOff, 0);
+
+        return new DataPagePayload(dataOff + PAYLOAD_LEN_SIZE + (fragmented ? LINK_SIZE : 0),
+            payloadSize,
+            nextLink);
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param idx Item index.
+     * @return Item.
+     */
+    private short getItem(long pageAddr, int idx) {
+        return PageUtils.getShort(pageAddr, itemOffset(idx));
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param idx Item index.
+     * @param item Item.
+     */
+    private void setItem(long pageAddr, int idx, short item) {
+        PageUtils.putShort(pageAddr, itemOffset(idx), item);
+    }
+
+    /**
+     * @param idx Index of the item.
+     * @return Offset in buffer.
+     */
+    private int itemOffset(int idx) {
+        assert checkIndex(idx) : idx;
+
+        return ITEMS_OFF + idx * ITEM_SIZE;
+    }
+
+    /**
+     * @param directItem Direct item.
+     * @return Offset of an entry payload inside of the page.
+     */
+    private int directItemToOffset(short directItem) {
+        return directItem & 0xFFFF;
+    }
+
+    /**
+     * @param dataOff Data offset.
+     * @return Direct item.
+     */
+    private short directItemFromOffset(int dataOff) {
+        assert dataOff >= ITEMS_OFF + ITEM_SIZE && dataOff < Short.MAX_VALUE : dataOff;
+
+        return (short)dataOff;
+    }
+
+    /**
+     * @param indirectItem Indirect item.
+     * @return Index of corresponding direct item.
+     */
+    private int directItemIndex(short indirectItem) {
+        return indirectItem & 0xFF;
+    }
+
+    /**
+     * @param indirectItem Indirect item.
+     * @return Fixed item ID (the index used for referencing an entry from the outside).
+     */
+    private int itemId(short indirectItem) {
+        return (indirectItem & 0xFFFF) >>> 8;
+    }
+
+    /**
+     * @param itemId Fixed item ID (the index used for referencing an entry from the outside).
+     * @param directItemIdx Index of corresponding direct item.
+     * @return Indirect item.
+     */
+    private short indirectItem(int itemId, int directItemIdx) {
+        assert checkIndex(itemId) : itemId;
+        assert checkIndex(directItemIdx) : directItemIdx;
+
+        return (short)((itemId << 8) | directItemIdx);
+    }
+
+    /**
+     * Move the last direct item to the free slot and reference it with indirect item on the same place.
+     *
+     * @param pageAddr Page address.
+     * @param freeDirectIdx Free slot.
+     * @param directCnt Direct items count.
+     * @param indirectCnt Indirect items count.
+     * @return {@code true} If the last direct item already had corresponding indirect item.
+     */
+    private boolean moveLastItem(long pageAddr, int freeDirectIdx, int directCnt, int indirectCnt) {
+        int lastIndirectId = findIndirectIndexForLastDirect(pageAddr, directCnt, indirectCnt);
+
+        int lastItemId = directCnt - 1;
+
+        assert lastItemId != freeDirectIdx;
+
+        short indirectItem = indirectItem(lastItemId, freeDirectIdx);
+
+        assert itemId(indirectItem) == lastItemId && directItemIndex(indirectItem) == freeDirectIdx;
+
+        setItem(pageAddr, freeDirectIdx, getItem(pageAddr, lastItemId));
+        setItem(pageAddr, lastItemId, indirectItem);
+
+        assert getItem(pageAddr, lastItemId) == indirectItem;
+
+        if (lastIndirectId != -1) { // Fix pointer to direct item.
+            setItem(pageAddr, lastIndirectId, indirectItem(itemId(getItem(pageAddr, lastIndirectId)), freeDirectIdx));
+
+            return true;
+        }
+
+        return false;
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param directCnt Direct items count.
+     * @param indirectCnt Indirect items count.
+     * @return Index of indirect item for the last direct item.
+     */
+    private int findIndirectIndexForLastDirect(long pageAddr, int directCnt, int indirectCnt) {
+        int lastDirectId = directCnt - 1;
+
+        for (int i = directCnt, end = directCnt + indirectCnt; i < end; i++) {
+            short item = getItem(pageAddr, i);
+
+            if (directItemIndex(item) == lastDirectId)
+                return i;
+        }
+
+        return -1;
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param itemId Item ID.
+     * @param pageSize Page size.
+     * @param payload Row data.
+     * @param row Row.
+     * @param rowSize Row size.
+     * @return {@code True} if entry is not fragmented.
+     * @throws IgniteCheckedException If failed.
+     */
+    public boolean updateRow(
+        final long pageAddr,
+        int itemId,
+        int pageSize,
+        @Nullable byte[] payload,
+        @Nullable T row,
+        final int rowSize) throws IgniteCheckedException {
+        assert checkIndex(itemId) : itemId;
+        assert row != null ^ payload != null;
+
+        final int dataOff = getDataOffset(pageAddr, itemId, pageSize);
+
+        if (isFragmented(pageAddr, dataOff))
+            return false;
+
+        if (row != null)
+            writeRowData(pageAddr, dataOff, rowSize, row, false);
+        else
+            writeRowData(pageAddr, dataOff, payload);
+
+        return true;
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param itemId Fixed item ID (the index used for referencing an entry from the outside).
+     * @param pageSize Page size.
+     * @return Next link for fragmented entries or {@code 0} if none.
+     * @throws IgniteCheckedException If failed.
+     */
+    public long removeRow(long pageAddr, int itemId, int pageSize) throws IgniteCheckedException {
+        assert checkIndex(itemId) : itemId;
+
+        final int dataOff = getDataOffset(pageAddr, itemId, pageSize);
+        final long nextLink = isFragmented(pageAddr, dataOff) ? getNextFragmentLink(pageAddr, dataOff) : 0;
+
+        // Record original counts to calculate delta in free space in the end of remove.
+        final int directCnt = getDirectCount(pageAddr);
+        final int indirectCnt = getIndirectCount(pageAddr);
+
+        int curIndirectCnt = indirectCnt;
+
+        assert directCnt > 0 : directCnt; // Direct count always represents overall number of live items.
+
+        // Remove the last item on the page.
+        if (directCnt == 1) {
+            assert (indirectCnt == 0 && itemId == 0) ||
+                (indirectCnt == 1 && itemId == itemId(getItem(pageAddr, 1))) : itemId;
+
+            setEmptyPage(pageAddr, pageSize);
+        }
+        else {
+            // Get the entry size before the actual remove.
+            int rmvEntrySize = getPageEntrySize(pageAddr, dataOff, SHOW_PAYLOAD_LEN | SHOW_LINK);
+
+            int indirectId = 0;
+
+            if (itemId >= directCnt) { // Need to remove indirect item.
+                assert indirectCnt > 0;
+
+                indirectId = findIndirectItemIndex(pageAddr, itemId, directCnt, indirectCnt);
+
+                assert indirectId >= directCnt;
+
+                itemId = directItemIndex(getItem(pageAddr, indirectId));
+
+                assert itemId < directCnt;
+            }
+
+            boolean dropLast = true;
+
+            if (itemId + 1 < directCnt) // It is not the last direct item.
+                dropLast = moveLastItem(pageAddr, itemId, directCnt, indirectCnt);
+
+            if (indirectId == 0) {// For the last direct item with no indirect item.
+                if (dropLast)
+                    moveItems(pageAddr, directCnt, indirectCnt, -1, pageSize);
+                else
+                    curIndirectCnt++;
+            }
+            else {
+                if (dropLast)
+                    moveItems(pageAddr, directCnt, indirectId - directCnt, -1, pageSize);
+
+                moveItems(pageAddr, indirectId + 1, directCnt + indirectCnt - indirectId - 1, dropLast ? -2 : -1, pageSize);
+
+                if (dropLast)
+                    curIndirectCnt--;
+            }
+
+            setIndirectCount(pageAddr, curIndirectCnt);
+            setDirectCount(pageAddr, directCnt - 1);
+
+            assert getIndirectCount(pageAddr) <= getDirectCount(pageAddr);
+
+            // Increase free space.
+            setRealFreeSpace(pageAddr,
+                getRealFreeSpace(pageAddr) + rmvEntrySize + ITEM_SIZE * (directCnt - getDirectCount(pageAddr) + indirectCnt - getIndirectCount(pageAddr)),
+                pageSize);
+        }
+
+        return nextLink;
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param idx Index.
+     * @param cnt Count.
+     * @param step Step.
+     * @param pageSize Page size.
+     */
+    private void moveItems(long pageAddr, int idx, int cnt, int step, int pageSize) {
+        assert cnt >= 0 : cnt;
+
+        if (cnt != 0)
+            moveBytes(pageAddr, itemOffset(idx), cnt * ITEM_SIZE, step * ITEM_SIZE, pageSize);
+    }
+
+    /**
+     * @param newEntryFullSize New entry full size (with item, length and link).
+     * @param firstEntryOff First entry data offset.
+     * @param directCnt Direct items count.
+     * @param indirectCnt Indirect items count.
+     * @return {@code true} If there is enough space for the entry.
+     */
+    private boolean isEnoughSpace(int newEntryFullSize, int firstEntryOff, int directCnt, int indirectCnt) {
+        return ITEMS_OFF + ITEM_SIZE * (directCnt + indirectCnt) <= firstEntryOff - newEntryFullSize;
+    }
+
+    /**
+     * Adds row to this data page and sets respective link to the given row object.
+     *
+     * @param pageAddr Page address.
+     * @param row Data row.
+     * @param rowSize Row size.
+     * @param pageSize Page size.
+     * @throws IgniteCheckedException If failed.
+     */
+    public void addRow(
+        final long pageAddr,
+        T row,
+        final int rowSize,
+        final int pageSize
+    ) throws IgniteCheckedException {
+        assert rowSize <= getFreeSpace(pageAddr) : "can't call addRow if not enough space for the whole row";
+
+        int fullEntrySize = getPageEntrySize(rowSize, SHOW_PAYLOAD_LEN | SHOW_ITEM);
+
+        int directCnt = getDirectCount(pageAddr);
+        int indirectCnt = getIndirectCount(pageAddr);
+
+        int dataOff = getDataOffsetForWrite(pageAddr, fullEntrySize, directCnt, indirectCnt, pageSize);
+
+        writeRowData(pageAddr, dataOff, rowSize, row, true);
+
+        int itemId = addItem(pageAddr, fullEntrySize, directCnt, indirectCnt, dataOff, pageSize);
+
+        setLink(row, pageAddr, itemId);
+    }
+
+    /**
+     * Adds row to this data page and sets respective link to the given row object.
+     *
+     * @param pageAddr Page address.
+     * @param payload Payload.
+     * @param pageSize Page size.
+     * @throws IgniteCheckedException If failed.
+     */
+    public void addRow(
+        long pageAddr,
+        byte[] payload,
+        int pageSize
+    ) throws IgniteCheckedException {
+        assert payload.length <= getFreeSpace(pageAddr) : "can't call addRow if not enough space for the whole row";
+
+        int fullEntrySize = getPageEntrySize(payload.length, SHOW_PAYLOAD_LEN | SHOW_ITEM);
+
+        int directCnt = getDirectCount(pageAddr);
+        int indirectCnt = getIndirectCount(pageAddr);
+
+        int dataOff = getDataOffsetForWrite(pageAddr, fullEntrySize, directCnt, indirectCnt, pageSize);
+
+        writeRowData(pageAddr, dataOff, payload);
+
+        addItem(pageAddr, fullEntrySize, directCnt, indirectCnt, dataOff, pageSize);
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param entryFullSize New entry full size (with item, length and link).
+     * @param directCnt Direct items count.
+     * @param indirectCnt Indirect items count.
+     * @param dataOff First entry offset.
+     * @param pageSize Page size.
+     * @return First entry offset after compaction.
+     */
+    private int compactIfNeed(
+        final long pageAddr,
+        final int entryFullSize,
+        final int directCnt,
+        final int indirectCnt,
+        int dataOff,
+        int pageSize
+    ) {
+        if (!isEnoughSpace(entryFullSize, dataOff, directCnt, indirectCnt)) {
+            dataOff = compactDataEntries(pageAddr, directCnt, pageSize);
+
+            assert isEnoughSpace(entryFullSize, dataOff, directCnt, indirectCnt);
+        }
+
+        return dataOff;
+    }
+
+    /**
+     * Put item reference on entry.
+     *
+     * @param pageAddr Page address.
+     * @param fullEntrySize Full entry size (with link, payload size and item).
+     * @param directCnt Direct items count.
+     * @param indirectCnt Indirect items count.
+     * @param dataOff Data offset.
+     * @param pageSize Page size.
+     * @return Item ID.
+     */
+    private int addItem(final long pageAddr,
+        final int fullEntrySize,
+        final int directCnt,
+        final int indirectCnt,
+        final int dataOff,
+        final int pageSize) {
+        setFirstEntryOffset(pageAddr, dataOff, pageSize);
+
+        int itemId = insertItem(pageAddr, dataOff, directCnt, indirectCnt, pageSize);
+
+        assert checkIndex(itemId) : itemId;
+        assert getIndirectCount(pageAddr) <= getDirectCount(pageAddr);
+
+        // Update free space. If number of indirect items changed, then we were able to reuse an item slot.
+        setRealFreeSpace(pageAddr,
+            getRealFreeSpace(pageAddr) - fullEntrySize + (getIndirectCount(pageAddr) != indirectCnt ? ITEM_SIZE : 0),
+            pageSize);
+
+        return itemId;
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param fullEntrySize Full entry size.
+     * @param directCnt Direct items count.
+     * @param indirectCnt Indirect items count.
+     * @param pageSize Page size.
+     * @return Offset in the buffer where the entry must be written.
+     */
+    private int getDataOffsetForWrite(long pageAddr, int fullEntrySize, int directCnt, int indirectCnt, int pageSize) {
+        int dataOff = getFirstEntryOffset(pageAddr);
+
+        // Compact if we do not have enough space for entry.
+        dataOff = compactIfNeed(pageAddr, fullEntrySize, directCnt, indirectCnt, dataOff, pageSize);
+
+        // We will write data right before the first entry.
+        dataOff -= fullEntrySize - ITEM_SIZE;
+
+        return dataOff;
+    }
+
+    /**
+     * Adds maximum possible fragment of the given row to this data page and sets respective link to the row.
+     *
+     * @param pageMem Page memory.
+     * @param pageAddr Page address.
+     * @param row Data row.
+     * @param written Number of bytes of row size that was already written.
+     * @param rowSize Row size.
+     * @param pageSize Page size.
+     * @return Written payload size.
+     * @throws IgniteCheckedException If failed.
+     */
+    public int addRowFragment(
+        PageMemory pageMem,
+        long pageAddr,
+        T row,
+        int written,
+        int rowSize,
+        int pageSize
+    ) throws IgniteCheckedException {
+        return addRowFragment(pageMem, pageAddr, written, rowSize, row.link(), row, null, pageSize);
+    }
+
+    /**
+     * Adds this payload as a fragment to this data page.
+     *
+     * @param pageAddr Page address.
+     * @param payload Payload bytes.
+     * @param lastLink Link to the previous written fragment (link to the tail).
+     * @param pageSize Page size.
+     * @throws IgniteCheckedException If failed.
+     */
+    public void addRowFragment(
+        long pageAddr,
+        byte[] payload,
+        long lastLink,
+        int pageSize
+    ) throws IgniteCheckedException {
+        addRowFragment(null, pageAddr, 0, 0, lastLink, null, payload, pageSize);
+    }
+
+    /**
+     * Adds maximum possible fragment of the given row to this data page and sets respective link to the row.
+     *
+     * @param pageMem Page memory.
+     * @param pageAddr Page address.
+     * @param written Number of bytes of row size that was already written.
+     * @param rowSize Row size.
+     * @param lastLink Link to the previous written fragment (link to the tail).
+     * @param row Row.
+     * @param payload Payload bytes.
+     * @param pageSize Page size.
+     * @return Written payload size.
+     * @throws IgniteCheckedException If failed.
+     */
+    private int addRowFragment(
+        PageMemory pageMem,
+        long pageAddr,
+        int written,
+        int rowSize,
+        long lastLink,
+        T row,
+        byte[] payload,
+        int pageSize
+    ) throws IgniteCheckedException {
+        assert payload == null ^ row == null;
+
+        int directCnt = getDirectCount(pageAddr);
+        int indirectCnt = getIndirectCount(pageAddr);
+
+        int payloadSize = payload != null ? payload.length :
+            Math.min(rowSize - written, getFreeSpace(pageAddr));
+
+        int fullEntrySize = getPageEntrySize(payloadSize, SHOW_PAYLOAD_LEN | SHOW_LINK | SHOW_ITEM);
+        int dataOff = getDataOffsetForWrite(pageAddr, fullEntrySize, directCnt, indirectCnt, pageSize);
+
+        if (payload == null) {
+            ByteBuffer buf = pageMem.pageBuffer(pageAddr);
+
+            buf.position(dataOff);
+
+            short p = (short)(payloadSize | FRAGMENTED_FLAG);
+
+            buf.putShort(p);
+            buf.putLong(lastLink);
+
+            int rowOff = rowSize - written - payloadSize;
+
+            writeFragmentData(row, buf, rowOff, payloadSize);
+        }
+        else {
+            PageUtils.putShort(pageAddr, dataOff, (short)(payloadSize | FRAGMENTED_FLAG));
+
+            PageUtils.putLong(pageAddr, dataOff + 2, lastLink);
+
+            PageUtils.putBytes(pageAddr, dataOff + 10, payload);
+        }
+
+        int itemId = addItem(pageAddr, fullEntrySize, directCnt, indirectCnt, dataOff, pageSize);
+
+        if (row != null)
+            setLink(row, pageAddr, itemId);
+
+        return payloadSize;
+    }
+
+    /**
+     * @param row Row to set link to.
+     * @param pageAddr Page address.
+     * @param itemId Item ID.
+     */
+    private void setLink(Storable row, long pageAddr, int itemId) {
+        row.link(PageIdUtils.link(getPageId(pageAddr), itemId));
+    }
+
+    /**
+     * Write row data fragment.
+     *
+     * @param row Row.
+     * @param buf Byte buffer.
+     * @param rowOff Offset in row data bytes.
+     * @param payloadSize Data length that should be written in a fragment.
+     * @throws IgniteCheckedException If failed.
+     */
+    protected abstract void writeFragmentData(
+        final T row,
+        final ByteBuffer buf,
+        final int rowOff,
+        final int payloadSize
+    ) throws IgniteCheckedException;
+
+    /**
+     * @param pageAddr Page address.
+     * @param dataOff Data offset.
+     * @param directCnt Direct items count.
+     * @param indirectCnt Indirect items count.
+     * @param pageSize Page size.
+     * @return Item ID (insertion index).
+     */
+    private int insertItem(long pageAddr, int dataOff, int directCnt, int indirectCnt, int pageSize) {
+        if (indirectCnt > 0) {
+            // If the first indirect item is on correct place to become the last direct item, do the transition
+            // and insert the new item into the free slot which was referenced by this first indirect item.
+            short item = getItem(pageAddr, directCnt);
+
+            if (itemId(item) == directCnt) {
+                int directItemIdx = directItemIndex(item);
+
+                setItem(pageAddr, directCnt, getItem(pageAddr, directItemIdx));
+                setItem(pageAddr, directItemIdx, directItemFromOffset(dataOff));
+
+                setDirectCount(pageAddr, directCnt + 1);
+                setIndirectCount(pageAddr, indirectCnt - 1);
+
+                return directItemIdx;
+            }
+        }
+
+        // Move all the indirect items forward to make a free slot and insert new item at the end of direct items.
+        moveItems(pageAddr, directCnt, indirectCnt, +1, pageSize);
+
+        setItem(pageAddr, directCnt, directItemFromOffset(dataOff));
+
+        setDirectCount(pageAddr, directCnt + 1);
+        assert getDirectCount(pageAddr) == directCnt + 1;
+
+        return directCnt; // Previous directCnt will be our itemId.
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param directCnt Direct items count.
+     * @param pageSize Page size.
+     * @return New first entry offset.
+     */
+    private int compactDataEntries(long pageAddr, int directCnt, int pageSize) {
+        assert checkCount(directCnt) : directCnt;
+
+        int[] offs = new int[directCnt];
+
+        for (int i = 0; i < directCnt; i++) {
+            int off = directItemToOffset(getItem(pageAddr, i));
+
+            offs[i] = (off << 8) | i; // This way we'll be able to sort by offset using Arrays.sort(...).
+        }
+
+        Arrays.sort(offs);
+
+        // Move right all of the entries if possible to make the page as compact as possible to its tail.
+        int prevOff = pageSize;
+
+        final int start = directCnt - 1;
+        int curOff = offs[start] >>> 8;
+        int curEntrySize = getPageEntrySize(pageAddr, curOff, SHOW_PAYLOAD_LEN | SHOW_LINK);
+
+        for (int i = start; i >= 0; i--) {
+            assert curOff < prevOff : curOff;
+
+            int delta = prevOff - (curOff + curEntrySize);
+
+            int off = curOff;
+            int entrySize = curEntrySize;
+
+            if (delta != 0) { // Move right.
+                assert delta > 0 : delta;
+
+                int itemId = offs[i] & 0xFF;
+
+                setItem(pageAddr, itemId, directItemFromOffset(curOff + delta));
+
+                for (int j = i - 1; j >= 0; j--) {
+                    int offNext = offs[j] >>> 8;
+                    int nextSize = getPageEntrySize(pageAddr, offNext, SHOW_PAYLOAD_LEN | SHOW_LINK);
+
+                    if (offNext + nextSize == off) {
+                        i--;
+
+                        off = offNext;
+                        entrySize += nextSize;
+
+                        itemId = offs[j] & 0xFF;
+                        setItem(pageAddr, itemId, directItemFromOffset(offNext + delta));
+                    }
+                    else {
+                        curOff = offNext;
+                        curEntrySize = nextSize;
+
+                        break;
+                    }
+                }
+
+                moveBytes(pageAddr, off, entrySize, delta, pageSize);
+
+                off += delta;
+            }
+            else if (i > 0) {
+                curOff = offs[i - 1] >>> 8;
+                curEntrySize = getPageEntrySize(pageAddr, curOff, SHOW_PAYLOAD_LEN | SHOW_LINK);
+            }
+
+            prevOff = off;
+        }
+
+        return prevOff;
+    }
+
+    /**
+     * Full-scan free space calculation procedure.
+     *
+     * @param pageAddr Page to scan.
+     * @param pageSize Page size.
+     * @return Actual free space in the buffer.
+     */
+    private int actualFreeSpace(long pageAddr, int pageSize) {
+        int directCnt = getDirectCount(pageAddr);
+
+        int entriesSize = 0;
+
+        for (int i = 0; i < directCnt; i++) {
+            int off = directItemToOffset(getItem(pageAddr, i));
+
+            int entrySize = getPageEntrySize(pageAddr, off, SHOW_PAYLOAD_LEN | SHOW_LINK);
+
+            entriesSize += entrySize;
+        }
+
+        return pageSize - ITEMS_OFF - entriesSize - (directCnt + getIndirectCount(pageAddr)) * ITEM_SIZE;
+    }
+
+    /**
+     * @param addr Address.
+     * @param off Offset.
+     * @param cnt Count.
+     * @param step Step.
+     * @param pageSize Page size.
+     */
+    private void moveBytes(long addr, int off, int cnt, int step, int pageSize) {
+        assert step != 0 : step;
+        assert off + step >= 0;
+        assert off + step + cnt <= pageSize : "[off=" + off + ", step=" + step + ", cnt=" + cnt +
+            ", cap=" + pageSize + ']';
+
+        PageHandler.copyMemory(addr, off, addr, off + step, cnt);
+    }
+
+    /**
+     * @param pageAddr Page address.
+     * @param dataOff Data offset.
+     * @param payloadSize Payload size.
+     * @param row Data row.
+     * @param newRow {@code False} if existing cache entry is updated, in this case skip key data write.
+     * @throws IgniteCheckedException If failed.
+     */
+    protected abstract void writeRowData(
+        long pageAddr,
+        int dataOff,
+        int payloadSize,
+        T row,
+        boolean newRow
+    ) throws IgniteCheckedException;
+
+    /**
+     * @param pageAddr Page address.
+     * @param dataOff Data offset.
+     * @param payload Payload
+     */
+    protected void writeRowData(
+        long pageAddr,
+        int dataOff,
+        byte[] payload
+    ) {
+        PageUtils.putShort(pageAddr, dataOff, (short)payload.length);
+        dataOff += 2;
+
+        PageUtils.putBytes(pageAddr, dataOff, payload);
+    }
+
+    /**
+     * @param row Row.
+     * @return Row size in page.
+     * @throws IgniteCheckedException if failed.
+     */
+    public abstract int getRowSize(T row) throws IgniteCheckedException;
+
+    /**
+     * Defines closure interface for applying computations to data page items.
+     *
+     * @param <T> Closure return type.
+     */
+    public interface CC<T> {
+        /**
+         * Closure body.
+         *
+         * @param link Link to item.
+         * @return Closure return value.
+         * @throws IgniteCheckedException In case of error in closure body.
+         */
+        public T apply(long link) throws IgniteCheckedException;
+    }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java
index 628ff38..8a04749 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/DataPageIO.java
@@ -18,74 +18,23 @@
 package org.apache.ignite.internal.processors.cache.persistence.tree.io;
 
 import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.List;
 import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.pagemem.PageIdUtils;
-import org.apache.ignite.internal.pagemem.PageMemory;
 import org.apache.ignite.internal.pagemem.PageUtils;
 import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
 import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
-import org.apache.ignite.internal.processors.cache.persistence.tree.util.PageHandler;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
 import org.apache.ignite.internal.util.GridStringBuilder;
-import org.apache.ignite.internal.util.typedef.internal.SB;
-import org.jetbrains.annotations.Nullable;
 
 /**
  * Data pages IO.
  */
-public class DataPageIO extends PageIO {
+public class DataPageIO extends AbstractDataPageIO<CacheDataRow> {
     /** */
     public static final IOVersions<DataPageIO> VERSIONS = new IOVersions<>(
         new DataPageIO(1)
     );
 
-    /** */
-    private static final int SHOW_ITEM = 0b0001;
-
-    /** */
-    private static final int SHOW_PAYLOAD_LEN = 0b0010;
-
-    /** */
-    private static final int SHOW_LINK = 0b0100;
-
-    /** */
-    private static final int FREE_LIST_PAGE_ID_OFF = COMMON_HEADER_END;
-
-    /** */
-    private static final int FREE_SPACE_OFF = FREE_LIST_PAGE_ID_OFF + 8;
-
-    /** */
-    private static final int DIRECT_CNT_OFF = FREE_SPACE_OFF + 2;
-
-    /** */
-    private static final int INDIRECT_CNT_OFF = DIRECT_CNT_OFF + 1;
-
-    /** */
-    private static final int FIRST_ENTRY_OFF = INDIRECT_CNT_OFF + 1;
-
-    /** Offset of items (internal page pointers) within data page */
-    public static final int ITEMS_OFF = FIRST_ENTRY_OFF + 2;
-
-    /** */
-    private static final int ITEM_SIZE = 2;
-
-    /** */
-    private static final int PAYLOAD_LEN_SIZE = 2;
-
-    /** */
-    private static final int LINK_SIZE = 8;
-
-    /** */
-    private static final int FRAGMENTED_FLAG = 0b10000000_00000000;
-
-    /** */
-    public static final int MIN_DATA_PAGE_OVERHEAD = ITEMS_OFF + ITEM_SIZE + PAYLOAD_LEN_SIZE + LINK_SIZE;
-
     /**
      * @param ver Page format version.
      */
@@ -94,946 +43,8 @@
     }
 
     /** {@inheritDoc} */
-    @Override public void initNewPage(long pageAddr, long pageId, int pageSize) {
-        super.initNewPage(pageAddr, pageId, pageSize);
-
-        setEmptyPage(pageAddr, pageSize);
-        setFreeListPageId(pageAddr, 0L);
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param pageSize Page size.
-     */
-    private void setEmptyPage(long pageAddr, int pageSize) {
-        setDirectCount(pageAddr, 0);
-        setIndirectCount(pageAddr, 0);
-        setFirstEntryOffset(pageAddr, pageSize, pageSize);
-        setRealFreeSpace(pageAddr, pageSize - ITEMS_OFF, pageSize);
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param freeListPageId Free list page ID.
-     */
-    public void setFreeListPageId(long pageAddr, long freeListPageId) {
-        PageUtils.putLong(pageAddr, FREE_LIST_PAGE_ID_OFF, freeListPageId);
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @return Free list page ID.
-     */
-    public long getFreeListPageId(long pageAddr) {
-        return PageUtils.getLong(pageAddr, FREE_LIST_PAGE_ID_OFF);
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param dataOff Data offset.
-     * @param show What elements of data page entry to show in the result.
-     * @return Data page entry size.
-     */
-    private int getPageEntrySize(long pageAddr, int dataOff, int show) {
-        int payloadLen = PageUtils.getShort(pageAddr, dataOff) & 0xFFFF;
-
-        if ((payloadLen & FRAGMENTED_FLAG) != 0)
-            payloadLen &= ~FRAGMENTED_FLAG; // We are fragmented and have a link.
-        else
-            show &= ~SHOW_LINK; // We are not fragmented, never have a link.
-
-        return getPageEntrySize(payloadLen, show);
-    }
-
-    /**
-     * @param payloadLen Length of the payload, may be a full data row or a row fragment length.
-     * @param show What elements of data page entry to show in the result.
-     * @return Data page entry size.
-     */
-    private int getPageEntrySize(int payloadLen, int show) {
-        assert payloadLen > 0 : payloadLen;
-
-        int res = payloadLen;
-
-        if ((show & SHOW_LINK) != 0)
-            res += LINK_SIZE;
-
-        if ((show & SHOW_ITEM) != 0)
-            res += ITEM_SIZE;
-
-        if ((show & SHOW_PAYLOAD_LEN) != 0)
-            res += PAYLOAD_LEN_SIZE;
-
-        return res;
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param dataOff Entry data offset.
-     * @param pageSize Page size.
-     */
-    private void setFirstEntryOffset(long pageAddr, int dataOff, int pageSize) {
-        assert dataOff >= ITEMS_OFF + ITEM_SIZE && dataOff <= pageSize : dataOff;
-
-        PageUtils.putShort(pageAddr, FIRST_ENTRY_OFF, (short)dataOff);
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @return Entry data offset.
-     */
-    private int getFirstEntryOffset(long pageAddr) {
-        return PageUtils.getShort(pageAddr, FIRST_ENTRY_OFF) & 0xFFFF;
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param freeSpace Free space.
-     * @param pageSize Page size.
-     */
-    private void setRealFreeSpace(long pageAddr, int freeSpace, int pageSize) {
-        assert freeSpace == actualFreeSpace(pageAddr, pageSize) : freeSpace + " != " + actualFreeSpace(pageAddr, pageSize);
-
-        PageUtils.putShort(pageAddr, FREE_SPACE_OFF, (short)freeSpace);
-    }
-
-    /**
-     * Free space refers to a "max row size (without any data page specific overhead) which is
-     * guaranteed to fit into this data page".
-     *
-     * @param pageAddr Page address.
-     * @return Free space.
-     */
-    public int getFreeSpace(long pageAddr) {
-        if (getFreeItemSlots(pageAddr) == 0)
-            return 0;
-
-        int freeSpace = getRealFreeSpace(pageAddr);
-
-        // We reserve size here because of getFreeSpace() method semantics (see method javadoc).
-        // It means that we must be able to accommodate a row of size which is equal to getFreeSpace(),
-        // plus we will have data page overhead: header of the page as well as item, payload length and
-        // possibly a link to the next row fragment.
-        freeSpace -= ITEM_SIZE + PAYLOAD_LEN_SIZE + LINK_SIZE;
-
-        return freeSpace < 0 ? 0 : freeSpace;
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @return {@code true} If there is no useful data in this page.
-     */
-    public boolean isEmpty(long pageAddr) {
-        return getDirectCount(pageAddr) == 0;
-    }
-
-    /**
-     * Equivalent for {@link #actualFreeSpace(long, int)} but reads saved value.
-     *
-     * @param pageAddr Page address.
-     * @return Free space.
-     */
-    private int getRealFreeSpace(long pageAddr) {
-        return PageUtils.getShort(pageAddr, FREE_SPACE_OFF);
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param cnt Direct count.
-     */
-    private void setDirectCount(long pageAddr, int cnt) {
-        assert checkCount(cnt): cnt;
-
-        PageUtils.putByte(pageAddr, DIRECT_CNT_OFF, (byte)cnt);
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @return Direct count.
-     */
-    private int getDirectCount(long pageAddr) {
-        return PageUtils.getByte(pageAddr, DIRECT_CNT_OFF) & 0xFF;
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param c Closure.
-     * @param <T> Closure return type.
-     * @return Collection of closure results for all items in page.
-     * @throws IgniteCheckedException In case of error in closure body.
-     */
-    public <T> List<T> forAllItems(long pageAddr, CC<T> c) throws IgniteCheckedException {
-        long pageId = getPageId(pageAddr);
-
-        int cnt = getDirectCount(pageAddr);
-
-        List<T> res = new ArrayList<>(cnt);
-
-        for (int i = 0; i < cnt; i++) {
-            long link = PageIdUtils.link(pageId, i);
-
-            res.add(c.apply(link));
-        }
-
-        return res;
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param cnt Indirect count.
-     */
-    private void setIndirectCount(long pageAddr, int cnt) {
-        assert checkCount(cnt): cnt;
-
-        PageUtils.putByte(pageAddr, INDIRECT_CNT_OFF, (byte)cnt);
-    }
-
-    /**
-     * @param idx Index.
-     * @return {@code true} If the index is valid.
-     */
-    private boolean checkIndex(int idx) {
-        return idx >= 0 && idx < 0xFF;
-    }
-
-    /**
-     * @param cnt Counter value.
-     * @return {@code true} If the counter fits 1 byte.
-     */
-    private boolean checkCount(int cnt) {
-        return cnt >= 0 && cnt <= 0xFF;
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @return Indirect count.
-     */
-    private int getIndirectCount(long pageAddr) {
-        return PageUtils.getByte(pageAddr, INDIRECT_CNT_OFF) & 0xFF;
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @return Number of free entry slots.
-     */
-    private int getFreeItemSlots(long pageAddr) {
-        return 0xFF - getDirectCount(pageAddr);
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param itemId Fixed item ID (the index used for referencing an entry from the outside).
-     * @param directCnt Direct items count.
-     * @param indirectCnt Indirect items count.
-     * @return Found index of indirect item.
-     */
-    private int findIndirectItemIndex(long pageAddr, int itemId, int directCnt, int indirectCnt) {
-        int low = directCnt;
-        int high = directCnt + indirectCnt - 1;
-
-        while (low <= high) {
-            int mid = (low + high) >>> 1;
-
-            int cmp = Integer.compare(itemId(getItem(pageAddr, mid)), itemId);
-
-            if (cmp < 0)
-                low = mid + 1;
-            else if (cmp > 0)
-                high = mid - 1;
-            else
-                return mid; // found
-        }
-
-        throw new IllegalStateException("Item not found: " + itemId);
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param pageSize Page size.
-     * @return String representation.
-     */
-    private String printPageLayout(long pageAddr, int pageSize) {
-        SB b = new SB();
-
-        printPageLayout(pageAddr, pageSize, b);
-
-        return b.toString();
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param pageSize Page size.
-     * @param b B.
-     */
-    private void printPageLayout(long pageAddr, int pageSize, GridStringBuilder b) {
-        int directCnt = getDirectCount(pageAddr);
-        int indirectCnt = getIndirectCount(pageAddr);
-        int free = getRealFreeSpace(pageAddr);
-
-        boolean valid = directCnt >= indirectCnt;
-
-        b.appendHex(PageIO.getPageId(pageAddr)).a(" [");
-
-        int entriesSize = 0;
-
-        for (int i = 0; i < directCnt; i++) {
-            if (i != 0)
-                b.a(", ");
-
-            short item = getItem(pageAddr, i);
-
-            if (item < ITEMS_OFF || item >= pageSize)
-                valid = false;
-
-            entriesSize += getPageEntrySize(pageAddr, item, SHOW_PAYLOAD_LEN | SHOW_LINK);
-
-            b.a(item);
-        }
-
-        b.a("][");
-
-        Collection<Integer> set = new HashSet<>();
-
-        for (int i = directCnt; i < directCnt + indirectCnt; i++) {
-            if (i != directCnt)
-                b.a(", ");
-
-            short item = getItem(pageAddr, i);
-
-            int itemId = itemId(item);
-            int directIdx = directItemIndex(item);
-
-            if (!set.add(directIdx) || !set.add(itemId))
-                valid = false;
-
-            assert indirectItem(itemId, directIdx) == item;
-
-            if (itemId < directCnt || directIdx < 0 || directIdx >= directCnt)
-                valid = false;
-
-            if (i > directCnt && itemId(getItem(pageAddr, i - 1)) >= itemId)
-                valid = false;
-
-
-            b.a(itemId).a('^').a(directIdx);
-        }
-
-        b.a("][free=").a(free);
-
-        int actualFree = pageSize - ITEMS_OFF - (entriesSize + (directCnt + indirectCnt) * ITEM_SIZE);
-
-        if (free != actualFree) {
-            b.a(", actualFree=").a(actualFree);
-
-            valid = false;
-        }
-        else
-            b.a("]");
-
-        assert valid : b.toString();
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param itemId Fixed item ID (the index used for referencing an entry from the outside).
-     * @param pageSize Page size.
-     * @return Data entry offset in bytes.
-     */
-    private int getDataOffset(long pageAddr, int itemId, int pageSize) {
-        assert checkIndex(itemId): itemId;
-
-        int directCnt = getDirectCount(pageAddr);
-
-        assert directCnt > 0: "itemId=" + itemId + ", directCnt=" + directCnt + ", page=" + printPageLayout(pageAddr, pageSize);
-
-        if (itemId >= directCnt) { // Need to do indirect lookup.
-            int indirectCnt = getIndirectCount(pageAddr);
-
-            // Must have indirect items here.
-            assert indirectCnt > 0: "itemId=" + itemId + ", directCnt=" + directCnt + ", indirectCnt=" + indirectCnt +
-                ", page=" + printPageLayout(pageAddr, pageSize);
-
-            int indirectItemIdx = findIndirectItemIndex(pageAddr, itemId, directCnt, indirectCnt);
-
-            assert indirectItemIdx >= directCnt : indirectItemIdx + " " + directCnt;
-            assert indirectItemIdx < directCnt + indirectCnt: indirectItemIdx + " " + directCnt + " " + indirectCnt;
-
-            itemId = directItemIndex(getItem(pageAddr, indirectItemIdx));
-
-            assert itemId >= 0 && itemId < directCnt: itemId + " " + directCnt + " " + indirectCnt; // Direct item.
-        }
-
-        return directItemToOffset(getItem(pageAddr, itemId));
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param dataOff Points to the entry start.
-     * @return Link to the next entry fragment or 0 if no fragments left or if entry is not fragmented.
-     */
-    private long getNextFragmentLink(long pageAddr, int dataOff) {
-        assert isFragmented(pageAddr, dataOff);
-
-        return PageUtils.getLong(pageAddr, dataOff + PAYLOAD_LEN_SIZE);
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param dataOff Data offset.
-     * @return {@code true} If the data row is fragmented across multiple pages.
-     */
-    private boolean isFragmented(long pageAddr, int dataOff) {
-        return (PageUtils.getShort(pageAddr, dataOff) & FRAGMENTED_FLAG) != 0;
-    }
-
-    /**
-     * Sets position to start of actual fragment data and limit to it's end.
-     *
-     * @param pageAddr Page address.
-     * @param itemId Item to position on.
-     * @param pageSize Page size.
-     * @return {@link DataPagePayload} object.
-     */
-    public DataPagePayload readPayload(final long pageAddr, final int itemId, final int pageSize) {
-        int dataOff = getDataOffset(pageAddr, itemId, pageSize);
-
-        boolean fragmented = isFragmented(pageAddr, dataOff);
-        long nextLink = fragmented ? getNextFragmentLink(pageAddr, dataOff) : 0;
-        int payloadSize = getPageEntrySize(pageAddr, dataOff, 0);
-
-        return new DataPagePayload(dataOff + PAYLOAD_LEN_SIZE + (fragmented ? LINK_SIZE : 0),
-            payloadSize,
-            nextLink);
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param idx Item index.
-     * @return Item.
-     */
-    private short getItem(long pageAddr, int idx) {
-        return PageUtils.getShort(pageAddr, itemOffset(idx));
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param idx Item index.
-     * @param item Item.
-     */
-    private void setItem(long pageAddr, int idx, short item) {
-        PageUtils.putShort(pageAddr, itemOffset(idx), item);
-    }
-
-    /**
-     * @param idx Index of the item.
-     * @return Offset in buffer.
-     */
-    private int itemOffset(int idx) {
-        assert checkIndex(idx): idx;
-
-        return ITEMS_OFF + idx * ITEM_SIZE;
-    }
-
-    /**
-     * @param directItem Direct item.
-     * @return Offset of an entry payload inside of the page.
-     */
-    private int directItemToOffset(short directItem) {
-        return directItem & 0xFFFF;
-    }
-
-    /**
-     * @param dataOff Data offset.
-     * @return Direct item.
-     */
-    private short directItemFromOffset(int dataOff) {
-        assert dataOff >= ITEMS_OFF + ITEM_SIZE && dataOff < Short.MAX_VALUE: dataOff;
-
-        return (short)dataOff;
-    }
-
-    /**
-     * @param indirectItem Indirect item.
-     * @return Index of corresponding direct item.
-     */
-    private int directItemIndex(short indirectItem) {
-        return indirectItem & 0xFF;
-    }
-
-    /**
-     * @param indirectItem Indirect item.
-     * @return Fixed item ID (the index used for referencing an entry from the outside).
-     */
-    private int itemId(short indirectItem) {
-        return (indirectItem & 0xFFFF) >>> 8;
-    }
-
-    /**
-     * @param itemId Fixed item ID (the index used for referencing an entry from the outside).
-     * @param directItemIdx Index of corresponding direct item.
-     * @return Indirect item.
-     */
-    private short indirectItem(int itemId, int directItemIdx) {
-        assert checkIndex(itemId): itemId;
-        assert checkIndex(directItemIdx): directItemIdx;
-
-        return (short)((itemId << 8) | directItemIdx);
-    }
-
-    /**
-     * Move the last direct item to the free slot and reference it with indirect item on the same place.
-     *
-     * @param pageAddr Page address.
-     * @param freeDirectIdx Free slot.
-     * @param directCnt Direct items count.
-     * @param indirectCnt Indirect items count.
-     * @return {@code true} If the last direct item already had corresponding indirect item.
-     */
-    private boolean moveLastItem(long pageAddr, int freeDirectIdx, int directCnt, int indirectCnt) {
-        int lastIndirectId = findIndirectIndexForLastDirect(pageAddr, directCnt, indirectCnt);
-
-        int lastItemId = directCnt - 1;
-
-        assert lastItemId != freeDirectIdx;
-
-        short indirectItem = indirectItem(lastItemId, freeDirectIdx);
-
-        assert itemId(indirectItem) == lastItemId && directItemIndex(indirectItem) == freeDirectIdx;
-
-        setItem(pageAddr, freeDirectIdx, getItem(pageAddr, lastItemId));
-        setItem(pageAddr, lastItemId, indirectItem);
-
-        assert getItem(pageAddr, lastItemId) == indirectItem;
-
-        if (lastIndirectId != -1) { // Fix pointer to direct item.
-            setItem(pageAddr, lastIndirectId, indirectItem(itemId(getItem(pageAddr, lastIndirectId)), freeDirectIdx));
-
-            return true;
-        }
-
-        return false;
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param directCnt Direct items count.
-     * @param indirectCnt Indirect items count.
-     * @return Index of indirect item for the last direct item.
-     */
-    private int findIndirectIndexForLastDirect(long pageAddr, int directCnt, int indirectCnt) {
-        int lastDirectId = directCnt - 1;
-
-        for (int i = directCnt, end = directCnt + indirectCnt; i < end; i++) {
-            short item = getItem(pageAddr, i);
-
-            if (directItemIndex(item) == lastDirectId)
-                return i;
-        }
-
-        return -1;
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param itemId Item ID.
-     * @param pageSize Page size.
-     * @param payload Row data.
-     * @param row Row.
-     * @param rowSize Row size.
-     * @return {@code True} if entry is not fragmented.
-     * @throws IgniteCheckedException If failed.
-     */
-    public boolean updateRow(
-        final long pageAddr,
-        int itemId,
-        int pageSize,
-        @Nullable byte[] payload,
-        @Nullable CacheDataRow row,
-        final int rowSize) throws IgniteCheckedException {
-        assert checkIndex(itemId) : itemId;
-        assert row != null ^ payload != null;
-
-        final int dataOff = getDataOffset(pageAddr, itemId, pageSize);
-
-        if (isFragmented(pageAddr, dataOff))
-            return false;
-
-        if (row != null)
-            writeRowData(pageAddr, dataOff, rowSize, row, false);
-        else
-            writeRowData(pageAddr, dataOff, payload);
-
-        return true;
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param itemId Fixed item ID (the index used for referencing an entry from the outside).
-     * @param pageSize Page size.
-     * @return Next link for fragmented entries or {@code 0} if none.
-     * @throws IgniteCheckedException If failed.
-     */
-    public long removeRow(long pageAddr, int itemId, int pageSize) throws IgniteCheckedException {
-        assert checkIndex(itemId) : itemId;
-
-        final int dataOff = getDataOffset(pageAddr, itemId, pageSize);
-        final long nextLink = isFragmented(pageAddr, dataOff) ? getNextFragmentLink(pageAddr, dataOff) : 0;
-
-        // Record original counts to calculate delta in free space in the end of remove.
-        final int directCnt = getDirectCount(pageAddr);
-        final int indirectCnt = getIndirectCount(pageAddr);
-
-        int curIndirectCnt = indirectCnt;
-
-        assert directCnt > 0 : directCnt; // Direct count always represents overall number of live items.
-
-        // Remove the last item on the page.
-        if (directCnt == 1) {
-            assert (indirectCnt == 0 && itemId == 0) ||
-                (indirectCnt == 1 && itemId == itemId(getItem(pageAddr, 1))) : itemId;
-
-            setEmptyPage(pageAddr, pageSize);
-        }
-        else {
-            // Get the entry size before the actual remove.
-            int rmvEntrySize = getPageEntrySize(pageAddr, dataOff, SHOW_PAYLOAD_LEN | SHOW_LINK);
-
-            int indirectId = 0;
-
-            if (itemId >= directCnt) { // Need to remove indirect item.
-                assert indirectCnt > 0;
-
-                indirectId = findIndirectItemIndex(pageAddr, itemId, directCnt, indirectCnt);
-
-                assert indirectId >= directCnt;
-
-                itemId = directItemIndex(getItem(pageAddr, indirectId));
-
-                assert itemId < directCnt;
-            }
-
-            boolean dropLast = true;
-
-            if (itemId + 1 < directCnt) // It is not the last direct item.
-                dropLast = moveLastItem(pageAddr, itemId, directCnt, indirectCnt);
-
-            if (indirectId == 0) {// For the last direct item with no indirect item.
-                if (dropLast)
-                    moveItems(pageAddr, directCnt, indirectCnt, -1, pageSize);
-                else
-                    curIndirectCnt++;
-            }
-            else {
-                if (dropLast)
-                    moveItems(pageAddr, directCnt, indirectId - directCnt, -1, pageSize);
-
-                moveItems(pageAddr, indirectId + 1, directCnt + indirectCnt - indirectId - 1, dropLast ? -2 : -1, pageSize);
-
-                if (dropLast)
-                    curIndirectCnt--;
-            }
-
-            setIndirectCount(pageAddr, curIndirectCnt);
-            setDirectCount(pageAddr, directCnt - 1);
-
-            assert getIndirectCount(pageAddr) <= getDirectCount(pageAddr);
-
-            // Increase free space.
-            setRealFreeSpace(pageAddr,
-                getRealFreeSpace(pageAddr) + rmvEntrySize + ITEM_SIZE * (directCnt - getDirectCount(pageAddr) + indirectCnt - getIndirectCount(pageAddr)),
-                pageSize);
-        }
-
-        return nextLink;
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param idx Index.
-     * @param cnt Count.
-     * @param step Step.
-     * @param pageSize Page size.
-     */
-    private void moveItems(long pageAddr, int idx, int cnt, int step, int pageSize) {
-        assert cnt >= 0: cnt;
-
-        if (cnt != 0)
-            moveBytes(pageAddr, itemOffset(idx), cnt * ITEM_SIZE, step * ITEM_SIZE, pageSize);
-    }
-
-    /**
-     * @param newEntryFullSize New entry full size (with item, length and link).
-     * @param firstEntryOff First entry data offset.
-     * @param directCnt Direct items count.
-     * @param indirectCnt Indirect items count.
-     * @return {@code true} If there is enough space for the entry.
-     */
-    private boolean isEnoughSpace(int newEntryFullSize, int firstEntryOff, int directCnt, int indirectCnt) {
-        return ITEMS_OFF + ITEM_SIZE * (directCnt + indirectCnt) <= firstEntryOff - newEntryFullSize;
-    }
-
-    /**
-     * Adds row to this data page and sets respective link to the given row object.
-     *
-     * @param pageAddr Page address.
-     * @param row Cache data row.
-     * @param rowSize Row size.
-     * @param pageSize Page size.
-     * @throws IgniteCheckedException If failed.
-     */
-    public void addRow(
-        final long pageAddr,
-        CacheDataRow row,
-        final int rowSize,
-        final int pageSize
-    ) throws IgniteCheckedException {
-        assert rowSize <= getFreeSpace(pageAddr): "can't call addRow if not enough space for the whole row";
-
-        int fullEntrySize = getPageEntrySize(rowSize, SHOW_PAYLOAD_LEN | SHOW_ITEM);
-
-        int directCnt = getDirectCount(pageAddr);
-        int indirectCnt = getIndirectCount(pageAddr);
-
-        int dataOff = getDataOffsetForWrite(pageAddr, fullEntrySize, directCnt, indirectCnt, pageSize);
-
-        writeRowData(pageAddr, dataOff, rowSize, row, true);
-
-        int itemId = addItem(pageAddr, fullEntrySize, directCnt, indirectCnt, dataOff, pageSize);
-
-        setLink(row, pageAddr, itemId);
-    }
-
-    /**
-     * Adds row to this data page and sets respective link to the given row object.
-     *
-     * @param pageAddr Page address.
-     * @param payload Payload.
-     * @param pageSize Page size.
-     * @throws IgniteCheckedException If failed.
-     */
-    public void addRow(
-        long pageAddr,
-        byte[] payload,
-        int pageSize
-    ) throws IgniteCheckedException {
-        assert payload.length <= getFreeSpace(pageAddr): "can't call addRow if not enough space for the whole row";
-
-        int fullEntrySize = getPageEntrySize(payload.length, SHOW_PAYLOAD_LEN | SHOW_ITEM);
-
-        int directCnt = getDirectCount(pageAddr);
-        int indirectCnt = getIndirectCount(pageAddr);
-
-        int dataOff = getDataOffsetForWrite(pageAddr, fullEntrySize, directCnt, indirectCnt, pageSize);
-
-        writeRowData(pageAddr, dataOff, payload);
-
-        addItem(pageAddr, fullEntrySize, directCnt, indirectCnt, dataOff, pageSize);
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param entryFullSize New entry full size (with item, length and link).
-     * @param directCnt Direct items count.
-     * @param indirectCnt Indirect items count.
-     * @param dataOff First entry offset.
-     * @param pageSize Page size.
-     * @return First entry offset after compaction.
-     */
-    private int compactIfNeed(
-        final long pageAddr,
-        final int entryFullSize,
-        final int directCnt,
-        final int indirectCnt,
-        int dataOff,
-        int pageSize
-    ) {
-        if (!isEnoughSpace(entryFullSize, dataOff, directCnt, indirectCnt)) {
-            dataOff = compactDataEntries(pageAddr, directCnt, pageSize);
-
-            assert isEnoughSpace(entryFullSize, dataOff, directCnt, indirectCnt);
-        }
-
-        return dataOff;
-    }
-
-    /**
-     * Put item reference on entry.
-     *
-     * @param pageAddr Page address.
-     * @param fullEntrySize Full entry size (with link, payload size and item).
-     * @param directCnt Direct items count.
-     * @param indirectCnt Indirect items count.
-     * @param dataOff Data offset.
-     * @param pageSize Page size.
-     * @return Item ID.
-     */
-    private int addItem(final long pageAddr,
-        final int fullEntrySize,
-        final int directCnt,
-        final int indirectCnt,
-        final int dataOff,
-        final int pageSize)
-    {
-        setFirstEntryOffset(pageAddr, dataOff, pageSize);
-
-        int itemId = insertItem(pageAddr, dataOff, directCnt, indirectCnt, pageSize);
-
-        assert checkIndex(itemId): itemId;
-        assert getIndirectCount(pageAddr) <= getDirectCount(pageAddr);
-
-        // Update free space. If number of indirect items changed, then we were able to reuse an item slot.
-        setRealFreeSpace(pageAddr,
-            getRealFreeSpace(pageAddr) - fullEntrySize + (getIndirectCount(pageAddr) != indirectCnt ? ITEM_SIZE : 0),
-            pageSize);
-
-        return itemId;
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param fullEntrySize Full entry size.
-     * @param directCnt Direct items count.
-     * @param indirectCnt Indirect items count.
-     * @param pageSize Page size.
-     * @return Offset in the buffer where the entry must be written.
-     */
-    private int getDataOffsetForWrite(long pageAddr, int fullEntrySize, int directCnt, int indirectCnt, int pageSize) {
-        int dataOff = getFirstEntryOffset(pageAddr);
-
-        // Compact if we do not have enough space for entry.
-        dataOff = compactIfNeed(pageAddr, fullEntrySize, directCnt, indirectCnt, dataOff, pageSize);
-
-        // We will write data right before the first entry.
-        dataOff -= fullEntrySize - ITEM_SIZE;
-
-        return dataOff;
-    }
-
-    /**
-     * Adds maximum possible fragment of the given row to this data page and sets respective link to the row.
-     *
-     * @param pageMem Page memory.
-     * @param pageAddr Page address.
-     * @param row Cache data row.
-     * @param written Number of bytes of row size that was already written.
-     * @param rowSize Row size.
-     * @param pageSize Page size.
-     * @return Written payload size.
-     * @throws IgniteCheckedException If failed.
-     */
-    public int addRowFragment(
-        PageMemory pageMem,
-        long pageAddr,
-        CacheDataRow row,
-        int written,
-        int rowSize,
-        int pageSize
-    ) throws IgniteCheckedException {
-        return addRowFragment(pageMem, pageAddr, written, rowSize, row.link(), row, null, pageSize);
-    }
-
-    /**
-     * Adds this payload as a fragment to this data page.
-     *
-     * @param pageAddr Page address.
-     * @param payload Payload bytes.
-     * @param lastLink Link to the previous written fragment (link to the tail).
-     * @param pageSize Page size.
-     * @throws IgniteCheckedException If failed.
-     */
-    public void addRowFragment(
-        long pageAddr,
-        byte[] payload,
-        long lastLink,
-        int pageSize
-    ) throws IgniteCheckedException {
-        addRowFragment(null, pageAddr, 0, 0, lastLink, null, payload, pageSize);
-    }
-
-    /**
-     * Adds maximum possible fragment of the given row to this data page and sets respective link to the row.
-     *
-     * @param pageMem Page memory.
-     * @param pageAddr Page address.
-     * @param written Number of bytes of row size that was already written.
-     * @param rowSize Row size.
-     * @param lastLink Link to the previous written fragment (link to the tail).
-     * @param row Row.
-     * @param payload Payload bytes.
-     * @param pageSize Page size.
-     * @return Written payload size.
-     * @throws IgniteCheckedException If failed.
-     */
-    private int addRowFragment(
-        PageMemory pageMem,
-        long pageAddr,
-        int written,
-        int rowSize,
-        long lastLink,
-        CacheDataRow row,
-        byte[] payload,
-        int pageSize
-    ) throws IgniteCheckedException {
-        assert payload == null ^ row == null;
-
-        int directCnt = getDirectCount(pageAddr);
-        int indirectCnt = getIndirectCount(pageAddr);
-
-        int payloadSize = payload != null ? payload.length :
-            Math.min(rowSize - written, getFreeSpace(pageAddr));
-
-        int fullEntrySize = getPageEntrySize(payloadSize, SHOW_PAYLOAD_LEN | SHOW_LINK | SHOW_ITEM);
-        int dataOff = getDataOffsetForWrite(pageAddr, fullEntrySize, directCnt, indirectCnt, pageSize);
-
-        if (payload == null) {
-            ByteBuffer buf = pageMem.pageBuffer(pageAddr);
-
-            buf.position(dataOff);
-
-            short p = (short)(payloadSize | FRAGMENTED_FLAG);
-
-            buf.putShort(p);
-            buf.putLong(lastLink);
-
-            int rowOff = rowSize - written - payloadSize;
-
-            writeFragmentData(row, buf, rowOff, payloadSize);
-        }
-        else {
-            PageUtils.putShort(pageAddr, dataOff, (short)(payloadSize | FRAGMENTED_FLAG));
-
-            PageUtils.putLong(pageAddr, dataOff + 2, lastLink);
-
-            PageUtils.putBytes(pageAddr, dataOff + 10, payload);
-        }
-
-        int itemId = addItem(pageAddr, fullEntrySize, directCnt, indirectCnt, dataOff, pageSize);
-
-        if (row != null)
-            setLink(row, pageAddr, itemId);
-
-        return payloadSize;
-    }
-
-    /**
-     * @param row Row to set link to.
-     * @param pageAddr Page address.
-     * @param itemId Item ID.
-     */
-    private void setLink(CacheDataRow row, long pageAddr, int itemId) {
-        row.link(PageIdUtils.link(getPageId(pageAddr), itemId));
-    }
-
-    /**
-     * Write row data fragment.
-     *
-     * @param row Row.
-     * @param buf Byte buffer.
-     * @param rowOff Offset in row data bytes.
-     * @param payloadSize Data length that should be written in a fragment.
-     * @throws IgniteCheckedException If failed.
-     */
-    private void writeFragmentData(
+    @Override
+    protected void writeFragmentData(
         final CacheDataRow row,
         final ByteBuffer buf,
         final int rowOff,
@@ -1230,169 +241,9 @@
         CACHE_ID
     }
 
-    /**
-     * @param pageAddr Page address.
-     * @param dataOff Data offset.
-     * @param directCnt Direct items count.
-     * @param indirectCnt Indirect items count.
-     * @param pageSize Page size.
-     * @return Item ID (insertion index).
-     */
-    private int insertItem(long pageAddr, int dataOff, int directCnt, int indirectCnt, int pageSize) {
-        if (indirectCnt > 0) {
-            // If the first indirect item is on correct place to become the last direct item, do the transition
-            // and insert the new item into the free slot which was referenced by this first indirect item.
-            short item = getItem(pageAddr, directCnt);
-
-            if (itemId(item) == directCnt) {
-                int directItemIdx = directItemIndex(item);
-
-                setItem(pageAddr, directCnt, getItem(pageAddr, directItemIdx));
-                setItem(pageAddr, directItemIdx, directItemFromOffset(dataOff));
-
-                setDirectCount(pageAddr, directCnt + 1);
-                setIndirectCount(pageAddr, indirectCnt - 1);
-
-                return directItemIdx;
-            }
-        }
-
-        // Move all the indirect items forward to make a free slot and insert new item at the end of direct items.
-        moveItems(pageAddr, directCnt, indirectCnt, +1, pageSize);
-
-        setItem(pageAddr, directCnt, directItemFromOffset(dataOff));
-
-        setDirectCount(pageAddr, directCnt + 1);
-        assert getDirectCount(pageAddr) == directCnt + 1;
-
-        return directCnt; // Previous directCnt will be our itemId.
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param directCnt Direct items count.
-     * @param pageSize Page size.
-     * @return New first entry offset.
-     */
-    private int compactDataEntries(long pageAddr, int directCnt, int pageSize) {
-        assert checkCount(directCnt): directCnt;
-
-        int[] offs = new int[directCnt];
-
-        for (int i = 0; i < directCnt; i++) {
-            int off = directItemToOffset(getItem(pageAddr, i));
-
-            offs[i] = (off << 8) | i; // This way we'll be able to sort by offset using Arrays.sort(...).
-        }
-
-        Arrays.sort(offs);
-
-        // Move right all of the entries if possible to make the page as compact as possible to its tail.
-        int prevOff = pageSize;
-
-        final int start = directCnt - 1;
-        int curOff = offs[start] >>> 8;
-        int curEntrySize = getPageEntrySize(pageAddr, curOff, SHOW_PAYLOAD_LEN | SHOW_LINK);
-
-        for (int i = start; i >= 0; i--) {
-            assert curOff < prevOff : curOff;
-
-            int delta = prevOff - (curOff + curEntrySize);
-
-            int off = curOff;
-            int entrySize = curEntrySize;
-
-            if (delta != 0) { // Move right.
-                assert delta > 0: delta;
-
-                int itemId = offs[i] & 0xFF;
-
-                setItem(pageAddr, itemId, directItemFromOffset(curOff + delta));
-
-                for (int j = i - 1; j >= 0; j--) {
-                    int offNext = offs[j] >>> 8;
-                    int nextSize = getPageEntrySize(pageAddr, offNext, SHOW_PAYLOAD_LEN | SHOW_LINK);
-
-                    if (offNext + nextSize == off) {
-                        i--;
-
-                        off = offNext;
-                        entrySize += nextSize;
-
-                        itemId = offs[j] & 0xFF;
-                        setItem(pageAddr, itemId, directItemFromOffset(offNext + delta));
-                    }
-                    else {
-                        curOff = offNext;
-                        curEntrySize = nextSize;
-
-                        break;
-                    }
-                }
-
-                moveBytes(pageAddr, off, entrySize, delta, pageSize);
-
-                off += delta;
-            }
-            else if (i > 0) {
-                curOff = offs[i - 1] >>> 8;
-                curEntrySize = getPageEntrySize(pageAddr, curOff, SHOW_PAYLOAD_LEN | SHOW_LINK);
-            }
-
-            prevOff = off;
-        }
-
-        return prevOff;
-    }
-
-    /**
-     * Full-scan free space calculation procedure.
-     *
-     * @param pageAddr Page to scan.
-     * @param pageSize Page size.
-     * @return Actual free space in the buffer.
-     */
-    private int actualFreeSpace(long pageAddr, int pageSize) {
-        int directCnt = getDirectCount(pageAddr);
-
-        int entriesSize = 0;
-
-        for (int i = 0; i < directCnt; i++) {
-            int off = directItemToOffset(getItem(pageAddr, i));
-
-            int entrySize = getPageEntrySize(pageAddr, off, SHOW_PAYLOAD_LEN | SHOW_LINK);
-
-            entriesSize += entrySize;
-        }
-
-        return pageSize - ITEMS_OFF - entriesSize - (directCnt + getIndirectCount(pageAddr)) * ITEM_SIZE;
-    }
-
-    /**
-     * @param addr Address.
-     * @param off Offset.
-     * @param cnt Count.
-     * @param step Step.
-     * @param pageSize Page size.
-     */
-    private void moveBytes(long addr, int off, int cnt, int step, int pageSize) {
-        assert step != 0: step;
-        assert off + step >= 0;
-        assert off + step + cnt <= pageSize : "[off=" + off + ", step=" + step + ", cnt=" + cnt +
-            ", cap=" + pageSize + ']';
-
-        PageHandler.copyMemory(addr, off, addr, off + step, cnt);
-    }
-
-    /**
-     * @param pageAddr Page address.
-     * @param dataOff Data offset.
-     * @param payloadSize Payload size.
-     * @param row Data row.
-     * @param newRow {@code False} if existing cache entry is updated, in this case skip key data write.
-     * @throws IgniteCheckedException If failed.
-     */
-    private void writeRowData(
+    /** {@inheritDoc} */
+    @Override
+    protected void writeRowData(
         long pageAddr,
         int dataOff,
         int payloadSize,
@@ -1426,12 +277,9 @@
         PageUtils.putLong(addr, 0, row.expireTime());
     }
 
-    /**
-     * @param pageAddr Page address.
-     * @param dataOff Data offset.
-     * @param payload Payload
-     */
-    private void writeRowData(
+    /** {@inheritDoc} */
+    @Override
+    protected void writeRowData(
         long pageAddr,
         int dataOff,
         byte[] payload
@@ -1443,6 +291,11 @@
     }
 
     /** {@inheritDoc} */
+    @Override public int getRowSize(CacheDataRow row) throws IgniteCheckedException {
+        return getRowSize(row, row.cacheId() != 0);
+    }
+
+    /** {@inheritDoc} */
     @Override protected void printPage(long addr, int pageSize, GridStringBuilder sb) throws IgniteCheckedException {
         sb.a("DataPageIO [\n");
         printPageLayout(addr, pageSize, sb);
@@ -1450,18 +303,18 @@
     }
 
     /**
-     * Defines closure interface for applying computations to data page items.
-     *
-     * @param <T> Closure return type.
+     * @param row Row.
+     * @param withCacheId If {@code true} adds cache ID size.
+     * @return Entry size on page.
+     * @throws IgniteCheckedException If failed.
      */
-    public interface CC<T> {
-        /**
-         * Closure body.
-         *
-         * @param link Link to item.
-         * @return Closure return value.
-         * @throws IgniteCheckedException In case of error in closure body.
-         */
-        public T apply(long link) throws IgniteCheckedException;
+    public static int getRowSize(CacheDataRow row, boolean withCacheId) throws IgniteCheckedException {
+        KeyCacheObject key = row.key();
+        CacheObject val = row.value();
+
+        int keyLen = key.valueBytesLength(null);
+        int valLen = val.valueBytesLength(null);
+
+        return keyLen + valLen + CacheVersionIO.size(row.version(), false) + 8 + (withCacheId ? 4 : 0);
     }
 }
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageIO.java
index 60b1aaf..a5236c2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageIO.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageIO.java
@@ -25,9 +25,10 @@
 import org.apache.ignite.internal.pagemem.PageMemory;
 import org.apache.ignite.internal.pagemem.PageUtils;
 import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
-import org.apache.ignite.internal.processors.cache.persistence.MetadataStorage;
+import org.apache.ignite.internal.processors.cache.persistence.IndexStorageImpl;
 import org.apache.ignite.internal.processors.cache.persistence.freelist.io.PagesListMetaIO;
 import org.apache.ignite.internal.processors.cache.persistence.freelist.io.PagesListNodeIO;
+import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetastorageTree;
 import org.apache.ignite.internal.processors.cache.persistence.tree.util.PageHandler;
 import org.apache.ignite.internal.processors.cache.persistence.tree.util.PageLockListener;
 import org.apache.ignite.internal.processors.cache.tree.CacheIdAwareDataInnerIO;
@@ -181,6 +182,16 @@
     /** */
     public static final short T_PART_CNTRS = 20;
 
+    /** */
+    public static final short T_DATA_METASTORAGE = 21;
+
+    /** */
+    public static final short T_DATA_REF_METASTORAGE_INNER = 22;
+
+    /** */
+    public static final short T_DATA_REF_METASTORAGE_LEAF = 23;
+
+
     /** Index for payload == 1. */
     public static final short T_H2_EX_REF_LEAF_START = 10000;
 
@@ -460,6 +471,9 @@
             case T_PAGE_UPDATE_TRACKING:
                 return (Q)TrackingPageIO.VERSIONS.forVersion(ver);
 
+            case T_DATA_METASTORAGE:
+                return (Q)SimpleDataPageIO.VERSIONS.forVersion(ver);
+
             default:
                 return (Q)getBPlusIO(type, ver);
         }
@@ -518,10 +532,10 @@
                 return (Q)CacheIdAwareDataLeafIO.VERSIONS.forVersion(ver);
 
             case T_METASTORE_INNER:
-                return (Q)MetadataStorage.MetaStoreInnerIO.VERSIONS.forVersion(ver);
+                return (Q)IndexStorageImpl.MetaStoreInnerIO.VERSIONS.forVersion(ver);
 
             case T_METASTORE_LEAF:
-                return (Q)MetadataStorage.MetaStoreLeafIO.VERSIONS.forVersion(ver);
+                return (Q)IndexStorageImpl.MetaStoreLeafIO.VERSIONS.forVersion(ver);
 
             case T_PENDING_REF_INNER:
                 return (Q)PendingEntryInnerIO.VERSIONS.forVersion(ver);
@@ -535,6 +549,12 @@
             case T_CACHE_ID_AWARE_PENDING_REF_LEAF:
                 return (Q)CacheIdAwarePendingEntryLeafIO.VERSIONS.forVersion(ver);
 
+            case T_DATA_REF_METASTORAGE_INNER:
+                return (Q)MetastorageTree.MetastorageInnerIO.VERSIONS.forVersion(ver);
+
+            case T_DATA_REF_METASTORAGE_LEAF:
+                return (Q)MetastorageTree.MetastoreLeafIO.VERSIONS.forVersion(ver);
+
             default:
                 // For tests.
                 if (innerTestIO != null && innerTestIO.getType() == type && innerTestIO.getVersion() == ver)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/SimpleDataPageIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/SimpleDataPageIO.java
new file mode 100644
index 0000000..6fd201d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/SimpleDataPageIO.java
@@ -0,0 +1,127 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.tree.io;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.pagemem.PageUtils;
+import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetastorageDataRow;
+import org.apache.ignite.internal.util.GridStringBuilder;
+
+/**
+ * Data pages IO for Metastorage.
+ */
+public class SimpleDataPageIO extends AbstractDataPageIO<MetastorageDataRow> {
+    /** */
+    public static final IOVersions<SimpleDataPageIO> VERSIONS = new IOVersions<>(
+        new SimpleDataPageIO(1)
+    );
+
+    /**
+     * @param ver Page format version.
+     */
+    public SimpleDataPageIO(int ver) {
+        super(T_DATA_METASTORAGE, ver);
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    protected void writeFragmentData(
+        final MetastorageDataRow row,
+        final ByteBuffer buf,
+        final int rowOff,
+        final int payloadSize
+    ) throws IgniteCheckedException {
+        int written = writeSizeFragment(row, buf, rowOff, payloadSize);
+
+        if (payloadSize == written)
+            return;
+
+        int start = rowOff > 2 ? rowOff - 2 : 0;
+
+        final int len = Math.min(row.value().length - start, payloadSize - written);
+
+        if (len > 0) {
+            buf.put(row.value(), start, len);
+            written += len;
+        }
+
+        assert written == payloadSize;
+    }
+
+    /** */
+    private int writeSizeFragment(final MetastorageDataRow row, final ByteBuffer buf, final int rowOff,
+        final int payloadSize) {
+        final int size = 2;
+
+        if (rowOff >= size)
+            return 0;
+
+        if (rowOff == 0 && payloadSize >= size) {
+            buf.putShort((short)row.value().length);
+
+            return size;
+        }
+
+        ByteBuffer buf2 = ByteBuffer.allocate(size);
+        buf2.order(buf.order());
+
+        buf2.putShort((short)row.value().length);
+        int len = Math.min(size - rowOff, payloadSize);
+        buf.put(buf2.array(), rowOff, len);
+
+        return len;
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    protected void writeRowData(
+        long pageAddr,
+        int dataOff,
+        int payloadSize,
+        MetastorageDataRow row,
+        boolean newRow
+    ) throws IgniteCheckedException {
+        long addr = pageAddr + dataOff;
+
+        if (newRow)
+            PageUtils.putShort(addr, 0, (short)payloadSize);
+
+        PageUtils.putShort(addr, 2, (short)row.value().length);
+        PageUtils.putBytes(addr, 4, row.value());
+    }
+
+    public static byte[] readPayload(long link) {
+        int size = PageUtils.getShort(link, 0);
+
+        return PageUtils.getBytes(link, 2, size);
+    }
+
+    /** {@inheritDoc} */
+    @Override public int getRowSize(MetastorageDataRow row) throws IgniteCheckedException {
+        return 2 + row.value().length;
+    }
+
+
+    /** {@inheritDoc} */
+    @Override protected void printPage(long addr, int pageSize, GridStringBuilder sb) throws IgniteCheckedException {
+        sb.a("SimpleDataPageIO [\n");
+        printPageLayout(addr, pageSize, sb);
+        sb.a("\n]");
+    }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/IgniteWalRecoveryTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/IgniteWalRecoveryTest.java
index bf8cd85..a729e71 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/IgniteWalRecoveryTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/IgniteWalRecoveryTest.java
@@ -70,6 +70,8 @@
 import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
 import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
 import org.apache.ignite.internal.processors.cache.persistence.filename.PdsConsistentIdProcessor;
+import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage;
+import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetastorageDataRow;
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryEx;
 import org.apache.ignite.internal.processors.cache.persistence.tree.io.TrackingPageIO;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
@@ -895,6 +897,177 @@
     }
 
     /**
+     * @throws Exception If fail.
+     */
+    public void testMetastorage() throws Exception {
+        try {
+            int cnt = 5000;
+
+            IgniteEx ignite0 = (IgniteEx)startGrid("node1");
+            IgniteEx ignite1 = (IgniteEx)startGrid("node2");
+
+            ignite1.active(true);
+
+            GridCacheSharedContext<Object, Object> sharedCtx0 = ignite0.context().cache().context();
+            GridCacheSharedContext<Object, Object> sharedCtx1 = ignite1.context().cache().context();
+
+            MetaStorage storage0 = ((GridCacheDatabaseSharedManager)sharedCtx0.database()).metaStorage();
+            MetaStorage storage1 = ((GridCacheDatabaseSharedManager)sharedCtx1.database()).metaStorage();
+
+            assert storage0 != null;
+
+            for (int i = 0; i < cnt; i++) {
+                storage0.putData(String.valueOf(i), new byte[] {(byte)(i % 256), 2, 3});
+
+                byte[] b1 = new byte[i + 3];
+                b1[0] = 1;
+                b1[1] = 2;
+                b1[2] = 3;
+                storage1.putData(String.valueOf(i), b1);
+            }
+
+            for (int i = 0; i < cnt; i++) {
+                byte[] d1 = storage0.getData(String.valueOf(i)).value();
+                assertEquals(3, d1.length);
+                assertEquals((byte)(i % 256), d1[0]);
+                assertEquals(2, d1[1]);
+                assertEquals(3, d1[2]);
+
+                byte[] d2 = storage1.getData(String.valueOf(i)).value();
+                assertEquals(i + 3, d2.length);
+                assertEquals(1, d2[0]);
+                assertEquals(2, d2[1]);
+                assertEquals(3, d2[2]);
+            }
+
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
+
+    /**
+     * @throws Exception If fail.
+     */
+    public void testMetastorageRemove() throws Exception {
+        try {
+            int cnt = 400;
+
+            IgniteEx ignite0 = (IgniteEx)startGrid("node1");
+
+            ignite0.active(true);
+
+            GridCacheSharedContext<Object, Object> sharedCtx0 = ignite0.context().cache().context();
+
+            MetaStorage storage = ((GridCacheDatabaseSharedManager)sharedCtx0.database()).metaStorage();
+
+            assert storage != null;
+
+            for (int i = 0; i < cnt; i++)
+                storage.putData(String.valueOf(i), new byte[] {1, 2, 3});
+
+            for (int i = 0; i < 10; i++)
+                storage.removeData(String.valueOf(i));
+
+            for (int i = 10; i < cnt; i++) {
+                byte[] d1 = storage.getData(String.valueOf(i)).value();
+                assertEquals(3, d1.length);
+                assertEquals(1, d1[0]);
+                assertEquals(2, d1[1]);
+                assertEquals(3, d1[2]);
+            }
+
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
+
+    /**
+     * @throws Exception If fail.
+     */
+    public void testMetastorageUpdate() throws Exception {
+        try {
+            int cnt = 2000;
+
+            IgniteEx ignite0 = (IgniteEx)startGrid("node1");
+
+            ignite0.active(true);
+
+            GridCacheSharedContext<Object, Object> sharedCtx0 = ignite0.context().cache().context();
+
+            MetaStorage storage = ((GridCacheDatabaseSharedManager)sharedCtx0.database()).metaStorage();
+
+            assert storage != null;
+
+            for (int i = 0; i < cnt; i++)
+                storage.putData(String.valueOf(i), new byte[] {1, 2, 3});
+
+            for (int i = 0; i < cnt; i++)
+                storage.putData(String.valueOf(i), new byte[] {2, 2, 3, 4});
+
+            for (int i = 0; i < cnt; i++) {
+                byte[] d1 = storage.getData(String.valueOf(i)).value();
+                assertEquals(4, d1.length);
+                assertEquals(2, d1[0]);
+                assertEquals(2, d1[1]);
+                assertEquals(3, d1[2]);
+            }
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
+
+    /**
+     * @throws Exception If fail.
+     */
+    public void testMetastorageWalRestore() throws Exception {
+        try {
+            int cnt = 2000;
+
+            IgniteEx ignite0 = (IgniteEx)startGrid(0);
+
+            ignite0.active(true);
+
+            GridCacheSharedContext<Object, Object> sharedCtx0 = ignite0.context().cache().context();
+
+            MetaStorage storage = ((GridCacheDatabaseSharedManager)sharedCtx0.database()).metaStorage();
+
+            assert storage != null;
+
+            for (int i = 0; i < cnt; i++)
+                storage.putData(String.valueOf(i), new byte[] {1, 2, 3});
+
+            for (int i = 0; i < cnt; i++) {
+                MetastorageDataRow row = storage.getData(String.valueOf(i));
+                assert row != null;
+                assert row.value().length == 3;
+            }
+
+            stopGrid(0);
+
+            ignite0 = (IgniteEx)startGrid(0);
+
+            ignite0.active(true);
+
+            sharedCtx0 = ignite0.context().cache().context();
+
+            storage = ((GridCacheDatabaseSharedManager)sharedCtx0.database()).metaStorage();
+
+            assert storage != null;
+
+            for (int i = 0; i < cnt; i++) {
+                MetastorageDataRow row = storage.getData(String.valueOf(i));
+                assert row != null;
+            }
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
+
+    /**
      * @throws Exception if failed.
      */
     public void testApplyDeltaRecords() throws Exception {
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/WalRecoveryTxLogicalRecordsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/WalRecoveryTxLogicalRecordsTest.java
index f5d46e2..61edc56 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/WalRecoveryTxLogicalRecordsTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/WalRecoveryTxLogicalRecordsTest.java
@@ -48,13 +48,13 @@
 import org.apache.ignite.internal.processors.cache.GridCacheContext;
 import org.apache.ignite.internal.processors.cache.IgniteCacheOffheapManager;
 import org.apache.ignite.internal.processors.cache.IgniteRebalanceIterator;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
 import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
 import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
 import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager;
-import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeListImpl;
+import org.apache.ignite.internal.processors.cache.persistence.freelist.CacheFreeListImpl;
 import org.apache.ignite.internal.processors.cache.persistence.freelist.PagesList;
 import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseListImpl;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.T2;
 import org.apache.ignite.internal.util.typedef.internal.CU;
@@ -897,14 +897,14 @@
         boolean foundTails = false;
 
         for (GridDhtLocalPartition part : parts) {
-            FreeListImpl freeList = GridTestUtils.getFieldValue(part.dataStore(), "freeList");
+            CacheFreeListImpl freeList = GridTestUtils.getFieldValue(part.dataStore(), "freeList");
 
             if (freeList == null)
                 // Lazy store.
                 continue;
 
             AtomicReferenceArray<PagesList.Stripe[]> buckets = GridTestUtils.getFieldValue(freeList,
-                FreeListImpl.class, "buckets");
+                CacheFreeListImpl.class, "buckets");
             //AtomicIntegerArray cnts = GridTestUtils.getFieldValue(freeList, PagesList.class, "cnts");
 
             assertNotNull(buckets);
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/MetadataStoragePageMemoryImplTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IndexStoragePageMemoryImplTest.java
similarity index 96%
rename from modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/MetadataStoragePageMemoryImplTest.java
rename to modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IndexStoragePageMemoryImplTest.java
index a427c63..fdcc2c5 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/MetadataStoragePageMemoryImplTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IndexStoragePageMemoryImplTest.java
@@ -28,7 +28,7 @@
 import org.apache.ignite.internal.processors.cache.persistence.CheckpointLockStateChecker;
 import org.apache.ignite.internal.processors.cache.persistence.IgniteCacheDatabaseSharedManager;
 import org.apache.ignite.internal.processors.cache.persistence.MemoryMetricsImpl;
-import org.apache.ignite.internal.processors.database.MetadataStorageSelfTest;
+import org.apache.ignite.internal.processors.database.IndexStorageSelfTest;
 import org.apache.ignite.internal.util.lang.GridInClosure3X;
 import org.apache.ignite.internal.util.typedef.CIX3;
 import org.apache.ignite.internal.util.typedef.internal.U;
@@ -37,7 +37,7 @@
 /**
  *
  */
-public class MetadataStoragePageMemoryImplTest extends MetadataStorageSelfTest{
+public class IndexStoragePageMemoryImplTest extends IndexStorageSelfTest {
     /** Make sure page is small enough to trigger multiple pages in a linked list. */
     public static final int PAGE_SIZE = 1024;
 
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/NoOpPageStoreManager.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/NoOpPageStoreManager.java
index 40887e8c..8f72ed6 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/NoOpPageStoreManager.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/NoOpPageStoreManager.java
@@ -59,6 +59,11 @@
     }
 
     /** {@inheritDoc} */
+    @Override public void initializeForMetastorage() throws IgniteCheckedException {
+        // No-op.
+    }
+
+    /** {@inheritDoc} */
     @Override public void shutdownForCacheGroup(CacheGroupContext grp, boolean destroy) throws IgniteCheckedException {
         // No-op.
     }
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/database/FreeListImplSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/database/CacheFreeListImplSelfTest.java
similarity index 98%
rename from modules/core/src/test/java/org/apache/ignite/internal/processors/database/FreeListImplSelfTest.java
rename to modules/core/src/test/java/org/apache/ignite/internal/processors/database/CacheFreeListImplSelfTest.java
index c190b1d..bf11764 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/database/FreeListImplSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/database/CacheFreeListImplSelfTest.java
@@ -43,8 +43,8 @@
 import org.apache.ignite.internal.processors.cache.persistence.MemoryMetricsImpl;
 import org.apache.ignite.internal.processors.cache.persistence.MemoryPolicy;
 import org.apache.ignite.internal.processors.cache.persistence.evict.NoOpPageEvictionTracker;
+import org.apache.ignite.internal.processors.cache.persistence.freelist.CacheFreeListImpl;
 import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeList;
-import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeListImpl;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
 import org.apache.ignite.plugin.extensions.communication.MessageReader;
 import org.apache.ignite.plugin.extensions.communication.MessageWriter;
@@ -55,7 +55,7 @@
 /**
  *
  */
-public class FreeListImplSelfTest extends GridCommonAbstractTest {
+public class CacheFreeListImplSelfTest extends GridCommonAbstractTest {
     /** */
     private static final int CPUS = Runtime.getRuntime().availableProcessors();
 
@@ -347,7 +347,7 @@
 
         MemoryPolicy memPlc = new MemoryPolicy(pageMem, plcCfg, metrics, new NoOpPageEvictionTracker());
 
-        return new FreeListImpl(1, "freelist", metrics, memPlc, null, null, metaPageId, true);
+        return new CacheFreeListImpl(1, "freelist", metrics, memPlc, null, null, metaPageId, true);
     }
 
     /**
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/database/MetadataStorageSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/database/IndexStorageSelfTest.java
similarity index 93%
rename from modules/core/src/test/java/org/apache/ignite/internal/processors/database/MetadataStorageSelfTest.java
rename to modules/core/src/test/java/org/apache/ignite/internal/processors/database/IndexStorageSelfTest.java
index dcd4ce1..795cd5a 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/database/MetadataStorageSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/database/IndexStorageSelfTest.java
@@ -25,13 +25,13 @@
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.ignite.configuration.MemoryPolicyConfiguration;
 import org.apache.ignite.internal.mem.DirectMemoryProvider;
+import org.apache.ignite.internal.mem.file.MappedFileMemoryProvider;
 import org.apache.ignite.internal.pagemem.FullPageId;
 import org.apache.ignite.internal.pagemem.PageIdAllocator;
-import org.apache.ignite.internal.pagemem.impl.PageMemoryNoStoreImpl;
-import org.apache.ignite.internal.processors.cache.persistence.MemoryMetricsImpl;
-import org.apache.ignite.internal.processors.cache.persistence.MetadataStorage;
-import org.apache.ignite.internal.mem.file.MappedFileMemoryProvider;
 import org.apache.ignite.internal.pagemem.PageMemory;
+import org.apache.ignite.internal.pagemem.impl.PageMemoryNoStoreImpl;
+import org.apache.ignite.internal.processors.cache.persistence.IndexStorageImpl;
+import org.apache.ignite.internal.processors.cache.persistence.MemoryMetricsImpl;
 import org.apache.ignite.internal.processors.cache.persistence.RootPage;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
@@ -39,7 +39,7 @@
 /**
  *
  */
-public class MetadataStorageSelfTest extends GridCommonAbstractTest {
+public class IndexStorageSelfTest extends GridCommonAbstractTest {
     /** Make sure page is small enough to trigger multiple pages in a linked list. */
     private static final int PAGE_SIZE = 1024;
 
@@ -74,7 +74,7 @@
         mem.start();
 
         try {
-            final Map<Integer, MetadataStorage> storeMap = new HashMap<>();
+            final Map<Integer, IndexStorageImpl> storeMap = new HashMap<>();
 
             for (int i = 0; i < 1_000; i++) {
                 int cacheId = cacheIds[i % cacheIds.length];
@@ -93,10 +93,10 @@
                     idxName = randomName();
                 } while (idxMap.containsKey(idxName));
 
-                MetadataStorage metaStore = storeMap.get(cacheId);
+                IndexStorageImpl metaStore = storeMap.get(cacheId);
 
                 if (metaStore == null) {
-                    metaStore = new MetadataStorage(mem, null, new AtomicLong(), cacheId,
+                    metaStore = new IndexStorageImpl(mem, null, new AtomicLong(), cacheId,
                         PageIdAllocator.INDEX_PARTITION, PageMemory.FLAG_IDX,
                         null, mem.allocatePage(cacheId, PageIdAllocator.INDEX_PARTITION, PageMemory.FLAG_IDX), true);
 
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
index 5c4d7fd..3f32414 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
@@ -53,9 +53,9 @@
 import org.apache.ignite.internal.processors.database.BPlusTreeFakeReuseSelfTest;
 import org.apache.ignite.internal.processors.database.BPlusTreeReuseSelfTest;
 import org.apache.ignite.internal.processors.database.BPlusTreeSelfTest;
-import org.apache.ignite.internal.processors.database.FreeListImplSelfTest;
+import org.apache.ignite.internal.processors.database.CacheFreeListImplSelfTest;
+import org.apache.ignite.internal.processors.database.IndexStorageSelfTest;
 import org.apache.ignite.internal.processors.database.MemoryMetricsSelfTest;
-import org.apache.ignite.internal.processors.database.MetadataStorageSelfTest;
 import org.apache.ignite.internal.processors.database.SwapPathConstructionSelfTest;
 import org.apache.ignite.internal.processors.odbc.OdbcConfigurationValidationSelfTest;
 import org.apache.ignite.internal.processors.odbc.OdbcEscapeSequenceSelfTest;
@@ -173,8 +173,8 @@
         suite.addTestSuite(BPlusTreeSelfTest.class);
         suite.addTestSuite(BPlusTreeFakeReuseSelfTest.class);
         suite.addTestSuite(BPlusTreeReuseSelfTest.class);
-        suite.addTestSuite(MetadataStorageSelfTest.class);
-        suite.addTestSuite(FreeListImplSelfTest.class);
+        suite.addTestSuite(IndexStorageSelfTest.class);
+        suite.addTestSuite(CacheFreeListImplSelfTest.class);
         suite.addTestSuite(MemoryMetricsSelfTest.class);
         suite.addTestSuite(SwapPathConstructionSelfTest.class);
 
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite.java
index ef7682f..e4f7732 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite.java
@@ -28,7 +28,7 @@
 import org.apache.ignite.internal.processors.cache.persistence.db.file.IgnitePdsEvictionTest;
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.BPlusTreePageMemoryImplTest;
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.BPlusTreeReuseListPageMemoryImplTest;
-import org.apache.ignite.internal.processors.cache.persistence.pagemem.MetadataStoragePageMemoryImplTest;
+import org.apache.ignite.internal.processors.cache.persistence.pagemem.IndexStoragePageMemoryImplTest;
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryImplNoLoadTest;
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryImplTest;
 import org.apache.ignite.internal.processors.cache.persistence.pagemem.PagesWriteThrottleSmokeTest;
@@ -52,7 +52,7 @@
 
         // Basic PageMemory tests.
         suite.addTestSuite(PageMemoryImplNoLoadTest.class);
-        suite.addTestSuite(MetadataStoragePageMemoryImplTest.class);
+        suite.addTestSuite(IndexStoragePageMemoryImplTest.class);
         suite.addTestSuite(IgnitePdsEvictionTest.class);
         suite.addTestSuite(PageMemoryImplTest.class);