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

import java.util.ArrayList;
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.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import javax.cache.Cache;
import javax.cache.processor.EntryProcessor;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.pagemem.FullPageId;
import org.apache.ignite.internal.pagemem.wal.record.delta.DataPageMvccMarkUpdatedRecord;
import org.apache.ignite.internal.pagemem.wal.record.delta.DataPageMvccUpdateNewTxStateHintRecord;
import org.apache.ignite.internal.pagemem.wal.record.delta.DataPageMvccUpdateTxStateHintRecord;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.dht.colocated.GridDhtDetachedCacheEntry;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.CachePartitionPartialCountersMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtDemandedPartitionsMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteHistoricalIterator;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteRebalanceIteratorImpl;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtInvalidPartitionException;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshotWithoutTxs;
import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
import org.apache.ignite.internal.processors.cache.mvcc.MvccVersion;
import org.apache.ignite.internal.processors.cache.mvcc.txlog.TxState;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter;
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.tree.BPlusTree;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.BPlusIO;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
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.processors.cache.query.GridCacheQueryManager;
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.DataRow;
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.RowLinkIO;
import org.apache.ignite.internal.processors.cache.tree.SearchRow;
import org.apache.ignite.internal.processors.cache.tree.mvcc.data.MvccDataRow;
import org.apache.ignite.internal.processors.cache.tree.mvcc.data.MvccUpdateDataRow;
import org.apache.ignite.internal.processors.cache.tree.mvcc.data.MvccUpdateDataRowNative;
import org.apache.ignite.internal.processors.cache.tree.mvcc.data.MvccUpdateResult;
import org.apache.ignite.internal.processors.cache.tree.mvcc.data.ResultType;
import org.apache.ignite.internal.processors.cache.tree.mvcc.search.MvccFirstRowTreeClosure;
import org.apache.ignite.internal.processors.cache.tree.mvcc.search.MvccLinkAwareSearchRow;
import org.apache.ignite.internal.processors.cache.tree.mvcc.search.MvccMaxSearchRow;
import org.apache.ignite.internal.processors.cache.tree.mvcc.search.MvccMinSearchRow;
import org.apache.ignite.internal.processors.cache.tree.mvcc.search.MvccSnapshotSearchRow;
import org.apache.ignite.internal.processors.cache.tree.mvcc.search.MvccTreeClosure;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.query.GridQueryRowCacheCleaner;
import org.apache.ignite.internal.util.GridAtomicLong;
import org.apache.ignite.internal.util.GridCloseableIteratorAdapter;
import org.apache.ignite.internal.util.GridEmptyCloseableIterator;
import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.GridSpinBusyLock;
import org.apache.ignite.internal.util.GridStripedLock;
import org.apache.ignite.internal.util.IgniteTree;
import org.apache.ignite.internal.util.lang.GridCloseableIterator;
import org.apache.ignite.internal.util.lang.GridCursor;
import org.apache.ignite.internal.util.lang.GridIterator;
import org.apache.ignite.internal.util.lang.IgniteInClosure2X;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgnitePredicate;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

import static org.apache.ignite.internal.pagemem.PageIdAllocator.FLAG_IDX;
import static org.apache.ignite.internal.pagemem.PageIdAllocator.INDEX_PARTITION;
import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.INITIAL_VERSION;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_COUNTER_NA;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_CRD_COUNTER_NA;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_HINTS_BIT_OFF;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_OP_COUNTER_MASK;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_OP_COUNTER_NA;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.compare;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.compareNewVersion;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.isVisible;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.mvccVersionIsValid;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.state;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.unexpectedStateException;
import static org.apache.ignite.internal.processors.cache.persistence.GridCacheOffheapManager.EMPTY_CURSOR;
import static org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO.MVCC_INFO_SIZE;
import static org.apache.ignite.internal.util.IgniteTree.OperationType.NOOP;
import static org.apache.ignite.internal.util.IgniteTree.OperationType.PUT;

/**
 *
 */
