You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by pp...@apache.org on 2025/02/19 12:40:10 UTC
(ignite-3) 01/01: IGNITE-23646 Support cancellation token in thin client (wip).
This is an automated email from the ASF dual-hosted git repository.
ppa pushed a commit to branch ignite-23646
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 084ad943c6cc3ad6b53f4d861beb17a7b1c63fa5
Author: Pavel Pereslegin <xx...@gmail.com>
AuthorDate: Wed Feb 19 15:39:54 2025 +0300
IGNITE-23646 Support cancellation token in thin client (wip).
---
.../ignite/internal/client/proto/ClientOp.java | 3 ++
.../apache/ignite/client/handler/TestServer.java | 4 +-
.../ignite/client/handler/ClientHandlerModule.java | 10 ++++-
.../handler/ClientInboundMessageHandler.java | 17 +++++++-
.../client/handler/ClientQueryIdGenerator.java | 48 +++++++++++++++++++++
.../requests/sql/ClientSqlCancelRequest.java | 49 ++++++++++++++++++++++
.../requests/sql/ClientSqlExecuteRequest.java | 22 ++++++++--
.../ignite/internal/client/sql/ClientSql.java | 41 ++++++++++++++++++
.../ignite/client/TestClientHandlerModule.java | 4 +-
.../java/org/apache/ignite/client/TestServer.java | 4 +-
.../org/apache/ignite/lang/CancelHandleHelper.java | 4 ++
.../org/apache/ignite/internal/app/IgniteImpl.java | 3 +-
.../sql/api/ItSqlClientAsynchronousApiTest.java | 3 --
.../sql/api/ItSqlClientSynchronousApiTest.java | 11 ++---
.../internal/sql/api/ItSqlSynchronousApiTest.java | 4 +-
.../ignite/internal/sql/engine/QueryProperty.java | 4 ++
.../sql/engine/exec/fsm/QueryExecutor.java | 6 ++-
17 files changed, 213 insertions(+), 24 deletions(-)
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientOp.java b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientOp.java
index 30eaa0dd7cd..ac1d6756552 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientOp.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientOp.java
@@ -186,6 +186,9 @@ public class ClientOp {
/** Execute partitioned compute job. */
public static final int COMPUTE_EXECUTE_PARTITIONED = 69;
+ /** Aborts SQL query. */
+ public static final int SQL_CANCEL = 70;
+
/** Reserved for extensions: min. */
@SuppressWarnings("unused")
public static final int RESERVED_EXTENSION_RANGE_START = 1000;
diff --git a/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/TestServer.java b/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/TestServer.java
index a4391f25023..f436244931c 100644
--- a/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/TestServer.java
+++ b/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/TestServer.java
@@ -48,6 +48,7 @@ import org.apache.ignite.internal.security.configuration.SecurityConfiguration;
import org.apache.ignite.internal.sql.engine.QueryProcessor;
import org.apache.ignite.internal.table.IgniteTablesInternal;
import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.util.CompletableFutures;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.TestInfo;
import org.mockito.Mockito;
@@ -152,7 +153,8 @@ public class TestServer {
mock(PlacementDriver.class),
clientConnectorConfiguration,
new TestLowWatermark(),
- Runnable::run
+ Runnable::run,
+ id -> CompletableFutures.falseCompletedFuture()
);
module.startAsync(componentContext).join();
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java
index c1c677d3e8d..e07ff1bfcca 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java
@@ -41,6 +41,7 @@ import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Function;
import java.util.function.Supplier;
import org.apache.ignite.client.handler.configuration.ClientConnectorConfiguration;
import org.apache.ignite.client.handler.configuration.ClientConnectorView;
@@ -130,6 +131,8 @@ public class ClientHandlerModule implements IgniteComponent {
private final Executor partitionOperationsExecutor;
+ private final Function<String, CompletableFuture<Boolean>> sqlKillHandler;
+
@TestOnly
@SuppressWarnings("unused")
private volatile ClientInboundMessageHandler handler;
@@ -168,7 +171,8 @@ public class ClientHandlerModule implements IgniteComponent {
PlacementDriver placementDriver,
ClientConnectorConfiguration clientConnectorConfiguration,
LowWatermark lowWatermark,
- Executor partitionOperationsExecutor
+ Executor partitionOperationsExecutor,
+ Function<String, CompletableFuture<Boolean>> sqlKillHandler
) {
assert igniteTables != null;
assert queryProcessor != null;
@@ -205,6 +209,7 @@ public class ClientHandlerModule implements IgniteComponent {
lowWatermark);
this.clientConnectorConfiguration = clientConnectorConfiguration;
this.partitionOperationsExecutor = partitionOperationsExecutor;
+ this.sqlKillHandler = sqlKillHandler;
}
/** {@inheritDoc} */
@@ -407,7 +412,8 @@ public class ClientHandlerModule implements IgniteComponent {
primaryReplicaTracker,
partitionOperationsExecutor,
HandshakeUtils.EMPTY_FEATURES,
- Map.of()
+ Map.of(),
+ sqlKillHandler
);
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
index c71967e5196..e57ebb74af3 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
@@ -38,6 +38,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
+import java.util.function.Function;
import java.util.function.Supplier;
import javax.net.ssl.SSLException;
import org.apache.ignite.client.handler.configuration.ClientConnectorView;
@@ -63,6 +64,7 @@ import org.apache.ignite.client.handler.requests.jdbc.ClientJdbcPrimaryKeyMetada
import org.apache.ignite.client.handler.requests.jdbc.ClientJdbcSchemasMetadataRequest;
import org.apache.ignite.client.handler.requests.jdbc.ClientJdbcTableMetadataRequest;
import org.apache.ignite.client.handler.requests.jdbc.JdbcMetadataCatalog;
+import org.apache.ignite.client.handler.requests.sql.ClientSqlCancelRequest;
import org.apache.ignite.client.handler.requests.sql.ClientSqlCursorCloseRequest;
import org.apache.ignite.client.handler.requests.sql.ClientSqlCursorNextPageRequest;
import org.apache.ignite.client.handler.requests.sql.ClientSqlExecuteBatchRequest;
@@ -220,6 +222,10 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
private final Map<HandshakeExtension, Object> extensions;
+ private final ClientQueryIdGenerator queryIdGenerator;
+
+ private final Function<String, CompletableFuture<Boolean>> sqlKillHandler;
+
/**
* Constructor.
*
@@ -257,7 +263,8 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
ClientPrimaryReplicaTracker primaryReplicaTracker,
Executor partitionOperationsExecutor,
BitSet features,
- Map<HandshakeExtension, Object> extensions
+ Map<HandshakeExtension, Object> extensions,
+ Function<String, CompletableFuture<Boolean>> sqlKillHandler
) {
assert igniteTables != null;
assert txManager != null;
@@ -304,6 +311,9 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
this.features = features;
this.extensions = extensions;
+
+ this.queryIdGenerator = new ClientQueryIdGenerator(clusterService.nodeName(), connectionId);
+ this.sqlKillHandler = sqlKillHandler;
}
@Override
@@ -817,7 +827,10 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
return ClientClusterGetNodesRequest.process(out, clusterService);
case ClientOp.SQL_EXEC:
- return ClientSqlExecuteRequest.process(in, out, queryProcessor, resources, metrics);
+ return ClientSqlExecuteRequest.process(in, out, queryProcessor, resources, metrics, queryIdGenerator);
+
+ case ClientOp.SQL_CANCEL:
+ return ClientSqlCancelRequest.process(in, out, sqlKillHandler, queryIdGenerator);
case ClientOp.SQL_CURSOR_NEXT_PAGE:
return ClientSqlCursorNextPageRequest.process(in, out, resources);
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientQueryIdGenerator.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientQueryIdGenerator.java
new file mode 100644
index 00000000000..f9313aa3a42
--- /dev/null
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientQueryIdGenerator.java
@@ -0,0 +1,48 @@
+/*
+ * 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.client.handler;
+
+import java.util.UUID;
+
+/**
+ * Generates client query IDs.
+ * TODO more details
+ */
+public class ClientQueryIdGenerator {
+ private final long mostSigBits;
+ private final long connectionId;
+
+ ClientQueryIdGenerator(String nodeName, long connectionId) {
+ this.mostSigBits = 1L << 32 | nodeName.hashCode();
+ this.connectionId = toIntWithOverflow(connectionId);
+ }
+
+ public UUID generate(int requestNumber) {
+ return new UUID(mostSigBits, (long) requestNumber << 32 | connectionId);
+ }
+
+ private static long toIntWithOverflow(long value) {
+ assert value >= 0;
+
+ if (value > Integer.MAX_VALUE) {
+ return value % Integer.MAX_VALUE;
+ }
+
+ return value;
+ }
+}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlCancelRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlCancelRequest.java
new file mode 100644
index 00000000000..131951800cd
--- /dev/null
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlCancelRequest.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.client.handler.requests.sql;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import org.apache.ignite.client.handler.ClientQueryIdGenerator;
+import org.apache.ignite.internal.client.proto.ClientMessagePacker;
+import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.util.CompletableFutures;
+
+/**
+ * Request ot abort running SQL query.
+ */
+public class ClientSqlCancelRequest {
+ /**
+ * Processes the request.
+ */
+ public static CompletableFuture<Void> process(
+ ClientMessageUnpacker in,
+ ClientMessagePacker out,
+ Function<String, CompletableFuture<Boolean>> sqlKillHandler,
+ ClientQueryIdGenerator queryIdGenerator
+ ) {
+ UUID queryId = queryIdGenerator.generate(in.unpackInt());
+
+ return sqlKillHandler.apply(queryId.toString()).thenCompose(res -> {
+ out.packBoolean(res);
+
+ return CompletableFutures.nullCompletedFuture();
+ });
+ }
+}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteRequest.java
index 6b45bb1d7de..44752bd32ef 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteRequest.java
@@ -22,10 +22,12 @@ import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.
import static org.apache.ignite.internal.lang.SqlExceptionMapperUtil.mapToPublicSqlException;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.CompletionStage;
import org.apache.ignite.client.handler.ClientHandlerMetricSource;
+import org.apache.ignite.client.handler.ClientQueryIdGenerator;
import org.apache.ignite.client.handler.ClientResource;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
@@ -39,6 +41,7 @@ import org.apache.ignite.internal.sql.engine.QueryProcessor;
import org.apache.ignite.internal.sql.engine.QueryProperty;
import org.apache.ignite.internal.sql.engine.SqlQueryType;
import org.apache.ignite.internal.sql.engine.property.SqlProperties;
+import org.apache.ignite.internal.sql.engine.property.SqlProperties.Builder;
import org.apache.ignite.internal.sql.engine.property.SqlPropertiesHelper;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.util.ArrayUtils;
@@ -69,7 +72,8 @@ public class ClientSqlExecuteRequest {
ClientMessagePacker out,
QueryProcessor sql,
ClientResourceRegistry resources,
- ClientHandlerMetricSource metrics
+ ClientHandlerMetricSource metrics,
+ ClientQueryIdGenerator queryIdGenerator
) {
InternalTransaction tx = readTx(in, out, resources);
ClientSqlProperties props = new ClientSqlProperties(in);
@@ -83,9 +87,14 @@ public class ClientSqlExecuteRequest {
HybridTimestamp clientTs = HybridTimestamp.nullableHybridTimestamp(in.unpackLong());
+ // TODO Feature check
+ int queryNum = in.unpackInt();
+
+ UUID queryId = queryIdGenerator.generate(queryNum);
+
HybridTimestampTracker tsUpdater = HybridTimestampTracker.atomicTracker(clientTs);
- return executeAsync(tx, sql, tsUpdater, statement, props.pageSize(), props.toSqlProps(), arguments)
+ return executeAsync(tx, sql, tsUpdater, queryId, statement, props.pageSize(), props.toSqlProps(), arguments)
.thenCompose(asyncResultSet -> {
out.meta(tsUpdater.get());
@@ -155,13 +164,20 @@ public class ClientSqlExecuteRequest {
@Nullable Transaction transaction,
QueryProcessor qryProc,
HybridTimestampTracker timestampTracker,
+ @Nullable UUID queryId,
String query,
int pageSize,
SqlProperties props,
@Nullable Object... arguments
) {
try {
- SqlProperties properties = SqlPropertiesHelper.builderFromProperties(props)
+ Builder builder = SqlPropertiesHelper.builderFromProperties(props);
+
+ if (queryId != null) {
+ builder.set(QueryProperty.QUERY_ID, queryId);
+ }
+
+ SqlProperties properties = builder
.set(QueryProperty.ALLOWED_QUERY_TYPES, SqlQueryType.SINGLE_STMT_TYPES)
.build();
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/sql/ClientSql.java b/modules/client/src/main/java/org/apache/ignite/internal/client/sql/ClientSql.java
index 44f68576c07..a9a6e970ddf 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/sql/ClientSql.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/sql/ClientSql.java
@@ -27,6 +27,7 @@ import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.ignite.internal.binarytuple.BinaryTupleBuilder;
import org.apache.ignite.internal.client.PayloadOutputChannel;
import org.apache.ignite.internal.client.PayloadReader;
@@ -41,7 +42,9 @@ import org.apache.ignite.internal.sql.StatementBuilderImpl;
import org.apache.ignite.internal.sql.StatementImpl;
import org.apache.ignite.internal.sql.SyncResultSetAdapter;
import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.lang.CancelHandleHelper;
import org.apache.ignite.lang.CancellationToken;
+import org.apache.ignite.lang.ErrorGroups.Sql;
import org.apache.ignite.sql.BatchedArguments;
import org.apache.ignite.sql.IgniteSql;
import org.apache.ignite.sql.ResultSet;
@@ -67,6 +70,8 @@ public class ClientSql implements IgniteSql {
/** Marshallers provider. */
private final MarshallersProvider marshallers;
+ private static final AtomicInteger sqlRequestCounter = new AtomicInteger();
+
/**
* Constructor.
*
@@ -261,6 +266,28 @@ public class ClientSql implements IgniteSql {
w.out().packObjectArrayAsBinaryTuple(arguments);
w.out().packLong(ch.observableTimestamp());
+
+ int queryRequestNumber = sqlRequestCounter.getAndIncrement();
+
+ // TODO feature check
+ w.out().packInt(queryRequestNumber);
+
+ if (cancellationToken != null) {
+ CompletableFuture<Boolean> cancelFuture = new CompletableFuture<>();
+
+ // TODO sync properly, clean resources
+ if (CancelHandleHelper.isCancelled(cancellationToken)) {
+ throw new SqlException(Sql.EXECUTION_CANCELLED_ERR, "The query was cancelled while executing.");
+ }
+
+ CancelHandleHelper.addCancelAction(cancellationToken, () -> cancelQuery(queryRequestNumber).whenComplete((r, e) -> {
+ if (e != null) {
+ cancelFuture.completeExceptionally(e);
+ } else {
+ cancelFuture.complete(r);
+ }
+ }), cancelFuture);
+ }
};
PayloadReader<AsyncResultSet<T>> payloadReader = r -> new ClientAsyncResultSet<>(r.clientChannel(), marshallers, r.in(), mapper);
@@ -350,6 +377,20 @@ public class ClientSql implements IgniteSql {
throw new UnsupportedOperationException();
}
+ private CompletableFuture<Boolean> cancelQuery(int queryNum) {
+ PayloadWriter payloadWriter = w -> {
+ w.out().packInt(queryNum);
+ };
+
+ PayloadReader<Boolean> payloadReader = r -> {
+ ClientMessageUnpacker unpacker = r.in();
+
+ return unpacker.unpackBoolean();
+ };
+
+ return ch.serviceAsync(ClientOp.SQL_CANCEL, payloadWriter, payloadReader);
+ }
+
private static void packProperties(
PayloadOutputChannel w,
@Nullable Map<String, Object> statementProps) {
diff --git a/modules/client/src/test/java/org/apache/ignite/client/TestClientHandlerModule.java b/modules/client/src/test/java/org/apache/ignite/client/TestClientHandlerModule.java
index dba12755c0d..6f555167b47 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/TestClientHandlerModule.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/TestClientHandlerModule.java
@@ -63,6 +63,7 @@ import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.schema.AlwaysSyncedSchemaSyncService;
import org.apache.ignite.internal.security.authentication.AuthenticationManager;
import org.apache.ignite.internal.table.IgniteTablesInternal;
+import org.apache.ignite.internal.util.CompletableFutures;
import org.apache.ignite.lang.IgniteException;
import org.jetbrains.annotations.Nullable;
@@ -243,7 +244,8 @@ public class TestClientHandlerModule implements IgniteComponent {
),
Runnable::run,
BitSet.valueOf(new long[]{ThreadLocalRandom.current().nextLong()}),
- randomExtensions()
+ randomExtensions(),
+ id -> CompletableFutures.falseCompletedFuture()
)
);
}
diff --git a/modules/client/src/test/java/org/apache/ignite/client/TestServer.java b/modules/client/src/test/java/org/apache/ignite/client/TestServer.java
index c61dd13eb0a..1462e06d7c3 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/TestServer.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/TestServer.java
@@ -76,6 +76,7 @@ import org.apache.ignite.internal.security.authentication.AuthenticationManager;
import org.apache.ignite.internal.security.authentication.AuthenticationManagerImpl;
import org.apache.ignite.internal.security.configuration.SecurityConfiguration;
import org.apache.ignite.internal.table.IgniteTablesInternal;
+import org.apache.ignite.internal.util.CompletableFutures;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.NetworkAddress;
import org.jetbrains.annotations.Nullable;
@@ -274,7 +275,8 @@ public class TestServer implements AutoCloseable {
ignite.placementDriver(),
clientConnectorConfiguration,
new TestLowWatermark(),
- Runnable::run
+ Runnable::run,
+ id -> CompletableFutures.falseCompletedFuture()
);
module.startAsync(componentContext).join();
diff --git a/modules/core/src/main/java/org/apache/ignite/lang/CancelHandleHelper.java b/modules/core/src/main/java/org/apache/ignite/lang/CancelHandleHelper.java
index f10477d1b43..2773ee97532 100644
--- a/modules/core/src/main/java/org/apache/ignite/lang/CancelHandleHelper.java
+++ b/modules/core/src/main/java/org/apache/ignite/lang/CancelHandleHelper.java
@@ -78,6 +78,10 @@ public final class CancelHandleHelper {
addCancelAction(token, () -> completionFut.cancel(true), completionFut);
}
+ public static boolean isCancelled(CancellationToken token) {
+ return unwrapToken(token).isCancelled();
+ }
+
private static CancellationTokenImpl unwrapToken(CancellationToken token) {
if (token instanceof CancellationTokenImpl) {
return (CancellationTokenImpl) token;
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
index bcb8444749f..49134305ca7 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -1206,7 +1206,8 @@ public class IgniteImpl implements Ignite {
placementDriverMgr.placementDriver(),
clientConnectorConfiguration,
lowWatermark,
- threadPoolsManager.partitionOperationsExecutor()
+ threadPoolsManager.partitionOperationsExecutor(),
+ id -> killCommandHandler.handler(CancellableOperationType.QUERY).cancelAsync(id)
);
metricMessaging = new MetricMessaging(metricManager, clusterSvc.messagingService(), clusterSvc.topologyService());
diff --git a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlClientAsynchronousApiTest.java b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlClientAsynchronousApiTest.java
index 0df393fa2a4..5b58287b39b 100644
--- a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlClientAsynchronousApiTest.java
+++ b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlClientAsynchronousApiTest.java
@@ -43,14 +43,12 @@ public class ItSqlClientAsynchronousApiTest extends ItSqlAsynchronousApiTest {
}
@Test
- @Disabled("https://issues.apache.org/jira/browse/IGNITE-23646")
@Override
public void cancelQueryString() throws InterruptedException {
super.cancelQueryString();
}
@Test
- @Disabled("https://issues.apache.org/jira/browse/IGNITE-23646")
@Override
public void cancelStatement() throws InterruptedException {
super.cancelStatement();
@@ -64,7 +62,6 @@ public class ItSqlClientAsynchronousApiTest extends ItSqlAsynchronousApiTest {
}
@Test
- @Disabled("https://issues.apache.org/jira/browse/IGNITE-23646")
@Override
public void cancelLongRunningStatement() throws InterruptedException {
super.cancelLongRunningStatement();
diff --git a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlClientSynchronousApiTest.java b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlClientSynchronousApiTest.java
index abad0b274a5..3304fd9869b 100644
--- a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlClientSynchronousApiTest.java
+++ b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlClientSynchronousApiTest.java
@@ -43,17 +43,15 @@ public class ItSqlClientSynchronousApiTest extends ItSqlSynchronousApiTest {
}
@Test
- @Disabled("https://issues.apache.org/jira/browse/IGNITE-23646")
@Override
- public void cancelQueryString() throws InterruptedException {
- super.cancelQueryString();
+ public void cancelStatement() throws InterruptedException {
+ super.cancelStatement();
}
@Test
- @Disabled("https://issues.apache.org/jira/browse/IGNITE-23646")
@Override
- public void cancelStatement() throws InterruptedException {
- super.cancelStatement();
+ public void cancelQueryString() throws InterruptedException {
+ super.cancelQueryString();
}
@Test
@@ -64,7 +62,6 @@ public class ItSqlClientSynchronousApiTest extends ItSqlSynchronousApiTest {
}
@Test
- @Disabled("https://issues.apache.org/jira/browse/IGNITE-23646")
@Override
public void cancelLongRunningStatement() throws InterruptedException {
super.cancelLongRunningStatement();
diff --git a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlSynchronousApiTest.java b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlSynchronousApiTest.java
index 609d79daa22..59ef1484502 100644
--- a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlSynchronousApiTest.java
+++ b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlSynchronousApiTest.java
@@ -58,7 +58,7 @@ public class ItSqlSynchronousApiTest extends ItSqlApiBaseTest {
}
@Test
- public void cancelQueryString() throws InterruptedException {
+ public void cancelStatement() throws InterruptedException {
IgniteSql sql = igniteSql();
String query = "SELECT * FROM system_range(0, 10000000000)";
@@ -94,7 +94,7 @@ public class ItSqlSynchronousApiTest extends ItSqlApiBaseTest {
}
@Test
- public void cancelStatement() throws InterruptedException {
+ public void cancelQueryString() throws InterruptedException {
IgniteSql sql = igniteSql();
String query = "SELECT * FROM system_range(0, 10000000000)";
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/QueryProperty.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/QueryProperty.java
index e04b2d9267b..44e373f3771 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/QueryProperty.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/QueryProperty.java
@@ -22,6 +22,7 @@ import static org.apache.ignite.internal.sql.engine.property.SqlPropertiesHelper
import java.time.ZoneId;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
import org.apache.ignite.internal.sql.engine.property.Property;
import org.jetbrains.annotations.Nullable;
@@ -35,6 +36,9 @@ public final class QueryProperty {
public static final Property<String> DEFAULT_SCHEMA = new Property<>("default_schema", String.class);
public static final Property<ZoneId> TIME_ZONE_ID = new Property<>("time_zone_id", ZoneId.class);
+ /** Externally generated query identifier. */
+ public static final Property<UUID> QUERY_ID = new Property<>("query_id", cast(UUID.class));
+
private static final Map<String, Property<?>> propsByName = createPropsByNameMap(QueryProperty.class);
/** Returns a property for the given name or {@code null} if there is no property with such name. */
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/QueryExecutor.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/QueryExecutor.java
index f00847bb84c..20f1fdb3cfd 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/QueryExecutor.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/QueryExecutor.java
@@ -175,10 +175,14 @@ public class QueryExecutor implements LifecycleAware {
) {
SqlProperties properties0 = SqlPropertiesHelper.chain(properties, defaultProperties);
+ UUID queryId = properties0.hasProperty(QueryProperty.QUERY_ID)
+ ? properties0.get(QueryProperty.QUERY_ID)
+ : idGenerator.next();
+
Query query = new Query(
Instant.ofEpochMilli(clockService.now().getPhysical()),
this,
- idGenerator.next(),
+ queryId,
sql,
properties0,
txContext,