Fix

Author: timoninmaxim

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/CacheQuery.java

@@ -840,9 +840,12 @@ private Collection<ClusterNode> nodes() throws IgniteCheckedException {
     }
 
     /**
+     * Collects query data nodes matching specified {@code prj} and {@code part}.
+     *
      * @param cctx Cache context.
      * @param prj Projection (optional).
-     * @return Collection of data nodes in provided projection (if any).
+     * @param part Partition (optional).
+     * @return Collection of data nodes matching specified {@code prj} and {@code part}.
      * @throws IgniteCheckedException If partition number is invalid.
      */
     private static Collection<ClusterNode> nodes(final GridCacheContext<?, ?> cctx,
@@ -856,25 +859,17 @@ private static Collection<ClusterNode> nodes(final GridCacheContext<?, ?> cctx,
         if (prj == null && part == null)
             return affNodes;
 
-        if (part != null && part >= cctx.affinity().partitions())
-            throw new IgniteCheckedException("Invalid partition number: " + part);
-
-        final Set<ClusterNode> partHolders;
-
         if (part != null) {
-            if (cctx.config().getCacheMode() == CacheMode.PARTITIONED)
-                partHolders = Collections.singleton(cctx.affinity().primaryByPartition(part, topVer));
-            else
-                partHolders = new HashSet<>(cctx.topology().owners(part, topVer));
+            if (part >= cctx.affinity().partitions())
+                throw new IgniteCheckedException("Invalid partition number: " + part);
+
+            affNodes = cctx.topology().nodes(part, topVer);
         }
-        else
-            partHolders = Collections.emptySet();
 
         return F.view(affNodes, new P1<ClusterNode>() {
             @Override public boolean apply(ClusterNode n) {
                 return cctx.discovery().cacheAffinityNode(n, cctx.name()) &&
-                    (prj == null || prj.node(n.id()) != null) &&
-                    (part == null || partHolders.contains(n));
+                    (prj == null || prj.node(n.id()) != null);
             }
         });
     }

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryFuture.java

@@ -25,6 +25,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.ignite.IgniteCheckedException;
@@ -38,6 +39,7 @@
 import org.apache.ignite.internal.processors.cache.query.reducer.TextQueryReducer;
 import org.apache.ignite.internal.processors.cache.query.reducer.UnsortedCacheQueryReducer;
 import org.apache.ignite.internal.util.lang.GridPlainCallable;
+import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.U;
 
 import static org.apache.ignite.internal.processors.cache.query.GridCacheQueryType.INDEX;
@@ -92,7 +94,7 @@ protected GridCacheDistributedQueryFuture(
         qryMgr = (GridCacheDistributedQueryManager<K, V>)ctx.queries();
 
         if (qry.query().partition() != null)
-            nodes = Collections.singletonList(node(nodes));
+            nodes = Collections.singletonList(cctx.isReplicated() ? localOrRemoteNode(nodes) : F.first(nodes));
 
         streams = new ConcurrentHashMap<>(nodes.size());
 
@@ -118,19 +120,24 @@ protected GridCacheDistributedQueryFuture(
     }
 
     /**
-     * @return Nodes for query execution.
+     * @return A local node if available, otherwise a random node from the given collection.
      */
-    private ClusterNode node(Collection<ClusterNode> nodes) {
-        ClusterNode rmtNode = null;
+    private static ClusterNode localOrRemoteNode(Collection<ClusterNode> nodes) {
+        int remoteNodeIdx = ThreadLocalRandom.current().nextInt(nodes.size());
+
+        ClusterNode remoteNode = null;
 
         for (ClusterNode node : nodes) {
             if (node.isLocal())
                 return node;
 
-            rmtNode = node;
+            if (remoteNodeIdx-- == 0)
+                remoteNode = node;
         }
 
-        return rmtNode;
+        assert remoteNode != null;
+
+        return remoteNode;
     }
 
     /** {@inheritDoc} */
@@ -282,7 +289,7 @@ private void cancelPages(UUID nodeId) {
             GridCacheQueryRequest req = GridCacheQueryRequest.cancelRequest(cctx, reqId, fields());
 
             if (nodeId.equals(cctx.localNodeId())) {
-                // Process cancel query directly (without sending) for local node,
+                // Process cancel query directly (without sending) for local node.
                 cctx.closures().callLocalSafe(new GridPlainCallable<Object>() {
                     @Override public Object call() {
                         qryMgr.processQueryRequest(cctx.localNodeId(), req);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java

@@ -46,7 +46,6 @@
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.IgniteSystemProperties;
 import org.apache.ignite.cache.CacheEntry;
-import org.apache.ignite.cache.CacheMode;
 import org.apache.ignite.cache.QueryIndexType;
 import org.apache.ignite.cache.query.IndexQuery;
 import org.apache.ignite.cache.query.QueryMetrics;
@@ -1124,10 +1123,7 @@ public GridCloseableIterator indexQueryLocal(final CacheQuery qry) throws Ignite
             if (part != null) {
                 final GridDhtLocalPartition locPart = cctx.dht().topology().localPartition(part);
 
-                boolean nonLocPart = locPart == null || locPart.state() != OWNING ||
-                        (cctx.config().getCacheMode() == CacheMode.PARTITIONED && !locPart.primary(AffinityTopologyVersion.NONE));
-
-                if (nonLocPart) {
+                if (locPart == null || locPart.state() != OWNING) {
                     throw new CacheInvalidStateException("Failed to execute index query because required partition " +
                         "has not been found on local node [cacheName=" + cctx.name() + ", part=" + part + "]");
                 }