You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by se...@apache.org on 2015/03/03 22:39:08 UTC
[48/50] incubator-ignite git commit: ignite-sql-tests - leak fixes
ignite-sql-tests - leak fixes
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/9424b9cd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/9424b9cd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/9424b9cd
Branch: refs/heads/ignite-sql-tests
Commit: 9424b9cd3e9373034d3fde4218716fa7fe9c386e
Parents: 4e3b4cb
Author: S.Vladykin <sv...@gridgain.com>
Authored: Wed Mar 4 00:28:33 2015 +0300
Committer: S.Vladykin <sv...@gridgain.com>
Committed: Wed Mar 4 00:28:33 2015 +0300
----------------------------------------------------------------------
.../query/h2/twostep/GridMapQueryExecutor.java | 267 ++++++++++++++-----
.../query/h2/twostep/GridMergeIndex.java | 9 +-
.../h2/twostep/GridReduceQueryExecutor.java | 8 +-
.../messages/GridQueryCancelRequest.java | 49 ++++
.../cache/GridCacheCrossCacheQuerySelfTest.java | 38 ++-
5 files changed, 298 insertions(+), 73 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9424b9cd/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
index 8b09b16..ed0da58 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
@@ -32,10 +32,12 @@ import org.h2.result.*;
import org.h2.value.*;
import org.jdk8.backport.*;
+import javax.cache.*;
import java.lang.reflect.*;
import java.sql.*;
import java.util.*;
import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
import static org.apache.ignite.events.EventType.*;
@@ -95,9 +97,11 @@ public class GridMapQueryExecutor {
boolean processed = true;
if (msg instanceof GridQueryRequest)
- executeLocalQuery(node, (GridQueryRequest)msg);
+ onQueryRequest(node, (GridQueryRequest)msg);
else if (msg instanceof GridQueryNextPageRequest)
- sendNextPage(node, (GridQueryNextPageRequest)msg);
+ onNextPageRequest(node, (GridQueryNextPageRequest)msg);
+ else if (msg instanceof GridQueryCancelRequest)
+ onCancel(node, (GridQueryCancelRequest)msg);
else
processed = false;
@@ -115,33 +119,56 @@ public class GridMapQueryExecutor {
/**
* @param node Node.
- * @param req Query request.
+ * @param msg Message.
*/
- private void executeLocalQuery(ClusterNode node, GridQueryRequest req) {
- h2.setFilters(h2.backupFilter());
+ private void onCancel(ClusterNode node, GridQueryCancelRequest msg) {
+ ConcurrentMap<Long,QueryResults> nodeRess = resultsForNode(node.id());
- try {
- QueryResults qr = new QueryResults(req.requestId(), req.queries().size());
+ QueryResults results = nodeRess.remove(msg.queryRequestId());
- ConcurrentMap<Long, QueryResults> nodeRess = qryRess.get(node.id());
+ if (results == null)
+ return;
- if (nodeRess == null) {
- nodeRess = new ConcurrentHashMap8<>();
+ results.cancel();
+ }
- ConcurrentMap<Long, QueryResults> old = qryRess.putIfAbsent(node.id(), nodeRess);
+ /**
+ * @param nodeId Node ID.
+ * @return Results for node.
+ */
+ private ConcurrentMap<Long, QueryResults> resultsForNode(UUID nodeId) {
+ ConcurrentMap<Long, QueryResults> nodeRess = qryRess.get(nodeId);
- if (old != null)
- nodeRess = old;
- }
+ if (nodeRess == null) {
+ nodeRess = new ConcurrentHashMap8<>();
- QueryResults old = nodeRess.putIfAbsent(req.requestId(), qr);
+ ConcurrentMap<Long, QueryResults> old = qryRess.putIfAbsent(nodeId, nodeRess);
- assert old == null;
+ if (old != null)
+ nodeRess = old;
+ }
- // Prepare snapshots for all the needed tables before actual run.
- for (GridCacheSqlQuery qry : req.queries()) {
- // TODO
- }
+ return nodeRess;
+ }
+
+ /**
+ * Executing queries locally.
+ *
+ * @param node Node.
+ * @param req Query request.
+ */
+ private void onQueryRequest(ClusterNode node, GridQueryRequest req) {
+ ConcurrentMap<Long,QueryResults> nodeRess = resultsForNode(node.id());
+
+ QueryResults qr = new QueryResults(req.requestId(), req.queries().size());
+
+ if (nodeRess.putIfAbsent(req.requestId(), qr) != null)
+ throw new IllegalStateException();
+
+ h2.setFilters(h2.backupFilter());
+
+ try {
+ // TODO Prepare snapshots for all the needed tables before the run.
// Run queries.
int i = 0;
@@ -152,13 +179,6 @@ public class GridMapQueryExecutor {
ResultSet rs = h2.executeSqlQueryWithTimer(space, h2.connectionForSpace(space), qry.query(),
F.asList(qry.parameters()));
- assert rs instanceof JdbcResultSet : rs.getClass();
-
- ResultInterface res = (ResultInterface)RESULT_FIELD.get(rs);
-
- qr.results[i] = res;
- qr.resultSets[i] = rs;
-
if (ctx.event().isRecordable(EVT_CACHE_QUERY_EXECUTED)) {
ctx.event().record(new CacheQueryExecutedEvent<>(
node,
@@ -175,13 +195,27 @@ public class GridMapQueryExecutor {
null));
}
+ assert rs instanceof JdbcResultSet : rs.getClass();
+
+ qr.addResult(i, rs);
+
+ if (qr.canceled) {
+ qr.result(i).close();
+
+ throw new IgniteException("Query was canceled.");
+ }
+
// Send the first page.
- sendNextPage(node, qr, i, req.pageSize(), res.getRowCount());
+ sendNextPage(nodeRess, node, qr, i, req.pageSize());
i++;
}
}
catch (Throwable e) {
+ nodeRess.remove(req.requestId(), qr);
+
+ qr.cancel();
+
U.error(log, "Failed to execute local query: " + req, e);
sendError(node, req.requestId(), e);
@@ -212,16 +246,15 @@ public class GridMapQueryExecutor {
* @param node Node.
* @param req Request.
*/
- private void sendNextPage(ClusterNode node, GridQueryNextPageRequest req) {
+ private void onNextPageRequest(ClusterNode node, GridQueryNextPageRequest req) {
ConcurrentMap<Long, QueryResults> nodeRess = qryRess.get(node.id());
QueryResults qr = nodeRess == null ? null : nodeRess.get(req.queryRequestId());
- if (qr == null)
- sendError(node, req.queryRequestId(),
- new IllegalStateException("No query result found for request: " + req));
+ if (qr == null || qr.canceled)
+ sendError(node, req.queryRequestId(), new CacheException("No query result found for request: " + req));
else
- sendNextPage(node, qr, req.query(), req.pageSize(), -1);
+ sendNextPage(nodeRess, node, qr, req.query(), req.pageSize());
}
/**
@@ -229,36 +262,29 @@ public class GridMapQueryExecutor {
* @param qr Query results.
* @param qry Query.
* @param pageSize Page size.
- * @param allRows All rows count.
*/
- private void sendNextPage(ClusterNode node, QueryResults qr, int qry, int pageSize, int allRows) {
- int page;
-
- List<Value[]> rows = new ArrayList<>(Math.min(64, pageSize));
-
- ResultInterface res = qr.results[qry];
+ private void sendNextPage(ConcurrentMap<Long, QueryResults> nodeRess, ClusterNode node, QueryResults qr, int qry,
+ int pageSize) {
+ QueryResult res = qr.result(qry);
assert res != null;
- boolean last = false;
+ int page = res.page;
- synchronized (res) {
- page = qr.pages[qry]++;
+ List<Value[]> rows = new ArrayList<>(Math.min(64, pageSize));
- for (int i = 0 ; i < pageSize; i++) {
- if (!res.next()) {
- last = true;
+ boolean last = res.fetchNextPage(rows, pageSize);
- break;
- }
+ if (last) {
+ res.close();
- rows.add(res.currentRow());
- }
+ if (qr.isAllClosed())
+ nodeRess.remove(qr.qryReqId, qr);
}
try {
ctx.io().sendUserMessage(F.asList(node),
- new GridQueryNextPageResponse(qr.qryReqId, qry, page, allRows, last, rows),
+ new GridQueryNextPageResponse(qr.qryReqId, qry, page, page == 0 ? res.rowCount : -1, last, rows),
GridTopic.TOPIC_QUERY, false, 0);
}
catch (IgniteCheckedException e) {
@@ -271,29 +297,146 @@ public class GridMapQueryExecutor {
/**
*
*/
- private static class QueryResults {
- /** */
- private long qryReqId;
-
+ private class QueryResults {
/** */
- private ResultInterface[] results;
+ private final long qryReqId;
/** */
- private ResultSet[] resultSets;
+ private final AtomicReferenceArray<QueryResult> results;
/** */
- private int[] pages;
+ private volatile boolean canceled;
/**
* @param qryReqId Query request ID.
- * @param qrys Queries.
+ * @param qrys Number of queries.
*/
private QueryResults(long qryReqId, int qrys) {
this.qryReqId = qryReqId;
- results = new ResultInterface[qrys];
- resultSets = new ResultSet[qrys];
- pages = new int[qrys];
+ results = new AtomicReferenceArray<>(qrys);
+ }
+
+ /**
+ * @param qry Query result index.
+ * @return Query result.
+ */
+ QueryResult result(int qry) {
+ return results.get(qry);
+ }
+
+ /**
+ * @param qry Query result index.
+ * @param rs Result set.
+ */
+ void addResult(int qry, ResultSet rs) {
+ if (!results.compareAndSet(qry, null, new QueryResult(rs)))
+ throw new IllegalStateException();
+ }
+
+ /**
+ * @return {@code true} If all results are closed.
+ */
+ boolean isAllClosed() {
+ for (int i = 0; i < results.length(); i++) {
+ QueryResult res = results.get(i);
+
+ if (res == null || !res.closed)
+ return false;
+ }
+
+ return true;
+ }
+
+ void cancel() {
+ if (canceled)
+ return;
+
+ canceled = true;
+
+ for (int i = 0; i < results.length(); i++) {
+ QueryResult res = results.get(i);
+
+ if (res != null)
+ res.close();
+ }
+ }
+ }
+
+ /**
+ * Result for a single part of the query.
+ */
+ private class QueryResult implements AutoCloseable {
+ /** */
+ private final ResultInterface res;
+
+ /** */
+ private final ResultSet rs;
+
+ /** */
+ private int page;
+
+ /** */
+ private final int rowCount;
+
+ /** */
+ private volatile boolean closed;
+
+ /**
+ * @param rs Result set.
+ */
+ private QueryResult(ResultSet rs) {
+ this.rs = rs;
+
+ try {
+ res = (ResultInterface)RESULT_FIELD.get(rs);
+ }
+ catch (IllegalAccessException e) {
+ throw new IllegalStateException(e); // Must not happen.
+ }
+
+ rowCount = res.getRowCount();
+ }
+
+ /**
+ * @param rows Collection to fetch into.
+ * @param pageSize Page size.
+ * @return {@code true} If there are no more rows available.
+ */
+ synchronized boolean fetchNextPage(List<Value[]> rows, int pageSize) {
+ if (closed)
+ return true;
+
+ page++;
+
+ for (int i = 0 ; i < pageSize; i++) {
+ if (!res.next())
+ return true;
+
+ rows.add(res.currentRow());
+ }
+
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public synchronized void close() {
+ if (closed)
+ return;
+
+ closed = true;
+
+ Statement stmt;
+
+ try {
+ stmt = rs.getStatement();
+ }
+ catch (SQLException e) {
+ throw new IllegalStateException(e); // Must not happen.
+ }
+
+ U.close(rs, log);
+ U.close(stmt, log);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9424b9cd/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndex.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndex.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndex.java
index 71cc151..4632e0b 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndex.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndex.java
@@ -112,13 +112,20 @@ public abstract class GridMergeIndex extends BaseIndex {
if (fetched == null)
throw new IgniteException("Fetched result set was too large.");
- if (fetched.size() == cnt.get()) // We've fetched all the rows.
+ if (fetchedAll())
return findAllFetched(fetched, first, last);
return findInStream(first, last);
}
/**
+ * @return {@code true} If we have fetched all the remote rows.
+ */
+ public boolean fetchedAll() {
+ return fetched.size() == cnt.get();
+ }
+
+ /**
* @param first First row.
* @param last Last row.
* @return Cursor. Usually it must be {@link FetchingCursor} instance.
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9424b9cd/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
index 9aec279..2804d90 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
@@ -164,7 +164,7 @@ public class GridReduceQueryExecutor {
r.conn = h2.connectionForSpace(space);
// TODO Add topology version.
- Collection<ClusterNode> nodes = ctx.grid().cluster().forCacheNodes(space).nodes();
+ final Collection<ClusterNode> nodes = ctx.grid().cluster().forCacheNodes(space).nodes();
for (GridCacheSqlQuery mapQry : qry.mapQueries()) {
GridMergeTable tbl;
@@ -198,8 +198,12 @@ public class GridReduceQueryExecutor {
final ResultSet res = h2.executeSqlQueryWithTimer(space, r.conn, rdc.query(), F.asList(rdc.parameters()));
- for (GridMergeTable tbl : r.tbls)
+ for (GridMergeTable tbl : r.tbls) {
+ if (!tbl.getScanIndex(null).fetchedAll()) // We have to explicitly cancel queries on remote nodes.
+ ctx.io().sendUserMessage(nodes, new GridQueryCancelRequest(qryReqId), GridTopic.TOPIC_QUERY, false, 0);
+
dropTable(r.conn, tbl.getName());
+ }
return new QueryCursorImpl<>(new Iter(res));
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9424b9cd/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java
new file mode 100644
index 0000000..9bcb410
--- /dev/null
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java
@@ -0,0 +1,49 @@
+/*
+ * 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.processors.query.h2.twostep.messages;
+
+import org.apache.ignite.internal.util.typedef.internal.*;
+
+import java.io.*;
+
+/**
+ * Cancel request.
+ */
+public class GridQueryCancelRequest implements Serializable {
+ /** */
+ private long qryReqId;
+
+ /**
+ * @param qryReqId Query request ID.
+ */
+ public GridQueryCancelRequest(long qryReqId) {
+ this.qryReqId = qryReqId;
+ }
+
+ /**
+ * @return Query request ID.
+ */
+ public long queryRequestId() {
+ return qryReqId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(GridQueryCancelRequest.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9424b9cd/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheCrossCacheQuerySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheCrossCacheQuerySelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheCrossCacheQuerySelfTest.java
index 8445912..2c38f4e 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheCrossCacheQuerySelfTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheCrossCacheQuerySelfTest.java
@@ -70,6 +70,8 @@ public class GridCacheCrossCacheQuerySelfTest extends GridCommonAbstractTest {
/** {@inheritDoc} */
@Override protected void beforeTest() throws Exception {
ignite = startGridsMultiThreaded(3);
+
+ fillCaches();
}
/** {@inheritDoc} */
@@ -117,8 +119,6 @@ public class GridCacheCrossCacheQuerySelfTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testTwoStep() throws Exception {
- fillCaches();
-
String cache = "partitioned";
GridCacheQueriesEx<Integer, FactPurchase> qx =
@@ -141,8 +141,6 @@ public class GridCacheCrossCacheQuerySelfTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testTwoStepGroupAndAggregates() throws Exception {
- fillCaches();
-
GridCacheQueriesEx<Integer, FactPurchase> qx =
(GridCacheQueriesEx<Integer, FactPurchase>)((IgniteKernal)ignite)
.<Integer, FactPurchase>cache("partitioned").queries();
@@ -235,8 +233,6 @@ public class GridCacheCrossCacheQuerySelfTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testApiQueries() throws Exception {
- fillCaches();
-
IgniteCache<Object,Object> c = ignite.jcache("partitioned");
c.queryFields(new SqlFieldsQuery("select cast(? as varchar) from FactPurchase").setArgs("aaa")).getAll();
@@ -249,6 +245,34 @@ public class GridCacheCrossCacheQuerySelfTest extends GridCommonAbstractTest {
assertEquals(8, res.get(0).get(1));
}
+ public void _testLoop() throws InterruptedException {
+ IgniteCache<Object,Object> c = ignite.jcache("partitioned");
+
+ X.println("___ GET READY");
+
+ Thread.sleep(20000);
+
+ X.println("___ GO");
+
+ long start = System.currentTimeMillis();
+
+ for (int i = 0; i < 1000000; i++) {
+ if (i % 1000 == 0) {
+ long t = System.currentTimeMillis();
+
+ X.println("__ " + i + " -> " + (t - start));
+
+ start = t;
+ }
+
+ c.queryFields(new SqlFieldsQuery("select * from FactPurchase")).getAll();
+ }
+
+ X.println("___ OK");
+
+ Thread.sleep(300000);
+ }
+
/**
* @param l List.
* @param idx Index.
@@ -260,8 +284,6 @@ public class GridCacheCrossCacheQuerySelfTest extends GridCommonAbstractTest {
/** @throws Exception If failed. */
public void testOnProjection() throws Exception {
- fillCaches();
-
CacheProjection<Integer, FactPurchase> prj = ((IgniteKernal)ignite)
.<Integer, FactPurchase>cache("partitioned").projection(
new IgnitePredicate<Cache.Entry<Integer, FactPurchase>>() {