Savepoints

vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2SocketConnection.java

@@ -32,6 +32,7 @@
 import io.vertx.sqlclient.spi.connection.Connection;
 import io.vertx.sqlclient.spi.protocol.CommandBase;
 import io.vertx.sqlclient.spi.protocol.ExtendedQueryCommand;
+import io.vertx.sqlclient.spi.protocol.SavepointCommand;
 import io.vertx.sqlclient.spi.protocol.SimpleQueryCommand;
 import io.vertx.sqlclient.spi.protocol.TxCommand;
 
@@ -121,11 +122,27 @@ protected <R> void doSchedule(CommandBase<R> cmd, Completable<R> handler) {
         super.doSchedule(cmd2, (res, err) -> handler.complete(txCmd.result(), err));
 
       }
+    } else if (cmd instanceof SavepointCommand) {
+      SavepointCommand<R> savepoint = (SavepointCommand<R>) cmd;
+      SimpleQueryCommand<Void> cmd2 = new SimpleQueryCommand<>(savepointSql(savepoint), false, false,
+          SocketConnectionBase.NULL_COLLECTOR, QueryResultHandler.NOOP_HANDLER);
+      super.doSchedule(cmd2, (res, err) -> handler.complete(savepoint.result(), err));
     } else {
       super.doSchedule(cmd, handler);
     }
   }
 
