/*
 * 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;

import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.cache.processor.EntryProcessor;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.failure.FailureContext;
import org.apache.ignite.failure.FailureType;
import org.apache.ignite.internal.pagemem.FullPageId;
import org.apache.ignite.internal.pagemem.PageIdAllocator;
import org.apache.ignite.internal.pagemem.PageIdUtils;
import org.apache.ignite.internal.pagemem.PageMemory;
import org.apache.ignite.internal.pagemem.PageSupport;
import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
import org.apache.ignite.internal.pagemem.wal.WALIterator;
import org.apache.ignite.internal.pagemem.wal.WALPointer;
import org.apache.ignite.internal.pagemem.wal.record.DataEntry;
import org.apache.ignite.internal.pagemem.wal.record.DataRecord;
import org.apache.ignite.internal.pagemem.wal.record.PageSnapshot;
import org.apache.ignite.internal.pagemem.wal.record.WALRecord;
import org.apache.ignite.internal.pagemem.wal.record.delta.MetaPageInitRecord;
import org.apache.ignite.internal.pagemem.wal.record.delta.MetaPageUpdateNextSnapshotId;
import org.apache.ignite.internal.pagemem.wal.record.delta.MetaPageUpdatePartitionDataRecord;
import org.apache.ignite.internal.pagemem.wal.record.delta.PartitionDestroyRecord;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
import org.apache.ignite.internal.processors.cache.CacheGroupContext;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheTtlManager;
import org.apache.ignite.internal.processors.cache.IgniteCacheOffheapManagerImpl;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.CachePartitionPartialCountersMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteHistoricalIterator;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.mvcc.MvccVersion;
import org.apache.ignite.internal.processors.cache.persistence.freelist.CacheFreeListImpl;
import org.apache.ignite.internal.processors.cache.persistence.migration.UpgradePendingTreeToPerPartitionTask;
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;
import org.apache.ignite.internal.processors.cache.persistence.partstate.PartitionAllocationMap;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageMetaIO;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PagePartitionCountersIO;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PagePartitionMetaIO;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PagePartitionMetaIOV2;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseListImpl;
import org.apache.ignite.internal.processors.cache.persistence.tree.util.PageHandler;
import org.apache.ignite.internal.processors.cache.persistence.wal.FileWALPointer;
import org.apache.ignite.internal.processors.cache.tree.CacheDataRowStore;
import org.apache.ignite.internal.processors.cache.tree.CacheDataTree;
import org.apache.ignite.internal.processors.cache.tree.PendingEntriesTree;
import org.apache.ignite.internal.processors.cache.tree.PendingRow;
import org.apache.ignite.internal.processors.cache.tree.mvcc.data.MvccUpdateResult;
import org.apache.ignite.internal.processors.cache.tree.mvcc.search.MvccLinkAwareSearchRow;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.query.GridQueryRowCacheCleaner;
import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.lang.GridCursor;
import org.apache.ignite.internal.util.lang.IgniteInClosure2X;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiTuple;
import org.jetbrains.annotations.Nullable;

import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.MOVING;
import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING;
import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.RENTING;

/**
 * Used when persistence enabled.
 */