@SuppressWarnings("PublicInnerClass")
public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager {
    /**
     * Throttling timeout in millis which avoid excessive PendingTree access on unwind
     * if there is nothing to clean yet.
     */
    public static final long UNWIND_THROTTLING_TIMEOUT = Long.getLong(
        IgniteSystemProperties.IGNITE_UNWIND_THROTTLING_TIMEOUT, 500L);

    /** */
    protected GridCacheSharedContext ctx;

    /** */
    protected CacheGroupContext grp;

    /** */
    protected IgniteLogger log;

    /** */
    private CacheDataStore locCacheDataStore;

    /** */
    protected final ConcurrentMap<Integer, CacheDataStore> partDataStores = new ConcurrentHashMap<>();

    /** */
    private PendingEntriesTree pendingEntries;

    /** */
    protected volatile boolean hasPendingEntries;

    /** Timestamp when next clean try will be allowed. Used for throttling on per-group basis. */
    protected volatile long nextCleanTime;

    /** */
    private final GridAtomicLong globalRmvId = new GridAtomicLong(U.currentTimeMillis() * 1000_000);

    /** */
    protected final GridSpinBusyLock busyLock = new GridSpinBusyLock();

    /** */
    private int updateValSizeThreshold;

    /** */
    protected GridStripedLock partStoreLock = new GridStripedLock(Runtime.getRuntime().availableProcessors());

    /** {@inheritDoc} */
    @Override public GridAtomicLong globalRemoveId() {
        return globalRmvId;
    }

    /** {@inheritDoc} */
    @Override public void start(GridCacheSharedContext ctx, CacheGroupContext grp) throws IgniteCheckedException {
        this.ctx = ctx;
        this.grp = grp;
        this.log = ctx.logger(getClass());

        updateValSizeThreshold = ctx.database().pageSize() / 2;

        if (grp.affinityNode()) {
            ctx.database().checkpointReadLock();

            try {
                initDataStructures();

                if (grp.isLocal())
                    locCacheDataStore = createCacheDataStore(0);
            }
            finally {
                ctx.database().checkpointReadUnlock();
            }
        }
    }

    /** {@inheritDoc} */
    @Override public void onCacheStarted(GridCacheContext cctx) throws IgniteCheckedException {
        initPendingTree(cctx);
    }

    /**
     * @param cctx Cache context.
     * @throws IgniteCheckedException If failed.
     */
    protected void initPendingTree(GridCacheContext cctx) throws IgniteCheckedException {
        assert !cctx.group().persistenceEnabled();

        if (cctx.affinityNode() && cctx.ttl().eagerTtlEnabled() && pendingEntries == null) {
            String name = "PendingEntries";

            long rootPage = allocateForTree();

            pendingEntries = new PendingEntriesTree(
                grp,
                name,
                grp.dataRegion().pageMemory(),
                rootPage,
                grp.reuseList(),
                true);
        }
    }

    /**
     * @throws IgniteCheckedException If failed.
     */
    protected void initDataStructures() throws IgniteCheckedException {
        // No-op.
    }

    /** {@inheritDoc} */
    @Override public void stopCache(int cacheId, final boolean destroy) {
        if (destroy && grp.affinityNode())
            removeCacheData(cacheId);
    }

    /** {@inheritDoc} */
    @Override public void stop() {
        try {
            for (CacheDataStore store : cacheDataStores())
                destroyCacheDataStore(store);

            if (pendingEntries != null)
                pendingEntries.destroy();
        }
        catch (IgniteCheckedException e) {
            throw new IgniteException(e.getMessage(), e);
        }
    }

    /** {@inheritDoc} */
    @Override public void onKernalStop() {
        busyLock.block();
    }

    /**
     * @param cacheId Cache ID.
     */
    private void removeCacheData(int cacheId) {
        assert grp.affinityNode();

        try {
            if (grp.sharedGroup()) {
                assert cacheId != CU.UNDEFINED_CACHE_ID;

                for (CacheDataStore store : cacheDataStores())
                    store.clear(cacheId);

                // Clear non-persistent pending tree if needed.
                if (pendingEntries != null) {
                    PendingRow row = new PendingRow(cacheId);

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

                    while (cursor.next()) {
                        boolean res = pendingEntries.removex(cursor.get());

                        assert res;
                    }
                }
            }
        }
        catch (IgniteCheckedException e) {
            throw new IgniteException(e.getMessage(), e);
        }
    }

    /**
     * @param part Partition.
     * @return Data store for given entry.
     */
    @Override public CacheDataStore dataStore(GridDhtLocalPartition part) {
        if (grp.isLocal())
            return locCacheDataStore;
        else {
            assert part != null;

            return part.dataStore();
        }
    }

    /**
     * @param part Partition.
     * @return Data store for given entry.
     */
    public CacheDataStore dataStore(int part) {
        return grp.isLocal() ? locCacheDataStore : partDataStores.get(part);
    }

    /** {@inheritDoc} */
    @Override public long cacheEntriesCount(int cacheId) {
        long size = 0;

        for (CacheDataStore store : cacheDataStores())
            size += store.cacheSize(cacheId);

        return size;
    }

    /** {@inheritDoc} */
    @Override public long totalPartitionEntriesCount(int p) {
        if (grp.isLocal())
            return locCacheDataStore.fullSize();
        else {
            GridDhtLocalPartition part = grp.topology().localPartition(p, AffinityTopologyVersion.NONE, false, true);

            return part != null ? part.dataStore().fullSize() : 0;
        }
    }

    /**
     * @param p Partition.
     * @return Partition data.
     */
    @Nullable private CacheDataStore partitionData(int p) {
        if (grp.isLocal())
            return locCacheDataStore;
        else {
            GridDhtLocalPartition part = grp.topology().localPartition(p, AffinityTopologyVersion.NONE, false, true);

            return part != null ? part.dataStore() : null;
        }
    }

    /** {@inheritDoc} */
    @Override public long cacheEntriesCount(
        int cacheId,
        boolean primary,
        boolean backup,
        AffinityTopologyVersion topVer
    ) throws IgniteCheckedException {
        if (grp.isLocal())
            return cacheEntriesCount(cacheId, 0);
        else {
            long cnt = 0;

            Iterator<CacheDataStore> it = cacheData(primary, backup, topVer);

            while (it.hasNext())
                cnt += it.next().cacheSize(cacheId);

            return cnt;
        }
    }

    /** {@inheritDoc} */
    @Override public long cacheEntriesCount(int cacheId, int part) {
        CacheDataStore store = partitionData(part);

        return store == null ? 0 : store.cacheSize(cacheId);
    }

    /**
     * @param primary Primary data flag.
     * @param backup Primary data flag.
     * @param topVer Topology version.
     * @return Data stores iterator.
     */
    private Iterator<CacheDataStore> cacheData(boolean primary, boolean backup, AffinityTopologyVersion topVer) {
        assert primary || backup;

        if (grp.isLocal())
            return singletonIterator(locCacheDataStore);
        else {
            final Iterator<GridDhtLocalPartition> it = grp.topology().currentLocalPartitions().iterator();

            if (primary && backup) {
                return F.iterator(it, new IgniteClosure<GridDhtLocalPartition, CacheDataStore>() {
                    @Override public CacheDataStore apply(GridDhtLocalPartition part) {
                        return part.dataStore();
                    }
                }, true);
            }

            final Set<Integer> parts = primary ? grp.affinity().primaryPartitions(ctx.localNodeId(), topVer) :
                grp.affinity().backupPartitions(ctx.localNodeId(), topVer);

            return F.iterator(it, new IgniteClosure<GridDhtLocalPartition, CacheDataStore>() {
                    @Override public CacheDataStore apply(GridDhtLocalPartition part) {
                        return part.dataStore();
                    }
                }, true,
                new IgnitePredicate<GridDhtLocalPartition>() {
                    @Override public boolean apply(GridDhtLocalPartition part) {
                        return parts.contains(part.id());
                    }
                });
        }
    }

    /** {@inheritDoc} */
    @Override public void invoke(
        GridCacheContext cctx,
        KeyCacheObject key,
        GridDhtLocalPartition part,
        OffheapInvokeClosure c)
        throws IgniteCheckedException {
        dataStore(part).invoke(cctx, key, c);
    }

    /** {@inheritDoc} */
    @Override public void update(
        GridCacheContext cctx,
        KeyCacheObject key,
        CacheObject val,
        GridCacheVersion ver,
        long expireTime,
        GridDhtLocalPartition part,
        @Nullable CacheDataRow oldRow
    ) throws IgniteCheckedException {
        assert expireTime >= 0;

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

    /** {@inheritDoc} */
    @Override public boolean mvccInitialValue(
        GridCacheMapEntry entry,
        CacheObject val,
        GridCacheVersion ver,
        long expireTime,
        MvccVersion mvccVer,
        MvccVersion newMvccVer) throws IgniteCheckedException {
        return dataStore(entry.localPartition()).mvccInitialValue(
            entry.context(),
            entry.key(),
            val,
            ver,
            expireTime,
            mvccVer,
            newMvccVer);
    }

    /** {@inheritDoc} */
    @Override public boolean mvccInitialValueIfAbsent(
        GridCacheMapEntry entry,
        CacheObject val,
        GridCacheVersion ver,
        long expireTime,
        MvccVersion mvccVer,
        MvccVersion newMvccVer,
        byte txState,
        byte newTxState) throws IgniteCheckedException {
        return dataStore(entry.localPartition()).mvccInitialValueIfAbsent(
            entry.context(),
            entry.key(),
            val,
            ver,
            expireTime,
            mvccVer,
            newMvccVer,
            txState,
            newTxState);
    }

    /** {@inheritDoc} */
    @Override public boolean mvccUpdateRowWithPreloadInfo(
        GridCacheMapEntry entry,
        @Nullable CacheObject val,
        GridCacheVersion ver,
        long expireTime,
        MvccVersion mvccVer,
        MvccVersion newMvccVer,
        byte mvccTxState,
        byte newMvccTxState
    ) throws IgniteCheckedException {
        assert entry.lockedByCurrentThread();

        return dataStore(entry.localPartition()).mvccUpdateRowWithPreloadInfo(
            entry.context(),
            entry.key(),
            val,
            ver,
            expireTime,
            mvccVer,
            newMvccVer,
            mvccTxState,
            newMvccTxState
        );
    }

    /** {@inheritDoc} */
    @Override public MvccUpdateResult mvccUpdate(
        GridCacheMapEntry entry,
        CacheObject val,
        GridCacheVersion ver,
        long expireTime,
        MvccSnapshot mvccSnapshot,
        boolean primary,
        boolean needHistory,
        boolean noCreate,
        boolean needOldVal,
        @Nullable CacheEntryPredicate filter,
        boolean retVal,
        EntryProcessor entryProc,
        Object[] invokeArgs) throws IgniteCheckedException {
        if (entry.detached() || entry.isNear())
            return null;

        assert entry.lockedByCurrentThread();

        return dataStore(entry.localPartition()).mvccUpdate(entry.context(),
            entry.key(),
            val,
            ver,
            expireTime,
            mvccSnapshot,
            filter,
            entryProc,
            invokeArgs,
            primary,
            needHistory,
            noCreate,
            needOldVal,
            retVal);
    }

    /** {@inheritDoc} */
    @Override public MvccUpdateResult mvccRemove(
        GridCacheMapEntry entry,
        MvccSnapshot mvccSnapshot,
        boolean primary,
        boolean needHistory,
        boolean needOldVal,
        @Nullable CacheEntryPredicate filter,
        boolean retVal) throws IgniteCheckedException {
        if (entry.detached() || entry.isNear())
            return null;

        assert entry.lockedByCurrentThread();

        return dataStore(entry.localPartition()).mvccRemove(entry.context(),
            entry.key(),
            mvccSnapshot,
            filter,
            primary,
            needHistory,
            needOldVal,
            retVal);
    }

    /** {@inheritDoc} */
    @Override public GridLongList mvccUpdateNative(
        boolean primary,
        GridCacheMapEntry entry,
        CacheObject val,
        GridCacheVersion ver,
        long expireTime,
        MvccSnapshot mvccSnapshot) throws IgniteCheckedException {
        if (entry.detached() || entry.isNear())
            return null;

        return dataStore(entry.localPartition()).mvccUpdateNative(entry.context(),
            primary,
            entry.key(),
            val,
            ver,
            expireTime,
            mvccSnapshot);
    }

    /** {@inheritDoc} */
    @Override public GridLongList mvccRemoveNative(
        boolean primary,
        GridCacheMapEntry entry,
        MvccSnapshot mvccSnapshot
    ) throws IgniteCheckedException {
        if (entry.detached() || entry.isNear())
            return null;

        return dataStore(entry.localPartition()).mvccRemoveNative(entry.context(),
            primary,
            entry.key(),
            mvccSnapshot);
    }

    /** {@inheritDoc} */
    @Override public void mvccRemoveAll(GridCacheMapEntry entry) throws IgniteCheckedException {
        if (entry.detached() || entry.isNear())
            return;

        dataStore(entry.localPartition()).mvccRemoveAll(entry.context(), entry.key());
    }

    /** {@inheritDoc} */
    @Nullable @Override public MvccUpdateResult mvccLock(GridCacheMapEntry entry,
        MvccSnapshot mvccSnapshot) throws IgniteCheckedException {
        if (entry.detached() || entry.isNear())
            return null;

        assert entry.lockedByCurrentThread();

        return dataStore(entry.localPartition()).mvccLock(entry.context(), entry.key(), mvccSnapshot);
    }

    /** {@inheritDoc} */
    @Override public void mvccApplyUpdate(
        GridCacheContext cctx,
        KeyCacheObject key,
        CacheObject val,
        GridCacheVersion ver,
        long expireTime,
        GridDhtLocalPartition part,
        MvccVersion mvccVer) throws IgniteCheckedException {

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

    /** {@inheritDoc} */
    @Override public void remove(
        GridCacheContext cctx,
        KeyCacheObject key,
        int partId,
        GridDhtLocalPartition part
    ) throws IgniteCheckedException {
        dataStore(part).remove(cctx, key, partId);
    }

    /** {@inheritDoc} */
    @SuppressWarnings("unchecked")
    @Override @Nullable public CacheDataRow read(GridCacheMapEntry entry)
        throws IgniteCheckedException {
        KeyCacheObject key = entry.key();

        assert grp.isLocal() || entry.localPartition() != null : entry;

        return dataStore(entry.localPartition()).find(entry.context(), key);
    }

    /** {@inheritDoc} */
    @Nullable @Override public CacheDataRow read(GridCacheContext cctx, KeyCacheObject key)
        throws IgniteCheckedException {
        CacheDataStore dataStore = dataStore(cctx, key);

        CacheDataRow row = dataStore != null ? dataStore.find(cctx, key) : null;

        assert row == null || row.value() != null : row;

        return row;
    }

    /** {@inheritDoc} */
    @Nullable @Override public CacheDataRow mvccRead(GridCacheContext cctx, KeyCacheObject key, MvccSnapshot ver)
        throws IgniteCheckedException {
        assert ver != null;

        CacheDataStore dataStore = dataStore(cctx, key);

        CacheDataRow row = dataStore != null ? dataStore.mvccFind(cctx, key, ver) : null;

        assert row == null || row.value() != null : row;

        return row;
    }

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

        return dataStore != null ? dataStore.mvccFindAllVersions(cctx, key) :
            Collections.emptyList();
    }

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

        return dataStore != null ? dataStore.mvccAllVersionsCursor(cctx, key, x) : EMPTY_CURSOR;
    }

    /**
     * @param cctx Cache context.
     * @param key Key.
     * @return Data store.
     */
    @Nullable private CacheDataStore dataStore(GridCacheContext cctx, KeyCacheObject key) {
        if (grp.isLocal())
            return locCacheDataStore;

        GridDhtLocalPartition part = grp.topology().localPartition(cctx.affinity().partition(key), null, false);

        return part != null ? dataStore(part) : null;
    }

    /** {@inheritDoc} */
    @Override public boolean containsKey(GridCacheMapEntry entry) {
        try {
            return read(entry) != null;
        }
        catch (IgniteCheckedException e) {
            U.error(log, "Failed to read value", e);

            return false;
        }
    }

    /** {@inheritDoc} */
    @Override public void onPartitionCounterUpdated(int part, long cntr) {
        // No-op.
    }

    /** {@inheritDoc} */
    @Override public void onPartitionInitialCounterUpdated(int part, long cntr) {
        // No-op.
    }

    /** {@inheritDoc} */
    @Override public long lastUpdatedPartitionCounter(int part) {
        return 0;
    }

    /**
     * Clears offheap entries.
     *
     * @param readers {@code True} to clear readers.
     */
    @SuppressWarnings("unchecked")
    @Override public void clearCache(GridCacheContext cctx, boolean readers) {
        GridCacheVersion obsoleteVer = null;

        try (GridCloseableIterator<CacheDataRow> it = grp.isLocal() ? iterator(cctx.cacheId(), cacheDataStores().iterator(), null) :
            evictionSafeIterator(cctx.cacheId(), cacheDataStores().iterator())) {
            while (it.hasNext()) {
                cctx.shared().database().checkpointReadLock();

                try {
                    KeyCacheObject key = it.next().key();

                    try {
                        if (obsoleteVer == null)
                            obsoleteVer = ctx.versions().next();

                        GridCacheEntryEx entry = cctx.cache().entryEx(key);

                        entry.clear(obsoleteVer, readers);
                    }
                    catch (GridDhtInvalidPartitionException ignore) {
                        // Ignore.
                    }
                    catch (IgniteCheckedException e) {
                        U.error(log, "Failed to clear cache entry: " + key, e);
                    }
                }
                finally {
                    cctx.shared().database().checkpointReadUnlock();
                }
            }
        }
        catch (IgniteCheckedException e) {
            U.error(log, "Failed to close iterator", e);
        }
    }

    /** {@inheritDoc} */
    @Override public int onUndeploy(ClassLoader ldr) {
        // TODO: GG-11141.
        return 0;
    }

    /** {@inheritDoc} */
    @Override public long offHeapAllocatedSize() {
        // TODO GG-10884.
        return 0;
    }

    /**
     * @param primary {@code True} if need return primary entries.
     * @param backup {@code True} if need return backup entries.
     * @param topVer Topology version to use.
     * @return Entries iterator.
     * @throws IgniteCheckedException If failed.
     */
    @SuppressWarnings("unchecked")
    @Override public <K, V> GridCloseableIterator<Cache.Entry<K, V>> cacheEntriesIterator(
        final GridCacheContext cctx,
        final boolean primary,
        final boolean backup,
        final AffinityTopologyVersion topVer,
        final boolean keepBinary) throws IgniteCheckedException {
        final Iterator<CacheDataRow> it = cacheIterator(cctx.cacheId(), primary, backup, topVer, null);

        return new GridCloseableIteratorAdapter<Cache.Entry<K, V>>() {
            /** */
            private CacheEntryImplEx next;

            @Override protected Cache.Entry<K, V> onNext() {
                CacheEntryImplEx ret = next;

                next = null;

                return ret;
            }

            @Override protected boolean onHasNext() {
                if (next != null)
                    return true;

                CacheDataRow nextRow = null;

                if (it.hasNext())
                    nextRow = it.next();

                if (nextRow != null) {
                    KeyCacheObject key = nextRow.key();
                    CacheObject val = nextRow.value();

                    Object key0 = cctx.unwrapBinaryIfNeeded(key, keepBinary, false);
                    Object val0 = cctx.unwrapBinaryIfNeeded(val, keepBinary, false);

                    next = new CacheEntryImplEx(key0, val0, nextRow.version());

                    return true;
                }

                return false;
            }
        };
    }

    /** {@inheritDoc} */
    @Override public GridCloseableIterator<KeyCacheObject> cacheKeysIterator(int cacheId, final int part)
        throws IgniteCheckedException {
        CacheDataStore data = partitionData(part);

        if (data == null)
            return new GridEmptyCloseableIterator<>();

        final GridCursor<? extends CacheDataRow> cur =
            data.cursor(cacheId, null, null, CacheDataRowAdapter.RowData.KEY_ONLY);

        return new GridCloseableIteratorAdapter<KeyCacheObject>() {
            /** */
            private KeyCacheObject next;

            @Override protected KeyCacheObject onNext() {
                KeyCacheObject res = next;

                next = null;

                return res;
            }

            @Override protected boolean onHasNext() throws IgniteCheckedException {
                if (next != null)
                    return true;

                if (cur.next()) {
                    CacheDataRow row = cur.get();

                    next = row.key();
                }

                return next != null;
            }
        };
    }

    /** {@inheritDoc} */
    @Override public GridIterator<CacheDataRow> cacheIterator(
        int cacheId,
        boolean primary,
        boolean backups,
        final AffinityTopologyVersion topVer,
        @Nullable MvccSnapshot mvccSnapshot)
        throws IgniteCheckedException {
        return iterator(cacheId, cacheData(primary, backups, topVer), mvccSnapshot);
    }

    /** {@inheritDoc} */
    @Override public GridIterator<CacheDataRow> cachePartitionIterator(int cacheId, int part,
        @Nullable MvccSnapshot mvccSnapshot) throws IgniteCheckedException {
        CacheDataStore data = partitionData(part);

        if (data == null)
            return new GridEmptyCloseableIterator<>();

        return iterator(cacheId, singletonIterator(data), mvccSnapshot);
    }

    /** {@inheritDoc} */
    @Override public GridIterator<CacheDataRow> partitionIterator(int part) throws IgniteCheckedException {
        CacheDataStore data = partitionData(part);

        if (data == null)
            return new GridEmptyCloseableIterator<>();

        return iterator(CU.UNDEFINED_CACHE_ID, singletonIterator(data), null);
    }

    /**
     *
     * @param cacheId Cache ID.
     * @param dataIt Data store iterator.
     * @param mvccSnapshot Mvcc snapshot.
     * @return Rows iterator
     */
    private GridCloseableIterator<CacheDataRow> iterator(final int cacheId,
        final Iterator<CacheDataStore> dataIt,
        final MvccSnapshot mvccSnapshot)
    {
        return new GridCloseableIteratorAdapter<CacheDataRow>() {
            /** */
            private GridCursor<? extends CacheDataRow> cur;

            /** */
            private int curPart;

            /** */
            private CacheDataRow next;

            @Override protected CacheDataRow onNext() {
                CacheDataRow res = next;

                next = null;

                return res;
            }

            @Override protected boolean onHasNext() throws IgniteCheckedException {
                if (next != null)
                    return true;

                while (true) {
                    if (cur == null) {
                        if (dataIt.hasNext()) {
                            CacheDataStore ds = dataIt.next();

                            curPart = ds.partId();

                            if (mvccSnapshot == null)
                                cur = cacheId == CU.UNDEFINED_CACHE_ID ? ds.cursor() : ds.cursor(cacheId);
                            else {
                                cur = cacheId == CU.UNDEFINED_CACHE_ID ?
                                    ds.cursor(mvccSnapshot) : ds.cursor(cacheId, mvccSnapshot);
                            }
                        }
                        else
                            break;
                    }

                    if (cur.next()) {
                        next = cur.get();
                        next.key().partition(curPart);

                        break;
                    }
                    else
                        cur = null;
                }

                return next != null;
            }
        };
    }

    /**
     * @param cacheId Cache ID.
     * @param dataIt Data store iterator.
     * @return Rows iterator
     */
    private GridCloseableIterator<CacheDataRow> evictionSafeIterator(final int cacheId, final Iterator<CacheDataStore> dataIt) {
        return new GridCloseableIteratorAdapter<CacheDataRow>() {
            /** */
            private GridCursor<? extends CacheDataRow> cur;

            /** */
            private GridDhtLocalPartition curPart;

            /** */
            private CacheDataRow next;

            @Override protected CacheDataRow onNext() {
                CacheDataRow res = next;

                next = null;

                return res;
            }

            @Override protected boolean onHasNext() throws IgniteCheckedException {
                if (next != null)
                    return true;

                while (true) {
                    if (cur == null) {
                        if (dataIt.hasNext()) {
                            CacheDataStore ds = dataIt.next();

                            if (!reservePartition(ds.partId()))
                                continue;

                            cur = cacheId == CU.UNDEFINED_CACHE_ID ? ds.cursor() : ds.cursor(cacheId);
                        }
                        else
                            break;
                    }

                    if (cur.next()) {
                        next = cur.get();
                        next.key().partition(curPart.id());

                        break;
                    }
                    else {
                        cur = null;

                        releaseCurrentPartition();
                    }
                }

                return next != null;
            }

            /** */
            private void releaseCurrentPartition() {
                GridDhtLocalPartition p = curPart;

                assert p != null;

                curPart = null;

                p.release();
            }

            /**
             * @param partId Partition number.
             * @return {@code True} if partition was reserved.
             */
            private boolean reservePartition(int partId) {
                GridDhtLocalPartition p = grp.topology().localPartition(partId);

                if (p != null && p.reserve()) {
                    curPart = p;

                    return true;
                }

                return false;
            }

            /** {@inheritDoc} */
            @Override protected void onClose() throws IgniteCheckedException {
                if (curPart != null)
                    releaseCurrentPartition();
            }
        };
    }

    /**
     * @param item Item.
     * @return Single item iterator.
     */
    private <T> Iterator<T> singletonIterator(final T item) {
        return new Iterator<T>() {
            /** */
            private boolean hasNext = true;

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

            /** {@inheritDoc} */
            @Override public T next() {
                if (hasNext) {
                    hasNext = false;

                    return item;
                }

                throw new NoSuchElementException();
            }

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

    /**
     * @return Page ID.
     * @throws IgniteCheckedException If failed.
     */
    private long allocateForTree() throws IgniteCheckedException {
        ReuseList reuseList = grp.reuseList();

        long pageId;

        if (reuseList == null || (pageId = reuseList.takeRecycledPage()) == 0L)
            pageId = grp.dataRegion().pageMemory().allocatePage(grp.groupId(), INDEX_PARTITION, FLAG_IDX);

        return pageId;
    }

    /** {@inheritDoc} */
    @Override public RootPage rootPageForIndex(int cacheId, String idxName) throws IgniteCheckedException {
        long pageId = allocateForTree();

        return new RootPage(new FullPageId(pageId, grp.groupId()), true);
    }

    /** {@inheritDoc} */
    @Override public void dropRootPageForIndex(int cacheId, String idxName) throws IgniteCheckedException {
        // No-op.
    }

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

    /** {@inheritDoc} */
    @Override public GridCloseableIterator<CacheDataRow> reservedIterator(int part,
        AffinityTopologyVersion topVer) throws IgniteCheckedException {
        final GridDhtLocalPartition loc = grp.topology().localPartition(part, topVer, false);

        if (loc == null || !loc.reserve())
            return null;

        // It is necessary to check state after reservation to avoid race conditions.
        if (loc.state() != OWNING) {
            loc.release();

            return null;
        }

        CacheDataStore data = partitionData(part);

        final GridCursor<? extends CacheDataRow> cur = data.cursor();

        return new GridCloseableIteratorAdapter<CacheDataRow>() {
            /** */
            private CacheDataRow next;

            @Override protected CacheDataRow onNext() {
                CacheDataRow res = next;

                next = null;

                return res;
            }

            @Override protected boolean onHasNext() throws IgniteCheckedException {
                if (next != null)
                    return true;

                if (cur.next())
                    next = cur.get();

                return next != null;
            }

            @Override protected void onClose() throws IgniteCheckedException {
                assert loc != null && loc.state() == OWNING && loc.reservations() > 0
                    : "Partition should be in OWNING state and has at least 1 reservation: " + loc;

                loc.release();
            }
        };
    }

    /** {@inheritDoc} */
    @Override public IgniteRebalanceIterator rebalanceIterator(IgniteDhtDemandedPartitionsMap parts,
        final AffinityTopologyVersion topVer)
        throws IgniteCheckedException {

        final TreeMap<Integer, GridCloseableIterator<CacheDataRow>> iterators = new TreeMap<>();

        Set<Integer> missing = new HashSet<>();

        for (Integer p : parts.fullSet()) {
            GridCloseableIterator<CacheDataRow> partIter = reservedIterator(p, topVer);

            if (partIter == null) {
                missing.add(p);

                continue;
            }

            iterators.put(p, partIter);
        }

        IgniteHistoricalIterator historicalIterator = historicalIterator(parts.historicalMap(), missing);

        IgniteRebalanceIterator iter = new IgniteRebalanceIteratorImpl(iterators, historicalIterator);

        for (Integer p : missing)
            iter.setPartitionMissing(p);

        return iter;
    }

    /**
     * @param partCntrs Partition counters map.
     * @param missing Set of partitions need to populate if partition is missing or failed to reserve.
     * @return Historical iterator.
     */
    @Nullable protected IgniteHistoricalIterator historicalIterator(CachePartitionPartialCountersMap partCntrs, Set<Integer> missing)
        throws IgniteCheckedException {
        return null;
    }

    /** {@inheritDoc} */
    @Override public final CacheDataStore createCacheDataStore(int p) throws IgniteCheckedException {
        CacheDataStore dataStore;

        partStoreLock.lock(p);

        try {
            assert !partDataStores.containsKey(p);

            dataStore = createCacheDataStore0(p);

            partDataStores.put(p, dataStore);
        }
        finally {
            partStoreLock.unlock(p);
        }

        return dataStore;
    }

    /**
     * @param p Partition.
     * @return Cache data store.
     * @throws IgniteCheckedException If failed.
     */
    protected CacheDataStore createCacheDataStore0(int p)
        throws IgniteCheckedException {
        final long rootPage = allocateForTree();

        CacheDataRowStore rowStore = new CacheDataRowStore(grp, grp.freeList(), p);

        String idxName = treeName(p);

        CacheDataTree dataTree = new CacheDataTree(
            grp,
            idxName,
            grp.reuseList(),
            rowStore,
            rootPage,
            true);

        return new CacheDataStoreImpl(p, idxName, rowStore, dataTree);
    }

    /** {@inheritDoc} */
    @Override public Iterable<CacheDataStore> cacheDataStores() {
        if (grp.isLocal())
            return Collections.singleton(locCacheDataStore);

        return new Iterable<CacheDataStore>() {
            @Override public Iterator<CacheDataStore> iterator() {
                return partDataStores.values().iterator();
            }
        };
    }

    /** {@inheritDoc} */
    @Override public final void destroyCacheDataStore(CacheDataStore store) throws IgniteCheckedException {
        int p = store.partId();

        partStoreLock.lock(p);

        try {
            boolean removed = partDataStores.remove(p, store);

            assert removed;

            destroyCacheDataStore0(store);
        }
        catch (IgniteCheckedException e) {
            throw new IgniteException(e);
        }
        finally {
            partStoreLock.unlock(p);
        }
    }

    /**
     * @param store Cache data store.
     * @throws IgniteCheckedException If failed.
     */
    protected void destroyCacheDataStore0(CacheDataStore store) throws IgniteCheckedException {
        store.destroy();
    }

    /**
     * @param p Partition.
     * @return Tree name for given partition.
     */
    protected final String treeName(int p) {
        return BPlusTree.treeName("p-" + p, "CacheData");
    }

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

        assert pendingEntries != null;

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

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

        return amount != -1 && cleared >= amount;
    }

    /**
     * @param cctx Cache context.
     * @param c Closure.
     * @param amount Limit of processed entries by single call, {@code -1} for no limit.
     * @return cleared entries count.
     * @throws IgniteCheckedException If failed.
     */
    private int expireInternal(
        GridCacheContext cctx,
        IgniteInClosure2X<GridCacheEntryEx, GridCacheVersion> c,
        int amount
    ) throws IgniteCheckedException {
        long now = U.currentTimeMillis();

        GridCacheVersion obsoleteVer = null;

        GridCursor<PendingRow> cur;

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

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

        if (!busyLock.enterBusy())
            return 0;

        try {
            int cleared = 0;

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

                PendingRow row = cur.get();

                if (row.key.partition() == -1)
                    row.key.partition(cctx.affinity().partition(row.key));

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

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

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

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

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

            return cleared;
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /** {@inheritDoc} */
    @Override public long expiredSize() throws IgniteCheckedException {
        return pendingEntries != null ? pendingEntries.size() : 0;
    }

    /**
     *
     */
    protected class CacheDataStoreImpl implements CacheDataStore {
        /** */
        private final int partId;

        /** Tree name. */
        private String name;

        /** */
        private final CacheDataRowStore rowStore;

        /** */
        private final CacheDataTree dataTree;

        /** Update counter. */
        protected final PartitionUpdateCounter pCntr = new PartitionUpdateCounter(log);

        /** Partition size. */
        private final AtomicLong storageSize = new AtomicLong();

        /** */
        private final ConcurrentMap<Integer, AtomicLong> cacheSizes = new ConcurrentHashMap<>();

        /** Mvcc remove handler. */
        private final PageHandler<MvccVersion, Boolean> mvccUpdateMarker = new MvccMarkUpdatedHandler();

        /** Mvcc update tx state hint handler. */
        private final PageHandler<Void, Boolean> mvccUpdateTxStateHint = new MvccUpdateTxStateHintHandler();

        /** */
        private final PageHandler<MvccDataRow, Boolean> mvccApplyChanges = new MvccApplyChangesHandler();

        /**
         * @param partId Partition number.
         * @param name Name.
         * @param rowStore Row store.
         * @param dataTree Data tree.
         */
        public CacheDataStoreImpl(
            int partId,
            String name,
            CacheDataRowStore rowStore,
            CacheDataTree dataTree
        ) {
            this.partId = partId;
            this.name = name;
            this.rowStore = rowStore;
            this.dataTree = dataTree;
        }

        /**
         * @param cacheId Cache ID.
         */
        void incrementSize(int cacheId) {
            updateSize(cacheId, 1);
        }

        /**
         * @param cacheId Cache ID.
         */
        void decrementSize(int cacheId) {
            updateSize(cacheId, -1);
        }

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

        /** {@inheritDoc} */
        @Override public long cacheSize(int cacheId) {
            if (grp.sharedGroup()) {
                AtomicLong size = cacheSizes.get(cacheId);

                return size != null ? (int)size.get() : 0;
            }

            return storageSize.get();
        }

        /** {@inheritDoc} */
        @Override public Map<Integer, Long> cacheSizes() {
            if (!grp.sharedGroup())
                return null;

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

            for (Map.Entry<Integer, AtomicLong> e : cacheSizes.entrySet())
                res.put(e.getKey(), e.getValue().longValue());

            return res;
        }

        /** {@inheritDoc} */
        @Override public long fullSize() {
            return storageSize.get();
        }

        /** {@inheritDoc} */
        @Override public void updateSize(int cacheId, long delta) {
            storageSize.addAndGet(delta);

            if (grp.sharedGroup()) {
                AtomicLong size = cacheSizes.get(cacheId);

                if (size == null) {
                    AtomicLong old = cacheSizes.putIfAbsent(cacheId, size = new AtomicLong());

                    if (old != null)
                        size = old;
                }

                size.addAndGet(delta);
            }
        }

        /** {@inheritDoc} */
        @Override public long nextUpdateCounter() {
            return pCntr.next();
        }

        /** {@inheritDoc} */
        @Override public long initialUpdateCounter() {
            return pCntr.initial();
        }

        /** {@inheritDoc} */
        @Override public void updateInitialCounter(long cntr) {
            pCntr.updateInitial(cntr);
        }

        /** {@inheritDoc} */
        @Override public long getAndIncrementUpdateCounter(long delta) {
            return pCntr.getAndAdd(delta);
        }

        /** {@inheritDoc} */
        @Override public long updateCounter() {
            return pCntr.get();
        }

        /** {@inheritDoc} */
        @Override public void updateCounter(long val) {
            pCntr.update(val);
        }

        /** {@inheritDoc} */
        @Override public void updateCounter(long start, long delta) {
            pCntr.update(start, delta);
        }

        /** {@inheritDoc} */
        @Override public void finalizeUpdateCountres() {
            pCntr.finalizeUpdateCountres();
        }

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

        /**
         * @param cctx Cache context.
         * @param oldRow Old row.
         * @param dataRow New row.
         * @return {@code True} if it is possible to update old row data.
         * @throws IgniteCheckedException If failed.
         */
        private boolean canUpdateOldRow(GridCacheContext cctx, @Nullable CacheDataRow oldRow, DataRow dataRow)
            throws IgniteCheckedException {
            if (oldRow == null || cctx.queries().enabled() || grp.mvccEnabled())
                return false;

            if (oldRow.expireTime() != dataRow.expireTime())
                return false;

            // Use grp.sharedGroup() flag since it is possible cacheId is not yet set here.
            boolean sizeWithCacheId = grp.sharedGroup();

            int oldLen = oldRow.size();

            if (oldLen > updateValSizeThreshold)
                return false;

            int newLen = dataRow.size();

            return oldLen == newLen;
        }

        /** {@inheritDoc} */
        @Override public void invoke(GridCacheContext cctx, KeyCacheObject key, OffheapInvokeClosure c)
            throws IgniteCheckedException {
            int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

            invoke0(cctx, new SearchRow(cacheId, key), c);
        }

        /**
         * @param cctx Cache context.
         * @param row Search row.
         * @param c Closure.
         * @throws IgniteCheckedException If failed.
         */
        private void invoke0(GridCacheContext cctx, CacheSearchRow row, OffheapInvokeClosure c)
            throws IgniteCheckedException {
            if (!busyLock.enterBusy())
                throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

            try {
                assert cctx.shared().database().checkpointLockIsHeldByThread();

                dataTree.invoke(row, CacheDataRowAdapter.RowData.NO_KEY, c);

                switch (c.operationType()) {
                    case PUT: {
                        assert c.newRow() != null : c;

                        CacheDataRow oldRow = c.oldRow();

                        finishUpdate(cctx, c.newRow(), oldRow);

                        break;
                    }

                    case REMOVE: {
                        CacheDataRow oldRow = c.oldRow();

                        finishRemove(cctx, row.key(), oldRow);

                        break;
                    }

                    case NOOP:
                        break;

                    default:
                        assert false : c.operationType();
                }
            }
            finally {
                busyLock.leaveBusy();
            }
        }

        /** {@inheritDoc} */
        @Override public CacheDataRow createRow(
            GridCacheContext cctx,
            KeyCacheObject key,
            CacheObject val,
            GridCacheVersion ver,
            long expireTime,
            @Nullable CacheDataRow oldRow) throws IgniteCheckedException {
            int cacheId = grp.storeCacheIdInDataPage() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

            DataRow dataRow = makeDataRow(key, val, ver, expireTime, cacheId);

            if (canUpdateOldRow(cctx, oldRow, dataRow) && rowStore.updateRow(oldRow.link(), dataRow))
                dataRow.link(oldRow.link());
            else {
                CacheObjectContext coCtx = cctx.cacheObjectContext();

                key.valueBytes(coCtx);
                val.valueBytes(coCtx);

                rowStore.addRow(dataRow);
            }

            assert dataRow.link() != 0 : dataRow;

            if (grp.sharedGroup() && dataRow.cacheId() == CU.UNDEFINED_CACHE_ID)
                dataRow.cacheId(cctx.cacheId());

            return dataRow;
        }

        /**
         * @param key Cache key.
         * @param val Cache value.
         * @param ver Version.
         * @param expireTime Expired time.
         * @param cacheId Cache id.
         * @return Made data row.
         */
        @NotNull private DataRow makeDataRow(KeyCacheObject key, CacheObject val, GridCacheVersion ver, long expireTime,
            int cacheId) {
            if (key.partition() == -1)
                key.partition(partId);

            return new DataRow(key, val, ver, partId, expireTime, cacheId);
        }

        /** {@inheritDoc} */
        @Override public boolean mvccInitialValue(
            GridCacheContext cctx,
            KeyCacheObject key,
            @Nullable CacheObject val,
            GridCacheVersion ver,
            long expireTime,
            MvccVersion mvccVer,
            MvccVersion newMvccVer)
            throws IgniteCheckedException
        {
            assert mvccVer != null || newMvccVer == null : newMvccVer;

            if (!busyLock.enterBusy())
                throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

            try {
                CacheObjectContext coCtx = cctx.cacheObjectContext();

                // Make sure value bytes initialized.
                key.valueBytes(coCtx);

                // null is passed for loaded from store.
                if (mvccVer == null) {
                    mvccVer = INITIAL_VERSION;

                    // Clean all versions of row
                    mvccRemoveAll(cctx, key);
                }

                if (val != null) {
                    val.valueBytes(coCtx);

                    MvccDataRow updateRow = new MvccDataRow(
                        key,
                        val,
                        ver,
                        partId,
                        expireTime,
                        cctx.cacheId(),
                        mvccVer,
                        newMvccVer);

                    assert cctx.shared().database().checkpointLockIsHeldByThread();

                    if (!grp.storeCacheIdInDataPage() && updateRow.cacheId() != CU.UNDEFINED_CACHE_ID) {
                        updateRow.cacheId(CU.UNDEFINED_CACHE_ID);

                        rowStore.addRow(updateRow);

                        updateRow.cacheId(cctx.cacheId());
                    }
                    else
                        rowStore.addRow(updateRow);

                    dataTree.putx(updateRow);

                    incrementSize(cctx.cacheId());

                    if (cctx.queries().enabled())
                        cctx.queries().store(updateRow, null, true);

                    return true;
                }
            }
            finally {
                busyLock.leaveBusy();
            }

            return false;
        }

        /** {@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
        {
            assert mvccVer != null;

            if (!busyLock.enterBusy())
                throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

            try {
                CacheObjectContext coCtx = cctx.cacheObjectContext();

                // Make sure value bytes initialized.
                key.valueBytes(coCtx);

                if (val != null)
                    val.valueBytes(coCtx);

                assert cctx.shared().database().checkpointLockIsHeldByThread();

                MvccPutIfAbsentClosure clo = new MvccPutIfAbsentClosure(key,
                    val,
                    ver,
                    partId,
                    expireTime,
                    cctx.cacheId(),
                    mvccVer,
                    newMvccVer,
                    txState,
                    newTxState);

                dataTree.invoke(clo, CacheDataRowAdapter.RowData.LINK_ONLY, clo);

                if (clo.operationType() == PUT)
                    finishUpdate(cctx, clo, null);

                return clo.operationType() == PUT;
            }
            finally {
                busyLock.leaveBusy();
            }
        }

        /** {@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 {
            if (!busyLock.enterBusy())
                throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

            try {
                CacheObjectContext coCtx = cctx.cacheObjectContext();

                // Make sure value bytes initialized.
                key.valueBytes(coCtx);

                if (val != null)
                    val.valueBytes(coCtx);

                assert cctx.shared().database().checkpointLockIsHeldByThread();

                MvccUpdateRowWithPreloadInfoClosure clo = new MvccUpdateRowWithPreloadInfoClosure(cctx,
                    key,
                    val,
                    ver,
                    expireTime,
                    mvccVer,
                    newMvccVer,
                    mvccTxState,
                    newMvccTxState);

                invoke0(cctx, clo, clo);
            }
            finally {
                busyLock.leaveBusy();
            }

            return true;
        }

        /** {@inheritDoc} */
        @Override public MvccUpdateResult mvccUpdate(
            GridCacheContext cctx,
            KeyCacheObject key,
            CacheObject val,
            GridCacheVersion ver,
            long expireTime,
            MvccSnapshot mvccSnapshot,
            @Nullable CacheEntryPredicate filter,
            EntryProcessor entryProc,
            Object[] invokeArgs,
            boolean primary,
            boolean needHistory,
            boolean noCreate,
            boolean needOldVal,
            boolean retVal) throws IgniteCheckedException {
            assert mvccSnapshot != null;
            assert primary || !needHistory;

            if (!busyLock.enterBusy())
                throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

            try {
                int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

                CacheObjectContext coCtx = cctx.cacheObjectContext();

                // Make sure value bytes initialized.
                key.valueBytes(coCtx);

                if(val != null)
                    val.valueBytes(coCtx);

                 MvccUpdateDataRow updateRow = new MvccUpdateDataRow(
                    cctx,
                    key,
                    val,
                    ver,
                    partId,
                    expireTime,
                    mvccSnapshot,
                    null,
                    filter,
                    primary,
                    false,
                    needHistory,
                    // we follow fast update visit flow here if row cannot be created by current operation
                    noCreate,
                    needOldVal,
                    retVal || entryProc != null);

                assert cctx.shared().database().checkpointLockIsHeldByThread();

                dataTree.visit(new MvccMaxSearchRow(cacheId, key), new MvccMinSearchRow(cacheId, key), updateRow);

                ResultType res = updateRow.resultType();

                if (res == ResultType.LOCKED // cannot update locked
                    || res == ResultType.VERSION_MISMATCH) // cannot update on write conflict
                    return updateRow;
                else if (res == ResultType.VERSION_FOUND || // exceptional case
                        res == ResultType.FILTERED || // Operation should be skipped.
                        (res == ResultType.PREV_NULL && noCreate)  // No op.
                    ) {
                    // Do nothing, except cleaning up not needed versions
                    cleanup(cctx, updateRow.cleanupRows());

                    return updateRow;
                }

                CacheDataRow oldRow = null;

                if (res == ResultType.PREV_NOT_NULL) {
                    oldRow = updateRow.oldRow();

                    assert oldRow != null && oldRow.link() != 0 : oldRow;

                    oldRow.key(key);
                }
                else
                    assert res == ResultType.PREV_NULL;

                if (entryProc != null) {
                    CacheInvokeEntry.Operation op = applyEntryProcessor(cctx, key, ver, entryProc, invokeArgs, updateRow, oldRow);

                    if (op == CacheInvokeEntry.Operation.NONE) {
                        if (res == ResultType.PREV_NOT_NULL)
                            updateRow.value(oldRow.value()); // Restore prev. value.

                        updateRow.resultType(ResultType.FILTERED);

                        cleanup(cctx, updateRow.cleanupRows());

                        return updateRow;
                    }

                    // Mark old version as removed.
                    if (res == ResultType.PREV_NOT_NULL) {
                        rowStore.updateDataRow(oldRow.link(), mvccUpdateMarker, mvccSnapshot);

                        if (op == CacheInvokeEntry.Operation.REMOVE) {
                            updateRow.resultType(ResultType.REMOVED_NOT_NULL);

                            cleanup(cctx, updateRow.cleanupRows());

                            clearPendingEntries(cctx, oldRow);

                            return updateRow; // Won't create new version on remove.
                        }
                    }
                    else
                        assert op != CacheInvokeEntry.Operation.REMOVE;
                }
                else if (oldRow != null)
                    rowStore.updateDataRow(oldRow.link(), mvccUpdateMarker, mvccSnapshot);

                if (!grp.storeCacheIdInDataPage() && updateRow.cacheId() != CU.UNDEFINED_CACHE_ID) {
                    updateRow.cacheId(CU.UNDEFINED_CACHE_ID);

                    rowStore.addRow(updateRow);

                    updateRow.cacheId(cctx.cacheId());
                }
                else
                    rowStore.addRow(updateRow);

                if (needHistory) {
                    assert updateRow.link() != 0;

                    updateRow.history().add(new MvccLinkAwareSearchRow(cacheId,
                        key,
                        updateRow.mvccCoordinatorVersion(),
                        updateRow.mvccCounter(),
                        updateRow.mvccOperationCounter(),
                        updateRow.link()));
                }

                boolean old = dataTree.putx(updateRow);

                assert !old;

                GridCacheQueryManager qryMgr = cctx.queries();

                if (qryMgr.enabled())
                    qryMgr.store(updateRow, null, true);

                updatePendingEntries(cctx, updateRow, oldRow);

                cleanup(cctx, updateRow.cleanupRows());

                return updateRow;
            }
            finally {
                busyLock.leaveBusy();
            }
        }

        /**
         *
         * @param cctx Cache context.
         * @param key entry key.
         * @param ver Entry version.
         * @param entryProc Entry processor.
         * @param invokeArgs Entry processor invoke arguments.
         * @param updateRow Row for update.
         * @param oldRow Old row.
         * @return Entry processor operation.
         */
        @SuppressWarnings("unchecked")
        private CacheInvokeEntry.Operation applyEntryProcessor(GridCacheContext cctx, KeyCacheObject key, GridCacheVersion ver,
            EntryProcessor entryProc, Object[] invokeArgs, MvccUpdateDataRow updateRow,
            CacheDataRow oldRow) {
            Object procRes = null;
            Exception err = null;

            CacheObject oldVal = oldRow == null ? null : oldRow.value();

            CacheInvokeEntry invokeEntry = new CacheInvokeEntry<>(key, oldVal, ver, cctx.keepBinary(),
                new GridDhtDetachedCacheEntry(cctx, key));

            try {
                procRes = entryProc.process(invokeEntry, invokeArgs);

                if(invokeEntry.modified() && invokeEntry.op() != CacheInvokeEntry.Operation.REMOVE) {
                    Object val = invokeEntry.getValue(true);

                    CacheObject val0 = cctx.toCacheObject(val);

                    val0.prepareForCache(cctx.cacheObjectContext());

                    updateRow.value(val0);
                }
            }
            catch (Exception e) {
                log.error("Exception was thrown during entry processing.", e);

                err = e;
            }

            CacheInvokeResult invokeRes = err == null ? CacheInvokeResult.fromResult(procRes) :
                CacheInvokeResult.fromError(err);

            updateRow.invokeResult(invokeRes);

            return invokeEntry.op();
        }

        /** {@inheritDoc} */
        @Override public MvccUpdateResult mvccRemove(GridCacheContext cctx,
            KeyCacheObject key,
            MvccSnapshot mvccSnapshot,
            @Nullable CacheEntryPredicate filter,
            boolean primary,
            boolean needHistory,
            boolean needOldVal,
            boolean retVal) throws IgniteCheckedException {
            assert mvccSnapshot != null;
            assert primary || mvccSnapshot.activeTransactions().size() == 0 : mvccSnapshot;
            assert primary || !needHistory;

            if (!busyLock.enterBusy())
                throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

            try {
                int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

                CacheObjectContext coCtx = cctx.cacheObjectContext();

                // Make sure value bytes initialized.
                key.valueBytes(coCtx);

                MvccUpdateDataRow updateRow = new MvccUpdateDataRow(
                    cctx,
                    key,
                    null,
                    null,
                    partId,
                    0,
                    mvccSnapshot,
                    null,
                    filter,
                    primary,
                    false,
                    needHistory,
                    true,
                    needOldVal,
                    retVal);

                assert cctx.shared().database().checkpointLockIsHeldByThread();

                dataTree.visit(new MvccMaxSearchRow(cacheId, key), new MvccMinSearchRow(cacheId, key), updateRow);

                ResultType res = updateRow.resultType();

                if (res == ResultType.LOCKED // cannot update locked
                    || res == ResultType.VERSION_MISMATCH) // cannot update on write conflict
                    return updateRow;
                else if (res == ResultType.VERSION_FOUND || res == ResultType.FILTERED) {
                    // Do nothing, except cleaning up not needed versions
                    cleanup(cctx, updateRow.cleanupRows());

                    return updateRow;
                }
                else if (res == ResultType.PREV_NOT_NULL) {
                    CacheDataRow oldRow = updateRow.oldRow();

                    assert oldRow != null && oldRow.link() != 0 : oldRow;

                    rowStore.updateDataRow(oldRow.link(), mvccUpdateMarker, mvccSnapshot);

                    clearPendingEntries(cctx, oldRow);
                }

                cleanup(cctx, updateRow.cleanupRows());

                return updateRow;
            }
            finally {
                busyLock.leaveBusy();
            }
        }

        /** {@inheritDoc} */
        @Override public MvccUpdateResult mvccLock(GridCacheContext cctx, KeyCacheObject key,
            MvccSnapshot mvccSnapshot) throws IgniteCheckedException {
            assert mvccSnapshot != null;

            if (!busyLock.enterBusy())
                throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

            try {
                int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

                CacheObjectContext coCtx = cctx.cacheObjectContext();

                // Make sure value bytes initialized.
                key.valueBytes(coCtx);

                MvccUpdateDataRow updateRow = new MvccUpdateDataRow(
                    cctx,
                    key,
                    null,
                    null,
                    partId,
                    0,
                    mvccSnapshot,
                    null,
                    null,
                    true,
                    true,
                    false,
                    false,
                    false,
                    false);

                assert cctx.shared().database().checkpointLockIsHeldByThread();

                dataTree.visit(new MvccMaxSearchRow(cacheId, key), new MvccMinSearchRow(cacheId, key), updateRow);

                ResultType res = updateRow.resultType();

                // cannot update locked, cannot update on write conflict
                if (res == ResultType.LOCKED || res == ResultType.VERSION_MISMATCH)
                    return updateRow;

                // Do nothing, except cleaning up not needed versions
                cleanup(cctx, updateRow.cleanupRows());

                return updateRow;
            }
            finally {
                busyLock.leaveBusy();
            }
        }

        /** {@inheritDoc} */
        @Override public GridLongList mvccUpdateNative(
            GridCacheContext cctx,
            boolean primary,
            KeyCacheObject key,
            CacheObject val,
            GridCacheVersion ver,
            long expireTime,
            MvccSnapshot mvccSnapshot) throws IgniteCheckedException {
            assert mvccSnapshot != null;
            assert primary || mvccSnapshot.activeTransactions().size() == 0 : mvccSnapshot;

            if (!busyLock.enterBusy())
                throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

            try {
                int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

                CacheObjectContext coCtx = cctx.cacheObjectContext();

                // Make sure value bytes initialized.
                key.valueBytes(coCtx);
                val.valueBytes(coCtx);

                MvccUpdateDataRowNative updateRow = new MvccUpdateDataRowNative(
                    key,
                    val,
                    ver,
                    expireTime,
                    mvccSnapshot,
                    null,
                    partId,
                    cctx);

                assert cctx.shared().database().checkpointLockIsHeldByThread();

                dataTree.iterate(new MvccMaxSearchRow(cacheId, key), new MvccMinSearchRow(cacheId, key), updateRow);

                ResultType res = updateRow.resultType();

                if (res == ResultType.VERSION_FOUND) {
                    // Do nothing, except cleaning up not needed versions
                    cleanup(cctx, updateRow.cleanupRows());

                    return null;
                }

                CacheDataRow oldRow = null;

                if (res == ResultType.PREV_NOT_NULL) {
                    oldRow = updateRow.oldRow();

                    assert oldRow != null && oldRow.link() != 0 : oldRow;

                    oldRow.key(key);

                    rowStore.updateDataRow(oldRow.link(), mvccUpdateMarker, mvccSnapshot);
                }
                else
                    assert res == ResultType.PREV_NULL;

                if (!grp.storeCacheIdInDataPage() && updateRow.cacheId() != CU.UNDEFINED_CACHE_ID) {
                    updateRow.cacheId(CU.UNDEFINED_CACHE_ID);

                    rowStore.addRow(updateRow);

                    updateRow.cacheId(cctx.cacheId());
                }
                else
                    rowStore.addRow(updateRow);

                boolean old = dataTree.putx(updateRow);

                assert !old;

                incrementSize(cctx.cacheId());

                GridCacheQueryManager qryMgr = cctx.queries();

                if (qryMgr.enabled())
                    qryMgr.store(updateRow, null, true);

                updatePendingEntries(cctx, updateRow, oldRow);

                cleanup(cctx, updateRow.cleanupRows());

                return updateRow.activeTransactions();
            }
            finally {
                busyLock.leaveBusy();
            }
        }

        /** {@inheritDoc} */
        @Override public GridLongList mvccRemoveNative(GridCacheContext cctx,
            boolean primary,
            KeyCacheObject key,
            MvccSnapshot mvccSnapshot) throws IgniteCheckedException {
            assert mvccSnapshot != null;
            assert primary || mvccSnapshot.activeTransactions().size() == 0 : mvccSnapshot;

            if (!busyLock.enterBusy())
                throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

            try {
                int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

                CacheObjectContext coCtx = cctx.cacheObjectContext();

                // Make sure value bytes initialized.
                key.valueBytes(coCtx);

                MvccUpdateDataRowNative updateRow = new MvccUpdateDataRowNative(
                    key,
                    null,
                    null,
                    0,
                    mvccSnapshot,
                    null,
                    partId,
                    cctx);

                assert cctx.shared().database().checkpointLockIsHeldByThread();

                dataTree.iterate(new MvccMaxSearchRow(cacheId, key) , new MvccMinSearchRow(cacheId, key), updateRow);

                ResultType res = updateRow.resultType();

                if (res == ResultType.VERSION_FOUND) {
                    assert !primary : updateRow;

                    // Do nothing, except cleaning up not needed versions
                    cleanup(cctx, updateRow.cleanupRows());

                    return null;
                }
                else if (res == ResultType.PREV_NOT_NULL) {
                    CacheDataRow oldRow = updateRow.oldRow();

                    assert oldRow != null && oldRow.link() != 0 : oldRow;

                    rowStore.updateDataRow(oldRow.link(), mvccUpdateMarker, mvccSnapshot);

                    clearPendingEntries(cctx, oldRow);
                }

                cleanup(cctx, updateRow.cleanupRows());

                return updateRow.activeTransactions();
            }
            finally {
                busyLock.leaveBusy();
            }
        }

        /** {@inheritDoc} */
        @Override public void mvccRemoveAll(GridCacheContext cctx, KeyCacheObject key) throws IgniteCheckedException {
            key.valueBytes(cctx.cacheObjectContext());

            int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

            boolean cleanup = cctx.queries().enabled() || hasPendingEntries;

            assert cctx.shared().database().checkpointLockIsHeldByThread();

            GridCursor<CacheDataRow> cur = dataTree.find(
                new MvccMaxSearchRow(cacheId, key),
                new MvccMinSearchRow(cacheId, key),
                cleanup ? CacheDataRowAdapter.RowData.NO_KEY : CacheDataRowAdapter.RowData.LINK_ONLY
            );

            boolean first = true;

            while (cur.next()) {
                CacheDataRow row = cur.get();

                row.key(key);

                assert row.link() != 0 : row;

                boolean rmvd = dataTree.removex(row);

                assert rmvd : row;

                if (cleanup) {
                    if (cctx.queries().enabled())
                        cctx.queries().remove(key, row);

                    if (first)
                        clearPendingEntries(cctx, row);
                }

                rowStore.removeRow(row.link());

                if (first)
                    first = false;
            }

            // first == true means there were no row versions
            if (!first)
                decrementSize(cctx.cacheId());
        }

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

            if (cleanupRows != null) {
                GridCacheQueryManager qryMgr = cctx.queries();

                for (int i = 0; i < cleanupRows.size(); i++) {
                    MvccLinkAwareSearchRow cleanupRow = cleanupRows.get(i);

                    assert cleanupRow.link() != 0 : cleanupRow;

                    assert cctx.shared().database().checkpointLockIsHeldByThread();

                    CacheDataRow oldRow = dataTree.remove(cleanupRow);

                    if (oldRow != null) { // oldRow == null means it was cleaned by another cleanup process.
                        assert oldRow.mvccCounter() == cleanupRow.mvccCounter();

                        if (qryMgr.enabled())
                            qryMgr.remove(oldRow.key(), oldRow);

                        clearPendingEntries(cctx, oldRow);

                        rowStore.removeRow(cleanupRow.link());

                        res++;
                    }
                }
            }

            return res;
        }

        /** {@inheritDoc} */
        @Override public void updateTxState(GridCacheContext cctx, CacheSearchRow row)
            throws IgniteCheckedException {
            assert grp.mvccEnabled();
            assert mvccVersionIsValid(row.mvccCoordinatorVersion(), row.mvccCounter(), row.mvccOperationCounter()) : row;

            // Need an extra lookup because the row may be already cleaned by another thread.
            CacheDataRow row0 = dataTree.findOne(row, CacheDataRowAdapter.RowData.LINK_ONLY);

            if (row0 != null)
                rowStore.updateDataRow(row0.link(), mvccUpdateTxStateHint, null);
        }

        /** {@inheritDoc} */
        @Override public void update(GridCacheContext cctx,
            KeyCacheObject key,
            CacheObject val,
            GridCacheVersion ver,
            long expireTime,
            @Nullable CacheDataRow oldRow
        ) throws IgniteCheckedException {
            assert oldRow == null || oldRow.link() != 0L : oldRow;

            if (!busyLock.enterBusy())
                throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

            try {
                int cacheId = grp.storeCacheIdInDataPage() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

                assert oldRow == null || oldRow.cacheId() == cacheId : oldRow;

                DataRow dataRow = makeDataRow(key, val, ver, expireTime, cacheId);

                CacheObjectContext coCtx = cctx.cacheObjectContext();

                // Make sure value bytes initialized.
                key.valueBytes(coCtx);
                val.valueBytes(coCtx);

                CacheDataRow old;

                assert cctx.shared().database().checkpointLockIsHeldByThread();

                if (canUpdateOldRow(cctx, oldRow, dataRow) && rowStore.updateRow(oldRow.link(), dataRow)) {
                    old = oldRow;

                    dataRow.link(oldRow.link());
                }
                else {
                    rowStore.addRow(dataRow);

                    assert dataRow.link() != 0 : dataRow;

                    if (grp.sharedGroup() && dataRow.cacheId() == CU.UNDEFINED_CACHE_ID)
                        dataRow.cacheId(cctx.cacheId());

                    if (oldRow != null) {
                        old = oldRow;

                        dataTree.putx(dataRow);
                    }
                    else
                        old = dataTree.put(dataRow);
                }

                finishUpdate(cctx, dataRow, old);
            }
            finally {
                busyLock.leaveBusy();
            }
        }

        /** {@inheritDoc} */
        @Override public void mvccApplyUpdate(GridCacheContext cctx,
            KeyCacheObject key,
            CacheObject val,
            GridCacheVersion ver,
            long expireTime,
            MvccVersion mvccVer
        ) throws IgniteCheckedException {
            if (!busyLock.enterBusy())
                throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

            try {
                int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

                CacheObjectContext coCtx = cctx.cacheObjectContext();

                // Make sure value bytes initialized.
                key.valueBytes(coCtx);

                if (val != null)
                    val.valueBytes(coCtx);

                MvccSnapshotWithoutTxs mvccSnapshot = new MvccSnapshotWithoutTxs(mvccVer.coordinatorVersion(),
                    mvccVer.counter(), mvccVer.operationCounter(), MvccUtils.MVCC_COUNTER_NA);

                MvccUpdateDataRow updateRow = new MvccUpdateDataRow(
                    cctx,
                    key,
                    val,
                    ver,
                    partId,
                    0L,
                    mvccSnapshot,
                    null,
                    null,
                    false,
                    false,
                    false,
                    false,
                    false,
                    false);

                assert cctx.shared().database().checkpointLockIsHeldByThread();

                dataTree.visit(new MvccMaxSearchRow(cacheId, key), new MvccMinSearchRow(cacheId, key), updateRow);

                ResultType res = updateRow.resultType();

                assert res == ResultType.PREV_NULL || res == ResultType.PREV_NOT_NULL : res;

                if (res == ResultType.PREV_NOT_NULL) {
                    CacheDataRow oldRow = updateRow.oldRow();

                    assert oldRow != null && oldRow.link() != 0 : oldRow;

                    rowStore.updateDataRow(oldRow.link(), mvccUpdateMarker, mvccSnapshot);
                }

                if (val != null) {
                    if (!grp.storeCacheIdInDataPage() && updateRow.cacheId() != CU.UNDEFINED_CACHE_ID) {
                        updateRow.cacheId(CU.UNDEFINED_CACHE_ID);

                        rowStore.addRow(updateRow);

                        updateRow.cacheId(cctx.cacheId());
                    }
                    else
                        rowStore.addRow(updateRow);

                    boolean old = dataTree.putx(updateRow);

                    assert !old;

                    GridCacheQueryManager qryMgr = cctx.queries();

                    if (qryMgr.enabled())
                        qryMgr.store(updateRow, null, true);

                    cleanup(cctx, updateRow.cleanupRows());
                }
            }
            finally {
                busyLock.leaveBusy();
            }
        }

        /**
         * @param cctx Cache context.
         * @param newRow New row.
         * @param oldRow Old row if available.
         * @throws IgniteCheckedException If failed.
         */
        private void finishUpdate(GridCacheContext cctx, CacheDataRow newRow, @Nullable CacheDataRow oldRow)
            throws IgniteCheckedException {
            if (oldRow == null)
                incrementSize(cctx.cacheId());

            KeyCacheObject key = newRow.key();

            GridCacheQueryManager qryMgr = cctx.queries();

            if (qryMgr.enabled())
                qryMgr.store(newRow, oldRow, true);

            updatePendingEntries(cctx, newRow, oldRow);

            if (oldRow != null) {
                assert oldRow.link() != 0 : oldRow;

                if (newRow.link() != oldRow.link())
                    rowStore.removeRow(oldRow.link());
            }

            updateIgfsMetrics(cctx, key, (oldRow != null ? oldRow.value() : null), newRow.value());
        }

        /**
         * @param cctx Cache context.
         * @param newRow New row.
         * @param oldRow Old row.
         * @throws IgniteCheckedException If failed.
         */
        private void updatePendingEntries(GridCacheContext cctx, CacheDataRow newRow, @Nullable CacheDataRow oldRow)
            throws IgniteCheckedException
        {
            long expireTime = newRow.expireTime();

            int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

            if (oldRow != null) {
                assert oldRow.link() != 0 : oldRow;

                if (pendingTree() != null && oldRow.expireTime() != 0)
                    pendingTree().removex(new PendingRow(cacheId, oldRow.expireTime(), oldRow.link()));
            }

            if (pendingTree() != null && expireTime != 0) {
                pendingTree().putx(new PendingRow(cacheId, expireTime, newRow.link()));

                hasPendingEntries = true;
            }
        }

        /** {@inheritDoc} */
        @Override public void remove(GridCacheContext cctx, KeyCacheObject key, int partId) throws IgniteCheckedException {
            if (!busyLock.enterBusy())
                throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

            try {
                int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

                assert cctx.shared().database().checkpointLockIsHeldByThread();

                CacheDataRow oldRow = dataTree.remove(new SearchRow(cacheId, key));

                finishRemove(cctx, key, oldRow);
            }
            finally {
                busyLock.leaveBusy();
            }
        }

        /**
         * @param cctx Cache context.
         * @param key Key.
         * @param oldRow Removed row.
         * @throws IgniteCheckedException If failed.
         */
        private void finishRemove(GridCacheContext cctx, KeyCacheObject key, @Nullable CacheDataRow oldRow) throws IgniteCheckedException {
            if (oldRow != null) {
                clearPendingEntries(cctx, oldRow);

                decrementSize(cctx.cacheId());
            }

            GridCacheQueryManager qryMgr = cctx.queries();

            if (qryMgr.enabled())
                qryMgr.remove(key, oldRow);

            if (oldRow != null)
                rowStore.removeRow(oldRow.link());

            updateIgfsMetrics(cctx, key, (oldRow != null ? oldRow.value() : null), null);
        }

        /**
         * @param cctx Cache context.
         * @param oldRow Old row.
         * @throws IgniteCheckedException
         */
        private void clearPendingEntries(GridCacheContext cctx, CacheDataRow oldRow)
            throws IgniteCheckedException {
            int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

            assert oldRow.link() != 0 : oldRow;
            assert cacheId == CU.UNDEFINED_CACHE_ID || oldRow.cacheId() == cacheId :
                "Incorrect cache ID [expected=" + cacheId + ", actual=" + oldRow.cacheId() + "].";

            if (pendingTree() != null && oldRow.expireTime() != 0)
                pendingTree().removex(new PendingRow(cacheId, oldRow.expireTime(), oldRow.link()));
        }

        /** {@inheritDoc} */
        @Override public CacheDataRow find(GridCacheContext cctx, KeyCacheObject key) throws IgniteCheckedException {
            key.valueBytes(cctx.cacheObjectContext());

            int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

            CacheDataRow row;

            if (grp.mvccEnabled()) {
                MvccFirstRowTreeClosure clo = new MvccFirstRowTreeClosure(cctx);

                dataTree.iterate(
                    new MvccMaxSearchRow(cacheId, key),
                    new MvccMinSearchRow(cacheId, key),
                    clo
                );

                row = clo.row();
            }
            else
                row = dataTree.findOne(new SearchRow(cacheId, key), CacheDataRowAdapter.RowData.NO_KEY);

            afterRowFound(row, key);

            return row;
        }

        /** {@inheritDoc} */
        @Override public List<IgniteBiTuple<Object, MvccVersion>> mvccFindAllVersions(
            GridCacheContext cctx,
            KeyCacheObject key)
            throws IgniteCheckedException
        {
            assert grp.mvccEnabled();

            // Note: this method is intended for testing only.

            key.valueBytes(cctx.cacheObjectContext());

            int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

            GridCursor<CacheDataRow> cur = dataTree.find(
                new MvccMaxSearchRow(cacheId, key),
                new MvccMinSearchRow(cacheId, key)
            );

            List<IgniteBiTuple<Object, MvccVersion>> res = new ArrayList<>();

            long crd = MVCC_CRD_COUNTER_NA;
            long cntr = MVCC_COUNTER_NA;
            int opCntr = MVCC_OP_COUNTER_NA;

            while (cur.next()) {
                CacheDataRow row = cur.get();

                if (compareNewVersion(row, crd, cntr, opCntr) != 0) // deleted row
                    res.add(F.t(null, row.newMvccVersion()));

                res.add(F.t(row.value(), row.mvccVersion()));

                crd = row.mvccCoordinatorVersion();
                cntr = row.mvccCounter();
                opCntr = row.mvccOperationCounter();
            }

            return res;
        }

        /** {@inheritDoc} */
        @Override public GridCursor<CacheDataRow> mvccAllVersionsCursor(GridCacheContext cctx, KeyCacheObject key, Object x)
            throws IgniteCheckedException {
            int cacheId = cctx.cacheId();

            return dataTree.find(new MvccMaxSearchRow(cacheId, key), new MvccMinSearchRow(cacheId, key), x);
        }

        /** {@inheritDoc} */
        @Override public CacheDataRow mvccFind(GridCacheContext cctx,
            KeyCacheObject key,
            MvccSnapshot snapshot) throws IgniteCheckedException {
            key.valueBytes(cctx.cacheObjectContext());

            int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;

            MvccSnapshotSearchRow clo = new MvccSnapshotSearchRow(cctx, key, snapshot);

            dataTree.iterate(
                clo,
                new MvccMinSearchRow(cacheId, key),
                clo
            );

            CacheDataRow row = clo.row();

            afterRowFound(row, key);

            return row;
        }

        /**
         * @param row Row.
         * @param key Key.
         * @throws IgniteCheckedException If failed.
         */
        private void afterRowFound(@Nullable CacheDataRow row, KeyCacheObject key) throws IgniteCheckedException {
            if (row != null) {
                row.key(key);

                grp.dataRegion().evictionTracker().touchPage(row.link());
            }
        }

        /** {@inheritDoc} */
        @Override public GridCursor<? extends CacheDataRow> cursor() throws IgniteCheckedException {
            return dataTree.find(null, null);
        }

        /** {@inheritDoc} */
        @Override public GridCursor<? extends CacheDataRow> cursor(Object x) throws IgniteCheckedException {
            return dataTree.find(null, null, x);
        }

        /** {@inheritDoc} */
        @Override public GridCursor<? extends CacheDataRow> cursor(MvccSnapshot mvccSnapshot)
            throws IgniteCheckedException {

            if (mvccSnapshot != null) {
                assert grp.mvccEnabled();

                return dataTree.find(null, null,
                    new MvccFirstVisibleRowTreeClosure(grp.singleCacheContext(), mvccSnapshot), null);
            }

            return dataTree.find(null, null);
        }

        /** {@inheritDoc} */
        @Override public GridCursor<? extends CacheDataRow> cursor(int cacheId) throws IgniteCheckedException {
            return cursor(cacheId, null, null);
        }

        /** {@inheritDoc} */
        @Override public GridCursor<? extends CacheDataRow> cursor(int cacheId,
            MvccSnapshot mvccSnapshot) throws IgniteCheckedException {
            return cursor(cacheId, null, null, null, mvccSnapshot);
        }

        /** {@inheritDoc} */
        @Override public GridCursor<? extends CacheDataRow> cursor(int cacheId, KeyCacheObject lower,
            KeyCacheObject upper) throws IgniteCheckedException {
            return cursor(cacheId, lower, upper, null);
        }

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

        /** {@inheritDoc} */
        @Override public GridCursor<? extends CacheDataRow> cursor(int cacheId, KeyCacheObject lower,
            KeyCacheObject upper, Object x, MvccSnapshot snapshot) throws IgniteCheckedException {
            SearchRow lowerRow;
            SearchRow upperRow;

            if (grp.sharedGroup()) {
                assert cacheId != CU.UNDEFINED_CACHE_ID;

                lowerRow = lower != null ? new SearchRow(cacheId, lower) : new SearchRow(cacheId);
                upperRow = upper != null ? new SearchRow(cacheId, upper) : new SearchRow(cacheId);
            }
            else {
                lowerRow = lower != null ? new SearchRow(CU.UNDEFINED_CACHE_ID, lower) : null;
                upperRow = upper != null ? new SearchRow(CU.UNDEFINED_CACHE_ID, upper) : null;
            }

            if (snapshot != null) {
                assert grp.mvccEnabled();

                GridCacheContext cctx =
                    grp.sharedGroup() ? grp.shared().cacheContext(cacheId) : grp.singleCacheContext();

                return dataTree.find(lowerRow, upperRow, new MvccFirstVisibleRowTreeClosure(cctx, snapshot), x);
            }

            return dataTree.find(lowerRow, upperRow, x);
        }

        /** {@inheritDoc} */
        @Override public void destroy() throws IgniteCheckedException {
            final AtomicReference<IgniteCheckedException> exception = new AtomicReference<>();

            dataTree.destroy(new IgniteInClosure<CacheSearchRow>() {
                @Override public void apply(CacheSearchRow row) {
                    try {
                        rowStore.removeRow(row.link());
                    }
                    catch (IgniteCheckedException e) {
                        U.error(log, "Failed to remove row [link=" + row.link() + "]");

                        IgniteCheckedException ex = exception.get();

                        if (ex == null)
                            exception.set(e);
                        else
                            ex.addSuppressed(e);
                    }
                }
            });

            if (exception.get() != null)
                throw new IgniteCheckedException("Failed to destroy store", exception.get());
        }

        /** {@inheritDoc} */
        @Override public void clear(int cacheId) throws IgniteCheckedException {
            assert cacheId != CU.UNDEFINED_CACHE_ID;

            if (cacheSize(cacheId) == 0)
                return;

            Exception ex = null;

            GridCursor<? extends CacheDataRow> cur =
                cursor(cacheId, null, null, CacheDataRowAdapter.RowData.KEY_ONLY);

            while (cur.next()) {
                CacheDataRow row = cur.get();

                assert row.link() != 0 : row;

                try {
                    boolean res = dataTree.removex(row);

                    assert res : row;

                    rowStore.removeRow(row.link());

                    decrementSize(cacheId);
                }
                catch (IgniteCheckedException e) {
                    U.error(log, "Fail remove row [link=" + row.link() + "]");

                    if (ex == null)
                        ex = e;
                    else
                        ex.addSuppressed(e);
                }
            }

            if (ex != null)
                throw new IgniteCheckedException("Fail destroy store", ex);
        }

        /** {@inheritDoc} */
        @Override public RowStore rowStore() {
            return rowStore;
        }

        /** {@inheritDoc} */
        @Override public void setRowCacheCleaner(GridQueryRowCacheCleaner rowCacheCleaner) {
            rowStore().setRowCacheCleaner(rowCacheCleaner);
        }

        /** {@inheritDoc} */
        @Override public void init(long size, long updCntr, @Nullable Map<Integer, Long> cacheSizes) {
            pCntr.init(updCntr);

            storageSize.set(size);

            if (cacheSizes != null) {
                for (Map.Entry<Integer, Long> e : cacheSizes.entrySet())
                    this.cacheSizes.put(e.getKey(), new AtomicLong(e.getValue()));
            }
        }

        /** {@inheritDoc} */
        @Override public PendingEntriesTree pendingTree() {
            return pendingEntries;
        }

        /**
         * @param cctx Cache context.
         * @param key Key.
         * @param oldVal Old value.
         * @param newVal New value.
         */
        private void updateIgfsMetrics(
            GridCacheContext cctx,
            KeyCacheObject key,
            CacheObject oldVal,
            CacheObject newVal
        ) {
            GridCacheAdapter cache = cctx.cache();
            if (cache == null) {
                return;
            }

            // In case we deal with IGFS cache, count updated data
            if (cache.isIgfsDataCache() &&
                !cctx.isNear() &&
                ctx.kernalContext()
                    .igfsHelper()
                    .isIgfsBlockKey(key.value(cctx.cacheObjectContext(), false))) {
                int oldSize = valueLength(cctx, oldVal);
                int newSize = valueLength(cctx, newVal);

                int delta = newSize - oldSize;

                if (delta != 0)
                    cache.onIgfsDataSizeChanged(delta);
            }
        }

        /**
         * Isolated method to get length of IGFS block.
         *
         * @param cctx Cache context.
         * @param val Value.
         * @return Length of value.
         */
        private int valueLength(GridCacheContext cctx, @Nullable CacheObject val) {
            if (val == null)
                return 0;

            byte[] bytes = val.value(cctx.cacheObjectContext(), false);

            if (bytes != null)
                return bytes.length;
            else
                return 0;
        }

        /** */
        private final class MvccFirstVisibleRowTreeClosure implements MvccTreeClosure {
            /** */
            private final GridCacheContext cctx;

            /** */
            private final MvccSnapshot snapshot;

            /**
             *
             * @param cctx Cache context.
             * @param snapshot MVCC snapshot.
             */
            MvccFirstVisibleRowTreeClosure(GridCacheContext cctx, MvccSnapshot snapshot) {
                this.cctx = cctx;
                this.snapshot = snapshot;
            }

            /** {@inheritDoc} */
            @Override public boolean apply(BPlusTree<CacheSearchRow, CacheDataRow> tree, BPlusIO<CacheSearchRow> io,
                long pageAddr, int idx) throws IgniteCheckedException {
                RowLinkIO rowIo = (RowLinkIO)io;

                long rowCrdVer = rowIo.getMvccCoordinatorVersion(pageAddr, idx);
                long rowCntr = rowIo.getMvccCounter(pageAddr, idx);
                int rowOpCntr = rowIo.getMvccOperationCounter(pageAddr, idx);

                assert mvccVersionIsValid(rowCrdVer, rowCntr, rowOpCntr);

                return isVisible(cctx, snapshot, rowCrdVer, rowCntr, rowOpCntr, rowIo.getLink(pageAddr, idx));
            }
        }

        /**
         * Put row if it doesn't exist yet.
         */
        private class MvccPutIfAbsentClosure extends MvccDataRow implements IgniteTree.InvokeClosure<CacheDataRow> {
            /** */
            private IgniteTree.OperationType op;

            /**
             * @param key Key.
             * @param val Value.
             * @param ver Version.
             * @param part Partition.
             * @param expireTime Expire time.
             * @param cacheId Cache ID.
             * @param mvccVer Mvcc version.
             * @param newMvccVer New mvcc version.
             * @param txState Tx state hint for mvcc version.
             * @param newTxState Tx state hint for new mvcc version.
             */
            MvccPutIfAbsentClosure(KeyCacheObject key,
                CacheObject val,
                GridCacheVersion ver,
                int part,
                long expireTime,
                int cacheId,
                MvccVersion mvccVer,
                MvccVersion newMvccVer,
                byte txState,
                byte newTxState) {
                super(key, val, ver, part, expireTime, cacheId, mvccVer, newMvccVer);

                mvccTxState(txState);
                newMvccTxState(newTxState);
            }

            /** {@inheritDoc} */
            @Override public void call(@Nullable CacheDataRow old) throws IgniteCheckedException {
                if (old == null) {
                    op = PUT;

                    int cacheId = cacheId();

                    if (!grp.storeCacheIdInDataPage() && cacheId != CU.UNDEFINED_CACHE_ID)
                        cacheId(CU.UNDEFINED_CACHE_ID);

                    rowStore().addRow(this);

                    cacheId(cacheId);
                }
                else
                    op = NOOP;
            }

            /** {@inheritDoc} */
            @Override public MvccDataRow newRow() {
                return this;
            }

            /** {@inheritDoc} */
            @Override public IgniteTree.OperationType operationType() {
                return op;
            }
        }

        /**
         *
         */
        private class MvccUpdateRowWithPreloadInfoClosure extends MvccDataRow implements OffheapInvokeClosure {
            /** */
            private CacheDataRow oldRow;
            /** */
            private IgniteTree.OperationType op;

            /**
             * @param cctx Cache context.
             * @param key Key.
             * @param val Value.
             * @param ver Version.
             * @param expireTime Expire time.
             * @param mvccVer Mvcc created version.
             * @param newMvccVer Mvcc updated version.
             * @param mvccTxState Mvcc Tx state hint.
             * @param newMvccTxState New Mvcc Tx state hint.
             */
            MvccUpdateRowWithPreloadInfoClosure(GridCacheContext cctx,
                KeyCacheObject key,
                @Nullable CacheObject val,
                GridCacheVersion ver,
                long expireTime,
                MvccVersion mvccVer,
                MvccVersion newMvccVer,
                byte mvccTxState,
                byte newMvccTxState) {
                super(key,
                    val,
                    ver,
                    CacheDataStoreImpl.this.partId(),
                    expireTime,
                    cctx.cacheId(),
                    mvccVer,
                    newMvccVer);

                mvccTxState(mvccTxState);
                newMvccTxState(newMvccTxState);
            }

            /** {@inheritDoc} */
            @Nullable @Override public CacheDataRow oldRow() {
                return oldRow;
            }

            /** {@inheritDoc} */
            @Override public void call(@Nullable CacheDataRow oldRow) throws IgniteCheckedException {
                this.oldRow = oldRow;

                if (oldRow == null) {
                    op = PUT;

                    int cacheId = cacheId();

                    if (!grp.storeCacheIdInDataPage() && cacheId != CU.UNDEFINED_CACHE_ID)
                        cacheId(CU.UNDEFINED_CACHE_ID);

                    rowStore().addRow(this);

                    cacheId(cacheId);
                }
                else {
                    op = NOOP;

                    if (oldRow.mvccTxState() != mvccTxState() ||
                        oldRow.newMvccCoordinatorVersion() != newMvccCoordinatorVersion() ||
                        oldRow.newMvccCounter() != newMvccCounter() ||
                        oldRow.newMvccOperationCounter() != newMvccOperationCounter() ||
                        oldRow.newMvccTxState() != newMvccTxState()) {

                        rowStore().updateDataRow(oldRow.link(), mvccApplyChanges, this);
                    }
                }
            }

            /** {@inheritDoc} */
            @Override public CacheDataRow newRow() {
                return op == PUT ? this : null;
            }

            /** {@inheritDoc} */
            @Override public IgniteTree.OperationType operationType() {
                return op == null ? NOOP : op;
            }
        }
    }

    /**
     * Mvcc remove handler.
     */
    private final class MvccMarkUpdatedHandler extends PageHandler<MvccVersion, Boolean> {
        /** {@inheritDoc} */
        @Override public Boolean run(int cacheId, long pageId, long page, long pageAddr, PageIO io, Boolean walPlc,
            MvccVersion newVer, int itemId) throws IgniteCheckedException {
            assert grp.mvccEnabled();

            DataPageIO iox = (DataPageIO)io;

            int offset = iox.getPayloadOffset(pageAddr, itemId,
                grp.dataRegion().pageMemory().realPageSize(grp.groupId()), MVCC_INFO_SIZE);

            long newCrd = iox.newMvccCoordinator(pageAddr, offset);
            long newCntr = iox.newMvccCounter(pageAddr, offset);
            int newOpCntr = iox.newMvccOperationCounter(pageAddr, offset);

            assert newCrd == MVCC_CRD_COUNTER_NA || state(grp, newCrd, newCntr, newOpCntr) == TxState.ABORTED;

            iox.updateNewVersion(pageAddr, offset, newVer, TxState.NA);

            if (isWalDeltaRecordNeeded(grp.dataRegion().pageMemory(), cacheId, pageId, page, ctx.wal(), walPlc))
                ctx.wal().log(new DataPageMvccMarkUpdatedRecord(cacheId, pageId, itemId,
                    newVer.coordinatorVersion(), newVer.counter(), newVer.operationCounter()));

            return Boolean.TRUE;
        }
    }

    /**
     * Mvcc update operation counter hints handler.
     */
    private final class MvccUpdateTxStateHintHandler extends PageHandler<Void, Boolean> {
        /** {@inheritDoc} */
        @Override public Boolean run(int cacheId, long pageId, long page, long pageAddr, PageIO io,
            Boolean walPlc, Void ignore,
            int itemId) throws IgniteCheckedException {

            DataPageIO iox = (DataPageIO)io;

            int offset = iox.getPayloadOffset(pageAddr, itemId,
                grp.dataRegion().pageMemory().realPageSize(grp.groupId()), MVCC_INFO_SIZE);

            long crd = iox.mvccCoordinator(pageAddr, offset);
            long cntr = iox.mvccCounter(pageAddr, offset);
            int opCntr = iox.mvccOperationCounter(pageAddr, offset);
            byte txState = (byte)(opCntr >>> MVCC_HINTS_BIT_OFF);

            if (txState == TxState.NA) {
                byte state = state(grp, crd, cntr, opCntr);

                if (state == TxState.COMMITTED || state == TxState.ABORTED) {
                    iox.mvccOperationCounter(pageAddr, offset, opCntr | (state << MVCC_HINTS_BIT_OFF));

                    if (isWalDeltaRecordNeeded(grp.dataRegion().pageMemory(), cacheId, pageId, page, ctx.wal(), walPlc))
                        ctx.wal().log(new DataPageMvccUpdateTxStateHintRecord(cacheId, pageId, itemId, state));
                }
                else
                    throw unexpectedStateException(grp, state, crd, cntr, opCntr);
            }

            long newCrd = iox.newMvccCoordinator(pageAddr, offset);
            long newCntr = iox.newMvccCounter(pageAddr, offset);
            int newOpCntr = iox.newMvccOperationCounter(pageAddr, offset);
            byte newTxState = (byte)(newOpCntr >>> MVCC_HINTS_BIT_OFF);

            if (newCrd != MVCC_CRD_COUNTER_NA && newTxState == TxState.NA) {
                byte state = state(grp, newCrd, newCntr, newOpCntr);

                if (state == TxState.COMMITTED || state == TxState.ABORTED) {
                    iox.newMvccOperationCounter(pageAddr, offset, newOpCntr | (state << MVCC_HINTS_BIT_OFF));

                    if (isWalDeltaRecordNeeded(grp.dataRegion().pageMemory(), cacheId, pageId, page, ctx.wal(), walPlc))
                        ctx.wal().log(new DataPageMvccUpdateNewTxStateHintRecord(cacheId, pageId, itemId, state));
                }

                // We do not throw an exception here because new version may be updated by active Tx at this moment.
            }

            return Boolean.TRUE;
        }
    }

    /**
     * Applies changes to the row.
     */
    private final class MvccApplyChangesHandler extends PageHandler<MvccDataRow, Boolean> {
        /** {@inheritDoc} */
        @Override public Boolean run(int cacheId, long pageId, long page, long pageAddr, PageIO io, Boolean walPlc,
            MvccDataRow newRow, int itemId) throws IgniteCheckedException {
            assert grp.mvccEnabled();

            DataPageIO iox = (DataPageIO)io;

            int offset = iox.getPayloadOffset(pageAddr, itemId,
                grp.dataRegion().pageMemory().realPageSize(grp.groupId()), MVCC_INFO_SIZE);

            long crd = iox.mvccCoordinator(pageAddr, offset);
            long cntr = iox.mvccCounter(pageAddr, offset);
            int opCntrAndHint = iox.mvccOperationCounter(pageAddr, offset);
            int opCntr = opCntrAndHint & ~MVCC_OP_COUNTER_MASK;
            byte txState = (byte)(opCntrAndHint >>> MVCC_HINTS_BIT_OFF);

            long newCrd = iox.newMvccCoordinator(pageAddr, offset);
            long newCntr = iox.newMvccCounter(pageAddr, offset);
            int newOpCntrAndHint = iox.newMvccOperationCounter(pageAddr, offset);
            int newOpCntr = newOpCntrAndHint & ~MVCC_OP_COUNTER_MASK;
            byte newTxState = (byte)(newOpCntrAndHint >>> MVCC_HINTS_BIT_OFF);

            assert crd == newRow.mvccCoordinatorVersion();
            assert cntr == newRow.mvccCounter();
            assert opCntr == newRow.mvccOperationCounter();

            if (txState != newRow.mvccTxState() && newRow.mvccTxState() != TxState.NA) {
                assert txState == TxState.NA;

                iox.mvccOperationCounter(pageAddr, offset, opCntr | (newRow.mvccTxState() << MVCC_HINTS_BIT_OFF));

                if (isWalDeltaRecordNeeded(grp.dataRegion().pageMemory(), cacheId, pageId, page, ctx.wal(), walPlc))
                    ctx.wal().log(new DataPageMvccUpdateTxStateHintRecord(cacheId, pageId, itemId, newRow.mvccTxState()));
            }

            if (compare(newCrd,
                newCntr,
                newOpCntr,
                newRow.newMvccCoordinatorVersion(),
                newRow.newMvccCounter(),
                newRow.newMvccOperationCounter()) != 0) {

                iox.updateNewVersion(pageAddr, offset, newRow.newMvccVersion(), newRow.newMvccTxState());

                if (isWalDeltaRecordNeeded(grp.dataRegion().pageMemory(), cacheId, pageId, page, ctx.wal(), walPlc))
                    ctx.wal().log(new DataPageMvccMarkUpdatedRecord(cacheId, pageId, itemId,
                        newRow.newMvccCoordinatorVersion(), newRow.newMvccCounter(), newRow.newMvccOperationCounter()));
            }
            else if (newTxState != newRow.newMvccTxState() && newRow.newMvccTxState() != TxState.NA) {
                assert newTxState == TxState.NA;

                iox.newMvccOperationCounter(pageAddr, offset, newOpCntr | (newRow.newMvccTxState() << MVCC_HINTS_BIT_OFF));

                if (isWalDeltaRecordNeeded(grp.dataRegion().pageMemory(), cacheId, pageId, page, ctx.wal(), walPlc))
                    ctx.wal().log(new DataPageMvccUpdateNewTxStateHintRecord(cacheId, pageId, itemId, newRow.newMvccTxState()));
            }

            return Boolean.TRUE;
        }
    }
}