+  /**
+   * DB2 requires the {@code ON ROLLBACK RETAIN CURSORS} clause when a savepoint is
+   * created, the other statements follow the standard syntax.
+   */
+  private static String savepointSql(SavepointCommand<?> savepoint) {
+    if (savepoint.kind() == SavepointCommand.Kind.CREATE) {
+      return "SAVEPOINT " + savepoint.name() + " ON ROLLBACK RETAIN CURSORS";
+    }
+    return savepoint.sql();
+  }
+
   @Override
   public String system() {
     return "db2";

vertx-db2-client/src/main/java/io/vertx/db2client/spi/DB2Driver.java

@@ -82,6 +82,11 @@ public ConnectionFactory<DB2ConnectOptions> createConnectionFactory(Vertx vertx,
     return new DB2ConnectionFactory((VertxInternal) vertx, transportOptions);
   }
 
+  @Override
+  public boolean supportsSavepoints() {
+    return true;
+  }
+
   @Override
   public SqlConnectionInternal wrapConnection(ContextInternal context, ConnectionFactory<DB2ConnectOptions> factory, Connection connection) {
     return new DB2ConnectionImpl(context, factory, connection);

vertx-db2-client/src/test/java/io/vertx/tests/db2client/tck/DB2TransactionTest.java

@@ -12,6 +12,9 @@
 
 import io.vertx.db2client.DB2Builder;
 import io.vertx.db2client.DB2ConnectOptions;
+import io.vertx.ext.unit.Async;
+import io.vertx.sqlclient.Cursor;
+import org.junit.Test;
 import io.vertx.ext.unit.TestContext;
 import io.vertx.ext.unit.junit.VertxUnitRunner;
 import io.vertx.sqlclient.Pool;
@@ -73,4 +76,43 @@ protected void cleanTestTable(TestContext ctx) {
   protected String statement(String... parts) {
     return String.join("?", parts);
   }
+
+  @Override
+  protected boolean supportsSavepoints() {
+    return true;
+  }
+
+  /**
+   * DB2 creates savepoints with ON ROLLBACK RETAIN CURSORS, so a cursor opened before the
+   * rollback keeps returning the rows it was already positioned on.
+   */
+  @Test
+  public void testCursorSurvivesRollbackToSavepoint(TestContext ctx) {
+    Async async = ctx.async();
+    connector.accept(ctx.asyncAssertSuccess(res -> {
+      insertMutable(res.client, 1, "one")
+        .compose(v -> insertMutable(res.client, 2, "two"))
+        .compose(v -> res.client.prepare("SELECT id FROM mutable ORDER BY id"))
+        .compose(ps -> {
+          Cursor cursor = ps.cursor();
+          return cursor.read(1)
+            .compose(first -> {
+              ctx.assertEquals(1, first.size());
+              ctx.assertEquals(1, first.iterator().next().getInteger("id"));
+              return res.tx.createSavepoint();
+            })
+            .compose(sp -> insertMutable(res.client, 3, "three").compose(v -> sp.rollback()))
+            // the cursor was opened before the savepoint, it must still be readable
+            .compose(v -> cursor.read(1))
+            .compose(second -> {
+              ctx.assertEquals(1, second.size());
+              ctx.assertEquals(2, second.iterator().next().getInteger("id"));
+              return cursor.close();
+            });
+        })
+        .compose(v -> res.tx.commit())
+        .compose(v -> assertMutableIds(ctx, 1, 2))
+        .onComplete(ctx.asyncAssertSuccess(v -> async.complete()));
+    }));
+  }
 }

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLSocketConnection.java

@@ -42,6 +42,8 @@
 import java.util.Map;
 import java.util.function.Predicate;
 
+import io.vertx.sqlclient.spi.protocol.SavepointCommand;
+
 import static io.vertx.sqlclient.spi.protocol.TxCommand.Kind.BEGIN;
 
 public class MSSQLSocketConnection extends SocketConnectionBase {
@@ -158,6 +160,20 @@ public void init() {
     return MSSQLCommandMessage.wrap(command);
   }
 
+  /**
+   * Transact-SQL names the savepoint statements differently and has no statement to
+   * release one, {@link io.vertx.sqlclient.spi.Driver#supportsSavepointRelease()}
+   * reports that.
+   */
+  private static String savepointSql(SavepointCommand<?> savepoint) {
+    switch (savepoint.kind()) {
+      case CREATE:
+        return "SAVE TRANSACTION " + savepoint.name();
+      default:
+        return "ROLLBACK TRANSACTION " + savepoint.name();
+    }
+  }
+
   @Override
   protected <R> void doSchedule(CommandBase<R> cmd, Completable<R> handler) {
     if (cmd instanceof TxCommand) {
@@ -170,6 +186,21 @@ protected <R> void doSchedule(CommandBase<R> cmd, Completable<R> handler) {
         SocketConnectionBase.NULL_COLLECTOR,
         QueryResultHandler.NOOP_HANDLER);
       super.doSchedule(cmd2, (res, err) -> handler.complete(tx.result(), err));
+    } else if (cmd instanceof SavepointCommand) {
+      SavepointCommand<R> savepoint = (SavepointCommand<R>) cmd;
+      if (savepoint.kind() == SavepointCommand.Kind.RELEASE) {
+        // Guarded by MSSQLDriver#supportsSavepointRelease, fail rather than throw on the event loop
+        handler.fail(new UnsupportedOperationException(
+          "Releasing a savepoint is not supported by Microsoft SQL Server"));
+        return;
+      }
+      SimpleQueryCommand<Void> cmd2 = new SimpleQueryCommand<>(
+        savepointSql(savepoint),
+        false,
+        false,
+        SocketConnectionBase.NULL_COLLECTOR,
+        QueryResultHandler.NOOP_HANDLER);
+      super.doSchedule(cmd2, (res, err) -> handler.complete(savepoint.result(), err));
     } else {
       super.doSchedule(cmd, handler);
     }

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/spi/MSSQLDriver.java

@@ -67,6 +67,19 @@ public int appendQueryPlaceholder(StringBuilder queryBuilder, int index, int cur
     return index;
   }
 
+  @Override
+  public boolean supportsSavepoints() {
+    return true;
+  }
+
+  /**
+   * Transact-SQL has no statement that discards a savepoint without rolling back to it.
+   */
+  @Override
+  public boolean supportsSavepointRelease() {
+    return false;
+  }
+
   @Override
   public SqlConnectionInternal wrapConnection(ContextInternal context, ConnectionFactory<MSSQLConnectOptions> factory, Connection connection) {
     return new MSSQLConnectionImpl(context, factory, connection);

vertx-mssql-client/src/test/java/io/vertx/tests/mssqlclient/tck/MSSQLTransactionTest.java

@@ -10,7 +10,10 @@
  */
 package io.vertx.tests.mssqlclient.tck;
 
+import io.vertx.core.Future;
+import io.vertx.sqlclient.SqlConnection;
 import io.vertx.ext.unit.TestContext;
+import io.vertx.ext.unit.Async;
 import io.vertx.ext.unit.junit.VertxUnitRunner;
 import io.vertx.mssqlclient.MSSQLBuilder;
 import io.vertx.mssqlclient.MSSQLConnectOptions;
@@ -55,4 +58,56 @@ protected String statement(String... parts) {
   public void testDelayedCommit(TestContext ctx) {
     throw new AssumptionViolatedException("MSSQL holds write locks on inserted row with isolation level = 2");
   }
+
+  @Override
+  protected boolean supportsSavepoints() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportsSavepointRelease() {
+    return false;
+  }
+
+  /**
+   * SQL Server drops the savepoint once the transaction has been rolled back to it.
+   */
+  @Override
+  protected boolean supportsRepeatedRollbackToSavepoint() {
+    return false;
+  }
+
+  /**
+   * A savepoint is a mark inside the current transaction, it must not open a nested one:
+   * @@TRANCOUNT stays at 1 after SAVE TRANSACTION and after rolling back to it.
+   */
+  @Test
+  public void testSavepointDoesNotNestTheTransaction(TestContext ctx) {
+    Async async = ctx.async();
+    connector.accept(ctx.asyncAssertSuccess(res -> {
+      trancount(res.client)
+        .compose(before -> {
+          ctx.assertEquals(1, before, "the transaction should be the only one open");
+          return res.tx.createSavepoint();
+        })
+        .compose(sp -> trancount(res.client)
+          .compose(afterSave -> {
+            ctx.assertEquals(1, afterSave, "SAVE TRANSACTION must not nest a transaction");
+            return insertMutable(res.client, 1, "rolled-back");
+          })
+          .compose(v -> sp.rollback())
+          .compose(v -> trancount(res.client))
+          .compose(afterRollback -> {
+            ctx.assertEquals(1, afterRollback, "rolling back to a savepoint must keep the transaction open");
+            return insertMutable(res.client, 2, "kept");
+          })
+          .compose(v -> res.tx.commit()))
+        .compose(v -> assertMutableIds(ctx, 2))
+        .onComplete(ctx.asyncAssertSuccess(v -> async.complete()));
+    }));
+  }
+
+  private Future<Integer> trancount(SqlConnection client) {
+    return client.query("SELECT @@TRANCOUNT AS c").execute().map(rows -> rows.iterator().next().getInteger("c"));
+  }
 }

vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLSocketConnection.java

@@ -43,6 +43,7 @@
 import io.vertx.sqlclient.codec.SocketConnectionBase;
 import io.vertx.sqlclient.spi.protocol.CommandBase;
 import io.vertx.sqlclient.spi.protocol.ExtendedQueryCommand;
+import io.vertx.sqlclient.spi.protocol.SavepointCommand;
 import io.vertx.sqlclient.spi.protocol.SimpleQueryCommand;
 import io.vertx.sqlclient.spi.protocol.TxCommand;
 import io.vertx.sqlclient.spi.DatabaseMetadata;
@@ -127,6 +128,15 @@ protected <R> void doSchedule(CommandBase<R> cmd, Completable<R> handler) {
         SocketConnectionBase.NULL_COLLECTOR,
         QueryResultHandler.NOOP_HANDLER);
       super.doSchedule(cmd2, (res, err) -> handler.complete(tx.result(), err));
+    } else if (cmd instanceof SavepointCommand) {
+      SavepointCommand<R> savepoint = (SavepointCommand<R>) cmd;
+      SimpleQueryCommand<Void> cmd2 = new SimpleQueryCommand<>(
+        savepoint.sql(),
+        false,
+        false,
+        SocketConnectionBase.NULL_COLLECTOR,
+        QueryResultHandler.NOOP_HANDLER);
+      super.doSchedule(cmd2, (res, err) -> handler.complete(savepoint.result(), err));
     } else {
       super.doSchedule(cmd, handler);
     }

vertx-mysql-client/src/main/java/io/vertx/mysqlclient/spi/MySQLDriver.java

@@ -85,6 +85,11 @@ public ConnectionFactory<MySQLConnectOptions> createConnectionFactory(Vertx vert
     return new MySQLConnectionFactory((VertxInternal) vertx, transportOptions);
   }
 
+  @Override
+  public boolean supportsSavepoints() {
+    return true;
+  }
+
   @Override
   public SqlConnectionInternal wrapConnection(ContextInternal context, ConnectionFactory<MySQLConnectOptions> factory, Connection connection) {
     return new MySQLConnectionImpl(context, factory, connection);

vertx-mysql-client/src/test/java/io/vertx/tests/mysqlclient/tck/MySQLTransactionTest.java

@@ -44,4 +44,11 @@ protected Pool nonTxPool() {
   protected String statement(String... parts) {
     return String.join("?", parts);
   }
+
+  @Override
+  protected boolean supportsSavepoints() {
+    return true;
+  }
+
+
 }

vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleJdbcConnection.java

@@ -206,6 +206,8 @@ private OracleCommand wrap(CommandBase cmd) {
       action = forExtendedQuery((ExtendedQueryCommand) cmd);
     } else if (cmd instanceof TxCommand) {
       action = OracleTransactionCommand.create(connection, context, ((TxCommand) cmd));
+    } else if (cmd instanceof SavepointCommand) {
+      action = OracleSavepointCommand.create(connection, context, ((SavepointCommand) cmd));
     } else if (cmd instanceof CloseStatementCommand) {
       action = new OracleCloseStatementCommand(connection, context);
     } else if (cmd instanceof CloseCursorCommand) {

vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/commands/OracleSavepointCommand.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright (c) 2011-2025 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+package io.vertx.oracleclient.impl.commands;
+
+import io.vertx.core.Future;
+import io.vertx.core.internal.ContextInternal;
+import io.vertx.sqlclient.spi.protocol.SavepointCommand;
+import oracle.jdbc.OracleConnection;
+
+import java.sql.Statement;
+
+/**
+ * Runs a savepoint statement on the JDBC connection.
+ *
+ * <p>Oracle names savepoints with the standard syntax, but has no statement that
+ * releases one, so {@link SavepointCommand.Kind#RELEASE} never reaches this command.
+ */
+public class OracleSavepointCommand<R> extends OracleCommand<R> {
+
+  private final SavepointCommand<R> op;
+
+  private OracleSavepointCommand(OracleConnection oracleConnection, ContextInternal connectionContext, SavepointCommand<R> op) {
+    super(oracleConnection, connectionContext);
+    this.op = op;
+  }
+
+  public static <U> OracleSavepointCommand<U> create(OracleConnection oracleConnection, ContextInternal connectionContext, SavepointCommand<U> cmd) {
+    return new OracleSavepointCommand<>(oracleConnection, connectionContext, cmd);
+  }
+
+  @Override
+  protected Future<R> execute() {
+    if (op.kind() == SavepointCommand.Kind.RELEASE) {
+      return connectionContext.failedFuture(new UnsupportedOperationException(
+        "Releasing a savepoint is not supported by Oracle"));
+    }
+    String sql = op.sql();
+    return executeBlocking(() -> {
+      try (Statement statement = oracleConnection.createStatement()) {
+        statement.execute(sql);
+      }
+    }).map(op.result());
+  }
+}

vertx-oracle-client/src/main/java/io/vertx/oracleclient/spi/OracleDriver.java

@@ -62,6 +62,19 @@ public ConnectionFactory<OracleConnectOptions> createConnectionFactory(Vertx ver
     return new OracleConnectionFactory();
   }
 
+  @Override
+  public boolean supportsSavepoints() {
+    return true;
+  }
+
+  /**
+   * Oracle has no statement that discards a savepoint without rolling back to it.
+   */
+  @Override
+  public boolean supportsSavepointRelease() {
+    return false;
+  }
+
   @Override
   public SqlConnectionInternal wrapConnection(ContextInternal context, ConnectionFactory<OracleConnectOptions> factory, Connection connection) {
     return new OracleConnectionImpl(context, factory, connection);