transactions: session savepoints and mutation-statement warning capture

xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/SqlMutationSession.java

@@ -1520,6 +1520,31 @@ public List<String> execute(List<String> statements) {
     return sqlSession.execute(statements);
   }
 
+  @Override
+  public List<String> drainWarnings() {
+    return sqlSession.drainWarnings();
+  }
+
+  @Override
+  public boolean supportsSavepoints() {
+    return true;
+  }
+
+  @Override
+  public void savepoint() {
+    sqlSession.savepoint();
+  }
+
+  @Override
+  public void releaseSavepoint() {
+    sqlSession.releaseSavepoint();
+  }
+
+  @Override
+  public void rollbackToSavepoint() {
+    sqlSession.rollbackToSavepoint();
+  }
+
   @Override
   public void commit() {
     sqlSession.commit();

xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlSession.java

@@ -56,6 +56,44 @@ String run(
    */
   List<String> execute(List<String> statements);
 
+  /**
+   * Returns the non-fatal SQL warnings (e.g. PostgreSQL {@code RAISE WARNING} / {@code RAISE
+   * NOTICE}) that mutation statements run via {@link #run} / {@link #runReturning} have produced
+   * since the last call, and clears them. Warnings produced by {@link #execute} are returned by
+   * that method directly and do not show up here. The default implementation returns an empty list
+   * for sessions that do not collect warnings.
+   */
+  default List<String> drainWarnings() {
+    return List.of();
+  }
+
+  /**
+   * Marks a recoverable point in this session's open transaction ({@code SAVEPOINT}). At most one
+   * savepoint may be active at a time; it is consumed by either {@link #releaseSavepoint()} or
+   * {@link #rollbackToSavepoint()}. The default implementation throws {@link
+   * UnsupportedOperationException} for sessions that have not adopted the API.
+   */
+  default void savepoint() {
+    throw new UnsupportedOperationException("Savepoints are not supported by this SQL session");
+  }
+
+  /**
+   * Releases the active savepoint ({@code RELEASE SAVEPOINT}), keeping all changes made since
+   * {@link #savepoint()} as part of the enclosing transaction. Throws when no savepoint is active.
+   */
+  default void releaseSavepoint() {
+    throw new UnsupportedOperationException("Savepoints are not supported by this SQL session");
+  }
+
+  /**
+   * Undoes all changes made since {@link #savepoint()} ({@code ROLLBACK TO SAVEPOINT}) and releases
+   * the savepoint, leaving the enclosing transaction usable for further statements. Throws when no
+   * savepoint is active.
+   */
+  default void rollbackToSavepoint() {
+    throw new UnsupportedOperationException("Savepoints are not supported by this SQL session");
+  }
+
   /** Commits all mutations performed against this session. */
   void commit();
 

xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSession.java

@@ -14,6 +14,7 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.SQLWarning;
+import java.sql.Savepoint;
 import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.List;
@@ -34,6 +35,10 @@ class JdbcSqlSession implements SqlSession {
 
   private final Connection connection;
   private boolean finalised;
+  private Savepoint activeSavepoint;
+  // Non-fatal SQL warnings (e.g. PostgreSQL RAISE WARNING / RAISE NOTICE) emitted by mutation
+  // statements, accumulated until the caller drains them.
+  private final List<String> pendingWarnings = new ArrayList<>();
 
   JdbcSqlSession(Connection connection) {
     this.connection = connection;
@@ -99,6 +104,7 @@ public String run(
         }
         try (Statement statement = connection.createStatement()) {
           boolean hasResultSet = statement.execute(sql);
+          harvestWarnings(statement);
           String returnedId = null;
           if (hasResultSet) {
             try (ResultSet rs = statement.getResultSet()) {
@@ -144,6 +150,7 @@ public List<String> runReturning(String sql) {
     }
     try (Statement statement = connection.createStatement()) {
       boolean hasResultSet = statement.execute(sql);
+      harvestWarnings(statement);
       if (!hasResultSet) {
         return List.of();
       }
@@ -196,6 +203,32 @@ private static boolean isBatchable(String sql) {
     return trimmed.toLowerCase(Locale.ROOT).endsWith("returning null");
   }
 
+  /**
+   * Collects the non-fatal SQL warning chain of a just-executed statement into {@link
+   * #pendingWarnings} and clears it — the batch statement is reused across flushes, so its chain
+   * would otherwise be harvested again on the next flush.
+   */
+  private void harvestWarnings(Statement statement) {
+    try {
+      for (SQLWarning w = statement.getWarnings(); w != null; w = w.getNextWarning()) {
+        pendingWarnings.add(w.getMessage());
+      }
+      statement.clearWarnings();
+    } catch (SQLException e) {
+      LOGGER.debug("Reading SQL warnings failed: {}", e.getMessage());
+    }
+  }
+
+  @Override
+  public List<String> drainWarnings() {
+    if (pendingWarnings.isEmpty()) {
+      return List.of();
+    }
+    List<String> drained = List.copyOf(pendingWarnings);
+    pendingWarnings.clear();
+    return drained;
+  }
+
   private void flushBatch(
       Statement batchStmt, List<String> batchedSql, List<Consumer<String>> batchedConsumers) {
     if (batchedSql.isEmpty()) {
@@ -208,6 +241,7 @@ private void flushBatch(
     try {
       batchStmt.executeBatch();
       batchStmt.clearBatch();
+      harvestWarnings(batchStmt);
     } catch (SQLException e) {
       throw new IllegalStateException(
           "Batched mutation failed: " + e.getMessage() + " — first statement: " + batchedSql.get(0),
@@ -225,6 +259,56 @@ private void flushBatch(
     batchedConsumers.clear();
   }
 
+  @Override
+  public void savepoint() {
+    if (finalised) {
+      throw new IllegalStateException("SQL session is closed");
+    }
+    if (activeSavepoint != null) {
+      throw new IllegalStateException("A savepoint is already active on this SQL session");
+    }
+    try {
+      activeSavepoint = connection.setSavepoint();
+    } catch (SQLException e) {
+      throw new IllegalStateException("Savepoint failed: " + e.getMessage(), e);
+    }
+  }
+
+  @Override
+  public void releaseSavepoint() {
+    if (finalised) {
+      throw new IllegalStateException("SQL session is closed");
+    }
+    if (activeSavepoint == null) {
+      throw new IllegalStateException("No savepoint is active on this SQL session");
+    }
+    try {
+      connection.releaseSavepoint(activeSavepoint);
+    } catch (SQLException e) {
+      throw new IllegalStateException("Savepoint release failed: " + e.getMessage(), e);
+    } finally {
+      activeSavepoint = null;
+    }
+  }
+
+  @Override
+  public void rollbackToSavepoint() {
+    if (finalised) {
+      throw new IllegalStateException("SQL session is closed");
+    }
+    if (activeSavepoint == null) {
+      throw new IllegalStateException("No savepoint is active on this SQL session");
+    }
+    try {
+      connection.rollback(activeSavepoint);
+      connection.releaseSavepoint(activeSavepoint);
+    } catch (SQLException e) {
+      throw new IllegalStateException("Savepoint rollback failed: " + e.getMessage(), e);
+    } finally {
+      activeSavepoint = null;
+    }
+  }
+
   @Override
   public void commit() {
     if (finalised) {

xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/app/SqlMutationSessionSpec.groovy

@@ -80,6 +80,24 @@ class SqlMutationSessionSpec extends Specification {
         0 * sqlSession.rollback()
     }
 
+    def 'savepoint lifecycle delegates to the underlying SqlSession and is advertised as supported'() {
+        given:
+        def session = buildSession()
+
+        expect:
+        session.supportsSavepoints()
+
+        when:
+        session.savepoint()
+        session.releaseSavepoint()
+        session.rollbackToSavepoint()
+
+        then:
+        1 * sqlSession.savepoint()
+        1 * sqlSession.releaseSavepoint()
+        1 * sqlSession.rollbackToSavepoint()
+    }
+
     def 'deleteFeature for an unknown feature type fails fast with IAE (no SQL is run)'() {
         given:
         def session = buildSession([:])  // empty mappings

xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSessionSpec.groovy

@@ -12,6 +12,8 @@ import spock.lang.Specification
 import java.sql.Connection
 import java.sql.ResultSet
 import java.sql.SQLException
+import java.sql.SQLWarning
+import java.sql.Savepoint
 import java.sql.Statement
 import java.util.function.Consumer
 import java.util.function.Supplier
@@ -294,6 +296,152 @@ class JdbcSqlSessionSpec extends Specification {
         0 * connection.rollback()
     }
 
+    def 'run collects SQL warnings from executed statements; drainWarnings returns and clears them'() {
+        given:
+        def session = new JdbcSqlSession(connection)
+        statement.execute('INSERT 1 RETURNING id') >> true
+        statement.getResultSet() >> resultSet
+        resultSet.next() >> true
+        resultSet.getString(1) >> 'gen-1'
+        statement.getWarnings() >> new SQLWarning('notice: derived data updated')
+
+        when:
+        session.run(
+                [{ 'INSERT 1 RETURNING id' } as Supplier<String>],
+                [{ String s -> } as Consumer<String>],
+                Optional.empty())
+
+        then:
+        session.drainWarnings() == ['notice: derived data updated']
+
+        and: 'a second drain is empty'
+        session.drainWarnings().isEmpty()
+    }
+
+    def 'runReturning flattens a chained SQL warning list in order'() {
+        given:
+        def session = new JdbcSqlSession(connection)
+        def chain = new SQLWarning('first')
+        chain.setNextWarning(new SQLWarning('second'))
+        statement.execute('INSERT INTO feat VALUES (1), (2) RETURNING id') >> false
+        statement.getWarnings() >> chain
+
+        when:
+        session.runReturning('INSERT INTO feat VALUES (1), (2) RETURNING id')
+
+        then:
+        session.drainWarnings() == ['first', 'second']
+    }
+
+    def 'savepoint marks a recoverable point on the connection'() {
+        given:
+        def session = new JdbcSqlSession(connection)
+
+        when:
+        session.savepoint()
+
+        then:
+        1 * connection.setSavepoint() >> Mock(Savepoint)
+    }
+
+    def 'releaseSavepoint keeps the changes and consumes the savepoint'() {
+        given:
+        def session = new JdbcSqlSession(connection)
+        def sp = Mock(Savepoint)
+        connection.setSavepoint() >> sp
+        session.savepoint()
+
+        when:
+        session.releaseSavepoint()
+
+        then:
+        1 * connection.releaseSavepoint(sp)
+        0 * connection.rollback(_ as Savepoint)
+    }
+
+    def 'rollbackToSavepoint undoes the changes, then consumes the savepoint'() {
+        given:
+        def session = new JdbcSqlSession(connection)
+        def sp = Mock(Savepoint)
+        connection.setSavepoint() >> sp
+        session.savepoint()
+
+        when:
+        session.rollbackToSavepoint()
+
+        then:
+        1 * connection.rollback(sp)
+        1 * connection.releaseSavepoint(sp)
+    }
+
+    def 'a consumed savepoint allows a new one; a second active savepoint is rejected'() {
+        given:
+        def session = new JdbcSqlSession(connection)
+        connection.setSavepoint() >> Mock(Savepoint)
+        session.savepoint()
+
+        when: 'a second savepoint while one is active'
+        session.savepoint()
+
+        then:
+        thrown(IllegalStateException)
+
+        when: 'the active savepoint is consumed'
+        session.rollbackToSavepoint()
+        session.savepoint()
+
+        then:
+        noExceptionThrown()
+    }
+
+    def 'releaseSavepoint and rollbackToSavepoint without an active savepoint throw'() {
+        given:
+        def session = new JdbcSqlSession(connection)
+
+        when:
+        session.releaseSavepoint()
+
+        then:
+        thrown(IllegalStateException)
+
+        when:
+        session.rollbackToSavepoint()
+
+        then:
+        thrown(IllegalStateException)
+    }
+
+    def 'savepoint methods on a finalised session throw'() {
+        given:
+        def session = new JdbcSqlSession(connection)
+        session.commit()
+
+        when:
+        session.savepoint()
+
+        then:
+        thrown(IllegalStateException)
+    }
+
+    def 'a failed rollbackToSavepoint clears the savepoint and surfaces the cause'() {
+        given:
+        def session = new JdbcSqlSession(connection)
+        def sp = Mock(Savepoint)
+        connection.setSavepoint() >> sp
+        connection.rollback(sp) >> { throw new SQLException('connection lost') }
+        session.savepoint()
+
+        when:
+        session.rollbackToSavepoint()
+
+        then:
+        def ex = thrown(IllegalStateException)
+        ex.message.contains('connection lost')
+
+        and: 'the savepoint is consumed, so a new one can be created'
+        session.savepoint()
+    }
+
     def 'rollback swallows SQLException from connection.rollback (still finalises)'() {
         given:
         def session = new JdbcSqlSession(connection)