ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [28/50] [abbrv] incubator-ignite git commit: ignite-sql - fix after merge
Date Tue, 27 Jan 2015 16:12:30 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c0e7b6d4/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/GridCacheQueryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/GridCacheQueryManager.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/GridCacheQueryManager.java
deleted file mode 100644
index 9c02003..0000000
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/GridCacheQueryManager.java
+++ /dev/null
@@ -1,3197 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.gridgain.grid.kernal.processors.cache.query;
-
-import org.apache.ignite.*;
-import org.apache.ignite.cluster.*;
-import org.apache.ignite.events.*;
-import org.apache.ignite.lang.*;
-import org.apache.ignite.resources.*;
-import org.apache.ignite.spi.*;
-import org.apache.ignite.spi.indexing.*;
-import org.gridgain.grid.*;
-import org.gridgain.grid.cache.*;
-import org.gridgain.grid.cache.query.*;
-import org.gridgain.grid.kernal.*;
-import org.gridgain.grid.kernal.managers.eventstorage.*;
-import org.gridgain.grid.kernal.processors.cache.*;
-import org.gridgain.grid.kernal.processors.cache.datastructures.*;
-import org.gridgain.grid.kernal.processors.cache.distributed.dht.*;
-import org.gridgain.grid.kernal.processors.query.*;
-import org.gridgain.grid.kernal.processors.task.*;
-import org.gridgain.grid.util.*;
-import org.gridgain.grid.util.future.*;
-import org.gridgain.grid.util.io.*;
-import org.gridgain.grid.util.lang.*;
-import org.gridgain.grid.util.typedef.*;
-import org.gridgain.grid.util.typedef.internal.*;
-import org.jdk8.backport.*;
-import org.jetbrains.annotations.*;
-
-import java.io.*;
-import java.sql.*;
-import java.util.*;
-import java.util.concurrent.*;
-
-import static org.gridgain.grid.cache.GridCacheMode.*;
-import static org.apache.ignite.events.IgniteEventType.*;
-import static org.gridgain.grid.kernal.GridClosureCallMode.*;
-import static org.gridgain.grid.kernal.processors.cache.query.GridCacheQueryType.*;
-
-/**
- * Query and index manager.
- */
-@SuppressWarnings("FieldAccessedSynchronizedAndUnsynchronized")
-public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapter<K, V> {
-    /** */
-    protected GridQueryProcessor qryProc;
-
-    /** */
-    private String space;
-
-    /** */
-    private int maxIterCnt;
-
-    /** */
-    private volatile GridCacheQueryMetricsAdapter metrics = new GridCacheQueryMetricsAdapter();
-
-    /** */
-    private final ConcurrentMap<UUID, Map<Long, GridFutureAdapter<QueryResult<K, V>>>> qryIters =
-        new ConcurrentHashMap8<>();
-
-    /** */
-    private final ConcurrentMap<UUID, Map<Long, GridFutureAdapter<FieldsResult>>> fieldsQryRes =
-        new ConcurrentHashMap8<>();
-
-    /** */
-    private volatile ConcurrentMap<Object, CachedResult<?>> qryResCache = new ConcurrentHashMap8<>();
-
-    /** */
-    private final GridSpinBusyLock busyLock = new GridSpinBusyLock();
-
-    /** {@inheritDoc} */
-    @Override public void start0() throws IgniteCheckedException {
-        qryProc = cctx.kernalContext().query();
-        space = cctx.name();
-        maxIterCnt = cctx.config().getMaximumQueryIteratorCount();
-
-        cctx.events().addListener(new GridLocalEventListener() {
-            @Override public void onEvent(IgniteEvent evt) {
-                UUID nodeId = ((IgniteDiscoveryEvent)evt).eventNode().id();
-
-                Map<Long, GridFutureAdapter<QueryResult<K, V>>> futs = qryIters.remove(nodeId);
-
-                if (futs != null) {
-                    for (Map.Entry<Long, GridFutureAdapter<QueryResult<K, V>>> entry : futs.entrySet()) {
-                        final Object recipient = recipient(nodeId, entry.getKey());
-
-                        entry.getValue().listenAsync(new CIX1<IgniteFuture<QueryResult<K, V>>>() {
-                            @Override public void applyx(IgniteFuture<QueryResult<K, V>> f) throws IgniteCheckedException {
-                                f.get().closeIfNotShared(recipient);
-                            }
-                        });
-                    }
-                }
-
-                Map<Long, GridFutureAdapter<FieldsResult>> fieldsFuts = fieldsQryRes.remove(nodeId);
-
-                if (fieldsFuts != null) {
-                    for (Map.Entry<Long, GridFutureAdapter<FieldsResult>> entry : fieldsFuts.entrySet()) {
-                        final Object recipient = recipient(nodeId, entry.getKey());
-
-                        entry.getValue().listenAsync(new CIX1<IgniteFuture<FieldsResult>>() {
-                            @Override public void applyx(IgniteFuture<FieldsResult> f)
-                                throws IgniteCheckedException {
-                                f.get().closeIfNotShared(recipient);
-                            }
-                        });
-                    }
-                }
-            }
-        }, EVT_NODE_LEFT, EVT_NODE_FAILED);
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void onKernalStop0(boolean cancel) {
-        busyLock.block();
-
-        if (cancel)
-            onCancelAtStop();
-        else
-            onWaitAtStop();
-    }
-
-    /**
-     * @return {@code True} if entered busy state.
-     */
-    private boolean enterBusy() {
-        return busyLock.enterBusy();
-    }
-
-    /**
-     *  Leaves busy state.
-     */
-    private void leaveBusy() {
-        busyLock.leaveBusy();
-    }
-
-    /**
-     * Stops query manager.
-     *
-     * @param cancel Cancel queries.
-     */
-    @SuppressWarnings({"LockAcquiredButNotSafelyReleased"})
-    @Override public final void stop0(boolean cancel) {
-        if (log.isDebugEnabled())
-            log.debug("Stopped cache query manager.");
-    }
-
-    /**
-     * Gets number of objects of given type in index.
-     *
-     * @param valType Value type.
-     * @return Number of objects or -1 if type was not indexed at all.
-     * @throws IgniteCheckedException If failed.
-     */
-    public long size(Class<?> valType) throws IgniteCheckedException {
-        if (!enterBusy())
-            throw new IllegalStateException("Failed to get size (grid is stopping).");
-
-        try {
-            return qryProc.size(space, valType);
-        }
-        finally {
-            leaveBusy();
-        }
-    }
-
-    /**
-     * Rebuilds all search indexes of given value type.
-     *
-     * @param valType Value type.
-     * @return Future that will be completed when rebuilding of all indexes is finished.
-     */
-    public IgniteFuture<?> rebuildIndexes(Class<?> valType) {
-        return rebuildIndexes(valType.getName());
-    }
-
-    /**
-     * Rebuilds all search indexes of given value type.
-     *
-     * @param typeName Value type name.
-     * @return Future that will be completed when rebuilding of all indexes is finished.
-     */
-    public IgniteFuture<?> rebuildIndexes(String typeName) {
-        if (!enterBusy())
-            throw new IllegalStateException("Failed to rebuild indexes (grid is stopping).");
-
-        try {
-            return qryProc.rebuildIndexes(space, typeName);
-        }
-        finally {
-            leaveBusy();
-        }
-    }
-
-    /**
-     * Rebuilds all search indexes of all types.
-     *
-     * @return Future that will be completed when rebuilding of all indexes is finished.
-     */
-    public IgniteFuture<?> rebuildAllIndexes() {
-        if (!enterBusy())
-            throw new IllegalStateException("Failed to rebuild indexes (grid is stopping).");
-
-        try {
-            return qryProc.rebuildAllIndexes();
-        }
-        finally {
-            leaveBusy();
-        }
-    }
-
-    /**
-     * Marks this request as canceled.
-     *
-     * @param reqId Request id.
-     */
-    void onQueryFutureCanceled(long reqId) {
-        // No-op.
-    }
-
-    /**
-     * Cancel flag handler at stop.
-     */
-    void onCancelAtStop() {
-        // No-op.
-    }
-
-    /**
-     * Wait flag handler at stop.
-     */
-    void onWaitAtStop() {
-        // No-op.
-    }
-
-    /**
-     * Processes cache query request.
-     *
-     * @param sndId Sender node id.
-     * @param req Query request.
-     */
-    void processQueryRequest(UUID sndId, GridCacheQueryRequest req) {
-        // No-op.
-    }
-
-    /**
-     * Entry for given key unswapped.
-     *
-     * @param swapSpaceName Swap space name.
-     * @param key Key.
-     * @throws IgniteCheckedException If failed.
-     */
-    public void onSwap(String swapSpaceName, K key) throws IgniteCheckedException {
-        if (!enterBusy())
-            return; // Ignore index update when node is stopping.
-
-        try {
-            qryProc.onSwap(space, key);
-        }
-        finally {
-            leaveBusy();
-        }
-    }
-
-    /**
-     * Entry for given key unswapped.
-     *
-     * @param key Key.
-     * @param val Value
-     * @param valBytes Value bytes.
-     * @throws IgniteCheckedException If failed.
-     */
-    public void onUnswap(K key, V val, byte[] valBytes) throws IgniteCheckedException {
-        if (!enterBusy())
-            return; // Ignore index update when node is stopping.
-
-        try {
-            qryProc.onUnswap(space, key, val, valBytes);
-        }
-        finally {
-            leaveBusy();
-        }
-    }
-
-    /**
-     *
-     */
-    private void invalidateResultCache() {
-        if (!qryResCache.isEmpty())
-            qryResCache = new ConcurrentHashMap8<>();
-    }
-
-    /**
-     * Writes key-value pair to index.
-     *
-     * @param key Key.
-     * @param keyBytes Byte array with key data.
-     * @param val Value.
-     * @param valBytes Value bytes.
-     * @param ver Cache entry version.
-     * @param expirationTime Expiration time or 0 if never expires.
-     * @throws IgniteCheckedException In case of error.
-     */
-    public void store(K key, @Nullable byte[] keyBytes, @Nullable V val, @Nullable byte[] valBytes,
-        GridCacheVersion ver, long expirationTime)
-        throws IgniteCheckedException {
-        assert key != null;
-        assert val != null || valBytes != null;
-
-        if (!cctx.config().isQueryIndexEnabled() && !(key instanceof GridCacheInternal))
-            return; // No-op.
-
-        if (!enterBusy())
-            return; // Ignore index update when node is stopping.
-
-        try {
-            if (val == null)
-                val = cctx.marshaller().unmarshal(valBytes, cctx.deploy().globalLoader());
-
-            qryProc.store(space, key, keyBytes, val, valBytes, CU.versionToBytes(ver), expirationTime);
-        }
-        finally {
-            invalidateResultCache();
-
-            leaveBusy();
-        }
-    }
-
-    /**
-     * @param key Key.
-     * @param keyBytes Byte array with key value.
-     * @throws IgniteCheckedException Thrown in case of any errors.
-     */
-    @SuppressWarnings("SimplifiableIfStatement")
-    public void remove(K key, @Nullable byte[] keyBytes) throws IgniteCheckedException {
-        assert key != null;
-
-        if (!cctx.config().isQueryIndexEnabled() && !(key instanceof GridCacheInternal))
-            return; // No-op.
-
-        if (!enterBusy())
-            return; // Ignore index update when node is stopping.
-
-        try {
-            qryProc.remove(space, key);
-        }
-        finally {
-            invalidateResultCache();
-
-            leaveBusy();
-        }
-    }
-
-    /**
-     * Undeploys given class loader.
-     *
-     * @param ldr Class loader to undeploy.
-     */
-    public void onUndeploy(ClassLoader ldr) {
-        if (!enterBusy())
-            return; // Ignore index update when node is stopping.
-
-        try {
-            qryProc.onUndeploy(space, ldr);
-        }
-        catch (IgniteCheckedException e) {
-            throw new IgniteException(e);
-        }
-        finally {
-            invalidateResultCache();
-
-            leaveBusy();
-        }
-    }
-
-    /**
-     * Executes local query.
-     *
-     * @param qry Query.
-     * @return Query future.
-     */
-    public abstract GridCacheQueryFuture<?> queryLocal(GridCacheQueryBean qry);
-
-    /**
-     * Executes distributed query.
-     *
-     * @param qry Query.
-     * @param nodes Nodes.
-     * @return Query future.
-     */
-    public abstract GridCacheQueryFuture<?> queryDistributed(GridCacheQueryBean qry, Collection<ClusterNode> nodes);
-
-    /**
-     * Loads page.
-     *
-     * @param id Query ID.
-     * @param qry Query.
-     * @param nodes Nodes.
-     * @param all Whether to load all pages.
-     */
-    public abstract void loadPage(long id, GridCacheQueryAdapter<?> qry, Collection<ClusterNode> nodes, boolean all);
-
-    /**
-     * Executes distributed fields query.
-     *
-     * @param qry Query.
-     * @return Query future.
-     */
-    public abstract GridCacheQueryFuture<?> queryFieldsLocal(GridCacheQueryBean qry);
-
-    /**
-     * Executes distributed fields query.
-     *
-     * @param qry Query.
-     * @param nodes Nodes.
-     * @return Query future.
-     */
-    public abstract GridCacheQueryFuture<?> queryFieldsDistributed(GridCacheQueryBean qry, Collection<ClusterNode> nodes);
-
-    /**
-     * Performs query.
-     *
-     * @param qry Query.
-     * @param args Arguments.
-     * @param loc Local query or not.
-     * @param subjId Security subject ID.
-     * @param taskName Task name.
-     * @param recipient ID of the recipient.
-     * @return Collection of found keys.
-     * @throws IgniteCheckedException In case of error.
-     */
-    private QueryResult<K, V> executeQuery(GridCacheQueryAdapter<?> qry,
-        @Nullable Object[] args, boolean loc, @Nullable UUID subjId, @Nullable String taskName, Object recipient)
-        throws IgniteCheckedException {
-        if (qry.type() == null) {
-            assert !loc;
-
-            throw new IgniteCheckedException("Received next page request after iterator was removed. " +
-                "Consider increasing maximum number of stored iterators (see " +
-                "GridCacheConfiguration.getMaximumQueryIteratorCount() configuration property).");
-        }
-
-        QueryResult<K, V> res;
-
-        T3<String, String, List<Object>> resKey = null;
-
-        if (qry.type() == SQL) {
-            resKey = new T3<>(qry.queryClassName(), qry.clause(), F.asList(args));
-
-            res = (QueryResult<K, V>)qryResCache.get(resKey);
-
-            if (res != null && res.addRecipient(recipient))
-                return res;
-
-            res = new QueryResult<>(qry.type(), recipient);
-
-            if (qryResCache.putIfAbsent(resKey, res) != null)
-                resKey = null;
-        }
-        else
-            res = new QueryResult<>(qry.type(), recipient);
-
-        GridCloseableIterator<IgniteBiTuple<K, V>> iter;
-
-        try {
-            switch (qry.type()) {
-                case SQL:
-                    if (cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_EXECUTED)) {
-                        cctx.gridEvents().record(new IgniteCacheQueryExecutedEvent<>(
-                            cctx.localNode(),
-                            "SQL query executed.",
-                            EVT_CACHE_QUERY_EXECUTED,
-                            org.gridgain.grid.cache.query.GridCacheQueryType.SQL,
-                            cctx.namex(),
-                            qry.queryClassName(),
-                            qry.clause(),
-                            null,
-                            null,
-                            args,
-                            subjId,
-                            taskName));
-                    }
-
-                    iter = qryProc.query(space, qry.clause(), F.asList(args),
-                        qry.queryClassName(), filter(qry));
-
-                    break;
-
-                case SCAN:
-                    if (cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_EXECUTED)) {
-                        cctx.gridEvents().record(new IgniteCacheQueryExecutedEvent<>(
-                            cctx.localNode(),
-                            "Scan query executed.",
-                            EVT_CACHE_QUERY_EXECUTED,
-                            org.gridgain.grid.cache.query.GridCacheQueryType.SCAN,
-                            cctx.namex(),
-                            null,
-                            null,
-                            qry.scanFilter(),
-                            null,
-                            null,
-                            subjId,
-                            taskName));
-                    }
-
-                    iter = scanIterator(qry);
-
-                    break;
-
-                case TEXT:
-                    if (cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_EXECUTED)) {
-                        cctx.gridEvents().record(new IgniteCacheQueryExecutedEvent<>(
-                            cctx.localNode(),
-                            "Full text query executed.",
-                            EVT_CACHE_QUERY_EXECUTED,
-                            org.gridgain.grid.cache.query.GridCacheQueryType.FULL_TEXT,
-                            cctx.namex(),
-                            qry.queryClassName(),
-                            qry.clause(),
-                            null,
-                            null,
-                            null,
-                            subjId,
-                            taskName));
-                    }
-
-                    iter = qryProc.queryText(space, qry.clause(), qry.queryClassName(), filter(qry));
-
-                    break;
-
-                case SET:
-                    iter = setIterator(qry);
-
-                    break;
-
-                case SQL_FIELDS:
-                    assert false : "SQL fields query is incorrectly processed.";
-
-                default:
-                    throw new IgniteCheckedException("Unknown query type: " + qry.type());
-            }
-
-            res.onDone(iter);
-        }
-        catch (Exception e) {
-            res.onDone(e);
-        }
-        finally {
-            if (resKey != null)
-                qryResCache.remove(resKey, res);
-        }
-
-        return res;
-    }
-
-    /**
-     * Performs fields query.
-     *
-     * @param qry Query.
-     * @param args Arguments.
-     * @param loc Local query or not.
-     * @param subjId Security subject ID.
-     * @param taskName Task name.
-     * @param recipient ID of the recipient.
-     * @return Collection of found keys.
-     * @throws IgniteCheckedException In case of error.
-     */
-    private FieldsResult executeFieldsQuery(GridCacheQueryAdapter<?> qry, @Nullable Object[] args,
-        boolean loc, @Nullable UUID subjId, @Nullable String taskName, Object recipient) throws IgniteCheckedException {
-        assert qry != null;
-
-        FieldsResult res;
-
-        T2<String, List<Object>> resKey = null;
-
-        if (qry.type() == SQL_FIELDS) {
-            if (qry.clause() == null) {
-                assert !loc;
-
-                throw new IgniteCheckedException("Received next page request after iterator was removed. " +
-                    "Consider increasing maximum number of stored iterators (see " +
-                    "GridCacheConfiguration.getMaximumQueryIteratorCount() configuration property).");
-            }
-
-            if (cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_EXECUTED)) {
-                cctx.gridEvents().record(new IgniteCacheQueryExecutedEvent<>(
-                    cctx.localNode(),
-                    "SQL fields query executed.",
-                    EVT_CACHE_QUERY_EXECUTED,
-                    org.gridgain.grid.cache.query.GridCacheQueryType.SQL_FIELDS,
-                    cctx.namex(),
-                    null,
-                    qry.clause(),
-                    null,
-                    null,
-                    args,
-                    subjId,
-                    taskName));
-            }
-
-            // Attempt to get result from cache.
-            resKey = new T2<>(qry.clause(), F.asList(args));
-
-            res = (FieldsResult)qryResCache.get(resKey);
-
-            if (res != null && res.addRecipient(recipient))
-                return res; // Cached result found.
-
-            res = new FieldsResult(recipient);
-
-            if (qryResCache.putIfAbsent(resKey, res) != null)
-                resKey = null; // Failed to cache result.
-        }
-        else {
-            assert qry.type() == SPI;
-
-            if (cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_EXECUTED)) {
-                cctx.gridEvents().record(new IgniteCacheQueryExecutedEvent<>(
-                    cctx.localNode(),
-                    "SPI query executed.",
-                    EVT_CACHE_QUERY_EXECUTED,
-                    org.gridgain.grid.cache.query.GridCacheQueryType.SPI,
-                    cctx.namex(),
-                    null,
-                    null,
-                    null,
-                    null,
-                    args,
-                    subjId,
-                    taskName));
-            }
-
-            res = new FieldsResult(recipient);
-        }
-
-        try {
-            if (qry.type() == SPI) {
-                IgniteSpiCloseableIterator<?> iter = cctx.kernalContext().indexing().query(space, F.asList(args),
-                    filter(qry));
-
-                res.onDone(iter);
-            }
-            else {
-                assert qry.type() == SQL_FIELDS;
-
-                GridQueryFieldsResult qryRes = qryProc.queryFields(space, qry.clause(), F.asList(args), filter(qry));
-
-                res.metaData(qryRes.metaData());
-
-                res.onDone(qryRes.iterator());
-            }
-        }
-        catch (Exception e) {
-            res.onDone(e);
-        }
-        finally {
-            if (resKey != null)
-                qryResCache.remove(resKey, res);
-        }
-
-        return res;
-    }
-
-    /**
-     * @param qry Query.
-     * @return Cache set items iterator.
-     */
-    private GridCloseableIterator<IgniteBiTuple<K, V>> setIterator(GridCacheQueryAdapter<?> qry) {
-        final GridSetQueryPredicate filter = (GridSetQueryPredicate)qry.scanFilter();
-
-        filter.init(cctx);
-
-        IgniteUuid id = filter.setId();
-
-        Collection<GridCacheSetItemKey> data = cctx.dataStructures().setData(id);
-
-        if (data == null)
-            data = Collections.emptyList();
-
-        final GridIterator<IgniteBiTuple<K, V>> it = F.iterator(
-            data,
-            new C1<GridCacheSetItemKey, IgniteBiTuple<K, V>>() {
-                @Override public IgniteBiTuple<K, V> apply(GridCacheSetItemKey e) {
-                    return new IgniteBiTuple<>((K)e.item(), (V)Boolean.TRUE);
-                }
-            },
-            true,
-            new P1<GridCacheSetItemKey>() {
-                @Override public boolean apply(GridCacheSetItemKey e) {
-                    return filter.apply(e, null);
-                }
-            });
-
-        return new GridCloseableIteratorAdapter<IgniteBiTuple<K, V>>() {
-            @Override protected boolean onHasNext() {
-                return it.hasNext();
-            }
-
-            @Override protected IgniteBiTuple<K, V> onNext() {
-                return it.next();
-            }
-
-            @Override protected void onRemove() {
-                it.remove();
-            }
-
-            @Override protected void onClose() {
-                // No-op.
-            }
-        };
-    }
-
-    /**
-     * @param qry Query.
-     * @return Full-scan row iterator.
-     * @throws IgniteCheckedException If failed to get iterator.
-     */
-    @SuppressWarnings({"unchecked"})
-    private GridCloseableIterator<IgniteBiTuple<K, V>> scanIterator(final GridCacheQueryAdapter<?> qry)
-        throws IgniteCheckedException {
-        IgnitePredicate<GridCacheEntry<K, V>> filter = null;
-
-        if (qry.projectionFilter() != null) {
-            filter = new P1<GridCacheEntry<K, V>>() {
-                @Override public boolean apply(GridCacheEntry<K, V> e) {
-                    return qry.projectionFilter().apply((GridCacheEntry<Object, Object>)e);
-                }
-            };
-        }
-
-        GridCacheProjection<K, V> prj0 = filter != null ? cctx.cache().projection(filter) : cctx.cache();
-
-        if (qry.keepPortable())
-            prj0 = prj0.keepPortable();
-
-        final GridCacheProjection<K, V> prj = prj0;
-
-        final IgniteBiPredicate<K, V> keyValFilter = qry.scanFilter();
-
-        injectResources(keyValFilter);
-
-        GridIterator<IgniteBiTuple<K, V>> heapIt = new GridIteratorAdapter<IgniteBiTuple<K, V>>() {
-            private IgniteBiTuple<K, V> next;
-
-            private Iterator<K> iter = prj.keySet().iterator();
-
-            {
-                advance();
-            }
-
-            @Override public boolean hasNextX() {
-                return next != null;
-            }
-
-            @Override public IgniteBiTuple<K, V> nextX() {
-                if (next == null)
-                    throw new NoSuchElementException();
-
-                IgniteBiTuple<K, V> next0 = next;
-
-                advance();
-
-                return next0;
-            }
-
-            @Override public void removeX() {
-                throw new UnsupportedOperationException();
-            }
-
-            private void advance() {
-                IgniteBiTuple<K, V> next0 = null;
-
-                while (iter.hasNext()) {
-                    next0 = null;
-
-                    K key = iter.next();
-
-                    V val = prj.peek(key);
-
-                    if (val != null) {
-                        next0 = F.t(key, val);
-
-                        if (checkPredicate(next0))
-                            break;
-                        else
-                            next0 = null;
-                    }
-                }
-
-                next = next0 != null ?
-                    new IgniteBiTuple<>(next0.getKey(), next0.getValue()) :
-                    null;
-            }
-
-            private boolean checkPredicate(Map.Entry<K, V> e) {
-                if (keyValFilter != null) {
-                    Map.Entry<K, V> e0 = (Map.Entry<K, V>)cctx.unwrapPortableIfNeeded(e, qry.keepPortable());
-
-                    return keyValFilter.apply(e0.getKey(), e0.getValue());
-                }
-
-                return true;
-            }
-        };
-
-        final GridIterator<IgniteBiTuple<K, V>> it;
-
-        if (cctx.isSwapOrOffheapEnabled()) {
-            List<GridIterator<IgniteBiTuple<K, V>>> iters = new ArrayList<>(3);
-
-            iters.add(heapIt);
-
-            if (cctx.isOffHeapEnabled())
-                iters.add(offheapIterator(qry));
-
-            if (cctx.swap().swapEnabled())
-                iters.add(swapIterator(qry));
-
-            it = new CompoundIterator<>(iters);
-        }
-        else
-            it = heapIt;
-
-        return new GridCloseableIteratorAdapter<IgniteBiTuple<K, V>>() {
-            @Override protected boolean onHasNext() {
-                return it.hasNext();
-            }
-
-            @Override protected IgniteBiTuple<K, V> onNext() {
-                return it.next();
-            }
-
-            @Override protected void onRemove() {
-                it.remove();
-            }
-        };
-    }
-
-    /**
-     * @param qry Query.
-     * @return Swap iterator.
-     * @throws IgniteCheckedException If failed.
-     */
-    private GridIterator<IgniteBiTuple<K, V>> swapIterator(GridCacheQueryAdapter<?> qry)
-        throws IgniteCheckedException {
-        IgnitePredicate<GridCacheEntry<Object, Object>> prjPred = qry.projectionFilter();
-
-        IgniteBiPredicate<K, V> filter = qry.scanFilter();
-
-        Iterator<Map.Entry<byte[], byte[]>> it = cctx.swap().rawSwapIterator();
-
-        return scanIterator(it, prjPred, filter, qry.keepPortable());
-    }
-
-    /**
-     * @param qry Query.
-     * @return Offheap iterator.
-     */
-    private GridIterator<IgniteBiTuple<K, V>> offheapIterator(GridCacheQueryAdapter<?> qry) {
-        IgnitePredicate<GridCacheEntry<Object, Object>> prjPred = qry.projectionFilter();
-
-        IgniteBiPredicate<K, V> filter = qry.scanFilter();
-
-        if ((cctx.portableEnabled() && cctx.offheapTiered()) && (prjPred != null || filter != null)) {
-            OffheapIteratorClosure c = new OffheapIteratorClosure(prjPred, filter, qry.keepPortable());
-
-            return cctx.swap().rawOffHeapIterator(c);
-        }
-        else {
-            Iterator<Map.Entry<byte[], byte[]>> it = cctx.swap().rawOffHeapIterator();
-
-            return scanIterator(it, prjPred, filter, qry.keepPortable());
-        }
-    }
-
-    /**
-     * @param it Lazy swap or offheap iterator.
-     * @param prjPred Projection predicate.
-     * @param filter Scan filter.
-     * @param keepPortable Keep portable flag.
-     * @return Iterator.
-     */
-    private GridIteratorAdapter<IgniteBiTuple<K, V>> scanIterator(
-        @Nullable final Iterator<Map.Entry<byte[], byte[]>> it,
-        @Nullable final IgnitePredicate<GridCacheEntry<Object, Object>> prjPred,
-        @Nullable final IgniteBiPredicate<K, V> filter,
-        final boolean keepPortable) {
-        if (it == null)
-            return new GridEmptyCloseableIterator<>();
-
-        return new GridIteratorAdapter<IgniteBiTuple<K, V>>() {
-            private IgniteBiTuple<K, V> next;
-
-            {
-                advance();
-            }
-
-            @Override public boolean hasNextX() {
-                return next != null;
-            }
-
-            @Override public IgniteBiTuple<K, V> nextX() {
-                if (next == null)
-                    throw new NoSuchElementException();
-
-                IgniteBiTuple<K, V> next0 = next;
-
-                advance();
-
-                return next0;
-            }
-
-            @Override public void removeX() {
-                throw new UnsupportedOperationException();
-            }
-
-            private void advance() {
-                next = null;
-
-                while (it.hasNext()) {
-                    final LazySwapEntry e = new LazySwapEntry(it.next());
-
-                    if (prjPred != null) {
-                        GridCacheEntry<K, V> cacheEntry = new GridCacheScanSwapEntry(e);
-
-                        if (!prjPred.apply((GridCacheEntry<Object, Object>)cacheEntry))
-                            continue;
-                    }
-
-                    if (filter != null) {
-                        K key = (K)cctx.unwrapPortableIfNeeded(e.key(), keepPortable);
-                        V val = (V)cctx.unwrapPortableIfNeeded(e.value(), keepPortable);
-
-                        if (!filter.apply(key, val))
-                            continue;
-                    }
-
-                    next = new IgniteBiTuple<>(e.key(), e.value());
-
-                    break;
-                }
-            }
-        };
-    }
-
-    /**
-     * @param o Object to inject resources to.
-     * @throws IgniteCheckedException If failure occurred while injecting resources.
-     */
-    private void injectResources(@Nullable Object o) throws IgniteCheckedException {
-        if (o != null) {
-            GridKernalContext ctx = cctx.kernalContext();
-
-            ClassLoader ldr = o.getClass().getClassLoader();
-
-            if (ctx.deploy().isGlobalLoader(ldr))
-                ctx.resource().inject(ctx.deploy().getDeployment(ctx.deploy().getClassLoaderId(ldr)), o.getClass(), o);
-            else
-                ctx.resource().inject(ctx.deploy().getDeployment(o.getClass().getName()), o.getClass(), o);
-        }
-    }
-
-    /**
-     * Processes fields query request.
-     *
-     * @param qryInfo Query info.
-     */
-    protected void runFieldsQuery(GridCacheQueryInfo qryInfo) {
-        assert qryInfo != null;
-
-        if (!enterBusy()) {
-            if (cctx.localNodeId().equals(qryInfo.senderId()))
-                throw new IllegalStateException("Failed to process query request (grid is stopping).");
-
-            return; // Ignore remote requests when when node is stopping.
-        }
-
-        try {
-            if (log.isDebugEnabled())
-                log.debug("Running query: " + qryInfo);
-
-            boolean rmvRes = true;
-
-            try {
-                // Preparing query closures.
-                IgnitePredicate<GridCacheEntry<Object, Object>> prjFilter = qryInfo.projectionPredicate();
-                IgniteClosure<Object, Object> trans = (IgniteClosure<Object, Object>)qryInfo.transformer();
-                IgniteReducer<Object, Object> rdc = (IgniteReducer<Object, Object>)qryInfo.reducer();
-
-                injectResources(prjFilter);
-                injectResources(trans);
-                injectResources(rdc);
-
-                GridCacheQueryAdapter<?> qry = qryInfo.query();
-
-                int pageSize = qry.pageSize();
-
-                Collection<Object> data = null;
-                Collection<Object> entities = null;
-
-                if (qryInfo.local() || rdc != null || cctx.isLocalNode(qryInfo.senderId()))
-                    data = new ArrayList<>(pageSize);
-                else
-                    entities = new ArrayList<>(pageSize);
-
-                String taskName = cctx.kernalContext().task().resolveTaskName(qry.taskHash());
-
-                FieldsResult res = qryInfo.local() ?
-                    executeFieldsQuery(qry, qryInfo.arguments(), qryInfo.local(), qry.subjectId(), taskName,
-                    recipient(qryInfo.senderId(), qryInfo.requestId())) :
-                    fieldsQueryResult(qryInfo, taskName);
-
-                // If metadata needs to be returned to user and cleaned from internal fields - copy it.
-                List<GridQueryFieldMetadata> meta = qryInfo.includeMetaData() ?
-                    (res.metaData() != null ? new ArrayList<GridQueryFieldMetadata>(res.metaData()) : null) :
-                    res.metaData();
-
-                if (!qryInfo.includeMetaData())
-                    meta = null;
-
-                GridCloseableIterator<?> it = new GridSpiCloseableIteratorWrapper<Object>(
-                    res.iterator(recipient(qryInfo.senderId(), qryInfo.requestId())));
-
-                if (log.isDebugEnabled())
-                    log.debug("Received fields iterator [iterHasNext=" + it.hasNext() + ']');
-
-                if (!it.hasNext()) {
-                    if (rdc != null)
-                        data = Collections.singletonList(rdc.reduce());
-
-                    onFieldsPageReady(qryInfo.local(), qryInfo, meta, entities, data, true, null);
-
-                    return;
-                }
-
-                int cnt = 0;
-                boolean metaSent = false;
-
-                while (!Thread.currentThread().isInterrupted() && it.hasNext()) {
-                    Object row = it.next();
-
-                    // Query is cancelled.
-                    if (row == null) {
-                        onPageReady(qryInfo.local(), qryInfo, null, true, null);
-
-                        break;
-                    }
-
-                    if (cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ)) {
-                        cctx.gridEvents().record(new IgniteCacheQueryReadEvent<K, V>(
-                            cctx.localNode(),
-                            "SQL fields query result set row read.",
-                            EVT_CACHE_QUERY_OBJECT_READ,
-                            org.gridgain.grid.cache.query.GridCacheQueryType.SQL_FIELDS,
-                            cctx.namex(),
-                            null,
-                            qry.clause(),
-                            null,
-                            null,
-                            qryInfo.arguments(),
-                            qry.subjectId(),
-                            taskName,
-                            null,
-                            null,
-                            null,
-                            row));
-                    }
-
-                    if ((qryInfo.local() || rdc != null || cctx.isLocalNode(qryInfo.senderId()))) {
-                        // Reduce.
-                        if (rdc != null) {
-                            if (!rdc.collect(row))
-                                break;
-                        }
-                        else
-                            data.add(row);
-                    }
-                    else
-                        entities.add(row);
-
-                    if (rdc == null && ((!qryInfo.allPages() && ++cnt == pageSize) || !it.hasNext())) {
-                        onFieldsPageReady(qryInfo.local(), qryInfo, !metaSent ? meta : null,
-                            entities, data, !it.hasNext(), null);
-
-                        if (it.hasNext())
-                            rmvRes = false;
-
-                        if (!qryInfo.allPages())
-                            return;
-                    }
-                }
-
-                if (rdc != null) {
-                    onFieldsPageReady(qryInfo.local(), qryInfo, meta, null,
-                        Collections.singletonList(rdc.reduce()), true, null);
-                }
-            }
-            catch (IgniteCheckedException e) {
-                if (log.isDebugEnabled() || !e.hasCause(SQLException.class))
-                    U.error(log, "Failed to run fields query [qry=" + qryInfo + ", node=" + cctx.nodeId() + ']', e);
-                else {
-                    if (e.hasCause(SQLException.class))
-                        U.error(log, "Failed to run fields query [node=" + cctx.nodeId() +
-                            ", msg=" + e.getCause(SQLException.class).getMessage() + ']');
-                    else
-                        U.error(log, "Failed to run fields query [node=" + cctx.nodeId() +
-                            ", msg=" + e.getMessage() + ']');
-                }
-
-                onFieldsPageReady(qryInfo.local(), qryInfo, null, null, null, true, e);
-            }
-            catch (Throwable e) {
-                U.error(log, "Failed to run fields query [qry=" + qryInfo + ", node=" + cctx.nodeId() + "]", e);
-
-                onFieldsPageReady(qryInfo.local(), qryInfo, null, null, null, true, e);
-            }
-            finally {
-                if (rmvRes)
-                    removeFieldsQueryResult(qryInfo.senderId(), qryInfo.requestId());
-            }
-        }
-        finally {
-            leaveBusy();
-        }
-    }
-
-    /**
-     * Processes cache query request.
-     *
-     * @param qryInfo Query info.
-     */
-    @SuppressWarnings("unchecked")
-    protected void runQuery(GridCacheQueryInfo qryInfo) {
-        assert qryInfo != null;
-
-        if (!enterBusy()) {
-            if (cctx.localNodeId().equals(qryInfo.senderId()))
-                throw new IllegalStateException("Failed to process query request (grid is stopping).");
-
-            return; // Ignore remote requests when when node is stopping.
-        }
-
-        try {
-            boolean loc = qryInfo.local();
-
-            if (log.isDebugEnabled())
-                log.debug("Running query: " + qryInfo);
-
-            boolean rmvIter = true;
-
-            try {
-                // Preparing query closures.
-                IgnitePredicate<GridCacheEntry<Object, Object>> prjFilter = qryInfo.projectionPredicate();
-                IgniteClosure<Map.Entry<K, V>, Object> trans = (IgniteClosure<Map.Entry<K, V>, Object>)qryInfo.transformer();
-                IgniteReducer<Map.Entry<K, V>, Object> rdc = (IgniteReducer<Map.Entry<K, V>, Object>)qryInfo.reducer();
-
-                injectResources(prjFilter);
-                injectResources(trans);
-                injectResources(rdc);
-
-                GridCacheQueryAdapter<?> qry = qryInfo.query();
-
-                int pageSize = qry.pageSize();
-
-                boolean incBackups = qry.includeBackups();
-
-                String taskName = cctx.kernalContext().task().resolveTaskName(qry.taskHash());
-
-                IgniteSpiCloseableIterator<IgniteBiTuple<K, V>> iter;
-                GridCacheQueryType type;
-
-                QueryResult<K, V> res;
-
-                res = loc ?
-                    executeQuery(qry, qryInfo.arguments(), loc, qry.subjectId(), taskName,
-                    recipient(qryInfo.senderId(), qryInfo.requestId())) :
-                    queryResult(qryInfo, taskName);
-
-                iter = res.iterator(recipient(qryInfo.senderId(), qryInfo.requestId()));
-                type = res.type();
-
-                GridCacheAdapter<K, V> cache = cctx.cache();
-
-                if (log.isDebugEnabled())
-                    log.debug("Received index iterator [iterHasNext=" + iter.hasNext() +
-                        ", cacheSize=" + cache.size() + ']');
-
-                int cnt = 0;
-
-                boolean stop = false;
-                boolean pageSent = false;
-
-                Collection<Object> data = new ArrayList<>(pageSize);
-
-                long topVer = cctx.affinity().affinityTopologyVersion();
-
-                while (!Thread.currentThread().isInterrupted() && iter.hasNext()) {
-                    IgniteBiTuple<K, V> row = iter.next();
-
-                    // Query is cancelled.
-                    if (row == null) {
-                        onPageReady(loc, qryInfo, null, true, null);
-
-                        break;
-                    }
-
-                    K key = row.getKey();
-
-                    // Filter backups for SCAN queries. Other types are filtered in indexing manager.
-                    if (!cctx.isReplicated() && cctx.config().getCacheMode() != LOCAL && qry.type() == SCAN &&
-                        !incBackups && !cctx.affinity().primary(cctx.localNode(), key, topVer)) {
-                        if (log.isDebugEnabled())
-                            log.debug("Ignoring backup element [row=" + row +
-                                ", cacheMode=" + cctx.config().getCacheMode() + ", incBackups=" + incBackups +
-                                ", primary=" + cctx.affinity().primary(cctx.localNode(), key, topVer) + ']');
-
-                        continue;
-                    }
-
-                    V val = row.getValue();
-
-                    if (log.isDebugEnabled())
-                        log.debug("Record [key=" + key + ", val=" + val + ", incBackups=" +
-                            incBackups + "priNode=" + U.id8(CU.primaryNode(cctx, key).id()) +
-                            ", node=" + U.id8(cctx.grid().localNode().id()) + ']');
-
-                    if (val == null) {
-                        if (log.isDebugEnabled())
-                            log.debug("Unsuitable record value: " + val);
-
-                        continue;
-                    }
-
-                    switch (type) {
-                        case SQL:
-                            if (cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ)) {
-                                cctx.gridEvents().record(new IgniteCacheQueryReadEvent<>(
-                                    cctx.localNode(),
-                                    "SQL query entry read.",
-                                    EVT_CACHE_QUERY_OBJECT_READ,
-                                    org.gridgain.grid.cache.query.GridCacheQueryType.SQL,
-                                    cctx.namex(),
-                                    qry.queryClassName(),
-                                    qry.clause(),
-                                    null,
-                                    null,
-                                    qryInfo.arguments(),
-                                    qry.subjectId(),
-                                    taskName,
-                                    key,
-                                    val,
-                                    null,
-                                    null));
-                            }
-
-                            break;
-
-                        case TEXT:
-                            if (cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ)) {
-                                cctx.gridEvents().record(new IgniteCacheQueryReadEvent<>(
-                                    cctx.localNode(),
-                                    "Full text query entry read.",
-                                    EVT_CACHE_QUERY_OBJECT_READ,
-                                    org.gridgain.grid.cache.query.GridCacheQueryType.FULL_TEXT,
-                                    cctx.namex(),
-                                    qry.queryClassName(),
-                                    qry.clause(),
-                                    null,
-                                    null,
-                                    null,
-                                    qry.subjectId(),
-                                    taskName,
-                                    key,
-                                    val,
-                                    null,
-                                    null));
-                            }
-
-                            break;
-
-                        case SCAN:
-                            if (cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ)) {
-                                cctx.gridEvents().record(new IgniteCacheQueryReadEvent<>(
-                                    cctx.localNode(),
-                                    "Scan query entry read.",
-                                    EVT_CACHE_QUERY_OBJECT_READ,
-                                    org.gridgain.grid.cache.query.GridCacheQueryType.SCAN,
-                                    cctx.namex(),
-                                    null,
-                                    null,
-                                    qry.scanFilter(),
-                                    null,
-                                    null,
-                                    qry.subjectId(),
-                                    taskName,
-                                    key,
-                                    val,
-                                    null,
-                                    null));
-                            }
-
-                            break;
-                    }
-
-                    Map.Entry<K, V> entry = F.t(key, val);
-
-                    // Unwrap entry for reducer or transformer only.
-                    if (rdc != null || trans != null)
-                        entry = (Map.Entry<K, V>)cctx.unwrapPortableIfNeeded(entry, qry.keepPortable());
-
-                    // Reduce.
-                    if (rdc != null) {
-                        if (!rdc.collect(entry) || !iter.hasNext()) {
-                            onPageReady(loc, qryInfo, Collections.singletonList(rdc.reduce()), true, null);
-
-                            pageSent = true;
-
-                            break;
-                        }
-                        else
-                            continue;
-                    }
-
-                    data.add(trans != null ? trans.apply(entry) :
-                        !loc ? new GridCacheQueryResponseEntry<>(key, val) : F.t(key, val));
-
-                    if (!loc) {
-                        if (++cnt == pageSize || !iter.hasNext()) {
-                            boolean finished = !iter.hasNext();
-
-                            onPageReady(loc, qryInfo, data, finished, null);
-
-                            pageSent = true;
-
-                            if (!finished)
-                                rmvIter = false;
-
-                            if (!qryInfo.allPages())
-                                return;
-
-                            data = new ArrayList<>(pageSize);
-
-                            if (stop)
-                                break; // while
-                        }
-                    }
-                }
-
-                if (!pageSent) {
-                    if (rdc == null)
-                        onPageReady(loc, qryInfo, data, true, null);
-                    else
-                        onPageReady(loc, qryInfo, Collections.singletonList(rdc.reduce()), true, null);
-                }
-            }
-            catch (Throwable e) {
-                U.error(log, "Failed to run query [qry=" + qryInfo + ", node=" + cctx.nodeId() + "]", e);
-
-                onPageReady(loc, qryInfo, null, true, e);
-            }
-            finally {
-                if (rmvIter)
-                    removeQueryResult(qryInfo.senderId(), qryInfo.requestId());
-            }
-        }
-        finally {
-            leaveBusy();
-        }
-    }
-
-    /**
-     * @param qryInfo Info.
-     * @return Iterator.
-     * @throws IgniteCheckedException In case of error.
-     */
-    private QueryResult<K, V> queryResult(GridCacheQueryInfo qryInfo, String taskName) throws IgniteCheckedException {
-        final UUID sndId = qryInfo.senderId();
-
-        assert sndId != null;
-
-        Map<Long, GridFutureAdapter<QueryResult<K, V>>> futs = qryIters.get(sndId);
-
-        if (futs == null) {
-            futs = new LinkedHashMap<Long, GridFutureAdapter<QueryResult<K, V>>>(
-                16, 0.75f, true) {
-                @Override protected boolean removeEldestEntry(Map.Entry<Long, GridFutureAdapter<QueryResult<K, V>>> e) {
-                    boolean rmv = size() > maxIterCnt;
-
-                    if (rmv) {
-                        try {
-                            e.getValue().get().closeIfNotShared(recipient(sndId, e.getKey()));
-                        }
-                        catch (IgniteCheckedException ex) {
-                            U.error(log, "Failed to close query iterator.", ex);
-                        }
-                    }
-
-                    return rmv;
-                }
-            };
-
-            Map<Long, GridFutureAdapter<QueryResult<K, V>>> old = qryIters.putIfAbsent(sndId, futs);
-
-            if (old != null)
-                futs = old;
-        }
-
-        return queryResult(futs, qryInfo, taskName);
-    }
-
-    /**
-     * @param futs Futures map.
-     * @param qryInfo Info.
-     * @return Iterator.
-     * @throws IgniteCheckedException In case of error.
-     */
-    @SuppressWarnings({"SynchronizationOnLocalVariableOrMethodParameter",
-        "NonPrivateFieldAccessedInSynchronizedContext"})
-    private QueryResult<K, V> queryResult(Map<Long, GridFutureAdapter<QueryResult<K, V>>> futs,
-        GridCacheQueryInfo qryInfo, String taskName) throws IgniteCheckedException {
-        assert futs != null;
-        assert qryInfo != null;
-
-        GridFutureAdapter<QueryResult<K, V>> fut;
-
-        boolean exec = false;
-
-        synchronized (futs) {
-            fut = futs.get(qryInfo.requestId());
-
-            if (fut == null) {
-                futs.put(qryInfo.requestId(), fut = new GridFutureAdapter<>(cctx.kernalContext()));
-
-                exec = true;
-            }
-        }
-
-        if (exec) {
-            try {
-                fut.onDone(executeQuery(qryInfo.query(), qryInfo.arguments(), false,
-                    qryInfo.query().subjectId(), taskName, recipient(qryInfo.senderId(), qryInfo.requestId())));
-            }
-            catch (Error e) {
-                fut.onDone(e);
-
-                throw e;
-            }
-            catch (Throwable e) {
-                fut.onDone(e);
-            }
-        }
-
-        return fut.get();
-    }
-
-    /**
-     * @param sndId Sender node ID.
-     * @param reqId Request ID.
-     */
-    @SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
-    protected void removeQueryResult(@Nullable UUID sndId, long reqId) {
-        if (sndId == null)
-            return;
-
-        Map<Long, GridFutureAdapter<QueryResult<K, V>>> futs = qryIters.get(sndId);
-
-        if (futs != null) {
-            IgniteFuture<QueryResult<K, V>> fut;
-
-            synchronized (futs) {
-                fut = futs.remove(reqId);
-            }
-
-            if (fut != null) {
-                try {
-                    fut.get().closeIfNotShared(recipient(sndId, reqId));
-                }
-                catch (IgniteCheckedException e) {
-                    U.error(log, "Failed to close iterator.", e);
-                }
-            }
-        }
-    }
-
-    /**
-     * @param sndId Sender node ID.
-     * @param reqId Request ID.
-     * @return Recipient ID.
-     */
-    private static Object recipient(UUID sndId, long reqId) {
-        assert sndId != null;
-
-        return new IgniteBiTuple<>(sndId, reqId);
-    }
-
-    /**
-     * @param qryInfo Info.
-     * @return Iterator.
-     * @throws IgniteCheckedException In case of error.
-     */
-    private FieldsResult fieldsQueryResult(GridCacheQueryInfo qryInfo, String taskName)
-        throws IgniteCheckedException {
-        final UUID sndId = qryInfo.senderId();
-
-        assert sndId != null;
-
-        Map<Long, GridFutureAdapter<FieldsResult>> iters = fieldsQryRes.get(sndId);
-
-        if (iters == null) {
-            iters = new LinkedHashMap<Long, GridFutureAdapter<FieldsResult>>(16, 0.75f, true) {
-                @Override protected boolean removeEldestEntry(Map.Entry<Long,
-                    GridFutureAdapter<FieldsResult>> e) {
-                    boolean rmv = size() > maxIterCnt;
-
-                    if (rmv) {
-                        try {
-                            e.getValue().get().closeIfNotShared(recipient(sndId, e.getKey()));
-                        }
-                        catch (IgniteCheckedException ex) {
-                            U.error(log, "Failed to close fields query iterator.", ex);
-                        }
-                    }
-
-                    return rmv;
-                }
-
-                @Override public boolean equals(Object o) {
-                    return o == this;
-                }
-            };
-
-            Map<Long, GridFutureAdapter<FieldsResult>> old = fieldsQryRes.putIfAbsent(sndId, iters);
-
-            if (old != null)
-                iters = old;
-        }
-
-        return fieldsQueryResult(iters, qryInfo, taskName);
-    }
-
-    /**
-     * @param resMap Results map.
-     * @param qryInfo Info.
-     * @return Fields query result.
-     * @throws IgniteCheckedException In case of error.
-     */
-    @SuppressWarnings({"SynchronizationOnLocalVariableOrMethodParameter",
-        "NonPrivateFieldAccessedInSynchronizedContext"})
-    private FieldsResult fieldsQueryResult(Map<Long, GridFutureAdapter<FieldsResult>> resMap,
-        GridCacheQueryInfo qryInfo, String taskName) throws IgniteCheckedException {
-        assert resMap != null;
-        assert qryInfo != null;
-
-        GridFutureAdapter<FieldsResult> fut;
-
-        boolean exec = false;
-
-        synchronized (resMap) {
-            fut = resMap.get(qryInfo.requestId());
-
-            if (fut == null) {
-                resMap.put(qryInfo.requestId(), fut =
-                    new GridFutureAdapter<>(cctx.kernalContext()));
-
-                exec = true;
-            }
-        }
-
-        if (exec) {
-            try {
-                fut.onDone(executeFieldsQuery(qryInfo.query(), qryInfo.arguments(), false,
-                    qryInfo.query().subjectId(), taskName, recipient(qryInfo.senderId(), qryInfo.requestId())));
-            }
-            catch (IgniteCheckedException e) {
-                fut.onDone(e);
-            }
-        }
-
-        return fut.get();
-    }
-
-    /**
-     * @param sndId Sender node ID.
-     * @param reqId Request ID.
-     */
-    @SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
-    protected void removeFieldsQueryResult(@Nullable UUID sndId, long reqId) {
-        if (sndId == null)
-            return;
-
-        Map<Long, GridFutureAdapter<FieldsResult>> futs = fieldsQryRes.get(sndId);
-
-        if (futs != null) {
-            IgniteFuture<FieldsResult> fut;
-
-            synchronized (futs) {
-                fut = futs.remove(reqId);
-            }
-
-            if (fut != null) {
-                assert fut.isDone();
-
-                try {
-                    fut.get().closeIfNotShared(recipient(sndId, reqId));
-                }
-                catch (IgniteCheckedException e) {
-                    U.error(log, "Failed to close iterator.", e);
-                }
-            }
-        }
-    }
-
-    /**
-     * Called when data for page is ready.
-     *
-     * @param loc Local query or not.
-     * @param qryInfo Query info.
-     * @param data Result data.
-     * @param finished Last page or not.
-     * @param e Exception in case of error.
-     * @return {@code true} if page was processed right.
-     */
-    protected abstract boolean onPageReady(boolean loc, GridCacheQueryInfo qryInfo,
-        @Nullable Collection<?> data, boolean finished, @Nullable Throwable e);
-
-    /**
-     * @param loc Local query or not.
-     * @param qryInfo Query info.
-     * @param metaData Meta data.
-     * @param entities Indexing entities.
-     * @param data Data.
-     * @param finished Last page or not.
-     * @param e Exception in case of error.
-     * @return {@code true} if page was processed right.
-     */
-    protected abstract boolean onFieldsPageReady(boolean loc, GridCacheQueryInfo qryInfo,
-        @Nullable List<GridQueryFieldMetadata> metaData,
-        @Nullable Collection<?> entities,
-        @Nullable Collection<?> data,
-        boolean finished, @Nullable Throwable e);
-
-    /**
-     * Gets cache queries metrics.
-     *
-     * @return Cache queries metrics.
-     */
-    public GridCacheQueryMetrics metrics() {
-        return metrics.copy();
-    }
-
-    /**
-     * Resets metrics.
-     */
-    public void resetMetrics() {
-        metrics = new GridCacheQueryMetricsAdapter();
-    }
-
-    /**
-     * @param duration Execution duration.
-     * @param fail {@code true} if execution failed.
-     */
-    public void onMetricsUpdate(long duration, boolean fail) {
-        metrics.onQueryExecute(duration, fail);
-    }
-
-    /**
-     * Gets SQL metadata.
-     *
-     * @return SQL metadata.
-     * @throws IgniteCheckedException In case of error.
-     */
-    public Collection<GridCacheSqlMetadata> sqlMetadata() throws IgniteCheckedException {
-        if (!enterBusy())
-            throw new IllegalStateException("Failed to get metadata (grid is stopping).");
-
-        try {
-            Callable<Collection<CacheSqlMetadata>> job = new MetadataJob();
-
-            // Remote nodes that have current cache.
-            Collection<ClusterNode> nodes = F.view(cctx.discovery().remoteNodes(), new P1<ClusterNode>() {
-                @Override public boolean apply(ClusterNode n) {
-                    return U.hasCache(n, space);
-                }
-            });
-
-            Collection<Collection<CacheSqlMetadata>> res = new ArrayList<>(nodes.size() + 1);
-
-            IgniteFuture<Collection<Collection<CacheSqlMetadata>>> rmtFut = null;
-
-            // Get metadata from remote nodes.
-            if (!nodes.isEmpty())
-                rmtFut = cctx.closures().callAsyncNoFailover(BROADCAST, F.asSet(job), nodes, true);
-
-            // Get local metadata.
-            IgniteFuture<Collection<CacheSqlMetadata>> locFut = cctx.closures().callLocalSafe(job, true);
-
-            if (rmtFut != null)
-                res.addAll(rmtFut.get());
-
-            res.add(locFut.get());
-
-            Map<String, Collection<CacheSqlMetadata>> map = new HashMap<>();
-
-            for (Collection<CacheSqlMetadata> col : res) {
-                for (CacheSqlMetadata meta : col) {
-                    String name = meta.cacheName();
-
-                    Collection<CacheSqlMetadata> cacheMetas = map.get(name);
-
-                    if (cacheMetas == null)
-                        map.put(name, cacheMetas = new LinkedList<>());
-
-                    cacheMetas.add(meta);
-                }
-            }
-
-            Collection<GridCacheSqlMetadata> col = new ArrayList<>(map.size());
-
-            // Metadata for current cache must be first in list.
-            col.add(new CacheSqlMetadata(map.remove(space)));
-
-            for (Collection<CacheSqlMetadata> metas : map.values())
-                col.add(new CacheSqlMetadata(metas));
-
-            return col;
-        }
-        finally {
-            leaveBusy();
-        }
-    }
-
-    /**
-     * Gets projection filter for query.
-     *
-     * @param qry Query.
-     * @return Filter.
-     */
-    @SuppressWarnings("unchecked")
-    @Nullable private GridIndexingQueryFilter projectionFilter(GridCacheQueryAdapter<?> qry) {
-        assert qry != null;
-
-        final IgnitePredicate<GridCacheEntry<Object, Object>> prjFilter = qry.projectionFilter();
-
-        if (prjFilter == null || F.isAlwaysTrue(prjFilter))
-            return null;
-
-        return new GridIndexingQueryFilter() {
-            @Nullable @Override public IgniteBiPredicate<K, V> forSpace(String spaceName) {
-                if (!F.eq(space, spaceName))
-                    return null;
-
-                return new IgniteBiPredicate<K, V>() {
-                    @Override public boolean apply(K k, V v) {
-                        try {
-                            GridCacheEntry<K, V> entry = context().cache().entry(k);
-
-                            return entry != null && prjFilter.apply((GridCacheEntry<Object, Object>)entry);
-                        }
-                        catch (GridDhtInvalidPartitionException ignore) {
-                            return false;
-                        }
-                    }
-                };
-            }
-        };
-    }
-
-    /**
-     * @param <K> Key type.
-     * @param <V> Value type.
-     * @return Predicate.
-     * @param includeBackups Include backups.
-     */
-    @SuppressWarnings("unchecked")
-    @Nullable public <K, V> GridIndexingQueryFilter backupsFilter(boolean includeBackups) {
-        if (includeBackups)
-            return null;
-
-        return new GridIndexingQueryFilter() {
-            @Nullable @Override public IgniteBiPredicate<K, V> forSpace(final String spaceName) {
-                final GridKernalContext ctx = cctx.kernalContext();
-
-                final GridCacheAdapter<Object, Object> cache = ctx.cache().internalCache(spaceName);
-
-                if (cache.context().isReplicated() || cache.configuration().getBackups() == 0)
-                    return null;
-
-                return new IgniteBiPredicate<K, V>() {
-                    @Override public boolean apply(K k, V v) {
-                        return cache.context().affinity().primary(ctx.discovery().localNode(), k, -1);
-                    }
-                };
-            }
-        };
-    }
-
-    /**
-     * @param qry Query.
-     * @return Filter.
-     */
-    private GridIndexingQueryFilter filter(GridCacheQueryAdapter<?> qry) {
-        return and(backupsFilter(qry.includeBackups()), projectionFilter(qry));
-    }
-
-    /**
-     * @param f1 First filter.
-     * @param f2 Second filter.
-     * @return And filter of the given two.
-     */
-    @Nullable private static GridIndexingQueryFilter and(@Nullable final GridIndexingQueryFilter f1,
-        @Nullable final GridIndexingQueryFilter f2) {
-        if (f1 == null)
-            return f2;
-
-        if (f2 == null)
-            return f1;
-
-        return new GridIndexingQueryFilter() {
-            @Nullable @Override public <K, V> IgniteBiPredicate<K, V> forSpace(String spaceName) {
-                final IgniteBiPredicate<K, V> fltr1 = f1.forSpace(spaceName);
-                final IgniteBiPredicate<K, V> fltr2 = f2.forSpace(spaceName);
-
-                if (fltr1 == null)
-                    return fltr2;
-
-                if (fltr2 == null)
-                    return fltr1;
-
-                return new IgniteBiPredicate<K, V>() {
-                    @Override public boolean apply(K k, V v) {
-                        return fltr1.apply(k, v) && fltr2.apply(k, v);
-                    }
-                };
-            }
-        };
-    }
-
-    /**
-     * Prints memory statistics for debugging purposes.
-     */
-    @Override public void printMemoryStats() {
-        X.println(">>>");
-        X.println(">>> Query manager memory stats [grid=" + cctx.gridName() + ", cache=" + cctx.name() + ']');
-    }
-
-    /**
-     * FOR TESTING ONLY
-     *
-     * @return Indexing space for this query manager.
-     */
-    public String space() {
-        return space;
-    }
-
-    /**
-     * Metadata job.
-     */
-    @GridInternal
-    private static class MetadataJob implements IgniteCallable<Collection<CacheSqlMetadata>> {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** Grid */
-        @IgniteInstanceResource
-        private Ignite ignite;
-
-        /** {@inheritDoc} */
-        @Override public Collection<CacheSqlMetadata> call() {
-            final GridKernalContext ctx = ((GridKernal) ignite).context();
-
-            Collection<String> cacheNames = F.viewReadOnly(ctx.cache().caches(),
-                new C1<GridCache<?, ?>, String>() {
-                    @Override public String apply(GridCache<?, ?> c) {
-                        return c.name();
-                    }
-                },
-                new P1<GridCache<?, ?>>() {
-                    @Override public boolean apply(GridCache<?, ?> c) {
-                        return !CU.UTILITY_CACHE_NAME.equals(c.name());
-                    }
-                }
-            );
-
-            return F.transform(cacheNames, new C1<String, CacheSqlMetadata>() {
-                @Override public CacheSqlMetadata apply(String cacheName) {
-                    Collection<GridQueryTypeDescriptor> types = ctx.query().types(cacheName);
-
-                    Collection<String> names = U.newHashSet(types.size());
-                    Map<String, String> keyClasses = U.newHashMap(types.size());
-                    Map<String, String> valClasses = U.newHashMap(types.size());
-                    Map<String, Map<String, String>> fields = U.newHashMap(types.size());
-                    Map<String, Collection<GridCacheSqlIndexMetadata>> indexes = U.newHashMap(types.size());
-
-                    for (GridQueryTypeDescriptor type : types) {
-                        // Filter internal types (e.g., data structures).
-                        if (type.name().startsWith("GridCache"))
-                            continue;
-
-                        names.add(type.name());
-
-                        keyClasses.put(type.name(), type.keyClass().getName());
-                        valClasses.put(type.name(), type.valueClass().getName());
-
-                        int size = 2 + type.keyFields().size() + type.valueFields().size();
-
-                        Map<String, String> fieldsMap = U.newLinkedHashMap(size);
-
-                        // _KEY and _VAL are not included in GridIndexingTypeDescriptor.valueFields
-                        fieldsMap.put("_KEY", type.keyClass().getName());
-                        fieldsMap.put("_VAL", type.valueClass().getName());
-
-                        for (Map.Entry<String, Class<?>> e : type.keyFields().entrySet())
-                            fieldsMap.put(e.getKey().toUpperCase(), e.getValue().getName());
-
-                        for (Map.Entry<String, Class<?>> e : type.valueFields().entrySet())
-                            fieldsMap.put(e.getKey().toUpperCase(), e.getValue().getName());
-
-                        fields.put(type.name(), fieldsMap);
-
-                        Collection<GridCacheSqlIndexMetadata> indexesCol =
-                            new ArrayList<>(type.indexes().size());
-
-                        for (Map.Entry<String, GridQueryIndexDescriptor> e : type.indexes().entrySet()) {
-                            GridQueryIndexDescriptor desc = e.getValue();
-
-                            // Add only SQL indexes.
-                            if (desc.type() == GridQueryIndexType.SORTED) {
-                                Collection<String> idxFields = e.getValue().fields();
-                                Collection<String> descendings = new LinkedList<>();
-
-                                for (String idxField : idxFields)
-                                    if (desc.descending(idxField))
-                                        descendings.add(idxField);
-
-                                indexesCol.add(new CacheSqlIndexMetadata(e.getKey().toUpperCase(),
-                                    idxFields, descendings, false));
-                            }
-                        }
-
-                        indexes.put(type.name(), indexesCol);
-                    }
-
-                    return new CacheSqlMetadata(cacheName, names, keyClasses, valClasses, fields, indexes);
-                }
-            });
-        }
-    }
-
-    /**
-     * Cache metadata.
-     */
-    private static class CacheSqlMetadata implements GridCacheSqlMetadata {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** */
-        private String cacheName;
-
-        /** */
-        private Collection<String> types;
-
-        /** */
-        private Map<String, String> keyClasses;
-
-        /** */
-        private Map<String, String> valClasses;
-
-        /** */
-        private Map<String, Map<String, String>> fields;
-
-        /** */
-        private Map<String, Collection<GridCacheSqlIndexMetadata>> indexes;
-
-        /**
-         * Required by {@link Externalizable}.
-         */
-        public CacheSqlMetadata() {
-            // No-op.
-        }
-
-        /**
-         * @param cacheName Cache name.
-         * @param types Types.
-         * @param keyClasses Key classes map.
-         * @param valClasses Value classes map.
-         * @param fields Fields maps.
-         * @param indexes Indexes.
-         */
-        CacheSqlMetadata(@Nullable String cacheName, Collection<String> types, Map<String, String> keyClasses,
-            Map<String, String> valClasses, Map<String, Map<String, String>> fields,
-            Map<String, Collection<GridCacheSqlIndexMetadata>> indexes) {
-            assert types != null;
-            assert keyClasses != null;
-            assert valClasses != null;
-            assert fields != null;
-            assert indexes != null;
-
-            this.cacheName = cacheName;
-            this.types = types;
-            this.keyClasses = keyClasses;
-            this.valClasses = valClasses;
-            this.fields = fields;
-            this.indexes = indexes;
-        }
-
-        /**
-         * @param metas Meta data instances from different nodes.
-         */
-        CacheSqlMetadata(Iterable<CacheSqlMetadata> metas) {
-            types = new HashSet<>();
-            keyClasses = new HashMap<>();
-            valClasses = new HashMap<>();
-            fields = new HashMap<>();
-            indexes = new HashMap<>();
-
-            for (CacheSqlMetadata meta : metas) {
-                if (cacheName == null)
-                    cacheName = meta.cacheName;
-                else
-                    assert F.eq(cacheName, meta.cacheName);
-
-                types.addAll(meta.types);
-                keyClasses.putAll(meta.keyClasses);
-                valClasses.putAll(meta.valClasses);
-                fields.putAll(meta.fields);
-                indexes.putAll(meta.indexes);
-            }
-        }
-
-        /** {@inheritDoc} */
-        @Override public String cacheName() {
-            return cacheName;
-        }
-
-        /** {@inheritDoc} */
-        @Override public Collection<String> types() {
-            return types;
-        }
-
-        /** {@inheritDoc} */
-        @Override public String keyClass(String type) {
-            return keyClasses.get(type);
-        }
-
-        /** {@inheritDoc} */
-        @Override public String valueClass(String type) {
-            return valClasses.get(type);
-        }
-
-        /** {@inheritDoc} */
-        @Override public Map<String, String> fields(String type) {
-            return fields.get(type);
-        }
-
-        /** {@inheritDoc} */
-        @Override public Collection<GridCacheSqlIndexMetadata> indexes(String type) {
-            return indexes.get(type);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void writeExternal(ObjectOutput out) throws IOException {
-            U.writeString(out, cacheName);
-            U.writeCollection(out, types);
-            U.writeMap(out, keyClasses);
-            U.writeMap(out, valClasses);
-            U.writeMap(out, fields);
-            U.writeMap(out, indexes);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-            cacheName = U.readString(in);
-            types = U.readCollection(in);
-            keyClasses = U.readMap(in);
-            valClasses = U.readMap(in);
-            fields = U.readMap(in);
-            indexes = U.readMap(in);
-        }
-
-        /** {@inheritDoc} */
-        @Override public String toString() {
-            return S.toString(CacheSqlMetadata.class, this);
-        }
-    }
-
-    /**
-     * Cache metadata index.
-     */
-    private static class CacheSqlIndexMetadata implements GridCacheSqlIndexMetadata {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** */
-        private String name;
-
-        /** */
-        private Collection<String> fields;
-
-        /** */
-        private Collection<String> descendings;
-
-        /** */
-        private boolean unique;
-
-        /**
-         * Required by {@link Externalizable}.
-         */
-        public CacheSqlIndexMetadata() {
-            // No-op.
-        }
-
-        /**
-         * @param name Index name.
-         * @param fields Fields.
-         * @param descendings Descendings.
-         * @param unique Unique flag.
-         */
-        CacheSqlIndexMetadata(String name, Collection<String> fields, Collection<String> descendings,
-            boolean unique) {
-            assert name != null;
-            assert fields != null;
-            assert descendings != null;
-
-            this.name = name;
-            this.fields = fields;
-            this.descendings = descendings;
-            this.unique = unique;
-        }
-
-        /** {@inheritDoc} */
-        @Override public String name() {
-            return name;
-        }
-
-        /** {@inheritDoc} */
-        @Override public Collection<String> fields() {
-            return fields;
-        }
-
-        /** {@inheritDoc} */
-        @Override public boolean descending(String field) {
-            return descendings.contains(field);
-        }
-
-        /** {@inheritDoc} */
-        @Override public boolean unique() {
-            return unique;
-        }
-
-        /** {@inheritDoc} */
-        @Override public void writeExternal(ObjectOutput out) throws IOException {
-            U.writeString(out, name);
-            U.writeCollection(out, fields);
-            U.writeCollection(out, descendings);
-            out.writeBoolean(unique);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-            name = U.readString(in);
-            fields = U.readCollection(in);
-            descendings = U.readCollection(in);
-            unique = in.readBoolean();
-        }
-
-        /** {@inheritDoc} */
-        @Override public String toString() {
-            return S.toString(CacheSqlIndexMetadata.class, this);
-        }
-    }
-
-    /**
-     *
-     */
-    private static class QueryResult<K, V> extends CachedResult<IgniteBiTuple<K, V>> {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** */
-        private final GridCacheQueryType type;
-
-        /**
-         * @param type Query type.
-         * @param recipient ID of the recipient.
-         */
-        private QueryResult(GridCacheQueryType type, Object recipient) {
-            super(recipient);
-
-            this.type = type;
-        }
-
-        /**
-         * @return Type.
-         */
-        public GridCacheQueryType type() {
-            return type;
-        }
-    }
-
-    /**
-     *
-     */
-    private static class FieldsResult<Q> extends CachedResult<Q> {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** */
-        private List<GridQueryFieldMetadata> meta;
-
-        /**
-         * @param recipient ID of the recipient.
-         */
-        FieldsResult(Object recipient) {
-            super(recipient);
-        }
-
-        /**
-         * @return Metadata.
-         */
-        public List<GridQueryFieldMetadata> metaData() throws IgniteCheckedException {
-            get(); // Ensure that result is ready.
-
-            return meta;
-        }
-
-        /**
-         * @param meta Metadata.
-         */
-        public void metaData(List<GridQueryFieldMetadata> meta) {
-            this.meta = meta;
-        }
-    }
-
-    /**
-     *
-     */
-    private abstract class AbstractLazySwapEntry {
-        /** */
-        private K key;
-
-        /** */
-        private V val;
-
-        /**
-         * @return Key bytes.
-         */
-        protected abstract byte[] keyBytes();
-
-        /**
-         * @return Value.
-         * @throws IgniteCheckedException If failed.
-         */
-        protected abstract V unmarshalValue() throws IgniteCheckedException;
-
-        /**
-         * @return Key.
-         */
-        K key() {
-            try {
-                if (key != null)
-                    return key;
-
-                key = cctx.marshaller().unmarshal(keyBytes(), cctx.deploy().globalLoader());
-
-                return key;
-            }
-            catch (IgniteCheckedException e) {
-                throw new IgniteException(e);
-            }
-        }
-
-        /**
-         * @return Value.
-         */
-        V value() {
-            try {
-                if (val != null)
-                    return val;
-
-                val = unmarshalValue();
-
-                return val;
-            }
-            catch (IgniteCheckedException e) {
-                throw new IgniteException(e);
-            }
-        }
-
-        /**
-         * @return TTL.
-         */
-        abstract long timeToLive();
-
-        /**
-         * @return Expire time.
-         */
-        abstract long expireTime();
-
-        /**
-         * @return Version.
-         */
-        abstract GridCacheVersion version();
-    }
-
-    /**
-     *
-     */
-    private class LazySwapEntry extends AbstractLazySwapEntry {
-        /** */
-        private final Map.Entry<byte[], byte[]> e;
-
-        /**
-         * @param e Entry with
-         */
-        LazySwapEntry(Map.Entry<byte[], byte[]> e) {
-            this.e = e;
-        }
-
-        /** {@inheritDoc} */
-        @Override protected byte[] keyBytes() {
-            return e.getKey();
-        }
-
-        /** {@inheritDoc} */
-        @SuppressWarnings("IfMayBeConditional")
-        @Override protected V unmarshalValue() throws IgniteCheckedException {
-            byte[] bytes = e.getValue();
-
-            byte[] val = GridCacheSwapEntryImpl.getValueIfByteArray(bytes);
-
-            if (val != null)
-                return (V)val;
-
-            if (cctx.offheapTiered() && cctx.portableEnabled())
-                return (V)cctx.portable().unmarshal(bytes, GridCacheSwapEntryImpl.valueOffset(bytes));
-            else {
-                GridByteArrayInputStream in = new GridByteArrayInputStream(bytes,
-                    GridCacheSwapEntryImpl.valueOffset(bytes),
-                    bytes.length);
-
-                return cctx.marshaller().unmarshal(in, cctx.deploy().globalLoader());
-            }
-        }
-
-        /** {@inheritDoc} */
-        @Override long timeToLive() {
-            return GridCacheSwapEntryImpl.timeToLive(e.getValue());
-        }
-
-        /** {@inheritDoc} */
-        @Override long expireTime() {
-            return GridCacheSwapEntryImpl.expireTime(e.getValue());
-        }
-
-        /** {@inheritDoc} */
-        @Override GridCacheVersion version() {
-            return GridCacheSwapEntryImpl.version(e.getValue());
-        }
-    }
-
-    /**
-     *
-     */
-    private class LazyOffheapEntry extends AbstractLazySwapEntry {
-        /** */
-        private final T2<Long, Integer> keyPtr;
-
-        /** */
-        private final T2<Long, Integer> valPtr;
-
-        /**
-         * @param keyPtr Key address.
-         * @param valPtr Value address.
-         */
-        private LazyOffheapEntry(T2<Long, Integer> keyPtr, T2<Long, Integer> valPtr) {
-            assert keyPtr != null;
-            assert valPtr != null;
-
-            this.keyPtr = keyPtr;
-            this.valPtr = valPtr;
-        }
-
-        /** {@inheritDoc} */
-        @Override protected byte[] keyBytes() {
-            return U.copyMemory(keyPtr.get1(), keyPtr.get2());
-        }
-
-        /** {@inheritDoc} */
-        @Override protected V unmarshalValue() throws IgniteCheckedException {
-            long ptr = GridCacheOffheapSwapEntry.valueAddress(valPtr.get1(), valPtr.get2());
-
-            V val = (V)cctx.portable().unmarshal(ptr, false);
-
-            assert val != null;
-
-            return val;
-        }
-
-        /** {@inheritDoc} */
-        @Override long timeToLive() {
-            return GridCacheOffheapSwapEntry.timeToLive(valPtr.get1());
-        }
-
-        /** {@inheritDoc} */
-        @Override long expireTime() {
-            return GridCacheOffheapSwapEntry.expireTime(valPtr.get1());
-        }
-
-        /** {@inheritDoc} */
-        @Override GridCacheVersion version() {
-            return GridCacheOffheapSwapEntry.version(valPtr.get1());
-        }
-    }
-
-    /**
-     *
-     */
-    private class OffheapIteratorClosure
-        extends CX2<T2<Long, Integer>, T2<Long, Integer>, IgniteBiTuple<K, V>> {
-        /** */
-        private static final long serialVersionUID = 7410163202728985912L;
-
-        /** */
-        private IgnitePredicate<GridCacheEntry<Object, Object>> prjPred;
-
-        /** */
-        private IgniteBiPredicate<K, V> filter;
-
-        /** */
-        private boolean keepPortable;
-
-        /**
-         * @param prjPred Projection filter.
-         * @param filter Filter.
-         * @param keepPortable Keep portable flag.
-         */
-        private OffheapIteratorClosure(
-            @Nullable IgnitePredicate<GridCacheEntry<Object, Object>> prjPred,
-            @Nullable IgniteBiPredicate<K, V> filter,
-            boolean keepPortable) {
-            assert prjPred != null || filter != null;
-
-            this.prjPred = prjPred;
-            this.filter = filter;
-            this.keepPortable = keepPortable;
-        }
-
-        /** {@inheritDoc} */
-        @Nullable @Override public IgniteBiTuple<K, V> applyx(T2<Long, Integer> keyPtr,
-            T2<Long, Integer> valPtr)
-            throws IgniteCheckedException {
-            LazyOffheapEntry e = new LazyOffheapEntry(keyPtr, valPtr);
-
-            if (prjPred != null) {
-                GridCacheEntry<K, V> cacheEntry = new GridCacheScanSwapEntry(e);
-
-                if (!prjPred.apply((GridCacheEntry<Object, Object>)cacheEntry))
-                    return null;
-            }
-
-            if (filter != null) {
-                K key = (K)cctx.unwrapPortableIfNeeded(e.key(), keepPortable);
-                V val = (V)cctx.unwrapPortableIfNeeded(e.value(), keepPortable);
-
-                if (!filter.apply(key, val))
-                    return null;
-            }
-
-            return new IgniteBiTuple<>(e.key(), (V)cctx.unwrapTemporary(e.value())) ;
-        }
-    }
-
-    /**
-     *
-     */
-    private static class CompoundIterator<T> extends GridIteratorAdapter<T> {
-        /** */
-        private static final long serialVersionUID = 4585888051556166304L;
-
-        /** */
-        private final List<GridIterator<T>> iters;
-
-        /** */
-        private int idx;
-
-        /** */
-        private GridIterator<T> iter;
-
-        /**
-         * @param iters Iterators.
-         */
-        private CompoundIterator(List<GridIterator<T>> iters) {
-            if (iters.isEmpty())
-                throw new IllegalArgumentException();
-
-            this.iters = iters;
-
-            iter = F.first(iters);
-        }
-
-        /** {@inheritDoc} */
-        @Override public boolean hasNextX() throws IgniteCheckedException {
-            if (iter.hasNextX())
-                return true;
-
-            idx++;
-
-            while(idx < iters.size()) {
-                iter = iters.get(idx);
-
-                if (iter.hasNextX())
-                    return true;
-
-                idx++;
-            }
-
-            return false;
-        }
-
-        /** {@inheritDoc} */
-        @Override public T nextX() throws IgniteCheckedException {
-            if (!hasNextX())
-                throw new NoSuchElementException();
-
-            return iter.nextX();
-        }
-
-        /** {@inheritDoc} */
-        @Override public void removeX() throws IgniteCheckedException {
-            throw new UnsupportedOperationException();
-        }
-    }
-
-    /**
-     *
-     */
-    private class GridCacheScanSwapEntry implements GridCacheEntry<K, V> {
-        /** */
-        private static final long serialVersionUID = 1262515168518736214L;
-
-        /** */
-        private final AbstractLazySwapEntry e;
-
-        /**
-         * @param e Entry.
-         */
-        private GridCacheScanSwapEntry(AbstractLazySwapEntry e) {
-            this.e = e;
-        }
-
-        /** {@inheritDoc} */
-        @Nullable @Override public V peek() {
-            return e.value();
-        }
-
-        /** {@inheritDoc} */
-        @Override public Object version() {
-            return e.version();
-        }
-
-        /** {@inheritDoc} */
-        @Override public long expirationTime() {
-            return e.expireTime();
-        }
-
-        /** {@inheritDoc} */
-        @Override public long timeToLive() {
-            return e.timeToLive();
-        }
-
-        /** {@inheritDoc} */
-        @Nullable @Override public V getValue() {
-            return e.value();
-        }
-
-        /** {@inheritDoc} */
-        @Override public K getKey() {
-            return e.key();
-        }
-
-        /** {@inheritDoc} */
-        @Override public GridCacheProjection<K, V> projection() {
-            return null;
-        }
-
-        /** {@inheritDoc} */
-        @Nullable @Override public V peek(@Nullable Collection<GridCachePeekMode> modes) throws IgniteCheckedException {
-            return null;
-        }
-
-        /** {@inheritDoc} */
-        @Nullable @Override public V reload() {
-            throw new UnsupportedOperationException();
-        }
-
-        /** {@inheritDoc} */
-        @Override public IgniteFuture<V> reloadAsync() {
-            throw new UnsupportedOperationException();
-        }
-
-        /** {@inheritDoc} */
-        @Override public boolean isLocked() {
-            return false;
-        }
-
-        /** {@inheritDoc} */
-        @Override public boolean isLockedByThread() {
-            return false;
-        }
-
-        /** {@inheritDoc} */
-        @Override public void timeToLive(long ttl) {
-            throw new UnsupportedOperationException();
-        }
-
-        /** {@inheritDoc} */
-        @Override public boolean primary() {
-            return cctx.affinity().
-                primary(cctx.localNode(), getKey(), cctx.affinity().affinityTopologyVersion());
-        }
-
-        /** {@inheritDoc} */
-        @Override public boolean backup() {
-            return cctx.affinity().
-                backu

<TRUNCATED>

Mime
View raw message