public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl implements DbCheckpointListener {
    /** */
    private IndexStorage indexStorage;

    /** */
    private ReuseListImpl reuseList;

    /** {@inheritDoc} */
    @Override protected void initPendingTree(GridCacheContext cctx) throws IgniteCheckedException {
        // No-op. Per-partition PendingTree should be used.
    }

    /** {@inheritDoc} */
    @Override protected void initDataStructures() throws IgniteCheckedException {
        assert ctx.database().checkpointLockIsHeldByThread();

        Metas metas = getOrAllocateCacheMetas();

        RootPage reuseListRoot = metas.reuseListRoot;

        reuseList = new ReuseListImpl(grp.groupId(),
            grp.cacheOrGroupName(),
            grp.dataRegion().pageMemory(),
            ctx.wal(),
            reuseListRoot.pageId().pageId(),
            reuseListRoot.isAllocated());

        RootPage metastoreRoot = metas.treeRoot;

        indexStorage = new IndexStorageImpl(grp.dataRegion().pageMemory(),
            ctx.wal(),
            globalRemoveId(),
            grp.groupId(),
            PageIdAllocator.INDEX_PARTITION,
            PageIdAllocator.FLAG_IDX,
            reuseList,
            metastoreRoot.pageId().pageId(),
            metastoreRoot.isAllocated(),
            ctx.kernalContext().failure());

        ((GridCacheDatabaseSharedManager)ctx.database()).addCheckpointListener(this);
    }

    /**
     * Get internal IndexStorage.
     * See {@link UpgradePendingTreeToPerPartitionTask} for details.
     */
    public IndexStorage getIndexStorage() {
        return indexStorage;
    }

    /** {@inheritDoc} */
    @Override protected CacheDataStore createCacheDataStore0(int p) throws IgniteCheckedException {
        if (ctx.database() instanceof GridCacheDatabaseSharedManager)
            ((GridCacheDatabaseSharedManager) ctx.database()).cancelOrWaitPartitionDestroy(grp.groupId(), p);

        boolean exists = ctx.pageStore() != null && ctx.pageStore().exists(grp.groupId(), p);

        return new GridCacheDataStore(p, exists);
    }

    /** {@inheritDoc} */
    @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
        assert grp.dataRegion().pageMemory() instanceof PageMemoryEx;

        Executor execSvc = ctx.executor();

        boolean needSnapshot = ctx.nextSnapshot() && ctx.needToSnapshot(grp.cacheOrGroupName());

        boolean hasNonEmptyGroups = false;

        for (CacheDataStore store : partDataStores.values()) {
            if (notEmpty(store)) {
                hasNonEmptyGroups = true;

                break;
            }
        }

        if (needSnapshot && hasNonEmptyGroups) {
            if (execSvc == null)
                updateSnapshotTag(ctx);
            else {
                execSvc.execute(() -> {
                    try {
                        updateSnapshotTag(ctx);
                    }
                    catch (IgniteCheckedException e) {
                        throw new IgniteException(e);
                    }
                });
            }
        }

        if (execSvc == null) {
            reuseList.saveMetadata();

            for (CacheDataStore store : partDataStores.values())
                saveStoreMetadata(store, ctx, false, needSnapshot);
        }
        else {
            execSvc.execute(() -> {
                try {
                    reuseList.saveMetadata();
                }
                catch (IgniteCheckedException e) {
                    throw new IgniteException(e);
                }
            });

            for (CacheDataStore store : partDataStores.values())
                execSvc.execute(() -> {
                    try {
                        saveStoreMetadata(store, ctx, false, needSnapshot);
                    }
                    catch (IgniteCheckedException e) {
                        throw new IgniteException(e);
                    }
                });
        }
    }

    /**
     * @return {@code True} is group is not empty.
     */
    private boolean notEmpty(CacheDataStore store) {
        return store.rowStore() != null && (store.fullSize() > 0  || store.updateCounter() > 0);
    }

    /**
     * @param store Store to save metadata.
     * @throws IgniteCheckedException If failed.
     */
    private void saveStoreMetadata(
        CacheDataStore store,
        Context ctx,
        boolean beforeDestroy,
        boolean needSnapshot
    ) throws IgniteCheckedException {
        RowStore rowStore0 = store.rowStore();

        if (rowStore0 != null) {
            CacheFreeListImpl freeList = (CacheFreeListImpl)rowStore0.freeList();

            freeList.saveMetadata();

            long updCntr = store.updateCounter();
            long size = store.fullSize();
            long rmvId = globalRemoveId().get();

            PageMemoryEx pageMem = (PageMemoryEx)grp.dataRegion().pageMemory();
            IgniteWriteAheadLogManager wal = this.ctx.wal();

            if (size > 0 || updCntr > 0) {
                GridDhtPartitionState state = null;

                // localPartition will not acquire writeLock here because create=false.
                GridDhtLocalPartition part = null;

                if (!grp.isLocal()) {
                    if (beforeDestroy)
                        state = GridDhtPartitionState.EVICTED;
                    else {
                        part = getPartition(store);

                        if (part != null && part.state() != GridDhtPartitionState.EVICTED)
                            state = part.state();
                    }

                    // Do not save meta for evicted partitions on next checkpoints.
                    if (state == null)
                        return;
                }

                int grpId = grp.groupId();
                long partMetaId = pageMem.partitionMetaPageId(grpId, store.partId());

                long partMetaPage = pageMem.acquirePage(grpId, partMetaId);
                try {
                    long partMetaPageAddr = pageMem.writeLock(grpId, partMetaId, partMetaPage);

                    if (partMetaPageAddr == 0L) {
                        U.warn(log, "Failed to acquire write lock for meta page [metaPage=" + partMetaPage +
                            ", beforeDestroy=" + beforeDestroy + ", size=" + size +
                            ", updCntr=" + updCntr + ", state=" + state + ']');

                        return;
                    }

                    boolean changed = false;

                    try {
                        PagePartitionMetaIO io = PageIO.getPageIO(partMetaPageAddr);

                        changed |= io.setUpdateCounter(partMetaPageAddr, updCntr);
                        changed |= io.setGlobalRemoveId(partMetaPageAddr, rmvId);
                        changed |= io.setSize(partMetaPageAddr, size);

                        if (state != null)
                            changed |= io.setPartitionState(partMetaPageAddr, (byte)state.ordinal());
                        else
                            assert grp.isLocal() : grp.cacheOrGroupName();

                        long cntrsPageId;

                        if (grp.sharedGroup()) {
                            long initCntrPageId = io.getCountersPageId(partMetaPageAddr);

                            Map<Integer, Long> newSizes = store.cacheSizes();
                            Map<Integer, Long> prevSizes = readSharedGroupCacheSizes(pageMem, grpId, initCntrPageId);

                            if (prevSizes != null && prevSizes.equals(newSizes))
                                cntrsPageId = initCntrPageId; // Preventing modification of sizes pages for store
                            else {
                                cntrsPageId = writeSharedGroupCacheSizes(pageMem, grpId, initCntrPageId,
                                    store.partId(), newSizes);

                                if (initCntrPageId == 0 && cntrsPageId != 0) {
                                    io.setCountersPageId(partMetaPageAddr, cntrsPageId);

                                    changed = true;
                                }
                            }
                        }
                        else
                            cntrsPageId = 0L;

                        int pageCnt;

                        if (needSnapshot) {
                            pageCnt = this.ctx.pageStore().pages(grpId, store.partId());

                            io.setCandidatePageCount(partMetaPageAddr, size == 0 ? 0 : pageCnt);

                            if (state == OWNING) {
                                assert part != null;

                                if (!addPartition(
                                    part,
                                    ctx.partitionStatMap(),
                                    partMetaPageAddr,
                                    io,
                                    grpId,
                                    store.partId(),
                                    this.ctx.pageStore().pages(grpId, store.partId()),
                                    store.fullSize()
                                ))
                                    U.warn(log, "Partition was concurrently evicted grpId=" + grpId +
                                        ", partitionId=" + part.id());
                            }
                            else if (state == MOVING || state == RENTING) {
                                if (ctx.partitionStatMap().forceSkipIndexPartition(grpId)) {
                                    if (log.isInfoEnabled())
                                        log.info("Will not include SQL indexes to snapshot because there is " +
                                            "a partition not in " + OWNING + " state [grp=" + grp.cacheOrGroupName() +
                                            ", partId=" + store.partId() + ", state=" + state + ']');
                                }
                            }

                            changed = true;
                        }
                        else
                            pageCnt = io.getCandidatePageCount(partMetaPageAddr);

                        if (PageHandler.isWalDeltaRecordNeeded(pageMem, grpId, partMetaId, partMetaPage, wal, null))
                            wal.log(new MetaPageUpdatePartitionDataRecord(
                                grpId,
                                partMetaId,
                                updCntr,
                                rmvId,
                                (int)size, // TODO: Partition size may be long
                                cntrsPageId,
                                state == null ? -1 : (byte)state.ordinal(),
                                pageCnt
                            ));
                    }
                    finally {
                        pageMem.writeUnlock(grpId, partMetaId, partMetaPage, null, changed);
                    }
                }
                finally {
                    pageMem.releasePage(grpId, partMetaId, partMetaPage);
                }
            }
            else if (needSnapshot)
                tryAddEmptyPartitionToSnapshot(store, ctx);
        }
        else if (needSnapshot)
            tryAddEmptyPartitionToSnapshot(store, ctx);
    }

    /**
     * Check that we need to snapshot this partition and add it to map.
     *
     * @param store Store.
     * @param ctx Snapshot context.
     */
    private void tryAddEmptyPartitionToSnapshot(CacheDataStore store, Context ctx) {
        if (getPartition(store).state() == OWNING) {
            ctx.partitionStatMap().put(
                new GroupPartitionId(grp.groupId(), store.partId()),
                new PagesAllocationRange(0, 0));
        }
    }

    /**
     * @param store Store.
     *
     * @return corresponding to store local partition
     */
    private GridDhtLocalPartition getPartition(CacheDataStore store) {
        return grp.topology().localPartition(store.partId(),
            AffinityTopologyVersion.NONE, false, true);
    }

    /**
     * Loads cache sizes for all caches in shared group.
     *
     * @param pageMem page memory to perform operations on pages.
     * @param grpId Cache group ID.
     * @param cntrsPageId Counters page ID, if zero is provided that means no counters page exist.
     * @return Cache sizes if store belongs to group containing multiple caches and sizes are available in memory. May
     * return null if counter page does not exist.
     * @throws IgniteCheckedException If page memory operation failed.
     */
    @Nullable private static Map<Integer, Long> readSharedGroupCacheSizes(PageSupport pageMem, int grpId,
        long cntrsPageId) throws IgniteCheckedException {

        if (cntrsPageId == 0L)
            return null;

        Map<Integer, Long> cacheSizes = new HashMap<>();

        long nextId = cntrsPageId;

        while (true) {
            final long curId = nextId;
            final long curPage = pageMem.acquirePage(grpId, curId);

            try {
                final long curAddr = pageMem.readLock(grpId, curId, curPage);

                assert curAddr != 0;

                try {
                    PagePartitionCountersIO cntrsIO = PageIO.getPageIO(curAddr);

                    if (cntrsIO.readCacheSizes(curAddr, cacheSizes))
                        break;

                    nextId = cntrsIO.getNextCountersPageId(curAddr);

                    assert nextId != 0;
                }
                finally {
                    pageMem.readUnlock(grpId, curId, curPage);
                }
            }
            finally {
                pageMem.releasePage(grpId, curId, curPage);
            }
        }
        return cacheSizes;
    }

    /**
     * Saves cache sizes for all caches in shared group. Unconditionally marks pages as dirty.
     *
     * @param pageMem page memory to perform operations on pages.
     * @param grpId Cache group ID.
     * @param cntrsPageId Counters page ID, if zero is provided that means no counters page exist.
     * @param partId Partition ID.
     * @param sizes Cache sizes of all caches in group. Not null.
     * @return new counter page Id. Same as {@code cntrsPageId} or new value if cache size pages were initialized.
     * @throws IgniteCheckedException if page memory operation failed.
     */
    private static long writeSharedGroupCacheSizes(PageMemory pageMem, int grpId,
        long cntrsPageId, int partId, Map<Integer, Long> sizes) throws IgniteCheckedException {
        byte[] data = PagePartitionCountersIO.VERSIONS.latest().serializeCacheSizes(sizes);

        int items = data.length / PagePartitionCountersIO.ITEM_SIZE;
        boolean init = cntrsPageId == 0;

        if (init && !sizes.isEmpty())
            cntrsPageId = pageMem.allocatePage(grpId, partId, PageIdAllocator.FLAG_DATA);

        long nextId = cntrsPageId;
        int written = 0;

        while (written != items) {
            final long curId = nextId;
            final long curPage = pageMem.acquirePage(grpId, curId);

            try {
                final long curAddr = pageMem.writeLock(grpId, curId, curPage);

                assert curAddr != 0;

                try {
                    PagePartitionCountersIO partCntrIo;

                    if (init) {
                        partCntrIo = PagePartitionCountersIO.VERSIONS.latest();

                        partCntrIo.initNewPage(curAddr, curId, pageMem.realPageSize(grpId));
                    }
                    else
                        partCntrIo = PageIO.getPageIO(curAddr);

                    written += partCntrIo.writeCacheSizes(pageMem.realPageSize(grpId), curAddr, data, written);

                    nextId = partCntrIo.getNextCountersPageId(curAddr);

                    if (written != items && (init = nextId == 0)) {
                        //allocate new counters page
                        nextId = pageMem.allocatePage(grpId, partId, PageIdAllocator.FLAG_DATA);
                        partCntrIo.setNextCountersPageId(curAddr, nextId);
                    }
                }
                finally {
                    // Write full page
                    pageMem.writeUnlock(grpId, curId, curPage, Boolean.TRUE, true);
                }
            }
            finally {
                pageMem.releasePage(grpId, curId, curPage);
            }
        }

        return cntrsPageId;
    }

    /**
     * @param ctx Context.
     */
    private void updateSnapshotTag(Context ctx) throws IgniteCheckedException {
        int grpId = grp.groupId();
        PageMemoryEx pageMem = (PageMemoryEx)grp.dataRegion().pageMemory();
        IgniteWriteAheadLogManager wal = this.ctx.wal();

        long metaPageId = pageMem.metaPageId(grpId);
        long metaPage = pageMem.acquirePage(grpId, metaPageId);

        try {
            long metaPageAddr = pageMem.writeLock(grpId, metaPageId, metaPage);

            try {
                PageMetaIO metaIo = PageMetaIO.getPageIO(metaPageAddr);

                long nextSnapshotTag = metaIo.getNextSnapshotTag(metaPageAddr);

                metaIo.setNextSnapshotTag(metaPageAddr, nextSnapshotTag + 1);

                if (log != null && log.isDebugEnabled())
                    log.debug("Save next snapshot before checkpoint start for grId = " + grpId
                        + ", nextSnapshotTag = " + nextSnapshotTag);

                if (PageHandler.isWalDeltaRecordNeeded(pageMem, grpId, metaPageId,
                    metaPage, wal, null))
                    wal.log(new MetaPageUpdateNextSnapshotId(grpId, metaPageId,
                        nextSnapshotTag + 1));

                addPartition(
                    null,
                    ctx.partitionStatMap(),
                    metaPageAddr,
                    metaIo,
                    grpId,
                    PageIdAllocator.INDEX_PARTITION,
                    this.ctx.pageStore().pages(grpId, PageIdAllocator.INDEX_PARTITION),
                    -1);
            }
            finally {
                pageMem.writeUnlock(grpId, metaPageId, metaPage, null, true);
            }
        }
        finally {
            pageMem.releasePage(grpId, metaPageId, metaPage);
        }
    }

    /**
     * @param part Local partition.
     * @param map Map to add values to.
     * @param metaPageAddr Meta page address
     * @param io Page Meta IO
     * @param grpId Cache Group ID.
     * @param currAllocatedPageCnt total number of pages allocated for partition <code>[partition, grpId]</code>
     */
    private static boolean addPartition(
        GridDhtLocalPartition part,
        final PartitionAllocationMap map,
        final long metaPageAddr,
        final PageMetaIO io,
        final int grpId,
        final int partId,
        final int currAllocatedPageCnt,
        final long partSize
    ) {
        if (part != null) {
            boolean reserved = part.reserve();

            if (!reserved)
                return false;
        }
        else
            assert partId == PageIdAllocator.INDEX_PARTITION : partId;

        assert PageIO.getPageId(metaPageAddr) != 0;

        int lastAllocatedPageCnt = io.getLastAllocatedPageCount(metaPageAddr);

        int curPageCnt = partSize == 0 ? 0 : currAllocatedPageCnt;

        map.put(
            new GroupPartitionId(grpId, partId),
            new PagesAllocationRange(lastAllocatedPageCnt, curPageCnt));

        return true;
    }

    /** {@inheritDoc} */
    @Override protected void destroyCacheDataStore0(CacheDataStore store) throws IgniteCheckedException {
        assert ctx.database() instanceof GridCacheDatabaseSharedManager
            : "Destroying cache data store when persistence is not enabled: " + ctx.database();

        int partId = store.partId();

        ctx.database().checkpointReadLock();

        try {
            saveStoreMetadata(store, null, true, false);
        }
        finally {
            ctx.database().checkpointReadUnlock();
        }

        ((GridCacheDatabaseSharedManager)ctx.database()).schedulePartitionDestroy(grp.groupId(), partId);
    }

    /**
     * Invalidates page memory for given partition. Destroys partition store.
     * <b>NOTE:</b> This method can be invoked only within checkpoint lock or checkpointer thread.
     *
     * @param grpId Group ID.
     * @param partId Partition ID.
     *
     * @throws IgniteCheckedException If destroy has failed.
     */
    public void destroyPartitionStore(int grpId, int partId) throws IgniteCheckedException {
        PageMemoryEx pageMemory = (PageMemoryEx)grp.dataRegion().pageMemory();

        int tag = pageMemory.invalidate(grp.groupId(), partId);

        if (grp.walEnabled())
            ctx.wal().log(new PartitionDestroyRecord(grp.groupId(), partId));

        ctx.pageStore().onPartitionDestroyed(grpId, partId, tag);
    }

    /** {@inheritDoc} */
    @Override public void onPartitionCounterUpdated(int part, long cntr) {
        CacheDataStore store = partDataStores.get(part);

        assert store != null;

        long oldCnt = store.updateCounter();

        if (oldCnt < cntr)
            store.updateCounter(cntr);
    }

    /** {@inheritDoc} */
    @Override public void onPartitionInitialCounterUpdated(int part, long cntr) {
        CacheDataStore store = partDataStores.get(part);

        assert store != null;

        long oldCnt = store.initialUpdateCounter();

        if (oldCnt < cntr)
            store.updateInitialCounter(cntr);
    }

    /** {@inheritDoc} */
    @Override public long lastUpdatedPartitionCounter(int part) {
        return partDataStores.get(part).updateCounter();
    }

    /** {@inheritDoc} */
    @Override public RootPage rootPageForIndex(int cacheId, String idxName) throws IgniteCheckedException {
        if (grp.sharedGroup())
            idxName = Integer.toString(cacheId) + "_" + idxName;

        return indexStorage.getOrAllocateForTree(idxName);
    }

    /** {@inheritDoc} */
    @Override public void dropRootPageForIndex(int cacheId, String idxName) throws IgniteCheckedException {
        if (grp.sharedGroup())
            idxName = Integer.toString(cacheId) + "_" + idxName;

        indexStorage.dropRootPage(idxName);
    }

    /** {@inheritDoc} */
    @Override public ReuseList reuseListForIndex(String idxName) {
        return reuseList;
    }

    /** {@inheritDoc} */
    @Override public void stop() {
        if (grp.affinityNode())
            ((GridCacheDatabaseSharedManager)ctx.database()).removeCheckpointListener(this);
    }

    /**
     * @return Meta root pages info.
     * @throws IgniteCheckedException If failed.
     */
    private Metas getOrAllocateCacheMetas() throws IgniteCheckedException {
        PageMemoryEx pageMem = (PageMemoryEx)grp.dataRegion().pageMemory();
        IgniteWriteAheadLogManager wal = ctx.wal();

        int grpId = grp.groupId();
        long metaId = pageMem.metaPageId(grpId);
        long metaPage = pageMem.acquirePage(grpId, metaId);

        try {
            final long pageAddr = pageMem.writeLock(grpId, metaId, metaPage);

            boolean allocated = false;

            try {
                long metastoreRoot, reuseListRoot;

                if (PageIO.getType(pageAddr) != PageIO.T_META) {
                    PageMetaIO pageIO = PageMetaIO.VERSIONS.latest();

                    pageIO.initNewPage(pageAddr, metaId, pageMem.realPageSize(grpId));

                    metastoreRoot = pageMem.allocatePage(grpId, PageIdAllocator.INDEX_PARTITION, PageMemory.FLAG_IDX);
                    reuseListRoot = pageMem.allocatePage(grpId, PageIdAllocator.INDEX_PARTITION, PageMemory.FLAG_IDX);

                    pageIO.setTreeRoot(pageAddr, metastoreRoot);
                    pageIO.setReuseListRoot(pageAddr, reuseListRoot);

                    if (PageHandler.isWalDeltaRecordNeeded(pageMem, grpId, metaId, metaPage, wal, null))
                        wal.log(new MetaPageInitRecord(
                            grpId,
                            metaId,
                            pageIO.getType(),
                            pageIO.getVersion(),
                            metastoreRoot,
                            reuseListRoot
                        ));

                    allocated = true;
                }
                else {
                    PageMetaIO pageIO = PageIO.getPageIO(pageAddr);

                    metastoreRoot = pageIO.getTreeRoot(pageAddr);
                    reuseListRoot = pageIO.getReuseListRoot(pageAddr);

                    assert reuseListRoot != 0L;
                }

                return new Metas(
                    new RootPage(new FullPageId(metastoreRoot, grpId), allocated),
                    new RootPage(new FullPageId(reuseListRoot, grpId), allocated),
                    null);
            }
            finally {
                pageMem.writeUnlock(grpId, metaId, metaPage, null, allocated);
            }
        }
        finally {
            pageMem.releasePage(grpId, metaId, metaPage);
        }
    }

    /** {@inheritDoc} */
    @Override @Nullable protected IgniteHistoricalIterator historicalIterator(
        CachePartitionPartialCountersMap partCntrs, Set<Integer> missing) throws IgniteCheckedException {
        if (partCntrs == null || partCntrs.isEmpty())
            return null;

        if (grp.mvccEnabled()) // TODO IGNITE-7384
            return super.historicalIterator(partCntrs, missing);

        GridCacheDatabaseSharedManager database = (GridCacheDatabaseSharedManager)grp.shared().database();

        FileWALPointer minPtr = null;

        for (int i = 0; i < partCntrs.size(); i++) {
            int p = partCntrs.partitionAt(i);
            long initCntr = partCntrs.initialUpdateCounterAt(i);

            FileWALPointer startPtr = (FileWALPointer)database.checkpointHistory().searchPartitionCounter(
                grp.groupId(), p, initCntr);

            if (startPtr == null)
                throw new IgniteCheckedException("Could not find start pointer for partition [part=" + p + ", partCntrSince=" + initCntr + "]");

            if (minPtr == null || startPtr.compareTo(minPtr) < 0)
                minPtr = startPtr;
        }

        WALIterator it = grp.shared().wal().replay(minPtr);

        WALHistoricalIterator iterator = new WALHistoricalIterator(grp, partCntrs, it);

        // Add historical partitions which are unabled to reserve to missing set.
        missing.addAll(iterator.missingParts);

        return iterator;
    }

    /** {@inheritDoc} */
    @Override public boolean expire(
        GridCacheContext cctx,
        IgniteInClosure2X<GridCacheEntryEx, GridCacheVersion> c,
        int amount
    ) throws IgniteCheckedException {
        assert !cctx.isNear() : cctx.name();

        if (!hasPendingEntries || nextCleanTime > U.currentTimeMillis())
            return false;

        // Prevent manager being stopped in the middle of pds operation.
        if (!busyLock.enterBusy())
            return false;

        try {
            int cleared = 0;

            for (CacheDataStore store : cacheDataStores()) {
                cleared += ((GridCacheDataStore)store).purgeExpired(cctx, c, amount - cleared);

                if (amount != -1 && cleared >= amount)
                    return true;
            }

            // Throttle if there is nothing to clean anymore.
            if (cleared < amount)
                nextCleanTime = U.currentTimeMillis() + UNWIND_THROTTLING_TIMEOUT;
        }
        finally {
            busyLock.leaveBusy();
        }

        return false;
    }

    /** {@inheritDoc} */
    @Override public long expiredSize() throws IgniteCheckedException {
        long size = 0;

        for (CacheDataStore store : cacheDataStores())
            size += ((GridCacheDataStore)store).expiredSize();

        return size;
    }

    /**
     * Calculates free space of all partition data stores - number of bytes available for use in allocated pages.
     *
     * @return Tuple (numenator, denominator).
     */
    long freeSpace() {
        long freeSpace = 0;

        for (CacheDataStore store : partDataStores.values()) {
            assert store instanceof GridCacheDataStore;

            CacheFreeListImpl freeList = ((GridCacheDataStore)store).freeList;

            if (freeList == null)
                continue;

            freeSpace += freeList.freeSpace();
        }

        return freeSpace;
    }

    /**
     *
     */
    private static class WALHistoricalIterator implements IgniteHistoricalIterator {
        /** */
        private static final long serialVersionUID = 0L;

        /** Cache context. */
        private final CacheGroupContext grp;

        /** Partition counters map. */
        private final CachePartitionPartialCountersMap partMap;

        /** Partitions marked as missing (unable to reserve or partition is not in OWNING state). */
        private final Set<Integer> missingParts = new HashSet<>();

        /** Partitions marked as done. */
        private final Set<Integer> doneParts = new HashSet<>();

        /** Cache IDs. This collection is stored as field to avoid re-calculation on each iteration. */
        private final Set<Integer> cacheIds;

        /** WAL iterator. */
        private WALIterator walIt;

        /** */
        private Iterator<DataEntry> entryIt;

        /** */
        private DataEntry next;

        /** Flag indicates that partition belongs to current {@link #next} is finished and no longer needs to rebalance. */
        private boolean reachedPartitionEnd;

        /** Flag indicates that update counters for requested partitions have been reached and done.
         *  It means that no further iteration is needed. */
        private boolean doneAllPartitions;

        /**
         * @param grp Cache context.
         * @param walIt WAL iterator.
         */
        private WALHistoricalIterator(CacheGroupContext grp, CachePartitionPartialCountersMap partMap, WALIterator walIt) {
            this.grp = grp;
            this.partMap = partMap;
            this.walIt = walIt;

            cacheIds = grp.cacheIds();

            reservePartitions();

            advance();
        }

        /** {@inheritDoc} */
        @Override public boolean contains(int partId) {
            return partMap.contains(partId);
        }

        /** {@inheritDoc} */
        @Override public boolean isDone(int partId) {
            return doneParts.contains(partId);
        }

        /** {@inheritDoc} */
        @Override public void close() throws IgniteCheckedException {
            walIt.close();
            releasePartitions();
        }

        /** {@inheritDoc} */
        @Override public boolean isClosed() {
            return walIt.isClosed();
        }

        /** {@inheritDoc} */
        @Override public boolean hasNextX() {
            return hasNext();
        }

        /** {@inheritDoc} */
        @Override public CacheDataRow nextX() throws IgniteCheckedException {
            return next();
        }

        /** {@inheritDoc} */
        @Override public void removeX() throws IgniteCheckedException {
            throw new UnsupportedOperationException();
        }

        /** {@inheritDoc} */
        @Override public Iterator<CacheDataRow> iterator() {
            return this;
        }

        /** {@inheritDoc} */
        @Override public boolean hasNext() {
            return next != null;
        }

        /** {@inheritDoc} */
        @Override public CacheDataRow next() {
            if (next == null)
                throw new NoSuchElementException();

            CacheDataRow val = new DataEntryRow(next);

            if (reachedPartitionEnd) {
                doneParts.add(next.partitionId());

                reachedPartitionEnd = false;

                if (doneParts.size() == partMap.size())
                    doneAllPartitions = true;
            }

            advance();

            return val;
        }

        /** {@inheritDoc} */
        @Override public void remove() {
            throw new UnsupportedOperationException();
        }

        /**
         * Reserve historical partitions.
         * If partition is unable to reserve, id of that partition is placed to {@link #missingParts} set.
         */
        private void reservePartitions() {
            for (int i = 0; i < partMap.size(); i++) {
                int p = partMap.partitionAt(i);
                GridDhtLocalPartition part = grp.topology().localPartition(p);

                if (part == null || !part.reserve()) {
                    missingParts.add(p);
                    continue;
                }

                if (part.state() != OWNING) {
                    part.release();
                    missingParts.add(p);
                }
            }
        }

        /**
         * Release historical partitions.
         */
        private void releasePartitions() {
            for (int i = 0; i < partMap.size(); i++) {
                int p = partMap.partitionAt(i);

                if (missingParts.contains(p))
                    continue;

                GridDhtLocalPartition part = grp.topology().localPartition(p);

                assert part != null && part.state() == OWNING && part.reservations() > 0
                    : "Partition should in OWNING state and has at least 1 reservation";

                part.release();
            }
        }

        /**
         *
         */
        private void advance() {
            next = null;

            if (doneAllPartitions)
                return;

            while (true) {
                if (entryIt != null) {
                    while (entryIt.hasNext()) {
                        DataEntry entry = entryIt.next();

                        if (cacheIds.contains(entry.cacheId())) {
                            int idx = partMap.partitionIndex(entry.partitionId());

                            if (idx < 0 || missingParts.contains(idx))
                                continue;

                            long from = partMap.initialUpdateCounterAt(idx);
                            long to = partMap.updateCounterAt(idx);

                            if (entry.partitionCounter() > from && entry.partitionCounter() <= to) {
                                if (entry.partitionCounter() == to)
                                    reachedPartitionEnd = true;

                                next = entry;

                                return;
                            }
                        }
                    }
                }

                entryIt = null;

                while (walIt.hasNext()) {
                    IgniteBiTuple<WALPointer, WALRecord> rec = walIt.next();

                    if (rec.get2() instanceof DataRecord) {
                        DataRecord data = (DataRecord)rec.get2();

                        entryIt = data.writeEntries().iterator();
                        // Move on to the next valid data entry.

                        break;
                    }
                }

                if (entryIt == null)
                    return;
            }
        }
    }

    /**
     * Data entry row.
     */
    private static class DataEntryRow implements CacheDataRow {
        /** */
        private final DataEntry entry;

        /**
         * @param entry Data entry.
         */
        private DataEntryRow(DataEntry entry) {
            this.entry = entry;
        }

        /** {@inheritDoc} */
        @Override public KeyCacheObject key() {
            return entry.key();
        }

        /** {@inheritDoc} */
        @Override public void key(KeyCacheObject key) {
            throw new IllegalStateException();
        }

        /** {@inheritDoc} */
        @Override public CacheObject value() {
            return entry.value();
        }

        /** {@inheritDoc} */
        @Override public GridCacheVersion version() {
            return entry.writeVersion();
        }

        /** {@inheritDoc} */
        @Override public long expireTime() {
            return entry.expireTime();
        }

        /** {@inheritDoc} */
        @Override public int partition() {
            return entry.partitionId();
        }

        /** {@inheritDoc} */
        @Override public int size() throws IgniteCheckedException {
            throw new UnsupportedOperationException();
        }

        /** {@inheritDoc} */
        @Override public int headerSize() {
            throw new UnsupportedOperationException();
        }

        /** {@inheritDoc} */
        @Override public long link() {
            return 0;
        }

        /** {@inheritDoc} */
        @Override public void link(long link) {
            throw new UnsupportedOperationException();
        }

        /** {@inheritDoc} */
        @Override public int hash() {
            return entry.key().hashCode();
        }

        /** {@inheritDoc} */
        @Override public int cacheId() {
            return entry.cacheId();
        }

        /** {@inheritDoc} */
        @Override public long mvccCoordinatorVersion() {
            return 0; // TODO IGNITE-7384
        }

        /** {@inheritDoc} */
        @Override public long mvccCounter() {
            return 0;  // TODO IGNITE-7384
        }

        /** {@inheritDoc} */
        @Override public int mvccOperationCounter() {
            return 0;  // TODO IGNITE-7384
        }

        /** {@inheritDoc} */
        @Override public long newMvccCoordinatorVersion() {
            return 0; // TODO IGNITE-7384
        }

        /** {@inheritDoc} */
        @Override public long newMvccCounter() {
            return 0; // TODO IGNITE-7384
        }

        /** {@inheritDoc} */
        @Override public int newMvccOperationCounter() {
            return 0;  // TODO IGNITE-7384
        }

        /** {@inheritDoc} */
        @Override public byte mvccTxState() {
            return 0;  // TODO IGNITE-7384
        }

        /** {@inheritDoc} */
        @Override public byte newMvccTxState() {
            return 0; // TODO IGNITE-7384
        }
    }

    /**
     *
     */
    private static class Metas {
        /** */
        @GridToStringInclude
        private final RootPage reuseListRoot;

        /** */
        @GridToStringInclude
        private final RootPage treeRoot;

        /** */
        @GridToStringInclude
        private final RootPage pendingTreeRoot;

        /**
         * @param treeRoot Metadata storage root.
         * @param reuseListRoot Reuse list root.
         */
        Metas(RootPage treeRoot, RootPage reuseListRoot, RootPage pendingTreeRoot) {
            this.treeRoot = treeRoot;
            this.reuseListRoot = reuseListRoot;
            this.pendingTreeRoot = pendingTreeRoot;
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(Metas.class, this);
        }
    }

    /**
     *
     */
    public class GridCacheDataStore implements CacheDataStore {
        /** */
        private final int partId;

        /** */
        private String name;

        /** */
        private volatile CacheFreeListImpl freeList;

        /** */
        private PendingEntriesTree pendingTree;

        /** */
        private volatile CacheDataStore delegate;

        /** Timestamp when next clean try will be allowed for current partition.
         * Used for fine-grained throttling on per-partition basis. */
        private volatile long nextStoreCleanTime;

        /** */
        private final boolean exists;

        /** */
        private final AtomicBoolean init = new AtomicBoolean();

        /** */
        private final CountDownLatch latch = new CountDownLatch(1);

        /**
         * @param partId Partition.
         * @param exists {@code True} if store for this index exists.
         */
        private GridCacheDataStore(int partId, boolean exists) {
            this.partId = partId;
            this.exists = exists;

            name = treeName(partId);
        }

        /**
         * @return Store delegate.
         * @throws IgniteCheckedException If failed.
         */
        private CacheDataStore init0(boolean checkExists) throws IgniteCheckedException {
            CacheDataStore delegate0 = delegate;

            if (delegate0 != null)
                return delegate0;

            if (checkExists) {
                if (!exists)
                    return null;
            }

            if (init.compareAndSet(false, true)) {
                IgniteCacheDatabaseSharedManager dbMgr = ctx.database();

                dbMgr.checkpointReadLock();

                try {
                    Metas metas = getOrAllocatePartitionMetas();

                    RootPage reuseRoot = metas.reuseListRoot;

                    freeList = new CacheFreeListImpl(
                        grp.groupId(),
                        grp.cacheOrGroupName() + "-" + partId,
                        grp.dataRegion().memoryMetrics(),
                        grp.dataRegion(),
                        null,
                        ctx.wal(),
                        reuseRoot.pageId().pageId(),
                        reuseRoot.isAllocated()) {
                        /** {@inheritDoc} */
                        @Override protected long allocatePageNoReuse() throws IgniteCheckedException {
                            assert grp.shared().database().checkpointLockIsHeldByThread();

                            return pageMem.allocatePage(grpId, partId, PageIdAllocator.FLAG_DATA);
                        }
                    };

                    CacheDataRowStore rowStore = new CacheDataRowStore(grp, freeList, partId);

                    RootPage treeRoot = metas.treeRoot;

                    CacheDataTree dataTree = new CacheDataTree(
                        grp,
                        name,
                        freeList,
                        rowStore,
                        treeRoot.pageId().pageId(),
                        treeRoot.isAllocated()) {
                        /** {@inheritDoc} */
                        @Override protected long allocatePageNoReuse() throws IgniteCheckedException {
                            assert grp.shared().database().checkpointLockIsHeldByThread();

                            return pageMem.allocatePage(grpId, partId, PageIdAllocator.FLAG_DATA);
                        }
                    };

                    RootPage pendingTreeRoot = metas.pendingTreeRoot;

                    final PendingEntriesTree pendingTree0 = new PendingEntriesTree(
                        grp,
                        "PendingEntries-" + partId,
                        grp.dataRegion().pageMemory(),
                        pendingTreeRoot.pageId().pageId(),
                        freeList,
                        pendingTreeRoot.isAllocated()) {
                        /** {@inheritDoc} */
                        @Override protected long allocatePageNoReuse() throws IgniteCheckedException {
                            assert grp.shared().database().checkpointLockIsHeldByThread();

                            return pageMem.allocatePage(grpId, partId, PageIdAllocator.FLAG_DATA);
                        }
                    };

                    PageMemoryEx pageMem = (PageMemoryEx)grp.dataRegion().pageMemory();

                    delegate0 = new CacheDataStoreImpl(partId, name, rowStore, dataTree) {
                        /** {@inheritDoc} */
                        @Override public PendingEntriesTree pendingTree() {
                            return pendingTree0;
                        }
                    };

                    pendingTree = pendingTree0;

                    if (!hasPendingEntries && pendingTree0.size() > 0)
                        hasPendingEntries = true;

                    int grpId = grp.groupId();
                    long partMetaId = pageMem.partitionMetaPageId(grpId, partId);
                    long partMetaPage = pageMem.acquirePage(grpId, partMetaId);

                    try {
                        long pageAddr = pageMem.readLock(grpId, partMetaId, partMetaPage);

                        try {
                            if (PageIO.getType(pageAddr) != 0) {
                                PagePartitionMetaIO io = PagePartitionMetaIO.VERSIONS.latest();

                                Map<Integer, Long> cacheSizes = null;

                                if (grp.sharedGroup())
                                    cacheSizes = readSharedGroupCacheSizes(pageMem, grpId, io.getCountersPageId(pageAddr));

                                delegate0.init(io.getSize(pageAddr), io.getUpdateCounter(pageAddr), cacheSizes);

                                globalRemoveId().setIfGreater(io.getGlobalRemoveId(pageAddr));
                            }
                        }
                        finally {
                            pageMem.readUnlock(grpId, partMetaId, partMetaPage);
                        }
                    }
                    finally {
                        pageMem.releasePage(grpId, partMetaId, partMetaPage);
                    }

                    delegate = delegate0;
                }
                catch (Throwable ex) {
                    U.error(log, "Unhandled exception during page store initialization. All further operations will " +
                        "be failed and local node will be stopped.", ex);

                    ctx.kernalContext().failure().process(new FailureContext(FailureType.CRITICAL_ERROR, ex));

                    throw ex;
                }
                finally {
                    latch.countDown();

                    dbMgr.checkpointReadUnlock();
                }
            }
            else {
                U.await(latch);

                delegate0 = delegate;

                if (delegate0 == null)
                    throw new IgniteCheckedException("Cache store initialization failed.");
            }

            return delegate0;
        }

        /**
         * @return Partition metas.
         */
        private Metas getOrAllocatePartitionMetas() throws IgniteCheckedException {
            PageMemoryEx pageMem = (PageMemoryEx)grp.dataRegion().pageMemory();
            IgniteWriteAheadLogManager wal = ctx.wal();

            int grpId = grp.groupId();
            long partMetaId = pageMem.partitionMetaPageId(grpId, partId);

            long partMetaPage = pageMem.acquirePage(grpId, partMetaId);
            try {
                boolean allocated = false;
                boolean pendingTreeAllocated = false;

                long pageAddr = pageMem.writeLock(grpId, partMetaId, partMetaPage);
                try {
                    long treeRoot, reuseListRoot, pendingTreeRoot;

                    // Initialize new page.
                    if (PageIO.getType(pageAddr) != PageIO.T_PART_META) {
                        PagePartitionMetaIO io = PagePartitionMetaIO.VERSIONS.latest();

                        io.initNewPage(pageAddr, partMetaId, pageMem.realPageSize(grpId));

                        treeRoot = pageMem.allocatePage(grpId, partId, PageMemory.FLAG_DATA);
                        reuseListRoot = pageMem.allocatePage(grpId, partId, PageMemory.FLAG_DATA);
                        pendingTreeRoot = pageMem.allocatePage(grpId, partId, PageMemory.FLAG_DATA);

                        assert PageIdUtils.flag(treeRoot) == PageMemory.FLAG_DATA;
                        assert PageIdUtils.flag(reuseListRoot) == PageMemory.FLAG_DATA;
                        assert PageIdUtils.flag(pendingTreeRoot) == PageMemory.FLAG_DATA;

                        io.setTreeRoot(pageAddr, treeRoot);
                        io.setReuseListRoot(pageAddr, reuseListRoot);
                        io.setPendingTreeRoot(pageAddr, pendingTreeRoot);

                        if (PageHandler.isWalDeltaRecordNeeded(pageMem, grpId, partMetaId, partMetaPage, wal, null)) {
                            wal.log(new PageSnapshot(new FullPageId(partMetaId, grpId), pageAddr,
                                pageMem.pageSize(), pageMem.realPageSize(grpId)));
                        }

                        allocated = true;
                    }
                    else {
                        PagePartitionMetaIO io = PageIO.getPageIO(pageAddr);

                        treeRoot = io.getTreeRoot(pageAddr);
                        reuseListRoot = io.getReuseListRoot(pageAddr);

                        int pageVersion = PagePartitionMetaIO.getVersion(pageAddr);

                        if (pageVersion < 2) {
                            assert pageVersion == 1;

                            if (log.isDebugEnabled())
                                log.info("Upgrade partition meta page version: [part=" + partId +
                                    ", grpId=" + grpId + ", oldVer=" + pageVersion +
                                    ", newVer=" + io.getVersion()
                                );

                            io = PagePartitionMetaIO.VERSIONS.latest();

                            ((PagePartitionMetaIOV2)io).upgradePage(pageAddr);

                            pendingTreeRoot = pageMem.allocatePage(grpId, partId, PageMemory.FLAG_DATA);

                            io.setPendingTreeRoot(pageAddr, pendingTreeRoot);

                            if (PageHandler.isWalDeltaRecordNeeded(pageMem, grpId, partMetaId, partMetaPage, wal,
                                null)) {
                                wal.log(new PageSnapshot(new FullPageId(partMetaId, grpId), pageAddr,
                                    pageMem.pageSize(), pageMem.realPageSize(grpId)));
                            }

                            pendingTreeAllocated = true;
                        }
                        else
                            pendingTreeRoot = io.getPendingTreeRoot(pageAddr);

                        if (PageIdUtils.flag(treeRoot) != PageMemory.FLAG_DATA)
                            throw new StorageException("Wrong tree root page id flag: treeRoot="
                                + U.hexLong(treeRoot) + ", part=" + partId + ", grpId=" + grpId);

                        if (PageIdUtils.flag(reuseListRoot) != PageMemory.FLAG_DATA)
                            throw new StorageException("Wrong reuse list root page id flag: reuseListRoot="
                                + U.hexLong(reuseListRoot) + ", part=" + partId + ", grpId=" + grpId);

                        if (PageIdUtils.flag(pendingTreeRoot) != PageMemory.FLAG_DATA)
                            throw new StorageException("Wrong pending tree root page id flag: reuseListRoot="
                                + U.hexLong(reuseListRoot) + ", part=" + partId + ", grpId=" + grpId);
                    }

                    return new Metas(
                        new RootPage(new FullPageId(treeRoot, grpId), allocated),
                        new RootPage(new FullPageId(reuseListRoot, grpId), allocated),
                        new RootPage(new FullPageId(pendingTreeRoot, grpId), allocated || pendingTreeAllocated));
                }
                finally {
                    pageMem.writeUnlock(grpId, partMetaId, partMetaPage, null, allocated || pendingTreeAllocated);
                }
            }
            finally {
                pageMem.releasePage(grpId, partMetaId, partMetaPage);
            }
        }

        /** {@inheritDoc} */
        @Override public int partId() {
            return partId;
        }

        /** {@inheritDoc} */
        @Override public String name() {
            return name;
        }

        /** {@inheritDoc} */
        @Override public RowStore rowStore() {
            CacheDataStore delegate0 = delegate;

            return delegate0 == null ? null : delegate0.rowStore();
        }

        /** {@inheritDoc} */
        @Override public long fullSize() {
            try {
                CacheDataStore delegate0 = init0(true);

                return delegate0 == null ? 0 : delegate0.fullSize();
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public long cacheSize(int cacheId) {
            try {
                CacheDataStore delegate0 = init0(true);

                return delegate0 == null ? 0 : delegate0.cacheSize(cacheId);
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public Map<Integer, Long> cacheSizes() {
            try {
                CacheDataStore delegate0 = init0(true);

                return delegate0 == null ? null : delegate0.cacheSizes();
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public void updateSize(int cacheId, long delta) {
            try {
                CacheDataStore delegate0 = init0(false);

                if (delegate0 != null)
                    delegate0.updateSize(cacheId, delta);
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public long updateCounter() {
            try {
                CacheDataStore delegate0 = init0(true);

                return delegate0 == null ? 0 : delegate0.updateCounter();
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public long getAndIncrementUpdateCounter(long delta) {
            try {
                CacheDataStore delegate0 = init0(true);

                return delegate0 == null ? 0 : delegate0.getAndIncrementUpdateCounter(delta);
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public void init(long size, long updCntr, @Nullable Map<Integer, Long> cacheSizes) {
            throw new IllegalStateException("Should be never called.");
        }

        /** {@inheritDoc} */
        @Override public void updateCounter(long val) {
            try {
                CacheDataStore delegate0 = init0(false);

                if (delegate0 != null)
                    delegate0.updateCounter(val);
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public void updateCounter(long start, long delta) {
            try {
                CacheDataStore delegate0 = init0(false);

                if (delegate0 != null)
                    delegate0.updateCounter(start, delta);
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public void finalizeUpdateCountres() {
            try {
                CacheDataStore delegate0 = init0(true);

                if (delegate0 != null)
                    delegate0.finalizeUpdateCountres();
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public long nextUpdateCounter() {
            try {
                CacheDataStore delegate0 = init0(false);

                return delegate0 == null ? 0 : delegate0.nextUpdateCounter();
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public long initialUpdateCounter() {
            try {
                CacheDataStore delegate0 = init0(true);

                return delegate0 == null ? 0 : delegate0.initialUpdateCounter();
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public void updateInitialCounter(long cntr) {
            try {
                CacheDataStore delegate0 = init0(true);

                if (delegate0 != null)
                    delegate0.updateInitialCounter(cntr);
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public void setRowCacheCleaner(GridQueryRowCacheCleaner rowCacheCleaner) {
            try {
                CacheDataStore delegate0 = init0(true);

                if (delegate0 != null)
                    delegate0.setRowCacheCleaner(rowCacheCleaner);
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public void update(
            GridCacheContext cctx,
            KeyCacheObject key,
            CacheObject val,
            GridCacheVersion ver,
            long expireTime,
            @Nullable CacheDataRow oldRow
        ) throws IgniteCheckedException {
            assert ctx.database().checkpointLockIsHeldByThread();

            CacheDataStore delegate = init0(false);

            delegate.update(cctx, key, val, ver, expireTime, oldRow);
        }

        /** {@inheritDoc} */
        @Override public boolean mvccInitialValue(
            GridCacheContext cctx,
            KeyCacheObject key,
            @Nullable CacheObject val,
            GridCacheVersion ver,
            long expireTime,
            MvccVersion mvccVer,
            MvccVersion newMvccVer)
            throws IgniteCheckedException
        {
            CacheDataStore delegate = init0(false);

            return delegate.mvccInitialValue(cctx, key, val, ver, expireTime, mvccVer, newMvccVer);
        }

        /** {@inheritDoc} */
        @Override public boolean mvccInitialValueIfAbsent(
            GridCacheContext cctx,
            KeyCacheObject key,
            @Nullable CacheObject val,
            GridCacheVersion ver,
            long expireTime,
            MvccVersion mvccVer,
            MvccVersion newMvccVer,
            byte txState,
            byte newTxState)
            throws IgniteCheckedException
        {
            CacheDataStore delegate = init0(false);

            return delegate.mvccInitialValueIfAbsent(cctx, key, val, ver, expireTime, mvccVer, newMvccVer,
                txState, newTxState);
        }

        /** {@inheritDoc} */
        @Override public boolean mvccUpdateRowWithPreloadInfo(
            GridCacheContext cctx,
            KeyCacheObject key,
            @Nullable CacheObject val,
            GridCacheVersion ver,
            long expireTime,
            MvccVersion mvccVer,
            MvccVersion newMvccVer,
            byte mvccTxState,
            byte newMvccTxState) throws IgniteCheckedException {

            CacheDataStore delegate = init0(false);

            return delegate.mvccUpdateRowWithPreloadInfo(cctx,
                key,
                val,
                ver,
                expireTime,
                mvccVer,
                newMvccVer,
                mvccTxState,
                newMvccTxState);
        }

        /** {@inheritDoc} */
        @Override public MvccUpdateResult mvccUpdate(
            GridCacheContext cctx,
            KeyCacheObject key,
            CacheObject val,
            GridCacheVersion ver,
            long expireTime,
            MvccSnapshot mvccVer,
            CacheEntryPredicate filter,
            EntryProcessor entryProc,
            Object[] invokeArgs,
            boolean primary,
            boolean needHistory,
            boolean noCreate,
            boolean needOldVal,
            boolean retVal) throws IgniteCheckedException {
            CacheDataStore delegate = init0(false);

            return delegate.mvccUpdate(cctx, key, val, ver, expireTime, mvccVer, filter, entryProc, invokeArgs, primary,
                needHistory, noCreate, needOldVal, retVal);
        }

        /** {@inheritDoc} */
        @Override public MvccUpdateResult mvccRemove(
            GridCacheContext cctx,
            KeyCacheObject key,
            MvccSnapshot mvccVer,
            CacheEntryPredicate filter,
            boolean primary,
            boolean needHistory,
            boolean needOldVal,
            boolean retVal) throws IgniteCheckedException {
            CacheDataStore delegate = init0(false);

            return delegate.mvccRemove(cctx, key, mvccVer,filter,  primary, needHistory, needOldVal, retVal);
        }

        /** {@inheritDoc} */
        @Override public MvccUpdateResult mvccLock(
            GridCacheContext cctx,
            KeyCacheObject key,
            MvccSnapshot mvccSnapshot) throws IgniteCheckedException {
            CacheDataStore delegate = init0(false);

            return delegate.mvccLock(cctx, key, mvccSnapshot);
        }

        /** {@inheritDoc} */
        @Override public GridLongList mvccUpdateNative(GridCacheContext cctx, boolean primary, KeyCacheObject key, CacheObject val, GridCacheVersion ver, long expireTime, MvccSnapshot mvccSnapshot) throws IgniteCheckedException {
            CacheDataStore delegate = init0(false);

            return delegate.mvccUpdateNative(cctx, primary, key, val, ver, expireTime, mvccSnapshot);
        }

        /** {@inheritDoc} */
        @Override public GridLongList mvccRemoveNative(GridCacheContext cctx, boolean primary, KeyCacheObject key, MvccSnapshot mvccSnapshot) throws IgniteCheckedException {
            CacheDataStore delegate = init0(false);

            return delegate.mvccRemoveNative(cctx, primary, key, mvccSnapshot);
        }

        /** {@inheritDoc} */
        @Override public void mvccRemoveAll(GridCacheContext cctx, KeyCacheObject key) throws IgniteCheckedException {
            CacheDataStore delegate = init0(false);

            delegate.mvccRemoveAll(cctx, key);
        }

        /** {@inheritDoc} */
        @Override public void mvccApplyUpdate(GridCacheContext cctx, KeyCacheObject key, CacheObject val, GridCacheVersion ver,
            long expireTime, MvccVersion mvccVer) throws IgniteCheckedException {
            CacheDataStore delegate = init0(false);

            delegate.mvccApplyUpdate(cctx, key, val, ver, expireTime, mvccVer);
        }

        /** {@inheritDoc} */
        @Override public CacheDataRow createRow(
            GridCacheContext cctx,
            KeyCacheObject key,
            CacheObject val,
            GridCacheVersion ver,
            long expireTime,
            @Nullable CacheDataRow oldRow) throws IgniteCheckedException {
            assert ctx.database().checkpointLockIsHeldByThread();

            CacheDataStore delegate = init0(false);

            return delegate.createRow(cctx, key, val, ver, expireTime, oldRow);
        }

        /** {@inheritDoc} */
        @Override public int cleanup(GridCacheContext cctx,
            @Nullable List<MvccLinkAwareSearchRow> cleanupRows) throws IgniteCheckedException {
            CacheDataStore delegate = init0(false);

            return delegate.cleanup(cctx, cleanupRows);
        }

        /** {@inheritDoc} */
        @Override public void updateTxState(GridCacheContext cctx, CacheSearchRow row) throws IgniteCheckedException {
            CacheDataStore delegate = init0(false);

            delegate.updateTxState(cctx, row);
        }

        /** {@inheritDoc} */
        @Override public void invoke(GridCacheContext cctx, KeyCacheObject key, OffheapInvokeClosure c)
            throws IgniteCheckedException {
            assert ctx.database().checkpointLockIsHeldByThread();

            CacheDataStore delegate = init0(false);

            delegate.invoke(cctx, key, c);
        }

        /** {@inheritDoc} */
        @Override public void remove(GridCacheContext cctx, KeyCacheObject key, int partId)
            throws IgniteCheckedException {
            assert ctx.database().checkpointLockIsHeldByThread();

            CacheDataStore delegate = init0(false);

            delegate.remove(cctx, key, partId);
        }

        /** {@inheritDoc} */
        @Override public CacheDataRow find(GridCacheContext cctx, KeyCacheObject key) throws IgniteCheckedException {
            CacheDataStore delegate = init0(true);

            if (delegate != null)
                return delegate.find(cctx, key);

            return null;
        }

        /** {@inheritDoc} */
        @Override public CacheDataRow mvccFind(GridCacheContext cctx, KeyCacheObject key, MvccSnapshot snapshot)
            throws IgniteCheckedException {
            CacheDataStore delegate = init0(true);

            if (delegate != null)
                return delegate.mvccFind(cctx, key, snapshot);

            return null;
        }

        /** {@inheritDoc} */
        @Override public List<IgniteBiTuple<Object, MvccVersion>> mvccFindAllVersions(GridCacheContext cctx, KeyCacheObject key)
            throws IgniteCheckedException {
            CacheDataStore delegate = init0(true);

            if (delegate != null)
                return delegate.mvccFindAllVersions(cctx, key);

            return Collections.emptyList();
        }

        /** {@inheritDoc} */
        @Override public GridCursor<CacheDataRow> mvccAllVersionsCursor(GridCacheContext cctx,
            KeyCacheObject key, Object x) throws IgniteCheckedException {
            CacheDataStore delegate = init0(true);

            if (delegate != null)
                return delegate.mvccAllVersionsCursor(cctx, key, x);

            return EMPTY_CURSOR;
        }


        /** {@inheritDoc} */
        @Override public GridCursor<? extends CacheDataRow> cursor() throws IgniteCheckedException {
            CacheDataStore delegate = init0(true);

            if (delegate != null)
                return delegate.cursor();

            return EMPTY_CURSOR;
        }

        /** {@inheritDoc} */
        @Override public GridCursor<? extends CacheDataRow> cursor(Object x) throws IgniteCheckedException {
            CacheDataStore delegate = init0(true);

            if (delegate != null)
                return delegate.cursor(x);

            return EMPTY_CURSOR;
        }

        /** {@inheritDoc} */
        @Override public GridCursor<? extends CacheDataRow> cursor(MvccSnapshot mvccSnapshot)
            throws IgniteCheckedException {
            CacheDataStore delegate = init0(true);

            if (delegate != null)
                return delegate.cursor(mvccSnapshot);

            return EMPTY_CURSOR;
        }

        /** {@inheritDoc} */
        @Override public GridCursor<? extends CacheDataRow> cursor(
            int cacheId,
            KeyCacheObject lower,
            KeyCacheObject upper) throws IgniteCheckedException {
            CacheDataStore delegate = init0(true);

            if (delegate != null)
                return delegate.cursor(cacheId, lower, upper);

            return EMPTY_CURSOR;
        }

        /** {@inheritDoc} */
        @Override public GridCursor<? extends CacheDataRow> cursor(int cacheId,
            KeyCacheObject lower,
            KeyCacheObject upper,
            Object x)
            throws IgniteCheckedException {
            CacheDataStore delegate = init0(true);

            if (delegate != null)
                return delegate.cursor(cacheId, lower, upper, x);

            return EMPTY_CURSOR;
        }

        /** {@inheritDoc} */
        @Override public GridCursor<? extends CacheDataRow> cursor(int cacheId,
            KeyCacheObject lower,
            KeyCacheObject upper,
            Object x,
            MvccSnapshot mvccSnapshot)
            throws IgniteCheckedException {
            CacheDataStore delegate = init0(true);

            if (delegate != null)
                return delegate.cursor(cacheId, lower, upper, x, mvccSnapshot);

            return EMPTY_CURSOR;
        }

        /** {@inheritDoc} */
        @Override public void destroy() throws IgniteCheckedException {
            // No need to destroy delegate.
        }

        /** {@inheritDoc} */
        @Override public GridCursor<? extends CacheDataRow> cursor(int cacheId) throws IgniteCheckedException {
            CacheDataStore delegate = init0(true);

            if (delegate != null)
                return delegate.cursor(cacheId);

            return EMPTY_CURSOR;
        }

        /** {@inheritDoc} */
        @Override public GridCursor<? extends CacheDataRow> cursor(int cacheId,
            MvccSnapshot mvccSnapshot) throws IgniteCheckedException {
            CacheDataStore delegate = init0(true);

            if (delegate != null)
                return delegate.cursor(cacheId, mvccSnapshot);

            return EMPTY_CURSOR;
        }

        /** {@inheritDoc} */
        @Override public void clear(int cacheId) throws IgniteCheckedException {
            CacheDataStore delegate0 = init0(true);

            if (delegate0 == null)
                return;

            ctx.database().checkpointReadLock();
            try {
                // Clear persistent pendingTree
                if (pendingTree != null) {
                    PendingRow row = new PendingRow(cacheId);

                    GridCursor<PendingRow> cursor = pendingTree.find(row, row, PendingEntriesTree.WITHOUT_KEY);

                    while (cursor.next()) {
                        PendingRow row0 = cursor.get();

                        assert row0.link != 0 : row;

                        boolean res = pendingTree.removex(row0);

                        assert res;
                    }
                }

                delegate0.clear(cacheId);
            }
            finally {
                ctx.database().checkpointReadUnlock();
            }
        }

        /**
         * Gets the number of entries pending expire.
         *
         * @return Number of pending entries.
         * @throws IgniteCheckedException If failed to get number of pending entries.
         */
        public long expiredSize() throws IgniteCheckedException {
            CacheDataStore delegate0 = init0(true);

            return delegate0 == null ? 0 : pendingTree.size();
        }

        /**
         * Try to remove expired entries from data store.
         *
         * @param cctx Cache context.
         * @param c Expiry closure that should be applied to expired entry. See {@link GridCacheTtlManager} for details.
         * @param amount Limit of processed entries by single call, {@code -1} for no limit.
         * @return cleared entries count.
         * @throws IgniteCheckedException If failed.
         */
        public int purgeExpired(GridCacheContext cctx,
            IgniteInClosure2X<GridCacheEntryEx, GridCacheVersion> c,
            int amount) throws IgniteCheckedException {
            CacheDataStore delegate0 = init0(true);

            long now = U.currentTimeMillis();

            if (delegate0 == null || nextStoreCleanTime > now)
                return 0;

            assert pendingTree != null : "Partition data store was not initialized.";

            int cleared = purgeExpiredInternal(cctx, c, amount);

            // Throttle if there is nothing to clean anymore.
            if (cleared < amount)
                nextStoreCleanTime = now + UNWIND_THROTTLING_TIMEOUT;

            return cleared;
        }

        /**
         * Removes expired entries from data store.
         *
         * @param cctx Cache context.
         * @param c Expiry closure that should be applied to expired entry. See {@link GridCacheTtlManager} for details.
         * @param amount Limit of processed entries by single call, {@code -1} for no limit.
         * @return cleared entries count.
         * @throws IgniteCheckedException If failed.
         */
        private int purgeExpiredInternal(GridCacheContext cctx,
            IgniteInClosure2X<GridCacheEntryEx, GridCacheVersion> c,
            int amount) throws IgniteCheckedException {

            GridDhtLocalPartition part = cctx.topology().localPartition(partId, AffinityTopologyVersion.NONE, false, false);

            // Skip non-owned partitions.
            if (part == null || part.state() != OWNING)
                return 0;

            cctx.shared().database().checkpointReadLock();
            try {
                if (!part.reserve())
                    return 0;

                try {
                    if (part.state() != OWNING)
                        return 0;

                    long now = U.currentTimeMillis();

                    GridCursor<PendingRow> cur;

                    if (grp.sharedGroup())
                        cur = pendingTree.find(new PendingRow(cctx.cacheId()), new PendingRow(cctx.cacheId(), now, 0));
                    else
                        cur = pendingTree.find(null, new PendingRow(CU.UNDEFINED_CACHE_ID, now, 0));

                    if (!cur.next())
                        return 0;

                    GridCacheVersion obsoleteVer = null;

                    int cleared = 0;

                    do {
                        PendingRow row = cur.get();

                        if (amount != -1 && cleared > amount)
                            return cleared;

                        assert row.key != null && row.link != 0 && row.expireTime != 0 : row;

                        row.key.partition(partId);

                        if (pendingTree.removex(row)) {
                            if (obsoleteVer == null)
                                obsoleteVer = ctx.versions().next();

                            GridCacheEntryEx e1 = cctx.cache().entryEx(row.key);

                            if (e1 != null)
                                c.apply(e1, obsoleteVer);
                        }

                        cleared++;
                    }
                    while (cur.next());

                    return cleared;
                }
                finally {
                    part.release();
                }
            }
            finally {
                cctx.shared().database().checkpointReadUnlock();
            }
        }

        /** {@inheritDoc} */
        @Override public PendingEntriesTree pendingTree() {
            try {
                CacheDataStore delegate0 = init0(true);

                return delegate0 == null ? null : pendingTree;
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }
    }

    /**
     *
     */
    public static final GridCursor<CacheDataRow> EMPTY_CURSOR = new GridCursor<CacheDataRow>() {
        /** {@inheritDoc} */
        @Override public boolean next() {
            return false;
        }

        /** {@inheritDoc} */
        @Override public CacheDataRow get() {
            return null;
        }
    };
}