IGNITE-27117 In a multistatement query if an error occurs do not abort preceding statements

Author: xtern

modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlMultiStatementTxTest.java

@@ -303,8 +303,6 @@ void dmlFailsOnReadOnlyTransaction() {
         assertThrowsSqlException(RUNTIME_ERR, "DML cannot be started by using read only transactions.",
                 () -> await(insCur.nextResult()));
 
-        expectQueryCancelled(new DrainCursor(insCur));
-
         verifyFinishedTxCount(1);
     }
 

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/MultiStatementHandler.java

@@ -278,8 +278,6 @@ CompletableFuture<AsyncSqlCursor<InternalSqlRow>> processNext() {
     }
 
     private void cancelAll(Throwable cause) {
-        query.cancel.cancel();
-
         for (ScriptStatement scriptStatement : statements) {
             CompletableFuture<AsyncSqlCursor<InternalSqlRow>> fut = scriptStatement.cursorFuture;
 

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ScriptTransactionWrapperImpl.java

@@ -20,7 +20,9 @@
 import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
 import static org.apache.ignite.lang.ErrorGroups.Sql.EXECUTION_CANCELLED_ERR;
 
+import java.util.Collection;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
@@ -32,6 +34,7 @@
 import org.apache.ignite.internal.sql.engine.exec.AsyncDataCursor;
 import org.apache.ignite.internal.sql.engine.exec.TransactionalOperationTracker;
 import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.AsyncCursor;
 import org.apache.ignite.sql.SqlException;
 import org.jetbrains.annotations.Nullable;
 
@@ -85,6 +88,7 @@ public CompletableFuture<Void> finalise() {
 
     @Override
     public CompletableFuture<Void> finalise(Throwable error) {
+        Collection<CompletableFuture<? extends AsyncCursor<?>>> cursorsToClose;
         assert error != null;
 
         synchronized (mux) {
@@ -94,6 +98,17 @@ public CompletableFuture<Void> finalise(Throwable error) {
 
             rollbackCause = error;
             txState = State.ROLLBACK;
+
+            cursorsToClose = List.copyOf(openedCursors.values());
+        }
+
+        // Close all associated cursors on error.
+        for (CompletableFuture<? extends AsyncCursor<?>> fut : cursorsToClose) {
+            fut.whenComplete((cursor, ex) -> {
+                if (cursor != null) {
+                    cursor.closeAsync();
+                }
+            });
         }
 
         completeTx();