GG-11360 - Implement SQL queries cancellation Fix for commit 80abd1b: for distributed joins need always send cancel request.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/b1c7c9bb Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/b1c7c9bb Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/b1c7c9bb Branch: refs/heads/master Commit: b1c7c9bb95c900083702d0ba0362edf3aea5a7b4 Parents: a62a013 Author: sboikov <[email protected]> Authored: Mon Nov 7 15:40:36 2016 +0300 Committer: sboikov <[email protected]> Committed: Mon Nov 7 15:40:36 2016 +0300 ---------------------------------------------------------------------- .../h2/twostep/GridReduceQueryExecutor.java | 25 ++++++++++++++------ 1 file changed, 18 insertions(+), 7 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ignite/blob/b1c7c9bb/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 3847373..3b851a6 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 @@ -756,7 +756,7 @@ public class GridReduceQueryExecutor { } finally { // Make sure any activity related to current attempt is cancelled. - cancelRemoteQueriesIfNeeded(nodes, r, qryReqId); + cancelRemoteQueriesIfNeeded(nodes, r, qryReqId, qry.distributedJoins()); if (!runs.remove(qryReqId, r)) U.warn(log, "Query run was already removed: " + qryReqId); @@ -793,15 +793,26 @@ public class GridReduceQueryExecutor { } /** + * @param nodes Query nodes. * @param r Query run. * @param qryReqId Query id. + * @param distributedJoins Distributed join flag. */ - private void cancelRemoteQueriesIfNeeded(Collection<ClusterNode> nodes, QueryRun r, long qryReqId) { - for (GridMergeIndex idx : r.idxs) { - if (!idx.fetchedAll()) { - send(nodes, new GridQueryCancelRequest(qryReqId), null, false); - - break; + private void cancelRemoteQueriesIfNeeded(Collection<ClusterNode> nodes, + QueryRun r, + long qryReqId, + boolean distributedJoins) + { + // For distributedJoins need always send cancel request to cleanup resources. + if (distributedJoins) + send(nodes, new GridQueryCancelRequest(qryReqId), null, false); + else { + for (GridMergeIndex idx : r.idxs) { + if (!idx.fetchedAll()) { + send(nodes, new GridQueryCancelRequest(qryReqId), null, false); + + break; + } } } }
