You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2015/04/10 19:07:09 UTC
[13/40] incubator-ignite git commit: # IGNITE-692 Implemented support
for GG and Ignite.
# IGNITE-692 Implemented support for GG and Ignite.
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/97a83a03
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/97a83a03
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/97a83a03
Branch: refs/heads/ignite-537
Commit: 97a83a034cd17fbef9e5f2a77410ee326e91a1a0
Parents: aa8f668
Author: AKuznetsov <ak...@gridgain.com>
Authored: Fri Apr 10 14:24:27 2015 +0700
Committer: AKuznetsov <ak...@gridgain.com>
Committed: Fri Apr 10 14:24:27 2015 +0700
----------------------------------------------------------------------
.../query/GridQueryFieldMetadata.java | 2 +-
.../visor/cache/VisorCacheConfiguration.java | 2 +-
.../internal/visor/cache/VisorCacheMetrics.java | 2 +-
.../internal/visor/query/VisorQueryArg.java | 69 ++++++
.../visor/query/VisorQueryCleanupTask.java | 2 +-
.../visor/query/VisorQueryCursorHolder.java | 64 ++++++
.../internal/visor/query/VisorQueryField.java | 66 ++++--
.../internal/visor/query/VisorQueryJob.java | 160 +++++++++++++
.../visor/query/VisorQueryNextPageTask.java | 46 ++--
.../internal/visor/query/VisorQueryResult.java | 6 +-
.../visor/query/VisorQueryResultEx.java | 16 +-
.../internal/visor/query/VisorQueryTask.java | 227 +------------------
.../internal/visor/query/VisorQueryUtils.java | 8 +-
.../resources/META-INF/classnames.properties | 6 +-
.../commands/cache/VisorCacheScanCommand.scala | 11 +-
15 files changed, 393 insertions(+), 294 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/97a83a03/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryFieldMetadata.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryFieldMetadata.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryFieldMetadata.java
index 9ebf768..f99466f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryFieldMetadata.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryFieldMetadata.java
@@ -34,7 +34,7 @@ public interface GridQueryFieldMetadata extends Externalizable {
/**
* Gets name of type to which this field belongs.
*
- * @return Gets type name.
+ * @return Type name.
*/
public String typeName();
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/97a83a03/modules/core/src/main/java/org/apache/ignite/internal/visor/cache/VisorCacheConfiguration.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/cache/VisorCacheConfiguration.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/cache/VisorCacheConfiguration.java
index 86a7f5f..d070872 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/cache/VisorCacheConfiguration.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/cache/VisorCacheConfiguration.java
@@ -146,7 +146,7 @@ public class VisorCacheConfiguration implements Serializable {
writerFactory = compactClass(ccfg.getCacheWriterFactory());
expiryPlcFactory = compactClass(ccfg.getExpiryPolicyFactory());
sys = ignite.context().cache().systemCache(ccfg.getName());
-
+
affinityCfg = VisorCacheAffinityConfiguration.from(ccfg);
rebalanceCfg = VisorCacheRebalanceConfiguration.from(ccfg);
evictCfg = VisorCacheEvictionConfiguration.from(ccfg);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/97a83a03/modules/core/src/main/java/org/apache/ignite/internal/visor/cache/VisorCacheMetrics.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/cache/VisorCacheMetrics.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/cache/VisorCacheMetrics.java
index fc30392..b0a1eb2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/cache/VisorCacheMetrics.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/cache/VisorCacheMetrics.java
@@ -245,7 +245,7 @@ public class VisorCacheMetrics implements Serializable {
}
/**
- * @return Cache mode.
+ * @return Cache mode.
*/
public CacheMode mode() {
return mode;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/97a83a03/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryArg.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryArg.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryArg.java
new file mode 100644
index 0000000..080bb35
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryArg.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.visor.query;
+
+import java.io.*;
+
+/**
+ * Arguments for {@link VisorQueryTask}.
+ */
+public class VisorQueryArg implements Serializable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Cache name for query. */
+ private final String cacheName;
+
+ /** Query text. */
+ private final String qryTxt;
+
+ /** Result batch size. */
+ private final Integer pageSize;
+
+ /**
+ * @param cacheName Cache name for query.
+ * @param qryTxt Query text.
+ * @param pageSize Result batch size.
+ */
+ public VisorQueryArg(String cacheName, String qryTxt, Integer pageSize) {
+ this.cacheName = cacheName;
+ this.qryTxt = qryTxt;
+ this.pageSize = pageSize;
+ }
+
+ /**
+ * @return Cache name.
+ */
+ public String cacheName() {
+ return cacheName;
+ }
+
+ /**
+ * @return Query txt.
+ */
+ public String queryTxt() {
+ return qryTxt;
+ }
+
+ /**
+ * @return Page size.
+ */
+ public Integer pageSize() {
+ return pageSize;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/97a83a03/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCleanupTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCleanupTask.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCleanupTask.java
index 1386638..d82f5fb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCleanupTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCleanupTask.java
@@ -86,7 +86,7 @@ public class VisorQueryCleanupTask extends VisorMultiNodeTask<Map<UUID, Collecti
/** {@inheritDoc} */
@Override protected Void run(Collection<String> qryIds) {
- ConcurrentMap<String, VisorQueryTask.VisorQueryCursorHolder> locMap = ignite.cluster().nodeLocalMap();
+ ConcurrentMap<String, VisorQueryCursorHolder> locMap = ignite.cluster().nodeLocalMap();
for (String qryId : qryIds)
locMap.remove(qryId);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/97a83a03/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCursorHolder.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCursorHolder.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCursorHolder.java
new file mode 100644
index 0000000..dec0253
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCursorHolder.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.visor.query;
+
+import java.io.*;
+
+/**
+ * ResultSet future holder.
+ */
+public class VisorQueryCursorHolder<T> implements Serializable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Query cursor. */
+ private final VisorQueryCursor<T> cur;
+
+ /** Flag indicating that this future was read from last check. */
+ private volatile boolean accessed;
+
+ /**
+ * @param cur Future.
+ * @param accessed {@code true} if query was accessed before remove timeout expired.
+ */
+ public VisorQueryCursorHolder(VisorQueryCursor<T> cur, boolean accessed) {
+ this.cur = cur;
+ this.accessed = accessed;
+ }
+
+ /**
+ * @return Query cursor.
+ */
+ public VisorQueryCursor<T> cursor() {
+ return cur;
+ }
+
+ /**
+ * @return Flag indicating that this future was read from last check..
+ */
+ public boolean accessed() {
+ return accessed;
+ }
+
+ /**
+ * @param accessed New accessed.
+ */
+ public void accessed(boolean accessed) {
+ this.accessed = accessed;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/97a83a03/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryField.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryField.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryField.java
index 78bcac6..cd97c7c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryField.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryField.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.visor.query;
+import org.apache.ignite.internal.util.typedef.*;
import org.apache.ignite.internal.util.typedef.internal.*;
import java.io.*;
@@ -28,35 +29,74 @@ public class VisorQueryField implements Serializable {
/** */
private static final long serialVersionUID = 0L;
- /** Column type. */
- private final String type;
+ /** Schema name. */
+ private String schemaName;
+
+ /** Type name. */
+ private String typeName;
/** Field name. */
- private final String field;
+ private String fieldName;
+
+ /** Field type name. */
+ private String fieldTypeName;
/**
* Create data transfer object with given parameters.
*
- * @param type Column type.
- * @param field Field name.
+ * @param schemaName Schema name.
+ * @param typeName Type name.
+ * @param fieldName Name.
+ * @param fieldTypeName Type.
+ */
+ public VisorQueryField(String schemaName, String typeName, String fieldName, String fieldTypeName) {
+ this.schemaName = schemaName;
+ this.typeName = typeName;
+ this.fieldName = fieldName;
+ this.fieldTypeName = fieldTypeName;
+ }
+
+ /**
+ * @return Schema name.
*/
- public VisorQueryField(String type, String field) {
- this.type = type;
- this.field = field;
+ public String schemaName() {
+ return schemaName;
}
/**
- * @return Column type.
+ * @return Type name.
*/
- public String type() {
- return type;
+ public String typeName() {
+ return typeName;
}
/**
* @return Field name.
*/
- public String field() {
- return field;
+ public String fieldName() {
+ return fieldName;
+ }
+
+ /**
+ * @return Field type name.
+ */
+ public String fieldTypeName() {
+ return fieldTypeName;
+ }
+
+ /**
+ * @param schema If {@code true} then add schema name to full name.
+ * @return Fully qualified field name with type name and schema name.
+ */
+ public String fullName(boolean schema) {
+ if (!F.isEmpty(typeName)) {
+ if (schema && !F.isEmpty(schemaName))
+ return schemaName + "." + typeName + "." + fieldName;
+
+ return typeName + "." + fieldName;
+ }
+
+ return fieldName;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/97a83a03/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryJob.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryJob.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryJob.java
new file mode 100644
index 0000000..2d8b32e
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryJob.java
@@ -0,0 +1,160 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.visor.query;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.query.*;
+import org.apache.ignite.internal.processors.cache.*;
+import org.apache.ignite.internal.processors.query.*;
+import org.apache.ignite.internal.processors.timeout.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.internal.visor.*;
+import org.apache.ignite.lang.*;
+
+import javax.cache.*;
+import java.sql.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+import static org.apache.ignite.internal.visor.query.VisorQueryUtils.*;
+
+/**
+ * Job for execute SCAN or SQL query and get first page of results.
+ */
+public class VisorQueryJob extends VisorJob<VisorQueryArg, IgniteBiTuple<? extends Exception, VisorQueryResultEx>> {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /**
+ * Create job with specified argument.
+ *
+ * @param arg Job argument.
+ * @param debug Debug flag.
+ */
+ protected VisorQueryJob(VisorQueryArg arg, boolean debug) {
+ super(arg, debug);
+ }
+
+ /**
+ * @param cacheName Cache name.
+ * @return Cache to execute query.
+ */
+ protected IgniteCache<Object, Object> cache(String cacheName) {
+ GridCacheProcessor cacheProcessor = ignite.context().cache();
+
+ return cacheProcessor.jcache(cacheName);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteBiTuple<? extends Exception, VisorQueryResultEx> run(VisorQueryArg arg) {
+ try {
+ boolean scan = arg.queryTxt().toUpperCase().startsWith("SCAN");
+
+ String qryId = (scan ? SCAN_QRY_NAME : SQL_QRY_NAME) + "-" +
+ UUID.randomUUID();
+
+ IgniteCache<Object, Object> c = cache(arg.cacheName());
+
+ if (scan) {
+ ScanQuery<Object, Object> qry = new ScanQuery<>(null);
+ qry.setPageSize(arg.pageSize());
+
+ long start = U.currentTimeMillis();
+
+ VisorQueryCursor<Cache.Entry<Object, Object>> cur = new VisorQueryCursor<>(c.query(qry));
+
+ List<Object[]> rows = fetchScanQueryRows(cur, arg.pageSize());
+
+ long duration = U.currentTimeMillis() - start; // Scan duration + fetch duration.
+
+ ignite.cluster().<String, VisorQueryCursorHolder>nodeLocalMap().put(qryId,
+ new VisorQueryCursorHolder<>(cur, false));
+
+ scheduleResultSetHolderRemoval(qryId);
+
+ return new IgniteBiTuple<>(null, new VisorQueryResultEx(ignite.localNode().id(), qryId,
+ SCAN_COL_NAMES, rows, cur.hasNext(), duration));
+ }
+ else {
+ SqlFieldsQuery qry = new SqlFieldsQuery(arg.queryTxt());
+ qry.setPageSize(arg.pageSize());
+
+ long start = U.currentTimeMillis();
+
+ VisorQueryCursor<List<?>> cur = new VisorQueryCursor<>(c.query(qry));
+
+ Collection<GridQueryFieldMetadata> meta = cur.fieldsMeta();
+
+ if (meta == null)
+ return new IgniteBiTuple<Exception, VisorQueryResultEx>(
+ new SQLException("Fail to execute query. No metadata available."), null);
+ else {
+ List<VisorQueryField> names = new ArrayList<>(meta.size());
+
+ for (GridQueryFieldMetadata col : meta)
+ names.add(new VisorQueryField(col.schemaName(), col.typeName(),
+ col.fieldName(), col.fieldTypeName()));
+
+ List<Object[]> rows = fetchSqlQueryRows(cur, arg.pageSize());
+
+ long duration = U.currentTimeMillis() - start; // Query duration + fetch duration.
+
+ ConcurrentMap<String, VisorQueryCursorHolder<List<?>>> storage = ignite.cluster().nodeLocalMap();
+ storage.put(qryId, new VisorQueryCursorHolder<>(cur, false));
+
+ scheduleResultSetHolderRemoval(qryId);
+
+ return new IgniteBiTuple<>(null, new VisorQueryResultEx(ignite.localNode().id(), qryId,
+ names, rows, cur.hasNext(), duration));
+ }
+ }
+ }
+ catch (Exception e) {
+ return new IgniteBiTuple<>(e, null);
+ }
+ }
+
+ /**
+ * @param id Unique query result id.
+ */
+ private void scheduleResultSetHolderRemoval(final String id) {
+ ignite.context().timeout().addTimeoutObject(new GridTimeoutObjectAdapter(RMV_DELAY) {
+ @Override public void onTimeout() {
+ ConcurrentMap<String, VisorQueryCursorHolder> storage = ignite.cluster().nodeLocalMap();
+
+ VisorQueryCursorHolder holder = storage.get(id);
+
+ if (holder != null) {
+ // If future was accessed since last scheduling, set access flag to false and reschedule.
+ if (holder.accessed()) {
+ holder.accessed(false);
+
+ scheduleResultSetHolderRemoval(id);
+ }
+ else
+ storage.remove(id); // Remove stored future otherwise.
+ }
+ }
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(VisorQueryJob.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/97a83a03/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryNextPageTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryNextPageTask.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryNextPageTask.java
index 320bdde..191f44d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryNextPageTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryNextPageTask.java
@@ -23,6 +23,7 @@ import org.apache.ignite.internal.util.typedef.internal.*;
import org.apache.ignite.internal.visor.*;
import org.apache.ignite.lang.*;
+import javax.cache.*;
import java.util.*;
import java.util.concurrent.*;
@@ -58,40 +59,33 @@ public class VisorQueryNextPageTask extends VisorOneNodeTask<IgniteBiTuple<Strin
/** {@inheritDoc} */
@Override protected VisorQueryResult run(IgniteBiTuple<String, Integer> arg) {
- try {
- return arg.get1().startsWith(VisorQueryUtils.SCAN_QRY_NAME) ? nextScanPage(arg) : nextSqlPage(arg);
- }
- catch (IgniteCheckedException e) {
- throw U.convertException(e);
- }
+ return arg.get1().startsWith(VisorQueryUtils.SCAN_QRY_NAME) ? nextScanPage(arg) : nextSqlPage(arg);
}
/**
* Collect data from SQL query.
*
- * @param arg
- * @return
- * @throws IgniteCheckedException
+ * @param arg Query name and page size.
+ * @return Query result with next page.
*/
- private VisorQueryResult nextSqlPage(IgniteBiTuple<String, Integer> arg) throws IgniteCheckedException {
+ private VisorQueryResult nextSqlPage(IgniteBiTuple<String, Integer> arg) {
long start = U.currentTimeMillis();
- ConcurrentMap<String, VisorQueryTask.VisorQueryCursorHolder> storage =
- ignite.cluster().nodeLocalMap();
+ ConcurrentMap<String, VisorQueryCursorHolder<List<?>>> storage = ignite.cluster().nodeLocalMap();
- VisorQueryTask.VisorQueryCursorHolder t = storage.get(arg.get1());
+ VisorQueryCursorHolder<List<?>> holder = storage.get(arg.get1());
- if (t == null)
+ if (holder == null)
throw new IgniteException("SQL query results are expired.");
- VisorQueryCursor cur = t.cursor();
+ VisorQueryCursor<List<?>> cur = holder.cursor();
List<Object[]> nextRows = VisorQueryUtils.fetchSqlQueryRows(cur, arg.get2());
boolean hasMore = cur.hasNext();
if (hasMore)
- storage.put(arg.get1(), new VisorQueryTask.VisorQueryCursorHolder(t.cursor(), true));
+ holder.accessed(true);
else
storage.remove(arg.get1());
@@ -101,29 +95,27 @@ public class VisorQueryNextPageTask extends VisorOneNodeTask<IgniteBiTuple<Strin
/**
* Collect data from SCAN query
*
- * @param arg
- * @return
- * @throws IgniteCheckedException
+ * @param arg Query name and page size.
+ * @return Next page with data.
*/
- private VisorQueryResult nextScanPage(IgniteBiTuple<String, Integer> arg) throws IgniteCheckedException {
+ private VisorQueryResult nextScanPage(IgniteBiTuple<String, Integer> arg) {
long start = U.currentTimeMillis();
- ConcurrentMap<String, VisorQueryTask.VisorQueryCursorHolder> storage =
- ignite.cluster().nodeLocalMap();
+ ConcurrentMap<String, VisorQueryCursorHolder<Cache.Entry<Object, Object>>> storage = ignite.cluster().nodeLocalMap();
- VisorQueryTask.VisorQueryCursorHolder t = storage.get(arg.get1());
+ VisorQueryCursorHolder<Cache.Entry<Object, Object>> holder = storage.get(arg.get1());
- if (t == null)
+ if (holder == null)
throw new IgniteException("Scan query results are expired.");
- VisorQueryCursor cur = t.cursor();
+ VisorQueryCursor<Cache.Entry<Object, Object>> cur = holder.cursor();
List<Object[]> rows = VisorQueryUtils.fetchScanQueryRows(cur, arg.get2());
- Boolean hasMore = cur.hasNext();
+ boolean hasMore = cur.hasNext();
if (hasMore)
- storage.put(arg.get1(), new VisorQueryTask.VisorQueryCursorHolder(cur, true));
+ holder.accessed(true);
else
storage.remove(arg.get1());
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/97a83a03/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryResult.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryResult.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryResult.java
index 1a8c9e9..48c26b9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryResult.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryResult.java
@@ -33,7 +33,7 @@ public class VisorQueryResult implements Serializable {
private final List<Object[]> rows;
/** Whether query has more rows to fetch. */
- private final Boolean hasMore;
+ private final boolean hasMore;
/** Query duration */
private final long duration;
@@ -45,7 +45,7 @@ public class VisorQueryResult implements Serializable {
* @param hasMore Whether query has more rows to fetch.
* @param duration Query duration.
*/
- public VisorQueryResult(List<Object[]> rows, Boolean hasMore, long duration) {
+ public VisorQueryResult(List<Object[]> rows, boolean hasMore, long duration) {
this.rows = rows;
this.hasMore = hasMore;
this.duration = duration;
@@ -61,7 +61,7 @@ public class VisorQueryResult implements Serializable {
/**
* @return Whether query has more rows to fetch.
*/
- public Boolean hasMore() {
+ public boolean hasMore() {
return hasMore;
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/97a83a03/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryResultEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryResultEx.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryResultEx.java
index 35ed790..35f08b3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryResultEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryResultEx.java
@@ -35,12 +35,12 @@ public class VisorQueryResultEx extends VisorQueryResult {
private final String qryId;
/** Query columns descriptors. */
- private final Collection<VisorQueryField> colNames;
+ private final Collection<VisorQueryField> cols;
/**
* @param resNodeId Node where query executed.
* @param qryId Query ID for future extraction in nextPage() access.
- * @param colNames Columns types and names.
+ * @param cols Columns descriptors.
* @param rows Rows fetched from query.
* @param hasMore Whether query has more rows to fetch.
* @param duration Query duration.
@@ -48,16 +48,16 @@ public class VisorQueryResultEx extends VisorQueryResult {
public VisorQueryResultEx(
UUID resNodeId,
String qryId,
- Collection<VisorQueryField> colNames,
+ Collection<VisorQueryField> cols,
List<Object[]> rows,
- Boolean hasMore,
+ boolean hasMore,
long duration
) {
super(rows, hasMore, duration);
this.resNodeId = resNodeId;
this.qryId = qryId;
- this.colNames = colNames;
+ this.cols = cols;
}
/**
@@ -75,10 +75,10 @@ public class VisorQueryResultEx extends VisorQueryResult {
}
/**
- * @return Columns names.
+ * @return Columns.
*/
- public Collection<VisorQueryField> columnNames() {
- return colNames;
+ public Collection<VisorQueryField> columns() {
+ return cols;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/97a83a03/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 cc05ba5..e01737f 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
@@ -18,38 +18,27 @@
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.*;
-import org.apache.ignite.internal.processors.timeout.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
import org.apache.ignite.internal.visor.*;
import org.apache.ignite.lang.*;
-import javax.cache.*;
-import java.io.*;
-import java.sql.*;
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.
*/
@GridInternal
-public class VisorQueryTask extends VisorOneNodeTask<VisorQueryTask.VisorQueryArg,
- IgniteBiTuple<? extends Exception, VisorQueryResultEx>> {
+public class VisorQueryTask extends VisorOneNodeTask<VisorQueryArg, IgniteBiTuple<? extends Exception, VisorQueryResultEx>> {
/** */
private static final long serialVersionUID = 0L;
/** {@inheritDoc} */
@Override protected Map<? extends ComputeJob, ClusterNode> map0(List<ClusterNode> subgrid,
- VisorTaskArgument<VisorQueryTask.VisorQueryArg> arg) {
+ VisorTaskArgument<VisorQueryArg> arg) {
String cacheName = taskArg.cacheName();
ClusterGroup prj = ignite.cluster().forDataNodes(cacheName);
@@ -71,216 +60,4 @@ public class VisorQueryTask extends VisorOneNodeTask<VisorQueryTask.VisorQueryAr
@Override protected VisorQueryJob job(VisorQueryArg arg) {
return new VisorQueryJob(arg, debug);
}
-
- /**
- * Arguments for {@link VisorQueryTask}.
- */
- @SuppressWarnings("PublicInnerClass")
- public static class VisorQueryArg implements Serializable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** Cache name for query. */
- private final String cacheName;
-
- /** Query text. */
- private final String qryTxt;
-
- /** Result batch size. */
- private final Integer pageSize;
-
- /**
- * @param cacheName Cache name for query.
- * @param qryTxt Query text.
- * @param pageSize Result batch size.
- */
- public VisorQueryArg(String cacheName, String qryTxt, Integer pageSize) {
- this.cacheName = cacheName;
- this.qryTxt = qryTxt;
- this.pageSize = pageSize;
- }
-
- /**
- * @return Cache name.
- */
- public String cacheName() {
- return cacheName;
- }
-
- /**
- * @return Query txt.
- */
- public String queryTxt() {
- return qryTxt;
- }
-
- /**
- * @return Page size.
- */
- public Integer pageSize() {
- return pageSize;
- }
- }
-
- /**
- * ResultSet future holder.
- */
- @SuppressWarnings("PublicInnerClass")
- public static class VisorQueryCursorHolder implements Serializable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** Query cursor. */
- private final VisorQueryCursor cur;
-
- /** Flag indicating that this future was read from last check. */
- private boolean accessed;
-
- /**
- * @param cur Future.
- * @param accessed {@code true} if query was accessed before remove timeout expired.
- */
- public VisorQueryCursorHolder(VisorQueryCursor cur, boolean accessed) {
- this.cur = cur;
- this.accessed = accessed;
- }
-
- /**
- * @return Query cursor.
- */
- public VisorQueryCursor cursor() {
- return cur;
- }
-
- /**
- * @return Flag indicating that this future was read from last check..
- */
- public Boolean accessed() {
- return accessed;
- }
-
- /**
- * @param accessed New accessed.
- */
- public void accessed(Boolean accessed) {
- this.accessed = accessed;
- }
- }
-
- /**
- * Job for execute SCAN or SQL query and get first page of results.
- */
- private static class VisorQueryJob extends
- VisorJob<VisorQueryArg, IgniteBiTuple<? extends Exception, VisorQueryResultEx>> {
- /** */
- private static final long serialVersionUID = 0L;
-
- /**
- * Create job with specified argument.
- *
- * @param arg Job argument.
- * @param debug Debug flag.
- */
- protected VisorQueryJob(VisorQueryArg arg, boolean debug) {
- super(arg, debug);
- }
-
- /** {@inheritDoc} */
- @Override protected IgniteBiTuple<? extends Exception, VisorQueryResultEx> run(VisorQueryArg arg) {
- try {
- Boolean scan = arg.queryTxt().toUpperCase().startsWith("SCAN");
-
- String qryId = (scan ? SCAN_QRY_NAME : SQL_QRY_NAME) + "-" +
- UUID.randomUUID();
-
- GridCacheProcessor cacheProcessor = ignite.context().cache();
-
- IgniteCache<Object, Object> c = cacheProcessor.jcache(arg.cacheName());
-
- if (scan) {
- ScanQuery<Object, Object> qry = new ScanQuery<>(null);
- qry.setPageSize(arg.pageSize());
-
- long start = U.currentTimeMillis();
-
- VisorQueryCursor<Cache.Entry<Object, Object>> cur = new VisorQueryCursor<>(c.query(qry));
-
- List<Object[]> rows = fetchScanQueryRows(cur, arg.pageSize());
-
- long duration = U.currentTimeMillis() - start; // Scan duration + fetch duration.
-
- ignite.cluster().<String, VisorQueryCursorHolder>nodeLocalMap().put(qryId,
- new VisorQueryCursorHolder(cur, false));
-
- scheduleResultSetHolderRemoval(qryId);
-
- return new IgniteBiTuple<>(null, new VisorQueryResultEx(ignite.localNode().id(), qryId,
- SCAN_COL_NAMES, rows, cur.hasNext(), duration));
- }
- else {
- SqlFieldsQuery qry = new SqlFieldsQuery(arg.queryTxt());
- qry.setPageSize(arg.pageSize());
-
- long start = U.currentTimeMillis();
-
- VisorQueryCursor<List<?>> cur = new VisorQueryCursor<>(c.query(qry));
-
- Collection<GridQueryFieldMetadata> meta = cur.fieldsMeta();
-
- if (meta == null)
- return new IgniteBiTuple<Exception, VisorQueryResultEx>(
- new SQLException("Fail to execute query. No metadata available."), null);
- else {
- List<VisorQueryField> names = new ArrayList<>(meta.size());
-
- for (GridQueryFieldMetadata col : meta)
- names.add(new VisorQueryField(col.typeName(), col.fieldName()));
-
- List<Object[]> rows = fetchSqlQueryRows(cur, arg.pageSize());
-
- long duration = U.currentTimeMillis() - start; // Query duration + fetch duration.
-
- ignite.cluster().<String, VisorQueryCursorHolder>nodeLocalMap().put(qryId, new VisorQueryCursorHolder(cur, false));
-
- scheduleResultSetHolderRemoval(qryId);
-
- return new IgniteBiTuple<>(null, new VisorQueryResultEx(ignite.localNode().id(), qryId,
- names, rows, cur.hasNext(), duration));
- }
- }
- }
- catch (Exception e) {
- return new IgniteBiTuple<>(e, null);
- }
- }
-
- /**
- * @param id Unique query result id.
- */
- private void scheduleResultSetHolderRemoval(final String id) {
- ignite.context().timeout().addTimeoutObject(new GridTimeoutObjectAdapter(RMV_DELAY) {
- @Override public void onTimeout() {
- ConcurrentMap<String, VisorQueryCursorHolder> storage = ignite.cluster().nodeLocalMap();
-
- VisorQueryCursorHolder holder = storage.get(id);
-
- if (holder != null) {
- // If future was accessed since last scheduling, set access flag to false and reschedule.
- if (holder.accessed()) {
- holder.accessed(false);
-
- scheduleResultSetHolderRemoval(id);
- }
- else
- storage.remove(id); // Remove stored future otherwise.
- }
- }
- });
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(VisorQueryJob.class, this);
- }
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/97a83a03/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryUtils.java
index f8f795d..be1d15d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryUtils.java
@@ -39,8 +39,8 @@ public class VisorQueryUtils {
/** Columns for SCAN queries. */
public static final Collection<VisorQueryField> SCAN_COL_NAMES = Arrays.asList(
- new VisorQueryField("", "Key Class"), new VisorQueryField("", "Key"),
- new VisorQueryField("", "Value Class"), new VisorQueryField("", "Value")
+ new VisorQueryField(null, null, "Key Class", ""), new VisorQueryField(null, null, "Key", ""),
+ new VisorQueryField(null, null, "Value Class", ""), new VisorQueryField(null, null, "Value", "")
);
/**
@@ -84,7 +84,7 @@ public class VisorQueryUtils {
StringBuilder sb = new StringBuilder();
- Boolean first = true;
+ boolean first = true;
for (Object v : arr) {
if (first)
@@ -141,7 +141,7 @@ public class VisorQueryUtils {
* @param obj Object instance to check.
* @return {@code true} if it is one of known types.
*/
- private static Boolean isKnownType(Object obj) {
+ private static boolean isKnownType(Object obj) {
return obj instanceof String ||
obj instanceof Boolean ||
obj instanceof Byte ||
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/97a83a03/modules/core/src/main/resources/META-INF/classnames.properties
----------------------------------------------------------------------
diff --git a/modules/core/src/main/resources/META-INF/classnames.properties b/modules/core/src/main/resources/META-INF/classnames.properties
index a855e12..305cdd8 100644
--- a/modules/core/src/main/resources/META-INF/classnames.properties
+++ b/modules/core/src/main/resources/META-INF/classnames.properties
@@ -1485,9 +1485,9 @@ org.apache.ignite.internal.visor.query.VisorQueryNextPageTask$VisorQueryNextPage
org.apache.ignite.internal.visor.query.VisorQueryResult
org.apache.ignite.internal.visor.query.VisorQueryResultEx
org.apache.ignite.internal.visor.query.VisorQueryTask
-org.apache.ignite.internal.visor.query.VisorQueryTask$VisorFutureResultSetHolder
-org.apache.ignite.internal.visor.query.VisorQueryTask$VisorQueryArg
-org.apache.ignite.internal.visor.query.VisorQueryTask$VisorQueryJob
+org.apache.ignite.internal.visor.query.VisorQueryCursorHolder
+org.apache.ignite.internal.visor.query.VisorQueryArg
+org.apache.ignite.internal.visor.query.VisorQueryJob
org.apache.ignite.internal.visor.util.VisorEventMapper
org.apache.ignite.internal.visor.util.VisorTaskUtils$4
org.apache.ignite.lang.IgniteBiClosure
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/97a83a03/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 7574be1..5eecf2f 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
@@ -17,16 +17,13 @@
package org.apache.ignite.visor.commands.cache
-import org.apache.ignite.internal.visor.query.VisorQueryTask.VisorQueryArg
-import org.apache.ignite.internal.visor.query.{VisorQueryNextPageTask, VisorQueryResult, VisorQueryTask}
-
import org.apache.ignite.cluster.ClusterNode
-import org.apache.ignite.internal.visor.util.VisorTaskUtils._
import org.apache.ignite.lang.IgniteBiTuple
-
import org.apache.ignite.visor.commands._
-import org.apache.ignite.visor.visor
-import visor._
+import org.apache.ignite.visor.visor._
+
+import org.apache.ignite.internal.visor.query._
+import org.apache.ignite.internal.visor.util.VisorTaskUtils._
import scala.collection.JavaConversions._