You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by am...@apache.org on 2019/04/18 13:18:55 UTC
[ignite] branch master updated: IGNITE-11564: SQL: Implement KILL
QUERY command. This closes #6391.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new b4b4be3 IGNITE-11564: SQL: Implement KILL QUERY command. This closes #6391.
b4b4be3 is described below
commit b4b4be3e21b8eddfdb85d13dd1b6542c056a1c2b
Author: Yuriy Gerzhedovich <yg...@gridgain.com>
AuthorDate: Thu Apr 18 16:18:28 2019 +0300
IGNITE-11564: SQL: Implement KILL QUERY command. This closes #6391.
---
.../communication/GridIoMessageFactory.java | 12 +
.../internal/processors/query/GridQueryCancel.java | 22 +-
.../processors/query/GridRunningQueryInfo.java | 13 +-
.../processors/query/QueryCancellable.java | 29 +
.../processors/query/QueryRunningFuture.java | 27 +
.../processors/query/RunningQueryManager.java | 14 +-
.../twostep/messages/GridQueryCancelRequest.java | 2 +-
.../GridQueryKillRequest.java} | 98 +-
.../GridQueryKillResponse.java} | 73 +-
.../org/apache/ignite/internal/sql/SqlKeyword.java | 6 +
.../org/apache/ignite/internal/sql/SqlParser.java | 26 +-
.../apache/ignite/internal/sql/SqlParserUtils.java | 10 +-
.../internal/sql/command/SqlKillQueryCommand.java | 121 +++
.../internal/sql/SqlParserDropIndexSelfTest.java | 4 +-
.../internal/sql/SqlParserKillQuerySelfTest.java | 144 +++
.../cache/query/RegisteredQueryCursor.java | 16 +-
.../processors/query/h2/CommandProcessor.java | 304 +++++-
.../processors/query/h2/IgniteH2Indexing.java | 41 +-
.../internal/processors/query/h2/KillQueryRun.java | 71 ++
.../internal/processors/query/h2/QueryParser.java | 27 +-
.../query/h2/twostep/GridReduceQueryExecutor.java | 18 +-
.../processors/query/KillQueryFromClientTest.java | 39 +
.../query/KillQueryFromNeighbourTest.java | 40 +
.../query/KillQueryOnClientDisconnectTest.java | 302 ++++++
.../internal/processors/query/KillQueryTest.java | 1010 ++++++++++++++++++++
.../IgniteBinaryCacheQueryTestSuite.java | 15 +-
26 files changed, 2363 insertions(+), 121 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
index b377270..6bb06cd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
@@ -182,6 +182,8 @@ import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQuery
import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryFailResponse;
import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryNextPageRequest;
import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryNextPageResponse;
+import org.apache.ignite.internal.processors.query.messages.GridQueryKillRequest;
+import org.apache.ignite.internal.processors.query.messages.GridQueryKillResponse;
import org.apache.ignite.internal.processors.query.schema.message.SchemaOperationStatusMessage;
import org.apache.ignite.internal.processors.rest.handlers.task.GridTaskResultRequest;
import org.apache.ignite.internal.processors.rest.handlers.task.GridTaskResultResponse;
@@ -1144,6 +1146,16 @@ public class GridIoMessageFactory implements MessageFactory {
break;
+ case GridQueryKillRequest.TYPE_CODE:
+ msg = new GridQueryKillRequest();
+
+ break;
+
+ case GridQueryKillResponse.TYPE_CODE:
+ msg = new GridQueryKillResponse();
+
+ break;
+
// [-3..119] [124..129] [-23..-28] [-36..-55] - this
// [120..123] - DR
// [-4..-22, -30..-35] - SQL
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryCancel.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryCancel.java
index d49bcc7..d1c7afb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryCancel.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryCancel.java
@@ -25,29 +25,27 @@ import org.apache.ignite.cache.query.QueryCancelledException;
*/
public class GridQueryCancel {
/** No-op runnable indicating cancelled state. */
- private static final Runnable CANCELLED = new Runnable() {
- @Override public void run() {
- // No-op.
- }
+ private static final QueryCancellable CANCELLED = () -> {
+ // No-op.
};
/** */
- private static final AtomicReferenceFieldUpdater<GridQueryCancel, Runnable> STATE_UPDATER =
- AtomicReferenceFieldUpdater.newUpdater(GridQueryCancel.class, Runnable.class, "clo");
+ private static final AtomicReferenceFieldUpdater<GridQueryCancel, QueryCancellable> STATE_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(GridQueryCancel.class, QueryCancellable.class, "clo");
/** */
- private volatile Runnable clo;
+ private volatile QueryCancellable clo;
/**
* Sets a cancel closure.
*
* @param clo Clo.
*/
- public void set(Runnable clo) throws QueryCancelledException {
+ public void set(QueryCancellable clo) throws QueryCancelledException {
assert clo != null;
while(true) {
- Runnable tmp = this.clo;
+ QueryCancellable tmp = this.clo;
if (tmp == CANCELLED)
throw new QueryCancelledException();
@@ -62,11 +60,11 @@ public class GridQueryCancel {
*/
public void cancel() {
while(true) {
- Runnable tmp = this.clo;
+ QueryCancellable tmp = clo;
if (STATE_UPDATER.compareAndSet(this, tmp, CANCELLED)) {
if (tmp != null)
- tmp.run();
+ tmp.doCancel();
return;
}
@@ -80,4 +78,4 @@ public class GridQueryCancel {
if (clo == CANCELLED)
throw new QueryCancelledException();
}
-}
\ No newline at end of file
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridRunningQueryInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridRunningQueryInfo.java
index e3b7e1b..09b40d1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridRunningQueryInfo.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridRunningQueryInfo.java
@@ -17,9 +17,8 @@
package org.apache.ignite.internal.processors.query;
-import org.apache.ignite.internal.processors.cache.query.GridCacheQueryType;
-
import java.util.UUID;
+import org.apache.ignite.internal.processors.cache.query.GridCacheQueryType;
/**
* Query descriptor.
@@ -49,6 +48,9 @@ public class GridRunningQueryInfo {
/** */
private final boolean loc;
+ /** */
+ private final QueryRunningFuture fut = new QueryRunningFuture();
+
/**
* Constructor.
*
@@ -141,6 +143,13 @@ public class GridRunningQueryInfo {
}
/**
+ * @return Query running future.
+ */
+ public QueryRunningFuture runningFuture(){
+ return fut;
+ }
+
+ /**
* @return {@code true} if query can be cancelled.
*/
public boolean cancelable() {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/QueryCancellable.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/QueryCancellable.java
new file mode 100644
index 0000000..841760d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/QueryCancellable.java
@@ -0,0 +1,29 @@
+/*
+ * 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;
+
+/**
+ * Cancellable query interface.
+ */
+@FunctionalInterface
+public interface QueryCancellable {
+
+ /** Query cancel handler. */
+ void doCancel();
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/QueryRunningFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/QueryRunningFuture.java
new file mode 100644
index 0000000..29c2324
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/QueryRunningFuture.java
@@ -0,0 +1,27 @@
+/*
+ * 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;
+
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+
+/**
+ * Query running future.
+ */
+public class QueryRunningFuture extends GridFutureAdapter<String> {
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/RunningQueryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/RunningQueryManager.java
index 807488c..ebb2701 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/RunningQueryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/RunningQueryManager.java
@@ -113,8 +113,11 @@ public class RunningQueryManager {
GridRunningQueryInfo qry = runs.remove(qryId);
//We need to collect query history only for SQL queries.
- if (qry != null && isSqlQuery(qry))
+ if (qry != null && isSqlQuery(qry)) {
+ qry.runningFuture().onDone();
+
qryHistTracker.collectMetrics(qry, failed);
+ }
}
/**
@@ -205,6 +208,15 @@ public class RunningQueryManager {
}
/**
+ * Gets info about running query by their id.
+ * @param qryId
+ * @return Running query info or {@code null} in case no running query for given id.
+ */
+ public @Nullable GridRunningQueryInfo runningQueryInfo(Long qryId) {
+ return runs.get(qryId);
+ }
+
+ /**
* Reset query history metrics.
*/
public void resetQueryHistoryMetrics() {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java
index 1ced7bc..d7be454 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java
@@ -24,7 +24,7 @@ import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
/**
- * Cancel request.
+ * Cancel map part of query request.
*/
public class GridQueryCancelRequest implements Message {
/** */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/messages/GridQueryKillRequest.java
similarity index 54%
copy from modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java
copy to modules/core/src/main/java/org/apache/ignite/internal/processors/query/messages/GridQueryKillRequest.java
index 1ced7bc..23e79d5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/messages/GridQueryKillRequest.java
@@ -13,9 +13,10 @@
* 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;
+package org.apache.ignite.internal.processors.query.messages;
import java.nio.ByteBuffer;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -24,44 +25,66 @@ import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
/**
- * Cancel request.
+ * Query kill request.
*/
-public class GridQueryCancelRequest implements Message {
+public class GridQueryKillRequest implements Message {
/** */
- private static final long serialVersionUID = 0L;
+ public static final short TYPE_CODE = 172;
/** */
- private long qryReqId;
+ private static final long serialVersionUID = 0L;
+
+ /** Request id. */
+ private long reqId;
+
+ /** Query id on a node. */
+ private long nodeQryId;
+
+ /** Async response flag. */
+ private boolean asyncRes;
/**
* Default constructor.
*/
- public GridQueryCancelRequest() {
+ public GridQueryKillRequest() {
// No-op.
}
/**
- * @param qryReqId Query request ID.
+ * @param reqId Request id.
+ * @param nodeQryId Query ID on a node.
+ * @param asyncRes {@code true} in case reposnse should be send asynchronous.
*/
- public GridQueryCancelRequest(long qryReqId) {
- this.qryReqId = qryReqId;
+ public GridQueryKillRequest(long reqId, long nodeQryId, boolean asyncRes) {
+ this.reqId = reqId;
+ this.nodeQryId = nodeQryId;
+ this.asyncRes = asyncRes;
}
/**
- * @return Query request ID.
+ * @return Request id.
*/
- public long queryRequestId() {
- return qryReqId;
+ public long requestId() {
+ return reqId;
}
- /** {@inheritDoc} */
- @Override public void onAckReceived() {
- // No-op.
+ /**
+ * @return Query id on a node.
+ */
+ public long nodeQryId() {
+ return nodeQryId;
+ }
+
+ /**
+ * @return {@code true} in case response should be send back asynchronous.
+ */
+ public boolean asyncResponse() {
+ return asyncRes;
}
/** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(GridQueryCancelRequest.class, this);
+ @Override public void onAckReceived() {
+ // No-op.
}
/** {@inheritDoc} */
@@ -77,11 +100,22 @@ public class GridQueryCancelRequest implements Message {
switch (writer.state()) {
case 0:
- if (!writer.writeLong("qryReqId", qryReqId))
+ if (!writer.writeLong("reqId", reqId))
+ return false;
+
+ writer.incrementState();
+
+ case 1:
+ if (!writer.writeLong("nodeQryId", nodeQryId))
return false;
writer.incrementState();
+ case 2:
+ if(!writer.writeBoolean("asyncRes", asyncRes))
+ return false;
+
+ writer.incrementState();
}
return true;
@@ -96,25 +130,45 @@ public class GridQueryCancelRequest implements Message {
switch (reader.state()) {
case 0:
- qryReqId = reader.readLong("qryReqId");
+ reqId = reader.readLong("reqId");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 1:
+ nodeQryId = reader.readLong("nodeQryId");
if (!reader.isLastRead())
return false;
reader.incrementState();
+ case 2:
+ asyncRes = reader.readBoolean("asyncRes");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
}
- return reader.afterMessageRead(GridQueryCancelRequest.class);
+ return reader.afterMessageRead(GridQueryKillRequest.class);
}
/** {@inheritDoc} */
@Override public short directType() {
- return 106;
+ return TYPE_CODE;
}
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 1;
+ return 3;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(GridQueryKillRequest.class, this);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/messages/GridQueryKillResponse.java
similarity index 64%
copy from modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java
copy to modules/core/src/main/java/org/apache/ignite/internal/processors/query/messages/GridQueryKillResponse.java
index 1ced7bc..d659e99 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/messages/GridQueryCancelRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/messages/GridQueryKillResponse.java
@@ -13,9 +13,10 @@
* 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;
+package org.apache.ignite.internal.processors.query.messages;
import java.nio.ByteBuffer;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -24,44 +25,56 @@ import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
/**
- * Cancel request.
+ * Query kill response.
*/
-public class GridQueryCancelRequest implements Message {
+public class GridQueryKillResponse implements Message {
/** */
- private static final long serialVersionUID = 0L;
+ public static final short TYPE_CODE = 173;
/** */
- private long qryReqId;
+ private static final long serialVersionUID = 0L;
+
+ /** Request id.*/
+ private long reqId;
+
+ /** Error text. */
+ private String errMsg;
+
/**
* Default constructor.
*/
- public GridQueryCancelRequest() {
+ public GridQueryKillResponse() {
// No-op.
}
/**
- * @param qryReqId Query request ID.
+ * @param reqId Request id.
+ * @param errMsg Error message.
*/
- public GridQueryCancelRequest(long qryReqId) {
- this.qryReqId = qryReqId;
+ public GridQueryKillResponse(long reqId, String errMsg) {
+ this.reqId = reqId;
+ this.errMsg = errMsg;
}
/**
- * @return Query request ID.
+ * @return Request id.
*/
- public long queryRequestId() {
- return qryReqId;
+ public long requestId() {
+ return reqId;
}
- /** {@inheritDoc} */
- @Override public void onAckReceived() {
- // No-op.
+ /**
+ * @return Error text or {@code null} if no error.
+ */
+ public String error(){
+ return errMsg;
}
+
/** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(GridQueryCancelRequest.class, this);
+ @Override public void onAckReceived() {
+ // No-op.
}
/** {@inheritDoc} */
@@ -77,11 +90,15 @@ public class GridQueryCancelRequest implements Message {
switch (writer.state()) {
case 0:
- if (!writer.writeLong("qryReqId", qryReqId))
+ if (!writer.writeLong("reqId", reqId))
return false;
writer.incrementState();
+ case 1:
+ if (!writer.writeString("errMsg", errMsg))
+ return false;
+ writer.incrementState();
}
return true;
@@ -96,25 +113,37 @@ public class GridQueryCancelRequest implements Message {
switch (reader.state()) {
case 0:
- qryReqId = reader.readLong("qryReqId");
+ reqId = reader.readLong("reqId");
if (!reader.isLastRead())
return false;
reader.incrementState();
+ case 1:
+ errMsg = reader.readString("errMsg");
+
+ if(!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
}
- return reader.afterMessageRead(GridQueryCancelRequest.class);
+ return reader.afterMessageRead(GridQueryKillResponse.class);
}
/** {@inheritDoc} */
@Override public short directType() {
- return 106;
+ return TYPE_CODE;
}
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 1;
+ return 2;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(GridQueryKillResponse.class, this);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/sql/SqlKeyword.java b/modules/core/src/main/java/org/apache/ignite/internal/sql/SqlKeyword.java
index c249a8f..5369650 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/sql/SqlKeyword.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/sql/SqlKeyword.java
@@ -26,6 +26,12 @@ import org.apache.ignite.internal.util.typedef.F;
* SQL keyword constants.
*/
public class SqlKeyword {
+ /** Keyword: KILL. */
+ public static final String KILL = "KILL";
+
+ /** Keyword: QUERY. */
+ public static final String QUERY = "QUERY";
+
/** Keyword: ALTER. */
public static final String ALTER = "ALTER";
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/sql/SqlParser.java b/modules/core/src/main/java/org/apache/ignite/internal/sql/SqlParser.java
index c5a0371..2f20009 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/sql/SqlParser.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/sql/SqlParser.java
@@ -28,6 +28,7 @@ import org.apache.ignite.internal.sql.command.SqlCreateIndexCommand;
import org.apache.ignite.internal.sql.command.SqlCreateUserCommand;
import org.apache.ignite.internal.sql.command.SqlDropIndexCommand;
import org.apache.ignite.internal.sql.command.SqlDropUserCommand;
+import org.apache.ignite.internal.sql.command.SqlKillQueryCommand;
import org.apache.ignite.internal.sql.command.SqlRollbackTransactionCommand;
import org.apache.ignite.internal.sql.command.SqlSetStreamingCommand;
import org.jetbrains.annotations.Nullable;
@@ -42,7 +43,9 @@ import static org.apache.ignite.internal.sql.SqlKeyword.GRANT;
import static org.apache.ignite.internal.sql.SqlKeyword.HASH;
import static org.apache.ignite.internal.sql.SqlKeyword.HELP;
import static org.apache.ignite.internal.sql.SqlKeyword.INDEX;
+import static org.apache.ignite.internal.sql.SqlKeyword.KILL;
import static org.apache.ignite.internal.sql.SqlKeyword.PRIMARY;
+import static org.apache.ignite.internal.sql.SqlKeyword.QUERY;
import static org.apache.ignite.internal.sql.SqlKeyword.REVOKE;
import static org.apache.ignite.internal.sql.SqlKeyword.ROLLBACK;
import static org.apache.ignite.internal.sql.SqlKeyword.SET;
@@ -183,6 +186,11 @@ public class SqlParser {
break;
+ case KILL:
+ cmd = processKill();
+
+ break;
+
case HELP:
cmd = processHelp();
@@ -217,7 +225,7 @@ public class SqlParser {
return cmd;
}
else
- throw errorUnexpectedToken(lex, BEGIN, COMMIT, CREATE, DROP, ROLLBACK, COPY, SET, ALTER, START);
+ throw errorUnexpectedToken(lex, BEGIN, COMMIT, CREATE, DROP, ROLLBACK, COPY, SET, ALTER, START, KILL);
case QUOTED:
case MINUS:
@@ -272,6 +280,22 @@ public class SqlParser {
}
/**
+ * Process KILL keyword.
+ *
+ * @return Command.
+ */
+ private SqlCommand processKill() {
+ if (lex.shift() && lex.tokenType() == SqlLexerTokenType.DEFAULT) {
+ switch (lex.token()) {
+ case QUERY:
+ return new SqlKillQueryCommand().parse(lex);
+ }
+ }
+
+ throw errorUnexpectedToken(lex, QUERY);
+ }
+
+ /**
* Processes COPY command.
*
* @return The {@link SqlBulkLoadCommand} command.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/sql/SqlParserUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/sql/SqlParserUtils.java
index 773fd43..894aebb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/sql/SqlParserUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/sql/SqlParserUtils.java
@@ -266,12 +266,18 @@ public class SqlParserUtils {
*
* @param lex Lexer.
* @param expKeyword Expected keyword.
+ * @return {@code true} In case token mathes, {@code false} otherwise.
*/
- static void skipIfMatchesOptionalKeyword(SqlLexer lex, String expKeyword) {
+ public static boolean skipIfMatchesOptionalKeyword(SqlLexer lex, String expKeyword) {
SqlLexerToken nextTok = lex.lookAhead();
- if (matchesKeyword(nextTok, expKeyword))
+ if (matchesKeyword(nextTok, expKeyword)) {
lex.shift();
+
+ return true;
+ }
+
+ return false;
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/sql/command/SqlKillQueryCommand.java b/modules/core/src/main/java/org/apache/ignite/internal/sql/command/SqlKillQueryCommand.java
new file mode 100644
index 0000000..604c27f
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/sql/command/SqlKillQueryCommand.java
@@ -0,0 +1,121 @@
+/*
+ * 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.sql.command;
+
+import java.util.UUID;
+import org.apache.ignite.internal.sql.SqlLexer;
+import org.apache.ignite.internal.sql.SqlLexerTokenType;
+import org.apache.ignite.internal.sql.SqlParserUtils;
+
+/**
+ * KILL QUERY command.
+ */
+public class SqlKillQueryCommand implements SqlCommand {
+ /** */
+ private static final String ASYNC = "ASYNC";
+
+ /** */
+ private static final String EXPECTED_GLOBAL_QRY_ID_FORMAT = "Global query id should have format " +
+ "'{node_id}_{query_id}', e.g. '6fa749ee-7cf8-4635-be10-36a1c75267a7_54321'";
+
+ /** Node query id. */
+ private long nodeQryId;
+
+ /** Node id. */
+ private UUID nodeId;
+
+ /** Async flag. */
+ private boolean async;
+
+ /** {@inheritDoc} */
+ @Override public SqlCommand parse(SqlLexer lex) {
+ async = SqlParserUtils.skipIfMatchesOptionalKeyword(lex, ASYNC);
+
+ parseGlobalQueryId(lex);
+
+ return this;
+ }
+
+ /**
+ * Parse global query id.
+ *
+ * @param lex Lexer.
+ */
+ private void parseGlobalQueryId(SqlLexer lex) {
+ if (lex.shift()) {
+ if (lex.tokenType() == SqlLexerTokenType.STRING) {
+ String tok = lex.token();
+
+ String[] ids = tok.split("_");
+
+ if (ids.length == 2) {
+ try {
+ nodeId = UUID.fromString(ids[0]);
+
+ nodeQryId = Long.parseLong(ids[1]);
+
+ return;
+ }
+ catch (Exception ignore) {
+ // Fall through.
+ }
+
+ throw SqlParserUtils.error(lex, EXPECTED_GLOBAL_QRY_ID_FORMAT);
+ }
+ }
+ }
+
+ if (async)
+ throw SqlParserUtils.error(lex, "Expected global query id. " + EXPECTED_GLOBAL_QRY_ID_FORMAT);
+ else
+ throw SqlParserUtils.error(lex, "Expected ASYNC token or global query id. "
+ + EXPECTED_GLOBAL_QRY_ID_FORMAT);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String schemaName() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void schemaName(String schemaName) {
+ // No-op.
+ }
+
+ /**
+ * @return Node query id.
+ */
+ public long nodeQueryId() {
+ return nodeQryId;
+ }
+
+ /**
+ * @return Node order id.
+ */
+ public UUID nodeId() {
+ return nodeId;
+ }
+
+ /**
+ * @return Async flag.
+ */
+ public boolean async() {
+ return async;
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/sql/SqlParserDropIndexSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/sql/SqlParserDropIndexSelfTest.java
index 518a502..684b701 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/sql/SqlParserDropIndexSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/sql/SqlParserDropIndexSelfTest.java
@@ -26,11 +26,9 @@ import org.junit.Test;
public class SqlParserDropIndexSelfTest extends SqlParserAbstractSelfTest {
/**
* Tests for DROP INDEX command.
- *
- * @throws Exception If failed.
*/
@Test
- public void testDropIndex() throws Exception {
+ public void testDropIndex() {
// Base.
parseValidate(null, "DROP INDEX idx", null, "IDX");
parseValidate(null, "DROP INDEX IDX", null, "IDX");
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/sql/SqlParserKillQuerySelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/sql/SqlParserKillQuerySelfTest.java
new file mode 100644
index 0000000..a7add82
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/sql/SqlParserKillQuerySelfTest.java
@@ -0,0 +1,144 @@
+/*
+ * 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.sql;
+
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+import org.apache.ignite.internal.sql.command.SqlCommand;
+import org.apache.ignite.internal.sql.command.SqlKillQueryCommand;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Tests for processing of KILL QUERY syntax.
+ */
+@RunWith(JUnit4.class)
+public class SqlParserKillQuerySelfTest extends SqlParserAbstractSelfTest {
+ /** */
+ private static final long TEST_QRY_ID = 341;
+
+ /**
+ * Tests for KILL QUERY command.
+ */
+ @Test
+ public void testKillQuery() {
+ UUID nodeId = UUID.randomUUID();
+
+ long qryId = ThreadLocalRandom.current().nextLong();
+
+ assertKillQuery("KILL QUERY '" + nodeId + "_" + qryId + "'", nodeId, qryId, false);
+
+ assertKillQuery("KILL QUERY '" + nodeId + "_" + TEST_QRY_ID + "'", nodeId, TEST_QRY_ID, false);
+
+ assertKillQuery("kill query '" + nodeId + "_" + TEST_QRY_ID + "'", nodeId, TEST_QRY_ID, false);
+
+ assertKillQuery("kIlL qUeRy '" + nodeId + "_" + TEST_QRY_ID + "'", nodeId, TEST_QRY_ID, false);
+
+ assertKillQuery("KILL QUERY '" + nodeId + "_" + TEST_QRY_ID + "'", nodeId, TEST_QRY_ID, false);
+
+ assertParseError("KILL QUERY '" + nodeId + "_*1'", expectedExceptionMessageIncorrectQueryId());
+
+ assertParseError("KILL QUERY '" + nodeId + "_a'", expectedExceptionMessageIncorrectQueryId());
+
+ assertParseError("KILL QUERY '" + nodeId + "_1 1'", expectedExceptionMessageIncorrectQueryId());
+
+ assertParseError("KILL QUERY 1'", expectedExceptionMessageNoAsyncAndQueryId());
+
+ assertParseError("KILL QUERY '" + nodeId + "_1' 1", "Unexpected token: \"1\"");
+
+ assertParseError("KILL '" + nodeId + "_" + TEST_QRY_ID + "'", "Unexpected token: \"" + nodeId + "_341\" (expected: \"QUERY\")");
+
+ assertParseError("KILL QUERY ", expectedExceptionMessageIncorrectQueryId());
+
+ assertParseError("KILL QUERY", expectedExceptionMessageIncorrectQueryId());
+
+ assertParseError("KILL QUERY " + nodeId + "_123", expectedExceptionMessageIncorrectQueryId());
+
+ assertKillQuery("KILL QUERY ASYNC '" + nodeId + "_" + qryId + "'", nodeId, qryId, true);
+
+ assertParseError("KILL QUERY ASYNC 1 '" + nodeId + "_" + qryId + "'", expectedExceptionMessageNoQueryId());
+
+ assertParseError("KILL QUERY ASYNC ", expectedExceptionMessageNoQueryId());
+
+ }
+
+ /**
+ * @return Expected exception message.
+ */
+ private String expectedExceptionMessageIncorrectQueryId(){
+ return "Global query id should have format '{node_id}_{query_id}', e.g. '6fa749ee-7cf8-4635-be10-36a1c75267a7_54321'";
+ }
+
+ /**
+ * @return Expected exception message.
+ */
+ private String expectedExceptionMessageNoAsyncAndQueryId(){
+ return "Expected ASYNC token or global query id. Global query id should have format '{node_id}_{query_id}', e.g. '6fa749ee-7cf8-4635-be10-36a1c75267a7_54321'";
+ }
+
+ /**
+ * @return Expected exception message.
+ */
+ private String expectedExceptionMessageNoQueryId(){
+ return "Expected global query id. Global query id should have format '{node_id}_{query_id}', e.g. '6fa749ee-7cf8-4635-be10-36a1c75267a7_54321'";
+ }
+
+ /**
+ * Make sure that parse error occurs.
+ *
+ * @param sql SQL.
+ * @param msg Expected error message.
+ */
+ private static void assertParseError(final String sql, String msg) {
+ assertParseError(null, sql, msg);
+ }
+
+ /**
+ * Test that given SQL is parsed as a KILL QUERY command and all parsed parameters have expected values.
+ *
+ * @param sql command.
+ * @param nodeIdExp Expected UUID.
+ * @param qryIdExp Expected query id.
+ */
+ private static void assertKillQuery(String sql, UUID nodeIdExp, long qryIdExp, boolean async) {
+ SqlCommand cmd = parse(sql);
+
+ Assert.assertTrue(cmd instanceof SqlKillQueryCommand);
+
+ SqlKillQueryCommand killQryCmd = (SqlKillQueryCommand)cmd;
+
+ Assert.assertEquals(nodeIdExp, killQryCmd.nodeId());
+
+ Assert.assertEquals(qryIdExp, killQryCmd.nodeQueryId());
+
+ Assert.assertEquals(async, killQryCmd.async());
+ }
+
+ /**
+ * Parse single SQL command.
+ *
+ * @param sql command.
+ * @return parsed command.
+ */
+ private static SqlCommand parse(String sql) {
+ return new SqlParser(null, sql).nextCommand();
+ }
+}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/cache/query/RegisteredQueryCursor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/cache/query/RegisteredQueryCursor.java
index 667c713..d376c1b 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/cache/query/RegisteredQueryCursor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/cache/query/RegisteredQueryCursor.java
@@ -20,9 +20,11 @@ package org.apache.ignite.internal.processors.cache.query;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.ignite.cache.query.QueryCancelledException;
import org.apache.ignite.internal.processors.cache.QueryCursorImpl;
import org.apache.ignite.internal.processors.query.GridQueryCancel;
import org.apache.ignite.internal.processors.query.RunningQueryManager;
+import org.apache.ignite.internal.util.typedef.X;
/**
* Query cursor for registered as running queries.
@@ -66,15 +68,25 @@ public class RegisteredQueryCursor<T> extends QueryCursorImpl<T> {
catch (Exception e) {
failed = true;
+ if (X.cause(e, QueryCancelledException.class) != null)
+ unregisterQuery();
+
throw e;
}
}
/** {@inheritDoc} */
@Override public void close() {
- if (unregistered.compareAndSet(false, true))
- runningQryMgr.unregister(qryId, failed);
+ unregisterQuery();
super.close();
}
+
+ /**
+ * Unregister query.
+ */
+ private void unregisterQuery(){
+ if (unregistered.compareAndSet(false, true))
+ runningQryMgr.unregister(qryId, failed);
+ }
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/CommandProcessor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/CommandProcessor.java
index 9f48eb9..630299a 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/CommandProcessor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/CommandProcessor.java
@@ -21,11 +21,17 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteCluster;
import org.apache.ignite.IgniteDataStreamer;
@@ -37,9 +43,16 @@ import org.apache.ignite.cache.QueryIndex;
import org.apache.ignite.cache.QueryIndexType;
import org.apache.ignite.cache.query.BulkLoadContextCursor;
import org.apache.ignite.cache.query.FieldsQueryCursor;
+import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.events.DiscoveryEvent;
+import org.apache.ignite.events.Event;
+import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.GridTopic;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
+import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.bulkload.BulkLoadAckClientParameters;
import org.apache.ignite.internal.processors.bulkload.BulkLoadCacheWriter;
import org.apache.ignite.internal.processors.bulkload.BulkLoadParser;
@@ -51,12 +64,12 @@ import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
import org.apache.ignite.internal.processors.query.GridQueryProperty;
import org.apache.ignite.internal.processors.query.GridQueryTypeDescriptor;
+import org.apache.ignite.internal.processors.query.GridRunningQueryInfo;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
import org.apache.ignite.internal.processors.query.NestedTxMode;
import org.apache.ignite.internal.processors.query.QueryEntityEx;
import org.apache.ignite.internal.processors.query.QueryField;
import org.apache.ignite.internal.processors.query.QueryUtils;
-import org.apache.ignite.internal.processors.query.RunningQueryManager;
import org.apache.ignite.internal.processors.query.SqlClientContext;
import org.apache.ignite.internal.processors.query.h2.dml.DmlBulkLoadDataConverter;
import org.apache.ignite.internal.processors.query.h2.dml.UpdatePlan;
@@ -70,6 +83,8 @@ import org.apache.ignite.internal.processors.query.h2.sql.GridSqlCreateTable;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlDropIndex;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlDropTable;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlStatement;
+import org.apache.ignite.internal.processors.query.messages.GridQueryKillRequest;
+import org.apache.ignite.internal.processors.query.messages.GridQueryKillResponse;
import org.apache.ignite.internal.processors.query.schema.SchemaOperationException;
import org.apache.ignite.internal.sql.command.SqlAlterTableCommand;
import org.apache.ignite.internal.sql.command.SqlAlterUserCommand;
@@ -82,13 +97,18 @@ import org.apache.ignite.internal.sql.command.SqlCreateUserCommand;
import org.apache.ignite.internal.sql.command.SqlDropIndexCommand;
import org.apache.ignite.internal.sql.command.SqlDropUserCommand;
import org.apache.ignite.internal.sql.command.SqlIndexColumn;
+import org.apache.ignite.internal.sql.command.SqlKillQueryCommand;
import org.apache.ignite.internal.sql.command.SqlRollbackTransactionCommand;
import org.apache.ignite.internal.sql.command.SqlSetStreamingCommand;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.lang.IgniteClosureX;
+import org.apache.ignite.internal.util.typedef.CIX2;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.lang.IgniteProductVersion;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.security.SecurityPermission;
import org.h2.command.Prepared;
import org.h2.command.ddl.AlterTableAlterColumn;
@@ -118,8 +138,8 @@ public class CommandProcessor {
/** Schema manager. */
private final SchemaManager schemaMgr;
- /** Running query manager. */
- private final RunningQueryManager runningQryMgr;
+ /** H2 Indexing. */
+ private final IgniteH2Indexing idx;
/** Logger. */
private final IgniteLogger log;
@@ -128,21 +148,222 @@ public class CommandProcessor {
private static final boolean handleUuidAsByte =
IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_SQL_UUID_DDL_BYTE_FORMAT, false);
+ /** Query cancel request counter. */
+ private final AtomicLong qryCancelReqCntr = new AtomicLong();
+
+ /** Cancellation runs. */
+ private ConcurrentMap<Long, KillQueryRun> cancellationRuns = new ConcurrentHashMap<>();
+
+ /** Flag indicate that node is stopped or not. */
+ private volatile boolean stopped;
+
+ /** */
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ /** KILL COMMAND support added since. */
+ private static final IgniteProductVersion KILL_COMMAND_SINCE_VER = IgniteProductVersion.fromString("2.8.0");
+
+ /** Local node message handler */
+ private final CIX2<ClusterNode, Message> locNodeMsgHnd = new CIX2<ClusterNode, Message>() {
+ @Override public void applyx(ClusterNode locNode, Message msg) {
+ onMessage(locNode.id(), msg);
+ }
+ };
+
/**
* Constructor.
*
* @param ctx Kernal context.
* @param schemaMgr Schema manager.
*/
- public CommandProcessor(GridKernalContext ctx, SchemaManager schemaMgr, RunningQueryManager runningQryMgr) {
+ public CommandProcessor(GridKernalContext ctx, SchemaManager schemaMgr, IgniteH2Indexing idx) {
this.ctx = ctx;
this.schemaMgr = schemaMgr;
- this.runningQryMgr = runningQryMgr;
+ this.idx = idx;
log = ctx.log(CommandProcessor.class);
}
/**
+ * Start executor.
+ */
+ public void start() {
+ ctx.io().addMessageListener(GridTopic.TOPIC_QUERY, (nodeId, msg, plc) -> onMessage(nodeId, msg));
+
+ ctx.event().addLocalEventListener(new GridLocalEventListener() {
+ @Override public void onEvent(final Event evt) {
+ UUID nodeId = ((DiscoveryEvent)evt).eventNode().id();
+
+ List<GridFutureAdapter<String>> futs = new ArrayList<>();
+
+ lock.writeLock().lock();
+
+ try {
+ Iterator<KillQueryRun> it = cancellationRuns.values().iterator();
+
+ while (it.hasNext()) {
+ KillQueryRun qryRun = it.next();
+
+ if (qryRun.nodeId().equals(nodeId)) {
+ futs.add(qryRun.cancelFuture());
+
+ it.remove();
+ }
+ }
+ }
+ finally {
+ lock.writeLock().unlock();
+ }
+
+ futs.forEach(f -> f.onDone("Query node has left the grid: [nodeId=" + nodeId + "]"));
+ }
+ }, EventType.EVT_NODE_FAILED, EventType.EVT_NODE_LEFT);
+ }
+
+ /**
+ * Close executor.
+ */
+ public void stop() {
+ stopped = true;
+
+ completeCancellationFutures("Local node is stopping: [nodeId=" + ctx.localNodeId() + "]");
+ }
+
+ /**
+ * Client disconnected callback.
+ */
+ public void onDisconnected() {
+ completeCancellationFutures("Failed to cancel query because local client node has been disconnected from the cluster");
+ }
+
+ /**
+ * @param err Text of error to complete futures.
+ */
+ private void completeCancellationFutures(@Nullable String err) {
+ lock.writeLock().lock();
+
+ try {
+ Iterator<KillQueryRun> it = cancellationRuns.values().iterator();
+
+ while (it.hasNext()) {
+ KillQueryRun qryRun = it.next();
+
+ qryRun.cancelFuture().onDone(err);
+
+ it.remove();
+ }
+ }
+ finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * @param nodeId Node ID.
+ * @param msg Message.
+ */
+ public void onMessage(UUID nodeId, Object msg) {
+ assert msg != null;
+
+ ClusterNode node = ctx.discovery().node(nodeId);
+
+ if (node == null)
+ return; // Node left, ignore.
+
+ boolean processed = true;
+
+ if (msg instanceof GridQueryKillRequest)
+ onQueryKillRequest((GridQueryKillRequest)msg, node);
+ if (msg instanceof GridQueryKillResponse)
+ onQueryKillResponse((GridQueryKillResponse)msg);
+ else
+ processed = false;
+
+ if (processed && log.isDebugEnabled())
+ log.debug("Processed response: " + nodeId + "->" + ctx.localNodeId() + " " + msg);
+ }
+
+ /**
+ * Process request to kill query.
+ *
+ * @param msg Message.
+ * @param node Cluster node.
+ */
+ private void onQueryKillRequest(GridQueryKillRequest msg, ClusterNode node) {
+ final long qryId = msg.nodeQryId();
+
+ String err = null;
+
+ GridRunningQueryInfo runningQryInfo = idx.runningQueryManager().runningQueryInfo(qryId);
+
+ if (runningQryInfo == null)
+ err = "Query with provided ID doesn't exist " +
+ "[nodeId=" + ctx.localNodeId() + ", qryId=" + qryId + "]";
+ else if (!runningQryInfo.cancelable())
+ err = "Query doesn't support cancellation " +
+ "[nodeId=" + ctx.localNodeId() + ", qryId=" + qryId + "]";
+
+ if (msg.asyncResponse() || err != null)
+ sendKillResponse(msg, node, err);
+
+ if (err == null) {
+ try {
+ runningQryInfo.cancel();
+ } catch (Exception e){
+ U.warn(log, "Cancellation of query failed: [qryId=" + qryId + "]", e);
+
+ if(!msg.asyncResponse())
+ sendKillResponse(msg, node, e.getMessage());
+
+ return;
+ }
+
+ if (!msg.asyncResponse())
+ runningQryInfo.runningFuture().listen((f) -> sendKillResponse(msg, node, f.result()));
+ }
+ }
+
+ /**
+ * @param msg Kill request message.
+ * @param node Initial kill request node.
+ * @param err Error message
+ */
+ private void sendKillResponse(GridQueryKillRequest msg, ClusterNode node, @Nullable String err) {
+ boolean snd = idx.send(GridTopic.TOPIC_QUERY,
+ GridTopic.TOPIC_QUERY.ordinal(),
+ Collections.singleton(node),
+ new GridQueryKillResponse(msg.requestId(), err),
+ null,
+ locNodeMsgHnd,
+ GridIoPolicy.MANAGEMENT_POOL,
+ false);
+
+ if (!snd)
+ U.warn(log, "Resposne on query cancellation wasn't send back: [qryId=" + msg.nodeQryId() + "]");
+ }
+
+ /**
+ * Process response to kill query request.
+ *
+ * @param msg Message.
+ */
+ private void onQueryKillResponse(GridQueryKillResponse msg) {
+ KillQueryRun qryRun;
+
+ lock.readLock().lock();
+
+ try {
+ qryRun = cancellationRuns.remove(msg.requestId());
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+
+ if (qryRun != null)
+ qryRun.cancelFuture().onDone(msg.error());
+ }
+
+ /**
* @param cmd Command.
* @return {@code True} if this is supported DDL command.
*/
@@ -186,6 +407,8 @@ public class CommandProcessor {
}
else if (cmdNative instanceof SqlSetStreamingCommand)
processSetStreamingCommand((SqlSetStreamingCommand)cmdNative, cliCtx);
+ else if(cmdNative instanceof SqlKillQueryCommand)
+ processKillQueryCommand((SqlKillQueryCommand) cmdNative);
else
processTxCommand(cmdNative, params);
}
@@ -199,6 +422,72 @@ public class CommandProcessor {
}
/**
+ * Process kill query command
+ *
+ * @param cmd Command.
+ */
+ private void processKillQueryCommand(SqlKillQueryCommand cmd) {
+ GridFutureAdapter<String> fut = new GridFutureAdapter<>();
+
+ lock.readLock().lock();
+
+ try {
+ if (stopped)
+ throw new IgniteSQLException("Failed to cancel query due to node is stopped [nodeId=" + cmd.nodeId() +
+ ",qryId=" + cmd.nodeQueryId() + "]");
+
+ ClusterNode node = ctx.discovery().node(cmd.nodeId());
+
+ if (node != null) {
+ if (node.version().compareTo(KILL_COMMAND_SINCE_VER) < 0)
+ throw new IgniteSQLException("Failed to cancel query: KILL QUERY operation is supported in " +
+ "versions 2.8.0 and newer");
+
+ KillQueryRun qryRun = new KillQueryRun(cmd.nodeId(), cmd.nodeQueryId(), fut);
+
+ long reqId = qryCancelReqCntr.incrementAndGet();
+
+ cancellationRuns.put(reqId, qryRun);
+
+ boolean snd = idx.send(GridTopic.TOPIC_QUERY,
+ GridTopic.TOPIC_QUERY.ordinal(),
+ Collections.singleton(node),
+ new GridQueryKillRequest(reqId, cmd.nodeQueryId(), cmd.async()),
+ null,
+ locNodeMsgHnd,
+ GridIoPolicy.MANAGEMENT_POOL,
+ false
+ );
+
+ if (!snd) {
+ cancellationRuns.remove(reqId);
+
+ throw new IgniteSQLException("Failed to cancel query due communication problem " +
+ "[nodeId=" + cmd.nodeId() + ",qryId=" + cmd.nodeQueryId() + "]");
+ }
+ }
+ else
+ throw new IgniteSQLException("Failed to cancel query, node is not alive [nodeId=" + cmd.nodeId() + ",qryId="
+ + cmd.nodeQueryId() + "]");
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+
+ try {
+ String err = fut.get();
+
+ if (err != null)
+ throw new IgniteSQLException("Failed to cancel query [nodeId=" + cmd.nodeId() + ",qryId="
+ + cmd.nodeQueryId() + ",err=" + err + "]");
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteSQLException("Failed to cancel query [nodeId=" + cmd.nodeId() + ",qryId="
+ + cmd.nodeQueryId() + ",err=" + e + "]", e);
+ }
+ }
+
+ /**
* Run DDL statement.
*
* @param sql Original SQL.
@@ -339,7 +628,6 @@ public class CommandProcessor {
* @param sql SQL.
* @param cmdH2 Command.
*/
- @SuppressWarnings({"unchecked"})
private void runCommandH2(String sql, GridSqlStatement cmdH2) {
IgniteInternalFuture fut = null;
@@ -996,10 +1284,10 @@ public class CommandProcessor {
BulkLoadParser inputParser = BulkLoadParser.createParser(cmd.inputFormat());
BulkLoadProcessor processor = new BulkLoadProcessor(inputParser, dataConverter, outputWriter,
- runningQryMgr, qryId);
+ idx.runningQueryManager(), qryId);
BulkLoadAckClientParameters params = new BulkLoadAckClientParameters(cmd.localFileName(), cmd.packetSize());
return new BulkLoadContextCursor(processor, params);
}
-}
\ No newline at end of file
+}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
index 0210fa5..765e1c1 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
@@ -26,7 +26,6 @@ import java.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
-import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
@@ -159,6 +158,7 @@ import org.h2.util.JdbcUtils;
import org.jetbrains.annotations.Nullable;
import static java.lang.Boolean.FALSE;
+import static java.util.Collections.singletonList;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_MVCC_TX_SIZE_CACHING_THRESHOLD;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccCachingManager.TX_SIZE_THRESHOLD;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.checkActive;
@@ -265,7 +265,6 @@ public class IgniteH2Indexing implements GridQueryIndexing {
/** H2 Connection manager. */
private LongRunningQueryManager longRunningQryMgr;
-
/**
* @return Kernal context.
*/
@@ -946,8 +945,6 @@ public class IgniteH2Indexing implements GridQueryIndexing {
Long qryId = registerRunningQuery(qryDesc, null);
- boolean fail = false;
-
CommandResult res = null;
try {
@@ -956,14 +953,12 @@ public class IgniteH2Indexing implements GridQueryIndexing {
return res.cursor();
}
catch (IgniteCheckedException e) {
- fail = true;
-
throw new IgniteSQLException("Failed to execute DDL statement [stmt=" + qryDesc.sql() +
", err=" + e.getMessage() + ']', e);
}
finally {
- if (fail || (res != null && res.unregisterRunningQuery()))
- runningQryMgr.unregister(qryId, fail);
+ if (res == null || res.unregisterRunningQuery())
+ runningQryMgr.unregister(qryId, res == null);
}
}
@@ -1140,10 +1135,10 @@ public class IgniteH2Indexing implements GridQueryIndexing {
cancel
);
- return Collections.singletonList(new QueryCursorImpl<>(new Iterable<List<?>>() {
+ return singletonList(new QueryCursorImpl<>(new Iterable<List<?>>() {
@SuppressWarnings("NullableProblems")
@Override public Iterator<List<?>> iterator() {
- return new IgniteSingletonIterator<>(Collections.singletonList(updRes.counter()));
+ return new IgniteSingletonIterator<>(singletonList(updRes.counter()));
}
}, cancel));
}
@@ -1176,8 +1171,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
boolean keepBinary,
GridQueryCancel cancel
) {
- if (cancel == null)
- cancel = new GridQueryCancel();
+ assert cancel != null;
// Register query.
Long qryId = registerRunningQuery(qryDesc, cancel);
@@ -1233,7 +1227,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
cursor.fieldsMeta(select.meta());
- return Collections.singletonList(cursor);
+ return singletonList(cursor);
}
catch (Exception e) {
runningQryMgr.unregister(qryId, true);
@@ -1916,7 +1910,8 @@ public class IgniteH2Indexing implements GridQueryIndexing {
runningQryMgr = new RunningQueryManager(ctx);
partExtractor = new PartitionExtractor(new H2PartitionResolver(this));
- cmdProc = new CommandProcessor(ctx, schemaMgr, runningQryMgr);
+ cmdProc = new CommandProcessor(ctx, schemaMgr, this);
+ cmdProc.start();
if (JdbcUtils.serializer != null)
U.warn(log, "Custom H2 serialization is already configured, will override.");
@@ -2058,6 +2053,8 @@ public class IgniteH2Indexing implements GridQueryIndexing {
longRunningQryMgr.stop();
connMgr.stop();
+ cmdProc.stop();
+
if (log.isDebugEnabled())
log.debug("Cache query index stopped.");
}
@@ -2163,6 +2160,8 @@ public class IgniteH2Indexing implements GridQueryIndexing {
/** {@inheritDoc} */
@Override public void onDisconnected(IgniteFuture<?> reconnectFut) {
rdcQryExec.onDisconnected(reconnectFut);
+
+ cmdProc.onDisconnected();
}
/**
@@ -2265,6 +2264,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
try {
List<List<List<?>>> cur = plan.createRows(argss);
+ //TODO: IGNITE-11176 - Need to support cancellation
ress = DmlUtils.processSelectResultBatched(plan, cur, qryParams.pageSize());
}
finally {
@@ -2320,8 +2320,8 @@ public class IgniteH2Indexing implements GridQueryIndexing {
for (UpdateResult res : ress) {
res.throwIfError();
- QueryCursorImpl<List<?>> resCur = (QueryCursorImpl<List<?>>)new QueryCursorImpl(Collections.singletonList
- (Collections.singletonList(res.counter())), cancel, false);
+ QueryCursorImpl<List<?>> resCur = (QueryCursorImpl<List<?>>)new QueryCursorImpl(singletonList
+ (singletonList(res.counter())), cancel, false);
resCur.fieldsMeta(UPDATE_RESULT_META);
@@ -2342,12 +2342,12 @@ public class IgniteH2Indexing implements GridQueryIndexing {
res.throwIfError();
- QueryCursorImpl<List<?>> resCur = (QueryCursorImpl<List<?>>)new QueryCursorImpl(Collections.singletonList
- (Collections.singletonList(res.counter())), cancel, false);
+ QueryCursorImpl<List<?>> resCur = (QueryCursorImpl<List<?>>)new QueryCursorImpl(singletonList
+ (singletonList(res.counter())), cancel, false);
resCur.fieldsMeta(UPDATE_RESULT_META);
- return Collections.singletonList(resCur);
+ return singletonList(resCur);
}
}
@@ -2536,6 +2536,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
int pageSize = loc ? 0 : qryParams.pageSize();
+ //TODO: IGNITE-11176 - Need to support cancellation
return DmlUtils.processSelectResult(plan, cur, pageSize);
}
@@ -2626,6 +2627,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
it = plan.iteratorForTransaction(connMgr, cur);
}
+ //TODO: IGNITE-11176 - Need to support cancellation
IgniteInternalFuture<Long> fut = tx.updateAsync(
cctx,
it,
@@ -2664,6 +2666,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
if (parts != null && parts.length == 0)
return new UpdateResult(0, X.EMPTY_OBJECT_ARRAY);
else {
+ //TODO: IGNITE-11176 - Need to support cancellation
IgniteInternalFuture<Long> fut = tx.updateAsync(
cctx,
ids,
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/KillQueryRun.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/KillQueryRun.java
new file mode 100644
index 0000000..a6c0565
--- /dev/null
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/KillQueryRun.java
@@ -0,0 +1,71 @@
+/*
+ * 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;
+
+import java.util.UUID;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+
+/**
+ * Kill Query run context.
+ */
+class KillQueryRun {
+ /** Node id. */
+ private final UUID nodeId;
+
+ /** Node query id. */
+ private final long nodeQryId;
+
+ /** Cancellation query future. */
+ private final GridFutureAdapter<String> cancelFut;
+
+ /**
+ * Constructor.
+ *
+ * @param nodeId Node id.
+ * @param cancelFut Cancellation query future.
+ */
+ public KillQueryRun(UUID nodeId, long nodeQryId, GridFutureAdapter<String> cancelFut) {
+ assert nodeId != null;
+
+ this.nodeId = nodeId;
+ this.nodeQryId = nodeQryId;
+ this.cancelFut = cancelFut;
+ }
+
+ /**
+ * @return Node id.
+ */
+ public UUID nodeId() {
+ return nodeId;
+ }
+
+ /**
+ * @return Node query id.
+ */
+ public long nodeQryId() {
+ return nodeQryId;
+ }
+
+ /**
+ * @return Cancellation query future.
+ */
+ public GridFutureAdapter<String> cancelFuture() {
+ return cancelFut;
+ }
+}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java
index 73e852d..0bb6922 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java
@@ -66,6 +66,7 @@ import org.apache.ignite.internal.sql.command.SqlCreateIndexCommand;
import org.apache.ignite.internal.sql.command.SqlCreateUserCommand;
import org.apache.ignite.internal.sql.command.SqlDropIndexCommand;
import org.apache.ignite.internal.sql.command.SqlDropUserCommand;
+import org.apache.ignite.internal.sql.command.SqlKillQueryCommand;
import org.apache.ignite.internal.sql.command.SqlRollbackTransactionCommand;
import org.apache.ignite.internal.sql.command.SqlSetStreamingCommand;
import org.apache.ignite.internal.util.GridBoundedConcurrentLinkedHashMap;
@@ -86,7 +87,7 @@ public class QueryParser {
/** A pattern for commands having internal implementation in Ignite. */
private static final Pattern INTERNAL_CMD_RE = Pattern.compile(
"^(create|drop)\\s+index|^alter\\s+table|^copy|^set|^begin|^commit|^rollback|^(create|alter|drop)\\s+user" +
- "|show|help|grant|revoke",
+ "|^kill\\s+query|show|help|grant|revoke",
Pattern.CASE_INSENSITIVE);
/** Indexing. */
@@ -179,12 +180,11 @@ public class QueryParser {
*
* @param schemaName Schema name.
* @param qry which sql text to parse.
- * @param remainingAllowed Whether multiple statements are allowed.
+ * @param remainingAllowed Whether multiple statements are allowed.
* @return Command or {@code null} if cannot parse this query.
*/
@SuppressWarnings("IfMayBeConditional")
- @Nullable
- private QueryParserResult parseNative(String schemaName, SqlFieldsQuery qry, boolean remainingAllowed) {
+ private @Nullable QueryParserResult parseNative(String schemaName, SqlFieldsQuery qry, boolean remainingAllowed) {
String sql = qry.getSql();
// Heuristic check for fast return.
@@ -208,7 +208,8 @@ public class QueryParser {
|| nativeCmd instanceof SqlSetStreamingCommand
|| nativeCmd instanceof SqlCreateUserCommand
|| nativeCmd instanceof SqlAlterUserCommand
- || nativeCmd instanceof SqlDropUserCommand)
+ || nativeCmd instanceof SqlDropUserCommand
+ || nativeCmd instanceof SqlKillQueryCommand)
)
return null;
@@ -217,10 +218,10 @@ public class QueryParser {
QueryDescriptor newPlanKey = queryDescriptor(schemaName, newQry);
SqlFieldsQuery remainingQry = null;
-
+
if (!F.isEmpty(parser.remainingSql())) {
checkRemainingAllowed(remainingAllowed);
-
+
remainingQry = cloneFieldsQuery(qry).setSql(parser.remainingSql()).setArgs(qry.getArgs());
}
@@ -263,7 +264,7 @@ public class QueryParser {
* @param schemaName Schema name.
* @param qry Query.
* @param batched Batched flag.
- * @param remainingAllowed Whether multiple statements are allowed.
+ * @param remainingAllowed Whether multiple statements are allowed.
* @return Parsing result.
*/
@SuppressWarnings("IfMayBeConditional")
@@ -304,10 +305,10 @@ public class QueryParser {
// Get remaining query and check if it is allowed.
SqlFieldsQuery remainingQry = null;
-
+
if (!F.isEmpty(prep.remainingSql())) {
checkRemainingAllowed(remainingAllowed);
-
+
remainingQry = cloneFieldsQuery(qry).setSql(prep.remainingSql());
}
@@ -540,13 +541,13 @@ public class QueryParser {
/**
* Throw exception is multiple statements are not allowed.
- *
+ *
* @param allowed Whether multiple statements are allowed.
*/
private static void checkRemainingAllowed(boolean allowed) {
if (allowed)
- return;
-
+ return;
+
throw new IgniteSQLException("Multiple statements queries are not supported.",
IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
}
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 1fe7ed4..3291295 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
@@ -622,11 +622,7 @@ public class GridReduceQueryExecutor {
final long qryReqId0 = qryReqId;
- cancel.set(new Runnable() {
- @Override public void run() {
- send(finalNodes, new GridQueryCancelRequest(qryReqId0), null, false);
- }
- });
+ cancel.set(() -> send(finalNodes, new GridQueryCancelRequest(qryReqId0), null, true));
boolean retry = false;
@@ -898,12 +894,10 @@ public class GridReduceQueryExecutor {
final Collection<ClusterNode> finalNodes = nodes;
- cancel.set(new Runnable() {
- @Override public void run() {
- r.future().onCancelled();
+ cancel.set(() -> {
+ r.future().onCancelled();
- send(finalNodes, new GridQueryCancelRequest(reqId), null, false);
- }
+ send(finalNodes, new GridQueryCancelRequest(reqId), null, true);
});
// send() logs the debug message
@@ -977,12 +971,12 @@ public class GridReduceQueryExecutor {
void releaseRemoteResources(Collection<ClusterNode> nodes, ReduceQueryRun r, long qryReqId,
boolean distributedJoins, MvccQueryTracker mvccTracker) {
if (distributedJoins)
- send(nodes, new GridQueryCancelRequest(qryReqId), null, false);
+ send(nodes, new GridQueryCancelRequest(qryReqId), null, true);
for (ReduceIndex idx : r.indexes()) {
if (!idx.fetchedAll()) {
if (!distributedJoins) // cancel request has been already sent for distributed join.
- send(nodes, new GridQueryCancelRequest(qryReqId), null, false);
+ send(nodes, new GridQueryCancelRequest(qryReqId), null, true);
r.setStateOnException(ctx.localNodeId(),
new CacheException("Query is canceled.", new QueryCancelledException()));
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/KillQueryFromClientTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/KillQueryFromClientTest.java
new file mode 100644
index 0000000..92fb3f9
--- /dev/null
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/KillQueryFromClientTest.java
@@ -0,0 +1,39 @@
+/*
+ * 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;
+
+import org.apache.ignite.internal.IgniteEx;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Test KILL QUERY requested from client node.
+ */
+
+@RunWith(JUnit4.class)
+public class KillQueryFromClientTest extends KillQueryTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteEx getKillRequestNode() {
+ IgniteEx clientNode = grid(NODES_COUNT - 1);
+
+ assertTrue(clientNode.context().clientNode());
+
+ return clientNode;
+ }
+}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/KillQueryFromNeighbourTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/KillQueryFromNeighbourTest.java
new file mode 100644
index 0000000..3b0a0ea
--- /dev/null
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/KillQueryFromNeighbourTest.java
@@ -0,0 +1,40 @@
+/*
+ * 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;
+
+import org.apache.ignite.internal.IgniteEx;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Test KILL QUERY requested from neighbour server node.
+ */
+@RunWith(JUnit4.class)
+public class KillQueryFromNeighbourTest extends KillQueryTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteEx getKillRequestNode() {
+ IgniteEx node = grid(1);
+
+ assertNotSame(node.localNode().id(), ignite.localNode().id());
+
+ assertFalse(node.context().clientNode());
+
+ return node;
+ }
+}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/KillQueryOnClientDisconnectTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/KillQueryOnClientDisconnectTest.java
new file mode 100644
index 0000000..e9c2c9d
--- /dev/null
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/KillQueryOnClientDisconnectTest.java
@@ -0,0 +1,302 @@
+/*
+ * 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;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import junit.framework.Assert;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.cache.query.annotations.QuerySqlFunction;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.processors.query.messages.GridQueryKillResponse;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.GridAbstractTest;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+
+/**
+ * Test KILL QUERY requested from client which disconnected from cluster during cancellation in progress.
+ */
+
+@RunWith(JUnit4.class)
+public class KillQueryOnClientDisconnectTest extends GridCommonAbstractTest {
+ /** IP finder. */
+ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+ /** Node configration conter. */
+ private static int cntr;
+
+ /** Statement. */
+ protected Statement stmt;
+
+ /** Cancellation processing timeout. */
+ public static final int TIMEOUT = 5000;
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ cntr = 0;
+
+ startGrids(2);
+
+ for (int i = 0; i < 1000; ++i)
+ grid(0).cache(GridAbstractTest.DEFAULT_CACHE_NAME).put(i, i);
+ }
+
+ /**
+ * Called before execution of every test method in class.
+ *
+ * @throws Exception If failed.
+ */
+ @Before
+ public void before() throws Exception {
+ TestSQLFunctions.init();
+
+ Connection conn = GridTestUtils.connect(grid(0), null);
+
+ conn.setSchema('"' + GridAbstractTest.DEFAULT_CACHE_NAME + '"');
+
+ stmt = conn.createStatement();
+ }
+
+ /** {@inheritDoc} */
+ protected IgniteEx clientNode() {
+ IgniteEx clientNode = grid(1);
+
+ assertTrue(clientNode.context().clientNode());
+
+ return clientNode;
+ }
+
+ /** {@inheritDoc} */
+ protected IgniteEx serverNode() {
+ IgniteEx srvNode = grid(0);
+
+ assertFalse(srvNode.context().clientNode());
+
+ return srvNode;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ CacheConfiguration<?, ?> cache = GridAbstractTest.defaultCacheConfiguration();
+
+ cache.setCacheMode(PARTITIONED);
+ cache.setBackups(1);
+ cache.setWriteSynchronizationMode(FULL_SYNC);
+ cache.setSqlFunctionClasses(TestSQLFunctions.class);
+ cache.setIndexedTypes(Integer.class, Integer.class);
+
+ cfg.setCacheConfiguration(cache);
+
+ TcpDiscoverySpi disco = new TcpDiscoverySpi();
+
+ disco.setIpFinder(IP_FINDER);
+
+ cfg.setDiscoverySpi(disco);
+
+ if (++cntr == 2)
+ cfg.setClientMode(true);
+
+ cfg.setCommunicationSpi(new TcpCommunicationSpi() {
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg,
+ IgniteInClosure<IgniteException> ackC) {
+ if (GridIoMessage.class.isAssignableFrom(msg.getClass())) {
+ Message gridMsg = ((GridIoMessage)msg).message();
+
+ //Don't send Kill response and disconnect client node.
+ if (GridQueryKillResponse.class.isAssignableFrom(gridMsg.getClass())) {
+ grid(0).configuration().getDiscoverySpi().failNode(clientNode().cluster().localNode().id(), null);
+
+ return;
+ }
+ }
+
+ super.sendMessage(node, msg, ackC);
+ }
+ });
+
+ return cfg;
+ }
+
+ /**
+ * Test client disconnect during cancellation request is processing.
+ *
+ * @throws Exception In case of failure.
+ */
+ @Test
+ public void clientDisconnectFromCluster() throws Exception {
+ IgniteInternalFuture cancelRes = cancelAndCheckClientDisconnect();
+
+ GridTestUtils.assertThrows(log, () -> {
+ stmt.executeQuery("select * from Integer where _key in " +
+ "(select _key from Integer where awaitLatchCancelled() = 0) and shouldNotBeCalledInCaseOfCancellation()");
+
+ return null;
+ }, SQLException.class, "The query was cancelled while executing.");
+
+ cancelRes.get(TIMEOUT);
+ }
+
+ /**
+ * Cancels current query, actual cancel will wait <code>cancelLatch</code> to be releaseds.
+ * Checked that Cancellation wasn't complete due to clint was disconnected.
+ *
+ * @return <code>IgniteInternalFuture</code> to check whether exception was thrown.
+ */
+ protected IgniteInternalFuture cancelAndCheckClientDisconnect() {
+ return GridTestUtils.runAsync(() -> {
+ try {
+ TestSQLFunctions.cancelLatch.await();
+
+ List<GridRunningQueryInfo> runningQueries = (List<GridRunningQueryInfo>)serverNode().context().query().runningQueries(-1);
+
+ assertEquals(1, runningQueries.size());
+
+ IgniteInternalFuture fut = GridTestUtils.runAsync(() -> {
+ clientNode().cache(DEFAULT_CACHE_NAME).query(new SqlFieldsQuery("KILL QUERY '" + runningQueries.get(0).globalQueryId() + "'"));
+ });
+
+ doSleep(500);
+
+ TestSQLFunctions.reqLatch.countDown();
+
+ GridTestUtils.assertThrows(log, () -> fut.get(TIMEOUT), IgniteCheckedException.class, "Failed to cancel query because local client node has been disconnected from the cluster");
+ }
+ catch (Exception e) {
+ log.error("Unexpected exception.", e);
+
+ Assert.fail("Unexpected exception");
+ }
+ });
+ }
+
+ /**
+ * Utility class with custom SQL functions.
+ */
+ public static class TestSQLFunctions {
+ /** Request latch. */
+ static CountDownLatch reqLatch;
+
+ /** Cancel latch. */
+ static CountDownLatch cancelLatch;
+
+ /** Suspend query latch. */
+ static CountDownLatch suspendQryLatch;
+
+ /**
+ * Recreate latches.
+ */
+ static void init() {
+ reqLatch = new CountDownLatch(1);
+
+ cancelLatch = new CountDownLatch(1);
+
+ suspendQryLatch = new CountDownLatch(1);
+ }
+
+ /**
+ * Releases cancelLatch that leeds to sending cancel Query and waits until cancel Query is fully processed.
+ *
+ * @return 0;
+ */
+ @QuerySqlFunction
+ public static long awaitLatchCancelled() {
+ try {
+ cancelLatch.countDown();
+ reqLatch.await();
+ }
+ catch (Exception ignored) {
+ // No-op.
+ }
+
+ return 0;
+ }
+
+ /**
+ * Waits latch release.
+ *
+ * @return 0;
+ */
+ @QuerySqlFunction
+ public static long awaitQuerySuspensionLatch() {
+ try {
+ suspendQryLatch.await();
+ }
+ catch (Exception ignored) {
+ // No-op.
+ }
+
+ return 0;
+ }
+
+ /**
+ * If called fails with corresponding message.
+ *
+ * @return 0;
+ */
+ @QuerySqlFunction
+ public static long shouldNotBeCalledInCaseOfCancellation() {
+ fail("Query wasn't actually cancelled.");
+
+ return 0;
+ }
+
+ /**
+ * @param v amount of milliseconds to sleep
+ * @return amount of milliseconds to sleep
+ */
+ @QuerySqlFunction
+ public static int sleep_func(int v) {
+ try {
+ Thread.sleep(v);
+ }
+ catch (InterruptedException ignored) {
+ // No-op
+ }
+ return v;
+ }
+ }
+}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/KillQueryTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/KillQueryTest.java
new file mode 100644
index 0000000..fa6553e
--- /dev/null
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/KillQueryTest.java
@@ -0,0 +1,1010 @@
+/*
+ * 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;
+
+import java.io.Serializable;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import javax.cache.CacheException;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.query.FieldsQueryCursor;
+import org.apache.ignite.cache.query.QueryCancelledException;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.cache.query.annotations.QuerySqlField;
+import org.apache.ignite.cache.query.annotations.QuerySqlFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.managers.discovery.CustomMessageWrapper;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch;
+import org.apache.ignite.internal.processors.query.schema.message.SchemaProposeDiscoveryMessage;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.GridAbstractTest;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.internal.util.IgniteUtils.resolveIgnitePath;
+
+/**
+ * Test KILL QUERY requested from the same node where quere was executed.
+ */
+@SuppressWarnings({"ThrowableNotThrown", "AssertWithSideEffects"})
+@RunWith(JUnit4.class)
+public class KillQueryTest extends GridCommonAbstractTest {
+ /** IP finder. */
+ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+ /** A CSV file with one record. */
+ private static final String BULKLOAD_20_000_LINE_CSV_FILE =
+ Objects.requireNonNull(resolveIgnitePath("/modules/clients/src/test/resources/bulkload20_000.csv")).
+ getAbsolutePath();
+
+ /** Max table rows. */
+ private static final int MAX_ROWS = 10000;
+
+ /** Cancellation processing timeout. */
+ public static final int TIMEOUT = 5000;
+
+ /** Nodes count. */
+ protected static final byte NODES_COUNT = 3;
+
+ /** Timeout for checking async result. */
+ public static final int CHECK_RESULT_TIMEOUT = 1_000;
+
+ /** Connection. */
+ private Connection conn;
+
+ /** Statement. */
+ private Statement stmt;
+
+ /** Ignite. */
+ protected IgniteEx ignite;
+
+ /** Ignite instance for kill request. */
+ private IgniteEx igniteForKillRequest;
+
+ /** Node configration conter. */
+ private static int cntr;
+
+ /** Table count. */
+ private static AtomicInteger tblCnt = new AtomicInteger();
+
+ /** Barrier. */
+ private static volatile CyclicBarrier barrier;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ CacheConfiguration<?, ?> cache = GridAbstractTest.defaultCacheConfiguration();
+
+ cache.setCacheMode(PARTITIONED);
+ cache.setBackups(1);
+ cache.setWriteSynchronizationMode(FULL_SYNC);
+ cache.setSqlFunctionClasses(TestSQLFunctions.class);
+ cache.setIndexedTypes(Integer.class, Integer.class, Long.class, Long.class, String.class, Person.class);
+
+ cfg.setCacheConfiguration(cache);
+
+ if (++cntr == NODES_COUNT)
+ cfg.setClientMode(true);
+
+ cfg.setDiscoverySpi(new TcpDiscoverySpi() {
+
+ @Override public void sendCustomEvent(DiscoverySpiCustomMessage msg) throws IgniteException {
+ if (CustomMessageWrapper.class.isAssignableFrom(msg.getClass())) {
+ DiscoveryCustomMessage delegate = ((CustomMessageWrapper)msg).delegate();
+
+ if (DynamicCacheChangeBatch.class.isAssignableFrom(delegate.getClass())) {
+ try {
+ awaitTimeout();
+ }
+ catch (Exception e) {
+ log.error(e.getMessage(), e);
+ }
+
+ }
+ else if (SchemaProposeDiscoveryMessage.class.isAssignableFrom(delegate.getClass())) {
+ try {
+ awaitTimeout();
+ }
+ catch (Exception e) {
+ log.error(e.getMessage(), e);
+ }
+ }
+ }
+
+ super.sendCustomEvent(msg);
+ }
+ }.setIpFinder(IP_FINDER));
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ cntr = 0;
+
+ startGrids(NODES_COUNT);
+
+ for (int i = 0; i < MAX_ROWS; ++i) {
+ grid(0).cache(GridAbstractTest.DEFAULT_CACHE_NAME).put(i, i);
+
+ grid(0).cache(GridAbstractTest.DEFAULT_CACHE_NAME).put((long)i, (long)i);
+ }
+ }
+
+ /**
+ * @return Ignite node which will be used to execute kill query.
+ */
+ protected IgniteEx getKillRequestNode() {
+ return grid(0);
+ }
+
+ /**
+ * Called before execution of every test method in class.
+ *
+ * @throws Exception If failed.
+ */
+ @Before
+ public void before() throws Exception {
+ TestSQLFunctions.init();
+
+ newBarrier(1);
+
+ tblCnt.incrementAndGet();
+
+ conn = GridTestUtils.connect(grid(0), null);
+
+ conn.setSchema('"' + GridAbstractTest.DEFAULT_CACHE_NAME + '"');
+
+ stmt = conn.createStatement();
+
+ ignite = grid(0);
+
+ igniteForKillRequest = getKillRequestNode();
+ }
+
+ /**
+ * Called after execution of every test method in class.
+ *
+ * @throws Exception If failed.
+ */
+ @After
+ public void after() throws Exception {
+ if (stmt != null && !stmt.isClosed()) {
+ stmt.close();
+
+ assert stmt.isClosed();
+ }
+
+ conn.close();
+
+ assertTrue(ignite.context().query().runningQueries(-1).isEmpty());
+ }
+
+ /**
+ * Trying to cancel COPY FROM command.
+ *
+ * @throws Exception In case of failure.
+ */
+ @Test
+ public void testBulkLoadCancellationUnsupported() throws Exception {
+ checkCreateTableCancellationUnsupported(false);
+ }
+
+ /**
+ * Trying to async cancel COPY FROM command.
+ *
+ * @throws Exception In case of failure.
+ */
+ @Test
+ public void testAsyncBulkLoadCancellationUnsupported() throws Exception {
+ checkCreateTableCancellationUnsupported(true);
+ }
+
+ /**
+ * Trying to cancel COPY FROM command.
+ *
+ * @param async execute cancellation in ASYNC mode.
+ * @throws Exception In case of failure.
+ */
+ public void checkBulkLoadCancellationUnsupported(boolean async) throws Exception {
+ String path = Objects.requireNonNull(resolveIgnitePath("/modules/clients/src/test/resources/bulkload1.csv"))
+ .getAbsolutePath();
+
+ String sqlPrepare = "CREATE TABLE " + currentTestTableName() +
+ "(id integer primary key, age integer, firstName varchar, lastname varchar)";
+ String sqlCmd = "COPY FROM '" + path + "'" +
+ " INTO " + currentTestTableName() +
+ " (_key, age, firstName, lastName)" +
+ " format csv charset 'ascii'";
+
+ checkCancellationUnsupported(
+ Arrays.asList(sqlPrepare),
+ sqlCmd,
+ async);
+ }
+
+ /**
+ * Trying to cancel CREATE TABLE command.
+ *
+ * @throws Exception In case of failure.
+ */
+ @Test
+ public void testCreateTableCancellationUnsupported() throws Exception {
+ checkCreateTableCancellationUnsupported(false);
+ }
+
+ /**
+ * Trying to async cancel CREATE TABLE command.
+ *
+ * @throws Exception In case of failure.
+ */
+ @Test
+ public void testAsyncCreateTableCancellationUnsupported() throws Exception {
+ checkCreateTableCancellationUnsupported(true);
+ }
+
+ /**
+ * Trying to cancel CREATE TABLE command.
+ *
+ * @param async execute cancellation in ASYNC mode.
+ * @throws Exception In case of failure.
+ */
+ public void checkCreateTableCancellationUnsupported(boolean async) throws Exception {
+ checkCancellationUnsupported(Collections.<String>emptyList(),
+ "CREATE TABLE " + currentTestTableName() + " (id INTEGER PRIMARY KEY, name VARCHAR)",
+ async);
+ }
+
+ /**
+ * Trying to cancel ALTER TABLE command.
+ *
+ * @throws Exception In case of failure.
+ */
+ @Test
+ public void testAlterTableCancellationUnsupported() throws Exception {
+ checkAlterTableCancellationUnsupported(false);
+ }
+
+ /**
+ * Trying to async cancel ALTER TABLE command.
+ *
+ * @throws Exception In case of failure.
+ */
+ @Test
+ public void testAsyncAlterTableCancellationUnsupported() throws Exception {
+ checkAlterTableCancellationUnsupported(true);
+ }
+
+ /**
+ * Trying to cancel ALTER TABLE command.
+ *
+ * @param async execute cancellation in ASYNC mode.
+ * @throws Exception In case of failure.
+ */
+ private void checkAlterTableCancellationUnsupported(boolean async) throws Exception {
+ checkCancellationUnsupported(Arrays.asList("CREATE TABLE " + currentTestTableName() + " (id INTEGER PRIMARY KEY, name VARCHAR)"),
+ "ALTER TABLE " + currentTestTableName() + " ADD COLUMN COL VARCHAR",
+ async);
+ }
+
+ /**
+ * Trying to cancel CREATE INDEX command.
+ *
+ * @throws Exception In case of failure.
+ */
+ @Test
+ public void testCreateIndexCancellationUnsupported() throws Exception {
+ checkCreateIndexCancellationUnsupported(false);
+ }
+
+ /**
+ * Trying to async cancel CREATE INDEX command.
+ *
+ * @throws Exception In case of failure.
+ */
+ @Test
+ public void testAsyncCreateIndexCancellationUnsupported() throws Exception {
+ checkCreateIndexCancellationUnsupported(true);
+ }
+
+ /**
+ * Trying to cancel CREATE INDEX command.
+ *
+ * @param async execute cancellation in ASYNC mode.
+ * @throws Exception In case of failure.
+ */
+ private void checkCreateIndexCancellationUnsupported(boolean async) throws Exception {
+ checkCancellationUnsupported(Arrays.asList("CREATE TABLE " + currentTestTableName() + " (id INTEGER PRIMARY KEY, name VARCHAR)"),
+ "CREATE INDEX " + currentTestTableName() + "_IDX ON " + currentTestTableName() + "(name, id)",
+ async);
+ }
+
+ /**
+ * Trying to cancel DROP INDEX command.
+ *
+ * @throws Exception In case of failure.
+ */
+ @Test
+ public void testDropIndexCancellationUnsupported() throws Exception {
+ checkDropIndexCancellationUnsupported(false);
+ }
+
+ /**
+ * Trying to async cancel DROP INDEX command.
+ *
+ * @throws Exception In case of failure.
+ */
+ @Test
+ public void testAsyncDropIndexCancellationUnsupported() throws Exception {
+ checkDropIndexCancellationUnsupported(true);
+ }
+
+ /**
+ * Trying to cancel DROP INDEX command.
+ *
+ * @param async execute cancellation in ASYNC mode.
+ * @throws Exception In case of failure.
+ */
+ private void checkDropIndexCancellationUnsupported(boolean async) throws Exception {
+ checkCancellationUnsupported(
+ Arrays.asList("CREATE TABLE " + currentTestTableName() + " (id INTEGER PRIMARY KEY, name VARCHAR)",
+ "CREATE INDEX " + currentTestTableName() + "_IDX ON " + currentTestTableName() + "(name, id)"),
+ "DROP INDEX " + currentTestTableName() + "_IDX",
+ async);
+ }
+
+ /**
+ * Get test table name unique for per tests, but stable within one test run.
+ *
+ * @return Generated test table name unique for per tests, but stable within one test run.
+ */
+ private String currentTestTableName() {
+ return "TST_TABLE_" + tblCnt.get();
+ }
+
+ /**
+ * Check that trying cancellation execution of {@code sqlCmd} can't be cancelled due to it's not supported.
+ *
+ * @param prepareSteps Preparation SQLs before start test.
+ * @param sqlCmd Command which can't be cancelled
+ * @param async Execute cancellation in ASYNC mode.
+ * @throws Exception In case of failure.
+ */
+ private void checkCancellationUnsupported(List<String> prepareSteps, String sqlCmd, boolean async) throws Exception {
+ for (String sql : prepareSteps) {
+ try {
+ stmt.execute(sql);
+ }
+ catch (SQLException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ newBarrier(2);
+
+ IgniteInternalFuture cancelRes = cancelQueryWithBarrier(sqlCmd,
+ "Query doesn't support cancellation",
+ async);
+
+ stmt.execute(sqlCmd);
+
+ cancelRes.get(TIMEOUT);
+ }
+
+ /**
+ * Trying to cancel unexist query.
+ */
+ @Test
+ public void testKillUnknownQry() {
+ checkKillUnknownQry(false);
+ }
+
+ /**
+ * Trying to async cancel unexist query.
+ */
+ @Test
+ public void testAsyncKillUnknownQry() {
+ checkKillUnknownQry(true);
+ }
+
+ /**
+ * Trying to cancel unexist query.
+ *
+ * @param async execute cancellation in ASYNC mode.
+ */
+ private void checkKillUnknownQry(boolean async) {
+ UUID nodeId = ignite.localNode().id();
+
+ GridTestUtils.assertThrows(log, () -> {
+ igniteForKillRequest.cache(DEFAULT_CACHE_NAME)
+ .query(createKillQuery(nodeId, Long.MAX_VALUE, async));
+
+ return null;
+ }, CacheException.class, "Query with provided ID doesn't exist [nodeId=" + nodeId);
+ }
+
+ /**
+ * Trying to cancel query on unknown node.
+ */
+ @Test
+ public void testKillQryUnknownNode() {
+ checkKillQryUnknownNode(false);
+ }
+
+ /**
+ * Trying to async cancel query on unknown node.
+ */
+ @Test
+ public void testAsyncKillQryUnknownNode() {
+ checkKillQryUnknownNode(true);
+ }
+
+ /**
+ * Trying to cancel query on unexist node.
+ *
+ * @param async execute cancellation in ASYNC mode.
+ */
+ private void checkKillQryUnknownNode(boolean async) {
+ GridTestUtils.assertThrows(log, () -> {
+ igniteForKillRequest.cache(DEFAULT_CACHE_NAME)
+ .query(createKillQuery(UUID.randomUUID(), Long.MAX_VALUE, async));
+
+ return null;
+ }, CacheException.class, "Failed to cancel query, node is not alive");
+ }
+
+ /**
+ * Trying to kill already killed query. No exceptions expected.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testKillAlreadyKilledQuery() throws Exception {
+ checkKillAlreadyKilledQuery(false);
+ }
+
+ /**
+ * Trying to kill already killed query. No exceptions expected.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testAsyncKillAlreadyKilledQuery() throws Exception {
+ checkKillAlreadyKilledQuery(true);
+ }
+
+ /**
+ * Trying to kill already killed query. No exceptions expected.
+ *
+ * @param async execute cancellation in ASYNC mode.
+ * @throws Exception If failed.
+ */
+ private void checkKillAlreadyKilledQuery(boolean async) throws Exception {
+ IgniteCache<Object, Object> cache = ignite.cache(DEFAULT_CACHE_NAME);
+
+ FieldsQueryCursor<List<?>> cur = cache.query(new SqlFieldsQuery("select * from Integer"));
+
+ List<GridRunningQueryInfo> runningQueries = (List<GridRunningQueryInfo>)ignite.context().query().runningQueries(-1);
+
+ GridRunningQueryInfo runQryInfo = runningQueries.get(0);
+
+ SqlFieldsQuery killQry = createKillQuery(runQryInfo.globalQueryId(), async);
+
+ IgniteCache<Object, Object> reqCache = igniteForKillRequest.cache(DEFAULT_CACHE_NAME);
+
+ IgniteInternalFuture killFut = GridTestUtils.runAsync(() -> reqCache.query(killQry));
+
+ GridTestUtils.assertThrows(log,
+ () -> cur.iterator().next(),
+ QueryCancelledException.class,
+ "The query was cancelled while executing");
+
+ killFut.get(CHECK_RESULT_TIMEOUT);
+
+ GridTestUtils.assertThrows(log,
+ () -> reqCache.query(killQry),
+ CacheException.class,
+ "Query with provided ID doesn't exist");
+
+ cur.close();
+
+ }
+
+ /**
+ * @param nodeId Node id.
+ * @param qryId Node query id.
+ */
+ private SqlFieldsQuery createKillQuery(UUID nodeId, long qryId, boolean async) {
+ return createKillQuery(nodeId + "_" + qryId, async);
+ }
+
+ /**
+ * @param globalQryId Global query id.
+ */
+ private SqlFieldsQuery createKillQuery(String globalQryId, boolean async) {
+ return new SqlFieldsQuery("KILL QUERY" + (async ? " ASYNC" : "") + " '" + globalQryId + "'");
+ }
+
+ /**
+ * Trying to cancel long running query. No exceptions expected.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testCancelQuery() throws Exception {
+ checkCancelQuery(false);
+ }
+
+ /**
+ * Trying to async cancel long running query. No exceptions expected.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testAsyncCancelQuery() throws Exception {
+ checkCancelQuery(true);
+ }
+
+ /**
+ * Trying to cancel long running query. No exceptions expected.
+ *
+ * @param async execute cancellation in ASYNC mode.
+ * @throws Exception If failed.
+ */
+ private void checkCancelQuery(boolean async) throws Exception {
+ IgniteInternalFuture cancelRes = cancel(1, async);
+
+ GridTestUtils.assertThrows(log, () -> {
+ stmt.executeQuery("select * from Integer where _key in " +
+ "(select _key from Integer where awaitLatchCancelled() = 0) and shouldNotBeCalledInCaseOfCancellation()");
+
+ return null;
+ }, SQLException.class, "The query was cancelled while executing.");
+
+ // Ensures that there were no exceptions within async cancellation process.
+ cancelRes.get(CHECK_RESULT_TIMEOUT);
+ }
+
+ /**
+ * Trying to cancel long running multiple statements query. No exceptions expected.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testKillMultipleStatementsQuery() throws Exception {
+ checkKillMultipleStatementsQuery(false);
+ }
+
+ /**
+ * Trying to async cancel long running multiple statements query. No exceptions expected.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testAsyncKillMultipleStatementsQuery() throws Exception {
+ checkKillMultipleStatementsQuery(true);
+ }
+
+ /**
+ * Trying to async cancel long running multiple statements query. No exceptions expected.
+ *
+ * @param async execute cancellation in ASYNC mode.
+ * @throws Exception If failed.
+ */
+ private void checkKillMultipleStatementsQuery(boolean async) throws Exception {
+ try (Statement anotherStatement = conn.createStatement()) {
+ anotherStatement.setFetchSize(1);
+
+ String sql = "select * from Integer";
+
+ ResultSet rs = anotherStatement.executeQuery(sql);
+
+ assert rs.next();
+
+ IgniteInternalFuture cancelRes = cancel(3, async, sql);
+
+ GridTestUtils.assertThrows(log, () -> {
+ // Executes multiple long running query
+ stmt.execute(
+ "select 100 from Integer;"
+ + "select _key from Integer where awaitLatchCancelled() = 0;");
+ return null;
+ }, SQLException.class, "The query was cancelled while executing");
+
+ assert rs.next() : "The other cursor mustn't be closed";
+
+ // Ensures that there were no exceptions within async cancellation process.
+ cancelRes.get(CHECK_RESULT_TIMEOUT);
+ }
+ }
+
+ /**
+ * Trying to cancel long running batch query. No exceptions expected. *
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testCancelBatchQuery() throws Exception {
+ checkCancelBatchQuery(false);
+ }
+
+ /**
+ * Trying to async cancel long running batch query. No exceptions expected. *
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testAsyncCancelBatchQuery() throws Exception {
+ checkCancelBatchQuery(true);
+ }
+
+ /**
+ * Trying to cancel long running batch query. No exceptions expected. *
+ *
+ * @param async execute cancellation in ASYNC mode.
+ * @throws Exception If failed.
+ */
+ private void checkCancelBatchQuery(boolean async) throws Exception {
+ try (Statement stmt2 = conn.createStatement()) {
+ stmt2.setFetchSize(1);
+
+ String sql = "SELECT * from Integer";
+
+ ResultSet rs = stmt2.executeQuery(sql);
+
+ Assert.assertTrue(rs.next());
+
+ IgniteInternalFuture cancelRes = cancel(2, async, sql);
+
+ GridTestUtils.assertThrows(log, () -> {
+ stmt.addBatch("update Long set _val = _val + 1 where _key < sleep_func (30)");
+ stmt.addBatch("update Long set _val = _val + 1 where awaitLatchCancelled() = 0");
+ stmt.addBatch("update Long set _val = _val + 1 where _key < sleep_func (30)");
+ stmt.addBatch("update Long set _val = _val + 1 where shouldNotBeCalledInCaseOfCancellation()");
+
+ stmt.executeBatch();
+ return null;
+ }, SQLException.class, "The query was cancelled while executing");
+
+ Assert.assertTrue("The other cursor mustn't be closed", rs.next());
+
+ // Ensures that there were no exceptions within async cancellation process.
+ cancelRes.get(CHECK_RESULT_TIMEOUT);
+ }
+ }
+
+ /**
+ * Cancels current query which wait on barrier.
+ *
+ * @param qry Query which need to try cancel.
+ * @param expErrMsg Expected error message during cancellation.
+ * @param async Execute cancellation in ASYNC mode.
+ * @return <code>IgniteInternalFuture</code> to check whether exception was thrown.
+ */
+ private IgniteInternalFuture cancelQueryWithBarrier(String qry, String expErrMsg, boolean async) {
+ return GridTestUtils.runAsync(() -> {
+ try {
+ List<GridRunningQueryInfo> runningQueries = new ArrayList<>();
+
+ GridTestUtils.waitForCondition(() -> {
+ List<GridRunningQueryInfo> r = (List<GridRunningQueryInfo>)ignite.context().query()
+ .runningQueries(-1);
+
+ runningQueries.addAll(r.stream().filter(q -> q.query().equals(qry)).collect(Collectors.toList()));
+
+ return !runningQueries.isEmpty();
+ }, TIMEOUT);
+
+ assertFalse(runningQueries.isEmpty());
+
+ for (GridRunningQueryInfo runningQuery : runningQueries) {
+ GridTestUtils.assertThrowsAnyCause(log,
+ () -> igniteForKillRequest.cache(DEFAULT_CACHE_NAME).query(createKillQuery(runningQuery.globalQueryId(), async)),
+ CacheException.class, expErrMsg);
+ }
+ }
+ catch (Exception e) {
+ log.error("Unexpected exception.", e);
+
+ Assert.fail("Unexpected exception");
+ }
+ finally {
+ try {
+ awaitTimeout();
+ }
+ catch (Exception e) {
+ log.error("Unexpected exception.", e);
+
+ Assert.fail("Unexpected exception");
+ }
+ }
+ });
+ }
+
+ /**
+ * Cancels current query, actual cancel will wait <code>cancelLatch</code> to be released.
+ *
+ * @return <code>IgniteInternalFuture</code> to check whether exception was thrown.
+ */
+ private IgniteInternalFuture cancel(int expQryNum, boolean async, String... skipSqls) {
+ return GridTestUtils.runAsync(() -> {
+ try {
+ TestSQLFunctions.cancelLatch.await();
+
+ List<GridRunningQueryInfo> runningQueries = (List<GridRunningQueryInfo>)ignite.context().query().runningQueries(-1);
+
+ List<IgniteInternalFuture> res = new ArrayList<>();
+
+ for (GridRunningQueryInfo runningQuery : runningQueries) {
+ if (Stream.of(skipSqls).noneMatch((skipSql -> runningQuery.query().equals(skipSql))))
+ res.add(GridTestUtils.runAsync(() -> {
+ igniteForKillRequest.cache(DEFAULT_CACHE_NAME).query(createKillQuery(runningQuery.globalQueryId(), async));
+ }
+ ));
+ }
+
+ doSleep(500);
+
+ assertEquals(expQryNum, runningQueries.size());
+
+ TestSQLFunctions.reqLatch.countDown();
+
+ for (IgniteInternalFuture fut : res) {
+ fut.get(TIMEOUT);
+ }
+ }
+ catch (Exception e) {
+ log.error("Unexpected exception.", e);
+
+ Assert.fail("Unexpected exception");
+ }
+ });
+ }
+
+ /**
+ * Fills Server Thread Pool with <code>qryCnt</code> queries. Given queries will wait for
+ * <code>suspendQryLatch</code> to be released.
+ *
+ * @param statements Statements.
+ * @param qryCnt Number of queries to execute.
+ * @return <code>IgniteInternalFuture</code> in order to check whether exception was thrown or not.
+ */
+ private IgniteInternalFuture<Long> fillServerThreadPool(List<Statement> statements, int qryCnt) {
+ AtomicInteger idx = new AtomicInteger(0);
+
+ return GridTestUtils.runMultiThreadedAsync(() -> {
+ try {
+ statements.get(idx.getAndIncrement()).executeQuery(
+ "select * from Integer where awaitQuerySuspensionLatch();");
+ }
+ catch (SQLException e) {
+ log.error("Unexpected exception.", e);
+
+ Assert.fail("Unexpected exception");
+ }
+ }, qryCnt, "ThreadName");
+ }
+
+ /**
+ * Create and set new CyclicBarrier for the function.
+ *
+ * @param parties the number of threads that must invoke await method before the barrier is tripped
+ */
+ private static void newBarrier(int parties) {
+ barrier = new CyclicBarrier(parties);
+ }
+
+ /**
+ * @throws InterruptedException In case of failure.
+ * @throws TimeoutException In case of failure.
+ * @throws BrokenBarrierException In case of failure.
+ */
+ private static void awaitTimeout() throws InterruptedException, TimeoutException, BrokenBarrierException {
+ barrier.await(TIMEOUT, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Utility class with custom SQL functions.
+ */
+ public static class TestSQLFunctions {
+ /** Request latch. */
+ static CountDownLatch reqLatch;
+
+ /** Cancel latch. */
+ static CountDownLatch cancelLatch;
+
+ /** Suspend query latch. */
+ static CountDownLatch suspendQryLatch;
+
+ /**
+ * Recreate latches.
+ */
+ static void init() {
+ reqLatch = new CountDownLatch(1);
+
+ cancelLatch = new CountDownLatch(1);
+
+ suspendQryLatch = new CountDownLatch(1);
+ }
+
+ /**
+ * Releases cancelLatch that leeds to sending cancel Query and waits until cancel Query is fully processed.
+ *
+ * @return 0;
+ */
+ @QuerySqlFunction
+ public static long awaitLatchCancelled() {
+ try {
+ cancelLatch.countDown();
+ reqLatch.await();
+ }
+ catch (Exception ignored) {
+ // No-op.
+ }
+
+ return 0;
+ }
+
+ /**
+ * Waits latch release.
+ *
+ * @return 0;
+ */
+ @QuerySqlFunction
+ public static long awaitQuerySuspensionLatch() {
+ try {
+ suspendQryLatch.await();
+ }
+ catch (Exception ignored) {
+ // No-op.
+ }
+
+ return 0;
+ }
+
+ /**
+ * If called fails with corresponding message.
+ *
+ * @return 0;
+ */
+ @QuerySqlFunction
+ public static long shouldNotBeCalledInCaseOfCancellation() {
+ fail("Query wasn't actually cancelled.");
+
+ return 0;
+ }
+
+ /**
+ * @param v amount of milliseconds to sleep
+ * @return amount of milliseconds to sleep
+ */
+ @QuerySqlFunction
+ public static int sleep_func(int v) {
+ try {
+ Thread.sleep(v);
+ }
+ catch (InterruptedException ignored) {
+ // No-op
+ }
+ return v;
+ }
+ }
+
+ /**
+ * Person.
+ */
+ static class Person implements Serializable {
+ /** ID. */
+ @QuerySqlField
+ private final int id;
+
+ /** First name. */
+ @QuerySqlField
+ private final String firstName;
+
+ /** Last name. */
+ @QuerySqlField
+ private final String lastName;
+
+ /** Age. */
+ @QuerySqlField
+ private final int age;
+
+ /**
+ * @param id ID.
+ * @param firstName First name.
+ * @param lastName Last name.
+ * @param age Age.
+ */
+ Person(int id, String firstName, String lastName, int age) {
+ assert !F.isEmpty(firstName);
+ assert !F.isEmpty(lastName);
+ assert age > 0;
+
+ this.id = id;
+ this.firstName = firstName;
+ this.lastName = lastName;
+ this.age = age;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ Person person = (Person)o;
+
+ if (id != person.id)
+ return false;
+ if (age != person.age)
+ return false;
+ if (!Objects.equals(firstName, person.firstName))
+ return false;
+ return Objects.equals(lastName, person.lastName);
+
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ int result = id;
+ result = 31 * result + (firstName != null ? firstName.hashCode() : 0);
+ result = 31 * result + (lastName != null ? lastName.hashCode() : 0);
+ result = 31 * result + age;
+ return result;
+ }
+ }
+}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java
index 33b5de1..6a78621 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java
@@ -196,6 +196,10 @@ import org.apache.ignite.internal.processors.query.IgniteSqlSegmentedIndexSelfTe
import org.apache.ignite.internal.processors.query.IgniteSqlSkipReducerOnUpdateDmlFlagSelfTest;
import org.apache.ignite.internal.processors.query.IgniteSqlSkipReducerOnUpdateDmlSelfTest;
import org.apache.ignite.internal.processors.query.IgniteSqlSplitterSelfTest;
+import org.apache.ignite.internal.processors.query.KillQueryFromClientTest;
+import org.apache.ignite.internal.processors.query.KillQueryFromNeighbourTest;
+import org.apache.ignite.internal.processors.query.KillQueryOnClientDisconnectTest;
+import org.apache.ignite.internal.processors.query.KillQueryTest;
import org.apache.ignite.internal.processors.query.MultipleStatementsSqlQuerySelfTest;
import org.apache.ignite.internal.processors.query.RunningQueriesTest;
import org.apache.ignite.internal.processors.query.SqlIllegalSchemaSelfTest;
@@ -239,6 +243,7 @@ import org.apache.ignite.internal.processors.sql.SqlConnectorConfigurationValida
import org.apache.ignite.internal.sql.SqlParserBulkLoadSelfTest;
import org.apache.ignite.internal.sql.SqlParserCreateIndexSelfTest;
import org.apache.ignite.internal.sql.SqlParserDropIndexSelfTest;
+import org.apache.ignite.internal.sql.SqlParserKillQuerySelfTest;
import org.apache.ignite.internal.sql.SqlParserMultiStatementSelfTest;
import org.apache.ignite.internal.sql.SqlParserSetStreamingSelfTest;
import org.apache.ignite.internal.sql.SqlParserTransactionalKeywordsSelfTest;
@@ -269,6 +274,7 @@ import org.junit.runners.Suite;
SqlParserTransactionalKeywordsSelfTest.class,
SqlParserBulkLoadSelfTest.class,
SqlParserSetStreamingSelfTest.class,
+ SqlParserKillQuerySelfTest.class,
SqlParserMultiStatementSelfTest.class,
SqlConnectorConfigurationValidationSelfTest.class,
@@ -556,7 +562,14 @@ import org.junit.runners.Suite;
SqlQueryHistorySelfTest.class,
SqlQueryHistoryFromClientSelfTest.class,
- SqlIncompatibleDataTypeExceptionTest.class
+ SqlIncompatibleDataTypeExceptionTest.class,
+
+ //Cancellation of queries.
+ KillQueryTest.class,
+ KillQueryFromNeighbourTest.class,
+ KillQueryFromClientTest.class,
+ KillQueryOnClientDisconnectTest.class,
+
})
public class IgniteBinaryCacheQueryTestSuite {
}