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 2016/07/07 14:01:55 UTC
[8/8] ignite git commit: IGNITE-3443 WIP
IGNITE-3443 WIP
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/2dcbe38b
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/2dcbe38b
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/2dcbe38b
Branch: refs/heads/ignite-3443
Commit: 2dcbe38b7625a3488043a460849866d4a3480e38
Parents: 920005d
Author: Alexey Kuznetsov <ak...@apache.org>
Authored: Thu Jul 7 21:01:30 2016 +0700
Committer: Alexey Kuznetsov <ak...@apache.org>
Committed: Thu Jul 7 21:01:30 2016 +0700
----------------------------------------------------------------------
.../processors/cache/IgniteCacheProxy.java | 6 ++--
.../cache/query/GridCacheQueryAdapter.java | 4 +--
.../query/GridCacheQueryFutureAdapter.java | 2 +-
.../cache/query/GridCacheQueryManager.java | 14 ++++----
.../processors/query/GridQueryProcessor.java | 34 ++++++++++----------
5 files changed, 31 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/2dcbe38b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
index 12ec8b8..84e5a49 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
@@ -467,7 +467,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
if (grp != null)
qry.projection(grp);
- final GridCloseableIterator<Entry<K, V>> iter = ctx.kernalContext().query().executeQuery(ctx,
+ final GridCloseableIterator<Entry<K, V>> iter = ctx.kernalContext().query().executeQuery("ScanQuery", ctx,
new IgniteOutClosureX<GridCloseableIterator<Entry<K,V>>>() {
@Override public GridCloseableIterator<Entry<K,V>> applyx() throws IgniteCheckedException {
final GridCloseableIterator<Map.Entry> iter0 = qry.executeScanQuery();
@@ -503,7 +503,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
if (grp != null)
qry.projection(grp);
- fut = ctx.kernalContext().query().executeQuery(ctx,
+ fut = ctx.kernalContext().query().executeQuery("FullTextQuery", ctx,
new IgniteOutClosureX<CacheQueryFuture<Map.Entry<K, V>>>() {
@Override public CacheQueryFuture<Map.Entry<K, V>> applyx() throws IgniteCheckedException {
return qry.execute();
@@ -516,7 +516,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
if (grp != null)
qry.projection(grp);
- fut = ctx.kernalContext().query().executeQuery(ctx,
+ fut = ctx.kernalContext().query().executeQuery("SpiQuery", ctx,
new IgniteOutClosureX<CacheQueryFuture<Map.Entry<K, V>>>() {
@Override public CacheQueryFuture<Map.Entry<K, V>> applyx() throws IgniteCheckedException {
return qry.execute(((SpiQuery)filter).getArgs());
http://git-wip-us.apache.org/repos/asf/ignite/blob/2dcbe38b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryAdapter.java
index 90e14f4..6a93056 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryAdapter.java
@@ -396,8 +396,8 @@ public class GridCacheQueryAdapter<T> implements CacheQuery<T> {
* @param startTime Start time.
* @param duration Duration.
*/
- public void onCompleted(Object res, Throwable err, long startTime, long duration) {
- GridQueryProcessor.onCompleted(cctx, res, err, startTime, duration, log);
+ public void onCompleted(String qry, Object res, Throwable err, long startTime, long duration) {
+ GridQueryProcessor.onCompleted(qry, cctx, res, err, startTime, duration, log);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/2dcbe38b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryFutureAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryFutureAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryFutureAdapter.java
index db519f5..779e7da 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryFutureAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryFutureAdapter.java
@@ -155,7 +155,7 @@ public abstract class GridCacheQueryFutureAdapter<K, V, R> extends GridFutureAda
@Override public boolean onDone(Collection<R> res, Throwable err) {
cctx.time().removeTimeoutObject(this);
- qry.query().onCompleted(res, err, startTime(), duration());
+ qry.query().onCompleted("onDone [clause=" + qry.query().clause() + ", type=" + qry.query().type() + "]", res, err, startTime(), duration());
return super.onDone(res, err);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2dcbe38b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
index d376df5..dec3273 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
@@ -1160,9 +1160,8 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
key = (K)cctx.unwrapBinaryIfNeeded(key, keepBinary);
- if (filter != null || locNode) {
+ if (filter != null || locNode)
val = (V)cctx.unwrapBinaryIfNeeded(val, keepBinary);
- }
if (filter != null && !filter.apply(key, val))
continue;
@@ -1723,6 +1722,7 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
throw new IllegalStateException("Failed to process query request (grid is stopping).");
final boolean statsEnabled = cctx.config().isStatisticsEnabled();
+ final String namex = cctx.namex();
boolean needUpdStatistics = updStatisticsIfNeeded && statsEnabled;
@@ -1736,7 +1736,6 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
final String taskName = cctx.kernalContext().task().resolveTaskName(qry.taskHash());
final IgniteBiPredicate filter = qry.scanFilter();
- final String namex = cctx.namex();
final ClusterNode locNode = cctx.localNode();
final UUID subjId = qry.subjectId();
@@ -1761,7 +1760,7 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
if (updStatisticsIfNeeded) {
needUpdStatistics = false;
- cctx.queries().onCompleted(U.currentTimeMillis() - startTime, false);
+ cctx.queries().onCompleted("SCAN cache " + namex, U.currentTimeMillis() - startTime, false);
}
final boolean readEvt = cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ);
@@ -1814,7 +1813,7 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
}
catch (Exception e) {
if (needUpdStatistics)
- cctx.queries().onCompleted(U.currentTimeMillis() - startTime, true);
+ cctx.queries().onCompleted("SCAN cache " + namex, U.currentTimeMillis() - startTime, true);
throw e;
}
@@ -2106,10 +2105,13 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
}
/**
+ * @qry Query that was completed.
* @param duration Execution duration.
* @param fail {@code true} if execution failed.
*/
- public void onCompleted(long duration, boolean fail) {
+ public void onCompleted(String qry, long duration, boolean fail) {
+ log.warning(qry);
+
metrics.onQueryCompleted(duration, fail);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2dcbe38b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
index a42eb98..87d9b71 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
@@ -743,7 +743,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
try {
final GridCacheContext<?, ?> cctx = ctx.cache().internalCache(space).context();
- return executeQuery(cctx, new IgniteOutClosureX<GridCloseableIterator<IgniteBiTuple<K, V>>>() {
+ return executeQuery("Query [space=" + space + ", clause=" + clause + "]", cctx, new IgniteOutClosureX<GridCloseableIterator<IgniteBiTuple<K, V>>>() {
@Override public GridCloseableIterator<IgniteBiTuple<K, V>> applyx() throws IgniteCheckedException {
TypeDescriptor type = typesByName.get(new TypeName(space, resType));
@@ -773,7 +773,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
try {
final GridCacheContext<Object, Object> cctx = ctx.cache().internalCache(space).context();
- return executeQuery(cctx, new IgniteOutClosureX<Iterable<List<?>>>() {
+ return executeQuery("TwoStepQuery", cctx, new IgniteOutClosureX<Iterable<List<?>>>() {
@Override public Iterable<List<?>> applyx() throws IgniteCheckedException {
return idx.queryTwoStep(
cctx,
@@ -802,7 +802,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
throw new IllegalStateException("Failed to execute query (grid is stopping).");
try {
- return executeQuery(cctx, new IgniteOutClosureX<QueryCursor<List<?>>>() {
+ return executeQuery("SqlFieldsQuery [sql=" + qry.getSql() + "]", cctx, new IgniteOutClosureX<QueryCursor<List<?>>>() {
@Override public QueryCursor<List<?>> applyx() throws IgniteCheckedException {
return idx.queryTwoStep(cctx, qry);
}
@@ -828,7 +828,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
throw new IllegalStateException("Failed to execute query (grid is stopping).");
try {
- return executeQuery(cctx, new IgniteOutClosureX<QueryCursor<Cache.Entry<K, V>>>() {
+ return executeQuery("SqlQuery [type=" + qry.getType() + ", sql=" + qry.getSql() + "]", cctx, new IgniteOutClosureX<QueryCursor<Cache.Entry<K, V>>>() {
@Override public QueryCursor<Cache.Entry<K, V>> applyx() throws IgniteCheckedException {
return idx.queryTwoStep(cctx, qry);
}
@@ -856,7 +856,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
throw new IllegalStateException("Failed to execute query (grid is stopping).");
try {
- return executeQuery(
+ return executeQuery("SqlQuery [type=" + qry.getType() + ", sql=" + qry.getSql() + "]",
cctx,
new IgniteOutClosureX<Iterator<Cache.Entry<K, V>>>() {
@Override public Iterator<Cache.Entry<K, V>> applyx() throws IgniteCheckedException {
@@ -957,7 +957,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
try {
final boolean keepBinary = cctx.keepBinary();
- return executeQuery(cctx, new IgniteOutClosureX<QueryCursor<List<?>>>() {
+ return executeQuery("SqlFieldsQuery [" + qry.getSql() + "]", cctx, new IgniteOutClosureX<QueryCursor<List<?>>>() {
@Override public QueryCursor<List<?>> applyx() throws IgniteCheckedException {
String space = cctx.name();
String sql = qry.getSql();
@@ -1073,14 +1073,13 @@ public class GridQueryProcessor extends GridProcessorAdapter {
* @return Type name.
*/
public static String typeName(String clsName) {
- int packageEnd = clsName.lastIndexOf('.');
+ int pkgEnd = clsName.lastIndexOf('.');
- if (packageEnd >= 0 && packageEnd < clsName.length() - 1)
- clsName = clsName.substring(packageEnd + 1);
+ if (pkgEnd >= 0 && pkgEnd < clsName.length() - 1)
+ clsName = clsName.substring(pkgEnd + 1);
- if (clsName.endsWith("[]")) {
+ if (clsName.endsWith("[]"))
clsName = clsName.substring(0, clsName.length() - 2) + "_array";
- }
int parentEnd = clsName.lastIndexOf('$');
@@ -1111,7 +1110,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
try {
final GridCacheContext<?, ?> cctx = ctx.cache().internalCache(space).context();
- return executeQuery(cctx, new IgniteOutClosureX<GridCloseableIterator<IgniteBiTuple<K, V>>>() {
+ return executeQuery("TextQuery [space=" + space + ", clause=" + clause + "]", cctx, new IgniteOutClosureX<GridCloseableIterator<IgniteBiTuple<K, V>>>() {
@Override public GridCloseableIterator<IgniteBiTuple<K, V>> applyx() throws IgniteCheckedException {
TypeDescriptor type = typesByName.get(new TypeName(space, resType));
@@ -1149,7 +1148,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
try {
final GridCacheContext<?, ?> cctx = ctx.cache().internalCache(space).context();
- return executeQuery(cctx, new IgniteOutClosureX<GridQueryFieldsResult>() {
+ return executeQuery("QueryFields [space=" + space + ", clause=" + clause + "]", cctx, new IgniteOutClosureX<GridQueryFieldsResult>() {
@Override public GridQueryFieldsResult applyx() throws IgniteCheckedException {
return idx.queryFields(space, clause, params, filters);
}
@@ -1754,11 +1753,12 @@ public class GridQueryProcessor extends GridProcessorAdapter {
}
/**
+ * @param qry
* @param cctx Cache context.
* @param clo Closure.
* @param complete Complete.
*/
- public <R> R executeQuery(GridCacheContext<?, ?> cctx, IgniteOutClosureX<R> clo, boolean complete)
+ public <R> R executeQuery(String qry, GridCacheContext<?, ?> cctx, IgniteOutClosureX<R> clo, boolean complete)
throws IgniteCheckedException {
final long startTime = U.currentTimeMillis();
@@ -1791,7 +1791,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
cctx.queries().onExecuted(err != null);
if (complete && err == null)
- onCompleted(cctx, res, null, startTime, U.currentTimeMillis() - startTime, log);
+ onCompleted(qry, cctx, res, null, startTime, U.currentTimeMillis() - startTime, log);
}
}
@@ -1803,11 +1803,11 @@ public class GridQueryProcessor extends GridProcessorAdapter {
* @param duration Duration.
* @param log Logger.
*/
- public static void onCompleted(GridCacheContext<?, ?> cctx, Object res, Throwable err,
+ public static void onCompleted(String qry, GridCacheContext<?, ?> cctx, Object res, Throwable err,
long startTime, long duration, IgniteLogger log) {
boolean fail = err != null;
- cctx.queries().onCompleted(duration, fail);
+ cctx.queries().onCompleted(qry, duration, fail);
if (log.isTraceEnabled())
log.trace("Query execution completed [startTime=" + startTime +