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:09 UTC
(ignite-3) branch ignite-23646 created (now 084ad943c6c)
This is an automated email from the ASF dual-hosted git repository.
ppa pushed a change to branch ignite-23646
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
at 084ad943c6c IGNITE-23646 Support cancellation token in thin client (wip).
This branch includes the following new commits:
new 084ad943c6c IGNITE-23646 Support cancellation token in thin client (wip).
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(ignite-3) 01/01: IGNITE-23646 Support cancellation token in thin client (wip).
Posted by pp...@apache.org.
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,