Savepoints

vertx-db2-client/src/main/asciidoc/savepoint_note.adoc

@@ -0,0 +1,4 @@
+DB2 supports savepoints fully.
+
+Savepoints are created with the `ON ROLLBACK RETAIN CURSORS` clause, so a cursor opened before the
+savepoint stays usable after a rollback.

vertx-db2-client/src/main/java/examples/SqlClientExamples.java

@@ -267,6 +267,51 @@ public void transaction03(Pool pool) {
       .onFailure(err -> System.out.println("Transaction failed: " + err.getMessage()));
   }
 
+  public void savepoint01(Pool pool) {
+
+    pool.withTransaction(client -> client
+      .query("INSERT INTO Users (first_name,last_name) VALUES ('Julien','Viet')")
+      .execute()
+      .flatMap(v -> client.transaction().createSavepoint())
+      .flatMap(savepoint -> client
+        .query("INSERT INTO Users (first_name,last_name) VALUES ('Emad','Alblueshi')")
+        .execute()
+        // Undo the second insert, the first one is still part of the transaction
+        .flatMap(v -> savepoint.rollback())))
+      .onSuccess(v -> System.out.println("Transaction committed with a single user"))
+      .onFailure(err -> System.out.println("Transaction failed: " + err.getMessage()));
+  }
+
+  public void savepoint02(Pool pool) {
+
+    pool.withTransaction(client -> client
+      .transaction()
+      .createSavepoint()
+      .flatMap(savepoint -> client
+        .query("INSERT INTO Users (first_name,last_name) VALUES ('Julien','Viet')")
+        .execute()
+        // The insert is kept, the savepoint cannot be used anymore
+        .flatMap(v -> savepoint.release())))
+      .onSuccess(v -> System.out.println("Transaction committed"))
+      .onFailure(err -> System.out.println("Transaction failed: " + err.getMessage()));
+  }
+
+  public void savepoint03(Pool pool) {
+
+    pool.withTransaction(client -> client
+      .transaction()
+      // The name shows up in database error messages
+      .createSavepoint("before_users")
+      .flatMap(savepoint -> client
+        .query("INSERT INTO Users (first_name,last_name) VALUES ('Julien','Viet')")
+        .execute()
+        .flatMap(v -> savepoint.rollback())))
+      .onSuccess(v -> System.out.println("Transaction committed without the user"))
+      .onFailure(err -> System.out.println("Transaction failed: " + err.getMessage()));
+  }
+
+
+
   public void usingCursors01(SqlConnection connection) {
     connection
       .prepare("SELECT * FROM users WHERE first_name LIKE ?")

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/asciidoc/savepoint_note.adoc

@@ -0,0 +1,6 @@
+Microsoft SQL Server has no statement that releases a savepoint, so
+{@link io.vertx.sqlclient.Savepoint#release} fails with an `UnsupportedOperationException`. The savepoint
+remains usable for a rollback.
+
+A savepoint is also dropped once the transaction has been rolled back to it, so it cannot be rolled back
+to a second time.

vertx-mssql-client/src/main/java/examples/SqlClientExamples.java

@@ -296,6 +296,51 @@ public void transaction03(Pool pool) {
       .onFailure(err -> System.out.println("Transaction failed: " + err.getMessage()));
   }
 
+  public void savepoint01(Pool pool) {
+
+    pool.withTransaction(client -> client
+      .query("INSERT INTO Users (first_name,last_name) VALUES ('Julien','Viet')")
+      .execute()
+      .flatMap(v -> client.transaction().createSavepoint())
+      .flatMap(savepoint -> client
+        .query("INSERT INTO Users (first_name,last_name) VALUES ('Emad','Alblueshi')")
+        .execute()
+        // Undo the second insert, the first one is still part of the transaction
+        .flatMap(v -> savepoint.rollback())))
+      .onSuccess(v -> System.out.println("Transaction committed with a single user"))
+      .onFailure(err -> System.out.println("Transaction failed: " + err.getMessage()));
+  }
+
+  public void savepoint02(Pool pool) {
+
+    pool.withTransaction(client -> client
+      .transaction()
+      .createSavepoint()
+      .flatMap(savepoint -> client
+        .query("INSERT INTO Users (first_name,last_name) VALUES ('Julien','Viet')")
+        .execute()
+        // The insert is kept, the savepoint cannot be used anymore
+        .flatMap(v -> savepoint.release())))
+      .onSuccess(v -> System.out.println("Transaction committed"))
+      .onFailure(err -> System.out.println("Transaction failed: " + err.getMessage()));
+  }
+
+  public void savepoint03(Pool pool) {
+
+    pool.withTransaction(client -> client
+      .transaction()
+      // The name shows up in database error messages
+      .createSavepoint("before_users")
+      .flatMap(savepoint -> client
+        .query("INSERT INTO Users (first_name,last_name) VALUES ('Julien','Viet')")
+        .execute()
+        .flatMap(v -> savepoint.rollback())))
+      .onSuccess(v -> System.out.println("Transaction committed without the user"))
+      .onFailure(err -> System.out.println("Transaction failed: " + err.getMessage()));
+  }
+
+
+
   public void usingCursors01(SqlConnection connection) {
     connection.prepare("SELECT * FROM users WHERE age > @p1")
       .onComplete(ar1 -> {

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,27 @@ public int appendQueryPlaceholder(StringBuilder queryBuilder, int index, int cur
     return index;
   }
 
+  /**
+   * Transact-SQL delimits identifiers with square brackets.
+   */
+  @Override
+  public StringBuilder appendQuotedIdentifier(StringBuilder sql, String identifier) {
+    return sql.append('[').append(identifier.replace("]", "]]")).append(']');
+  }
+
+  @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/asciidoc/savepoint_note.adoc

@@ -0,0 +1,4 @@
+MySQL supports savepoints fully.
+
+A failed statement does not fail the transaction, only the statement is rolled back and the transaction
+stays usable. A savepoint is still useful to discard a group of statements as a whole.