From e5d95c132ec31356cd5f1ff9d871829296f62ec9 Mon Sep 17 00:00:00 2001 From: Clemens Portele Date: Thu, 23 Jul 2026 12:54:34 +0200 Subject: [PATCH] transactions: session savepoints and mutation-statement warning capture - FeatureTransactions.Session: supportsSavepoints/savepoint/releaseSavepoint/ rollbackToSavepoint (single-active-savepoint contract, defaults throw for providers that have not adopted the API) and drainWarnings for non-fatal SQL warnings emitted by mutator calls - SqlSession: same savepoint methods and drainWarnings as defaults - JdbcSqlSession: JDBC savepoint lifecycle on the session connection; harvests Statement warning chains (RAISE WARNING / RAISE NOTICE) from run/runReturning, including batched statements, into a drainable list - SqlMutationSession: delegates savepoints and warning draining to the underlying SqlSession - specs for the savepoint lifecycle, warning harvesting, and delegation --- .../features/sql/app/SqlMutationSession.java | 25 +++ .../features/sql/domain/SqlSession.java | 38 +++++ .../features/sql/infra/db/JdbcSqlSession.java | 84 ++++++++++ .../sql/app/SqlMutationSessionSpec.groovy | 18 +++ .../sql/infra/db/JdbcSqlSessionSpec.groovy | 148 ++++++++++++++++++ .../features/domain/FeatureTransactions.java | 49 ++++++ 6 files changed, 362 insertions(+) diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/SqlMutationSession.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/SqlMutationSession.java index ab959e361..e54f7fa8a 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/SqlMutationSession.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/SqlMutationSession.java @@ -1520,6 +1520,31 @@ public List execute(List statements) { return sqlSession.execute(statements); } + @Override + public List 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(); diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlSession.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlSession.java index d1e313069..544c99f9b 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlSession.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlSession.java @@ -56,6 +56,44 @@ String run( */ List execute(List 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 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(); diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSession.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSession.java index 8252bd4c8..3512de43b 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSession.java +++ b/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 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 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 drainWarnings() { + if (pendingWarnings.isEmpty()) { + return List.of(); + } + List drained = List.copyOf(pendingWarnings); + pendingWarnings.clear(); + return drained; + } + private void flushBatch( Statement batchStmt, List batchedSql, List> 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) { diff --git a/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/app/SqlMutationSessionSpec.groovy b/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/app/SqlMutationSessionSpec.groovy index 9edba8670..1304f9291 100644 --- a/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/app/SqlMutationSessionSpec.groovy +++ b/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 diff --git a/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSessionSpec.groovy b/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSessionSpec.groovy index 1a1b3c041..94d163339 100644 --- a/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/JdbcSqlSessionSpec.groovy +++ b/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 s -> } as Consumer], + 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) diff --git a/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureTransactions.java b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureTransactions.java index 5c7c2b4dc..9cdd5c9fd 100644 --- a/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureTransactions.java +++ b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureTransactions.java @@ -303,6 +303,55 @@ default List execute(List statements) { return List.of(); } + /** + * Returns the non-fatal warnings the data source has emitted for the mutator calls since the + * last drain (e.g. PostgreSQL {@code RAISE WARNING} / {@code RAISE NOTICE} from triggers or + * functions the statements invoked), 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 providers that do not collect warnings. + */ + default List drainWarnings() { + return List.of(); + } + + /** + * Whether this session supports recoverable per-action scopes via {@link #savepoint()}. When + * {@code false} (the default), the savepoint methods throw {@link + * UnsupportedOperationException}. + */ + default boolean supportsSavepoints() { + return false; + } + + /** + * Marks a recoverable point in this session's open transaction. 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 providers that have not adopted the API. + */ + default void savepoint() { + throw new UnsupportedOperationException( + "Savepoints are not supported by this feature provider session"); + } + + /** + * Releases the active 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 feature provider session"); + } + + /** + * Undoes all changes made since {@link #savepoint()} and releases the savepoint, leaving the + * enclosing transaction usable for further mutations. Throws when no savepoint is active. + */ + default void rollbackToSavepoint() { + throw new UnsupportedOperationException( + "Savepoints are not supported by this feature provider session"); + } + /** Commits all mutations performed against this session. Throws if already finalised. */ void commit();