You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ko...@apache.org on 2024/12/18 10:09:48 UTC
(ignite-3) branch main updated: IGNITE-24005 Sql. Rollback implicit tx when query is timed out (#4907)
This is an automated email from the ASF dual-hosted git repository.
korlov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new fd16105e20 IGNITE-24005 Sql. Rollback implicit tx when query is timed out (#4907)
fd16105e20 is described below
commit fd16105e20f2aa1ea7a4d1e9a9a706368aeb3b5a
Author: korlov42 <ko...@apache.org>
AuthorDate: Wed Dec 18 12:09:43 2024 +0200
IGNITE-24005 Sql. Rollback implicit tx when query is timed out (#4907)
---
.../handler/requests/jdbc/JdbcQueryCursor.java | 4 +-
.../org/apache/ignite/client/fakes/FakeCursor.java | 2 +-
.../apache/ignite/internal/util/AsyncCursor.java | 12 +---
.../apache/ignite/internal/util/AsyncWrapper.java | 2 +-
modules/jdbc/build.gradle | 1 +
.../apache/ignite/jdbc/ItJdbcBatchSelfTest.java | 12 ++++
.../internal/sql/engine/AsyncSqlCursorImpl.java | 4 +-
...ecutionService.java => AsyncDataCursorExt.java} | 31 +++++-----
.../internal/sql/engine/exec/ExecutionService.java | 2 +-
.../sql/engine/exec/ExecutionServiceImpl.java | 65 +++++++++------------
.../sql/engine/exec/TxAwareAsyncCursor.java | 68 +++++++++-------------
.../exec/fsm/CursorInitializationPhaseHandler.java | 9 ++-
.../sql/engine/exec/fsm/QueryExecutor.java | 4 +-
.../sql/engine/exec/rel/AsyncRootNode.java | 8 +--
.../engine/tx/ScriptTransactionWrapperImpl.java | 5 +-
.../engine/util/IteratorToDataCursorAdapter.java | 8 ++-
.../sql/engine/exec/TxAwareCursorSelfTest.java | 9 ++-
17 files changed, 114 insertions(+), 132 deletions(-)
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/JdbcQueryCursor.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/JdbcQueryCursor.java
index 07058917bb..2356200da9 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/JdbcQueryCursor.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/JdbcQueryCursor.java
@@ -76,8 +76,8 @@ public class JdbcQueryCursor<T> implements AsyncSqlCursor<T> {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> closeAsync(boolean cancelled) {
- return cur.closeAsync(cancelled);
+ public CompletableFuture<Void> closeAsync() {
+ return cur.closeAsync();
}
/** {@inheritDoc} */
diff --git a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeCursor.java b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeCursor.java
index 887994da9e..377b44306c 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeCursor.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeCursor.java
@@ -110,7 +110,7 @@ public class FakeCursor implements AsyncSqlCursor<InternalSqlRow> {
}
@Override
- public CompletableFuture<Void> closeAsync(boolean cancelled) {
+ public CompletableFuture<Void> closeAsync() {
return nullCompletedFuture();
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/AsyncCursor.java b/modules/core/src/main/java/org/apache/ignite/internal/util/AsyncCursor.java
index 02b52df82d..1d0c6c5505 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/AsyncCursor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/AsyncCursor.java
@@ -45,17 +45,7 @@ public interface AsyncCursor<T> {
*
* @return A future which will be completed when the resources will be actually released.
*/
- default CompletableFuture<Void> closeAsync() {
- return closeAsync(false);
- }
-
- /**
- * Releases resources acquired by the cursor.
- *
- * @param cancelled Whether the cursor is closed due to query cancellation.
- * @return A future which will be completed when the resources will be actually released.
- */
- CompletableFuture<Void> closeAsync(boolean cancelled);
+ CompletableFuture<Void> closeAsync();
/**
* Batch of the items.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/AsyncWrapper.java b/modules/core/src/main/java/org/apache/ignite/internal/util/AsyncWrapper.java
index 3c5f235f70..f14fc9b43f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/AsyncWrapper.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/AsyncWrapper.java
@@ -120,7 +120,7 @@ public class AsyncWrapper<T> implements AsyncCursor<T> {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> closeAsync(boolean ignore) {
+ public CompletableFuture<Void> closeAsync() {
if (!cancelled) {
synchronized (lock) {
if (!cancelled) {
diff --git a/modules/jdbc/build.gradle b/modules/jdbc/build.gradle
index 5f5d993a33..a92f9774a7 100644
--- a/modules/jdbc/build.gradle
+++ b/modules/jdbc/build.gradle
@@ -46,6 +46,7 @@ dependencies {
integrationTestImplementation testFixtures(project(":ignite-sql-engine"))
integrationTestImplementation testFixtures(project(':ignite-jdbc'))
integrationTestImplementation testFixtures(project(':ignite-runner'))
+ integrationTestImplementation project(':ignite-transactions')
integrationTestImplementation project(":ignite-runner")
integrationTestImplementation project(":ignite-api")
integrationTestImplementation project(":ignite-system-view-api")
diff --git a/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcBatchSelfTest.java b/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcBatchSelfTest.java
index 138666a5a4..96f4a35ac4 100644
--- a/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcBatchSelfTest.java
+++ b/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcBatchSelfTest.java
@@ -39,10 +39,14 @@ import java.time.LocalDate;
import java.time.LocalTime;
import java.util.Arrays;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.app.IgniteImpl;
import org.apache.ignite.internal.jdbc.JdbcPreparedStatement;
import org.apache.ignite.internal.jdbc.JdbcStatement;
import org.apache.ignite.internal.jdbc.proto.IgniteQueryErrorCode;
import org.apache.ignite.internal.jdbc.proto.SqlStateCode;
+import org.apache.ignite.internal.restart.RestartProofIgnite;
+import org.apache.ignite.internal.tx.TxManager;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
@@ -104,6 +108,14 @@ public class ItJdbcBatchSelfTest extends AbstractJdbcSelfTest {
}
assertTrue(pstmt.isClosed());
+
+ long countOfPendingTransactions = CLUSTER.runningNodes()
+ .map(RestartProofIgnite.class::cast)
+ .map(rpi -> rpi.unwrap(IgniteImpl.class).txManager())
+ .collect(Collectors.summarizingInt(TxManager::pending))
+ .getSum();
+
+ assertEquals(0, countOfPendingTransactions);
}
@Test
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/AsyncSqlCursorImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/AsyncSqlCursorImpl.java
index e10735215a..c4dcfcb14c 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/AsyncSqlCursorImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/AsyncSqlCursorImpl.java
@@ -91,8 +91,8 @@ public class AsyncSqlCursorImpl<T> implements AsyncSqlCursor<T> {
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> closeAsync(boolean cancelled) {
- return dataCursor.closeAsync(cancelled);
+ public CompletableFuture<Void> closeAsync() {
+ return dataCursor.closeAsync();
}
/** {@inheritDoc} */
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionService.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/AsyncDataCursorExt.java
similarity index 54%
copy from modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionService.java
copy to modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/AsyncDataCursorExt.java
index 5c08a895d6..588d7eae82 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionService.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/AsyncDataCursorExt.java
@@ -18,23 +18,18 @@
package org.apache.ignite.internal.sql.engine.exec;
import java.util.concurrent.CompletableFuture;
-import org.apache.ignite.internal.sql.engine.InternalSqlRow;
-import org.apache.ignite.internal.sql.engine.SqlOperationContext;
-import org.apache.ignite.internal.sql.engine.prepare.QueryPlan;
-/**
- * SQL query plan execution interface.
- */
-public interface ExecutionService extends LifecycleAware {
- /**
- * Executes the given plan.
- *
- * @param plan Plan to execute.
- * @param operationContext Context of operation.
- * @return Future that will be completed when cursor is successfully initialized, implying for distributed plans all fragments have been
- * sent successfully.
- */
- CompletableFuture<AsyncDataCursor<InternalSqlRow>> executePlan(
- QueryPlan plan, SqlOperationContext operationContext
- );
+/** Extension over {@link AsyncDataCursor} which gives an ability to specify a reason of cancellation. */
+public interface AsyncDataCursorExt<T> extends AsyncDataCursor<T> {
+ CompletableFuture<Void> cancelAsync(CancellationReason reason);
+
+ /** Reason of cancellation. */
+ enum CancellationReason {
+ /** User's request. This reason denotes normal completion of an execution. */
+ CLOSE,
+ /** Execution must be terminated abruptly due to an external request or an error. */
+ CANCEL,
+ /** Execution must be terminated abruptly due to specified timeout. */
+ TIMEOUT
+ }
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionService.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionService.java
index 5c08a895d6..454ca87b80 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionService.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionService.java
@@ -34,7 +34,7 @@ public interface ExecutionService extends LifecycleAware {
* @return Future that will be completed when cursor is successfully initialized, implying for distributed plans all fragments have been
* sent successfully.
*/
- CompletableFuture<AsyncDataCursor<InternalSqlRow>> executePlan(
+ CompletableFuture<AsyncDataCursorExt<InternalSqlRow>> executePlan(
QueryPlan plan, SqlOperationContext operationContext
);
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
index 189be37a10..bc40d88a52 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.sql.engine.exec;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.sql.engine.externalize.RelJsonReader.fromJson;
+import static org.apache.ignite.internal.sql.engine.util.Commons.cast;
import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.ExceptionUtils.sneakyThrow;
@@ -80,6 +81,7 @@ import org.apache.ignite.internal.sql.engine.QueryCancelledException;
import org.apache.ignite.internal.sql.engine.SqlOperationContext;
import org.apache.ignite.internal.sql.engine.SqlQueryProcessor.PrefetchCallback;
import org.apache.ignite.internal.sql.engine.SqlQueryType;
+import org.apache.ignite.internal.sql.engine.exec.AsyncDataCursorExt.CancellationReason;
import org.apache.ignite.internal.sql.engine.exec.ddl.DdlCommandHandler;
import org.apache.ignite.internal.sql.engine.exec.exp.func.TableFunctionRegistry;
import org.apache.ignite.internal.sql.engine.exec.kill.KillCommand;
@@ -303,7 +305,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
return ddlCmdHnd;
}
- private CompletableFuture<AsyncDataCursor<InternalSqlRow>> executeQuery(
+ private CompletableFuture<AsyncDataCursorExt<InternalSqlRow>> executeQuery(
SqlOperationContext operationContext,
MultiStepPlan plan
) {
@@ -320,7 +322,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
// This call immediately triggers a cancellation exception if operation has timed out or it has already been cancelled.
cancelHandler.add(timeout -> {
- QueryCompletionReason reason = timeout ? QueryCompletionReason.TIMEOUT : QueryCompletionReason.CANCEL;
+ CancellationReason reason = timeout ? CancellationReason.TIMEOUT : CancellationReason.CANCEL;
queryManager.close(reason);
});
@@ -348,12 +350,14 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
Predicate<String> nodeExclusionFilter = operationContext.nodeExclusionFilter();
- CompletableFuture<AsyncDataCursor<InternalSqlRow>> f = queryManager.execute(tx, plan,
+ CompletableFuture<AsyncDataCursorExt<InternalSqlRow>> f = queryManager.execute(tx, plan,
nodeExclusionFilter).thenApply(dataCursor -> new TxAwareAsyncCursor<>(
txWrapper,
dataCursor,
- firstPageReady0
+ firstPageReady0,
+ queryManager::close
));
+
return f.whenComplete((r, t) -> {
if (t != null) {
txWrapper.rollback(t);
@@ -384,7 +388,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
/** {@inheritDoc} */
@Override
@SuppressWarnings("CastConflictsWithInstanceof") // IDEA incorrectly highlights casts in EXPLAIN and DDL branches
- public CompletableFuture<AsyncDataCursor<InternalSqlRow>> executePlan(
+ public CompletableFuture<AsyncDataCursorExt<InternalSqlRow>> executePlan(
QueryPlan plan, SqlOperationContext operationContext
) {
SqlQueryType queryType = plan.type();
@@ -393,18 +397,18 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
case DML:
case QUERY:
if (plan instanceof ExecutablePlan) {
- return completedFuture(executeExecutablePlan(operationContext, (ExecutablePlan) plan));
+ return cast(completedFuture(executeExecutablePlan(operationContext, (ExecutablePlan) plan)));
}
assert plan instanceof MultiStepPlan : plan.getClass();
return executeQuery(operationContext, (MultiStepPlan) plan);
case EXPLAIN:
- return completedFuture(executeExplain((ExplainPlan) plan));
+ return cast(completedFuture(executeExplain((ExplainPlan) plan)));
case DDL:
- return completedFuture(executeDdl(operationContext, (DdlPlan) plan));
+ return cast(completedFuture(executeDdl(operationContext, (DdlPlan) plan)));
case KILL:
- return completedFuture(executeKill(operationContext, (KillPlan) plan));
+ return cast(completedFuture(executeKill(operationContext, (KillPlan) plan)));
default:
throw new AssertionError("Unexpected query type: " + plan);
@@ -451,7 +455,8 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
return new TxAwareAsyncCursor<>(
txWrapper,
dataCursor,
- prefetchCallback.prefetchFuture()
+ prefetchCallback.prefetchFuture(),
+ reason -> nullCompletedFuture()
);
}
@@ -591,7 +596,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
DistributedQueryManager dqm = queryManagerMap.get(new ExecutionId(msg.queryId(), msg.executionToken()));
if (dqm != null) {
- dqm.close(QueryCompletionReason.CANCEL);
+ dqm.close(CancellationReason.CANCEL);
}
}
@@ -600,7 +605,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
public void stop() throws Exception {
CompletableFuture<Void> f = CompletableFuture.allOf(queryManagerMap.values().stream()
.filter(mgr -> mgr.rootFragmentId != null)
- .map(mgr -> mgr.close(QueryCompletionReason.CANCEL))
+ .map(mgr -> mgr.close(CancellationReason.CANCEL))
.toArray(CompletableFuture[]::new)
);
@@ -825,7 +830,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
var root = new CompletableFuture<AsyncRootNode<RowT, InternalSqlRow>>();
root.exceptionally(t -> {
- this.close(QueryCompletionReason.ERROR);
+ this.close(CancellationReason.CANCEL);
return null;
});
@@ -876,7 +881,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
root.thenAccept(root -> {
root.onError(ex);
- close(QueryCompletionReason.ERROR);
+ close(CancellationReason.CANCEL);
});
}
}
@@ -888,7 +893,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
root.thenAccept(root -> {
root.onError(ex);
- close(QueryCompletionReason.ERROR);
+ close(CancellationReason.CANCEL);
});
}
@@ -995,7 +1000,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
} catch (Exception e) {
LOG.info("Unable to send error message", e);
- close(QueryCompletionReason.ERROR);
+ close(CancellationReason.CANCEL);
}
}
@@ -1117,7 +1122,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
if (!root.completeExceptionally(error)) {
root.thenAccept(root -> root.onError(error));
- close(QueryCompletionReason.ERROR);
+ close(CancellationReason.CANCEL);
}
return cancelFut
@@ -1189,7 +1194,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
}.visit(mappedFragment.fragment().root());
}
- private CompletableFuture<Void> close(QueryCompletionReason reason) {
+ private CompletableFuture<Void> close(CancellationReason reason) {
if (!cancelled.compareAndSet(false, true)) {
return cancelFut;
}
@@ -1290,11 +1295,11 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
* @param closeReason Reason to use in {@link QueryCancelledException}.
* @return Completable future that should run asynchronously.
*/
- private CompletableFuture<Void> closeRootNode(QueryCompletionReason closeReason) {
+ private CompletableFuture<Void> closeRootNode(CancellationReason closeReason) {
assert root != null;
String message;
- if (closeReason == QueryCompletionReason.TIMEOUT) {
+ if (closeReason == CancellationReason.TIMEOUT) {
message = QueryCancelledException.TIMEOUT_MSG;
} else {
message = QueryCancelledException.CANCEL_MSG;
@@ -1307,7 +1312,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
if (!root.isCompletedExceptionally()) {
AsyncRootNode<RowT, InternalSqlRow> node = root.getNow(null);
- if (closeReason != QueryCompletionReason.CLOSE) {
+ if (closeReason != CancellationReason.CLOSE) {
node.onError(new QueryCancelledException(message));
}
@@ -1325,7 +1330,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
fut.thenAccept(batch -> {
if (!batch.hasMore()) {
- DistributedQueryManager.this.close(QueryCompletionReason.CLOSE);
+ DistributedQueryManager.this.close(CancellationReason.CLOSE);
}
});
@@ -1333,12 +1338,8 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
}
@Override
- public CompletableFuture<Void> closeAsync(boolean cancelled) {
- QueryCompletionReason reason = cancelled
- ? QueryCompletionReason.CANCEL
- : QueryCompletionReason.CLOSE;
-
- return DistributedQueryManager.this.close(reason);
+ public CompletableFuture<Void> closeAsync() {
+ return DistributedQueryManager.this.close(CancellationReason.CLOSE);
}
};
}
@@ -1390,12 +1391,4 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
return Objects.hash(catalogVersion, fragmentString);
}
}
-
- /** Represents reasons why a query was completed. */
- enum QueryCompletionReason {
- CLOSE,
- CANCEL,
- TIMEOUT,
- ERROR,
- }
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/TxAwareAsyncCursor.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/TxAwareAsyncCursor.java
index 4c73b7d625..db0779e630 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/TxAwareAsyncCursor.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/TxAwareAsyncCursor.java
@@ -35,22 +35,25 @@ import org.apache.ignite.internal.util.ExceptionUtils;
*
* @param <T> Type of the returned items.
*/
-class TxAwareAsyncCursor<T> implements AsyncDataCursor<T> {
+class TxAwareAsyncCursor<T> implements AsyncDataCursorExt<T> {
private final AtomicBoolean closed = new AtomicBoolean(false);
private final CompletableFuture<Void> closeResult = new CompletableFuture<>();
private final QueryTransactionWrapper txWrapper;
private final AsyncCursor<T> dataCursor;
private final CompletableFuture<Void> firstPageReady;
+ private final Function<CancellationReason, CompletableFuture<Void>> closeHandler;
TxAwareAsyncCursor(
QueryTransactionWrapper txWrapper,
AsyncCursor<T> dataCursor,
- CompletableFuture<Void> firstPageReady
+ CompletableFuture<Void> firstPageReady,
+ Function<CancellationReason, CompletableFuture<Void>> closeHandler
) {
this.txWrapper = txWrapper;
this.dataCursor = dataCursor;
this.firstPageReady = firstPageReady;
+ this.closeHandler = closeHandler;
}
@Override
@@ -71,52 +74,39 @@ class TxAwareAsyncCursor<T> implements AsyncDataCursor<T> {
}
@Override
- public CompletableFuture<Void> closeAsync(boolean cancelled) {
+ public CompletableFuture<Void> cancelAsync(CancellationReason reason) {
if (!closed.compareAndSet(false, true)) {
return closeResult;
}
- CompletableFuture<Void> closeFuture;
-
- if (cancelled) {
- closeFuture = txWrapper.rollback(new QueryCancelledException())
- .handle((none, rollbackErr) -> {
- return dataCursor.closeAsync(true)
- .handle((res, closeErr) -> {
- Throwable err = rollbackErr;
-
- if (closeErr != null) {
- if (rollbackErr != null) {
- closeErr.addSuppressed(rollbackErr);
- }
-
- err = closeErr;
- }
-
- if (err != null) {
- throw new CompletionException(err);
- }
-
- return res;
- });
- })
- .thenCompose(Function.identity());
- } else {
- closeFuture = dataCursor.closeAsync()
- .thenCompose(ignored -> txWrapper.commitImplicit());
- }
+ closeHandler.apply(reason)
+ .thenCompose(ignored -> {
+ if (reason != CancellationReason.CLOSE) {
+ String message = reason == CancellationReason.TIMEOUT
+ ? QueryCancelledException.TIMEOUT_MSG
+ : QueryCancelledException.CANCEL_MSG;
- closeFuture.whenComplete((r, e) -> {
- if (e != null) {
- closeResult.completeExceptionally(e);
- } else {
- closeResult.complete(null);
- }
- });
+ return txWrapper.rollback(new QueryCancelledException(message));
+ }
+
+ return txWrapper.commitImplicit();
+ })
+ .whenComplete((r, e) -> {
+ if (e != null) {
+ closeResult.completeExceptionally(e);
+ } else {
+ closeResult.complete(null);
+ }
+ });
return closeResult;
}
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ return cancelAsync(CancellationReason.CLOSE);
+ }
+
@Override
public CompletableFuture<Void> onClose() {
return closeResult;
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/CursorInitializationPhaseHandler.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/CursorInitializationPhaseHandler.java
index 4e828c7031..2ffb904432 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/CursorInitializationPhaseHandler.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/CursorInitializationPhaseHandler.java
@@ -24,6 +24,7 @@ import org.apache.ignite.internal.sql.engine.AsyncSqlCursorImpl;
import org.apache.ignite.internal.sql.engine.InternalSqlRow;
import org.apache.ignite.internal.sql.engine.SqlOperationContext;
import org.apache.ignite.internal.sql.engine.SqlQueryType;
+import org.apache.ignite.internal.sql.engine.exec.AsyncDataCursorExt.CancellationReason;
import org.apache.ignite.internal.sql.engine.prepare.QueryPlan;
import org.apache.ignite.internal.sql.engine.tx.QueryTransactionContext;
@@ -54,11 +55,9 @@ class CursorInitializationPhaseHandler implements ExecutionPhaseHandler {
query.cursor = cursor;
- query.cancel.add(timeout -> {
- if (!timeout) {
- cursor.closeAsync(true);
- }
- });
+ query.cancel.add(timeout -> dataCursor.cancelAsync(
+ timeout ? CancellationReason.TIMEOUT : CancellationReason.CANCEL
+ ));
QueryTransactionContext txContext = query.txContext;
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 01521b3893..ad56db7329 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
@@ -37,7 +37,7 @@ import org.apache.ignite.internal.sql.engine.InternalSqlRow;
import org.apache.ignite.internal.sql.engine.QueryCancelledException;
import org.apache.ignite.internal.sql.engine.QueryProperty;
import org.apache.ignite.internal.sql.engine.SqlOperationContext;
-import org.apache.ignite.internal.sql.engine.exec.AsyncDataCursor;
+import org.apache.ignite.internal.sql.engine.exec.AsyncDataCursorExt;
import org.apache.ignite.internal.sql.engine.exec.ExecutionService;
import org.apache.ignite.internal.sql.engine.exec.LifecycleAware;
import org.apache.ignite.internal.sql.engine.exec.TransactionTracker;
@@ -286,7 +286,7 @@ public class QueryExecutor implements LifecycleAware {
return clockService.now();
}
- CompletableFuture<AsyncDataCursor<InternalSqlRow>> executePlan(
+ CompletableFuture<AsyncDataCursorExt<InternalSqlRow>> executePlan(
SqlOperationContext ctx,
QueryPlan plan
) {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AsyncRootNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AsyncRootNode.java
index 5141f15d1a..6d014c3fd2 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AsyncRootNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AsyncRootNode.java
@@ -61,8 +61,6 @@ public class AsyncRootNode<InRowT, OutRowT> implements Downstream<InRowT>, Async
private volatile boolean closed = false;
- private volatile boolean cancelled = false;
-
/**
* Amount of rows which were requested from a source.
*
@@ -136,7 +134,7 @@ public class AsyncRootNode<InRowT, OutRowT> implements Downstream<InRowT>, Async
}
if (closed) {
- next.completeExceptionally(cancelled ? new QueryCancelledException() : new CursorClosedException());
+ next.completeExceptionally(new CursorClosedException());
return next;
}
@@ -151,11 +149,9 @@ public class AsyncRootNode<InRowT, OutRowT> implements Downstream<InRowT>, Async
/** {@inheritDoc} */
@Override
- public CompletableFuture<Void> closeAsync(boolean cancelled) {
+ public CompletableFuture<Void> closeAsync() {
if (!closed) {
synchronized (lock) {
- this.cancelled = cancelled;
-
if (!closed) {
Throwable th = ex.get();
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ScriptTransactionWrapperImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ScriptTransactionWrapperImpl.java
index 4f61717cf6..ccbce19c26 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ScriptTransactionWrapperImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ScriptTransactionWrapperImpl.java
@@ -30,7 +30,6 @@ import java.util.concurrent.CompletionException;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
import org.apache.ignite.internal.sql.engine.InternalSqlRow;
-import org.apache.ignite.internal.sql.engine.QueryCancelledException;
import org.apache.ignite.internal.sql.engine.SqlQueryType;
import org.apache.ignite.internal.sql.engine.exec.TransactionTracker;
import org.apache.ignite.internal.tx.InternalTransaction;
@@ -101,13 +100,11 @@ class ScriptTransactionWrapperImpl implements QueryTransactionWrapper {
cursorsToClose = List.copyOf(openedCursors.values());
}
- boolean cancelled = cause instanceof QueryCancelledException;
-
// Close all associated cursors on error.
for (CompletableFuture<? extends AsyncCursor<?>> fut : cursorsToClose) {
fut.whenComplete((cursor, ex) -> {
if (cursor != null) {
- cursor.closeAsync(cancelled);
+ cursor.closeAsync();
}
});
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/IteratorToDataCursorAdapter.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/IteratorToDataCursorAdapter.java
index eeeac2a1b6..3fb1a8b256 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/IteratorToDataCursorAdapter.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/IteratorToDataCursorAdapter.java
@@ -23,6 +23,7 @@ import java.util.Iterator;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import org.apache.ignite.internal.sql.engine.exec.AsyncDataCursor;
+import org.apache.ignite.internal.sql.engine.exec.AsyncDataCursorExt;
import org.apache.ignite.internal.util.AsyncWrapper;
/**
@@ -30,7 +31,7 @@ import org.apache.ignite.internal.util.AsyncWrapper;
*
* @param <T> Type of the items.
*/
-public class IteratorToDataCursorAdapter<T> extends AsyncWrapper<T> implements AsyncDataCursor<T> {
+public class IteratorToDataCursorAdapter<T> extends AsyncWrapper<T> implements AsyncDataCursorExt<T> {
private final CompletableFuture<Void> initialized = new CompletableFuture<>();
/**
@@ -71,4 +72,9 @@ public class IteratorToDataCursorAdapter<T> extends AsyncWrapper<T> implements A
public CompletableFuture<Void> onFirstPageReady() {
return initialized;
}
+
+ @Override
+ public CompletableFuture<Void> cancelAsync(CancellationReason reason) {
+ return nullCompletedFuture();
+ }
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/TxAwareCursorSelfTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/TxAwareCursorSelfTest.java
index 464cd1a321..e5e66516ab 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/TxAwareCursorSelfTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/TxAwareCursorSelfTest.java
@@ -57,7 +57,8 @@ public class TxAwareCursorSelfTest extends BaseIgniteAbstractTest {
AsyncCursor<Integer> cursor = new TxAwareAsyncCursor<>(
txWrapper,
new AsyncWrapper<>(CompletableFuture.completedFuture(list.iterator()), Runnable::run),
- nullCompletedFuture()
+ nullCompletedFuture(),
+ reason -> nullCompletedFuture()
);
int requestRows = 2;
@@ -83,7 +84,8 @@ public class TxAwareCursorSelfTest extends BaseIgniteAbstractTest {
AsyncCursor<Integer> cursor = new TxAwareAsyncCursor<>(
txWrapper,
new AsyncWrapper<>(CompletableFuture.failedFuture(err), Runnable::run),
- nullCompletedFuture()
+ nullCompletedFuture(),
+ reason -> nullCompletedFuture()
);
CompletionException t = assertThrows(CompletionException.class, () -> cursor.requestNextAsync(1).join());
@@ -103,7 +105,8 @@ public class TxAwareCursorSelfTest extends BaseIgniteAbstractTest {
AsyncCursor<Integer> cursor = new TxAwareAsyncCursor<>(
txWrapper,
data,
- nullCompletedFuture()
+ nullCompletedFuture(),
+ reason -> nullCompletedFuture()
);
cursor.closeAsync().join();