You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ak...@apache.org on 2015/04/10 17:48:51 UTC
[04/17] incubator-ignite git commit: # IGNITE-692 WIP rework of query
tasks.
# IGNITE-692 WIP rework of query tasks.
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/9fb5651e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/9fb5651e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/9fb5651e
Branch: refs/heads/ignite-sprint-3
Commit: 9fb5651ee3ae16805b6295d20424c1a5aaab71ae
Parents: 2576622
Author: AKuznetsov <ak...@gridgain.com>
Authored: Wed Apr 8 17:53:21 2015 +0700
Committer: AKuznetsov <ak...@gridgain.com>
Committed: Wed Apr 8 17:53:21 2015 +0700
----------------------------------------------------------------------
.../internal/visor/query/VisorQueryTask.java | 37 +++++++++++++-------
.../commands/cache/VisorCacheScanCommand.scala | 2 +-
2 files changed, 25 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9fb5651e/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryTask.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryTask.java
index ae63e3c..5f66a6d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryTask.java
@@ -19,6 +19,8 @@ package org.apache.ignite.internal.visor.query;
import org.apache.ignite.*;
import org.apache.ignite.cache.query.*;
+import org.apache.ignite.cluster.*;
+import org.apache.ignite.compute.*;
import org.apache.ignite.internal.processors.cache.*;
import org.apache.ignite.internal.processors.query.*;
import org.apache.ignite.internal.processors.task.*;
@@ -34,6 +36,7 @@ import java.util.*;
import java.util.concurrent.*;
import static org.apache.ignite.internal.visor.query.VisorQueryUtils.*;
+import static org.apache.ignite.internal.visor.util.VisorTaskUtils.*;
/**
* Task for execute SCAN or SQL query and get first page of results.
@@ -45,6 +48,26 @@ public class VisorQueryTask extends VisorOneNodeTask<VisorQueryTask.VisorQueryAr
private static final long serialVersionUID = 0L;
/** {@inheritDoc} */
+ @Override protected Map<? extends ComputeJob, ClusterNode> map0(List<ClusterNode> subgrid,
+ VisorTaskArgument<VisorQueryTask.VisorQueryArg> arg) {
+ String cacheName = taskArg.cacheName();
+
+ ClusterGroup prj = ignite.cluster().forDataNodes(cacheName);
+
+ if (prj.nodes().isEmpty())
+ throw new IgniteException("No data nodes for cache: " + escapeName(cacheName));
+
+ // First try to take local node to avoid network hop.
+ ClusterNode node = prj.node(ignite.localNode().id());
+
+ // Take any node from projection.
+ if (node == null)
+ node = prj.forRandom().node();
+
+ return Collections.singletonMap(job(taskArg), node);
+ }
+
+ /** {@inheritDoc} */
@Override protected VisorQueryJob job(VisorQueryArg arg) {
return new VisorQueryJob(arg, debug);
}
@@ -57,9 +80,6 @@ public class VisorQueryTask extends VisorOneNodeTask<VisorQueryTask.VisorQueryAr
/** */
private static final long serialVersionUID = 0L;
- /** Node ids for query. */
- private final Collection<UUID> proj;
-
/** Cache name for query. */
private final String cacheName;
@@ -70,26 +90,17 @@ public class VisorQueryTask extends VisorOneNodeTask<VisorQueryTask.VisorQueryAr
private final Integer pageSize;
/**
- * @param proj Node ids for query.
* @param cacheName Cache name for query.
* @param qryTxt Query text.
* @param pageSize Result batch size.
*/
- public VisorQueryArg(Collection<UUID> proj, String cacheName, String qryTxt, Integer pageSize) {
- this.proj = proj;
+ public VisorQueryArg(String cacheName, String qryTxt, Integer pageSize) {
this.cacheName = cacheName;
this.qryTxt = qryTxt;
this.pageSize = pageSize;
}
/**
- * @return Proj.
- */
- public Collection<UUID> proj() {
- return proj;
- }
-
- /**
* @return Cache name.
*/
public String cacheName() {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9fb5651e/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/cache/VisorCacheScanCommand.scala
----------------------------------------------------------------------
diff --git a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/cache/VisorCacheScanCommand.scala b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/cache/VisorCacheScanCommand.scala
index 4fb1896..7574be1 100644
--- a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/cache/VisorCacheScanCommand.scala
+++ b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/cache/VisorCacheScanCommand.scala
@@ -163,7 +163,7 @@ class VisorCacheScanCommand {
.withName("visor-cscan-task")
.withNoFailover()
.execute(classOf[VisorQueryTask],
- toTaskArgument(nid, new VisorQueryArg(proj, cacheName, "SCAN", pageSize)))
+ toTaskArgument(nid, new VisorQueryArg(cacheName, "SCAN", pageSize)))
match {
case x if x.get1() != null =>
error(x.get1())