Add streamingRequiresTransaction to correctly apply fetch size. Add tests. (#76)

Author: zantvoort

docs/queries.md

@@ -1029,8 +1029,8 @@ Inline records expand in GROUP BY the same way. This is particularly useful in c
 ```kotlin
 data class CityOrderCount(val city: City, val count: Long)
 
-val page = orm.query(Order::class)
-    .select(Order_.city, "COUNT(*)")
+val orders = orm.entity(Order::class)
+val slice = orders.select(CityOrderCount::class) { "${City::class}, COUNT(*)" }
     .groupBy(Order_.city)
     .slice(Order_.city.key(), 20)
 ```
@@ -1121,7 +1121,7 @@ Optional<User> user = orm.entity(User.class)
 ## Tips
 
 1. **Use the metamodel** -- `User_.email` catches typos at compile time; see [Metamodel](metamodel.md)
-2. **Kotlin: choose your style** -- quick queries (`orm.find`, `orm.findAll`) for simple cases, repository builder for complex operations
+2. **Kotlin: choose your style** -- quick queries (`orm.find`, `orm.findAll`) for simple cases, query builder for complex operations
 3. **Java: DSL or Templates** -- DSL for type-safe conditions, SQL Templates for complex SQL like CTEs, window functions, or database-specific features
 4. **Entity graphs load in one query** -- related entities marked with `@FK` are JOINed automatically, no N+1 problems
 5. **Close Java streams** -- always use try-with-resources with `Stream` results

storm-core/src/main/java/st/orm/core/template/SqlDialect.java

@@ -493,4 +493,19 @@ default boolean streamOnlyFetchSize() {
         return false;
     }
 
+    /**
+     * Returns whether streaming result sets require an active transaction for the fetch size hint to take effect.
+     *
+     * <p>Some JDBC drivers silently ignore {@link PreparedStatement#setFetchSize(int)} when the connection is in
+     * auto-commit mode. When this method returns {@code true} and a streaming method ({@code getResultStream()}) is
+     * called without an active transaction, a warning is logged to alert the caller that the entire result set will
+     * be buffered in memory instead of being fetched in batches.</p>
+     *
+     * @return {@code true} if an active transaction is required for cursor-based streaming.
+     * @since 1.10
+     */
+    default boolean streamingRequiresTransaction() {
+        return false;
+    }
+
 }

storm-core/src/main/java/st/orm/core/template/impl/PreparedQueryImpl.java

@@ -48,8 +48,9 @@ public PreparedQueryImpl(@Nonnull RefFactory refFactory,
                              boolean managed,
                              int defaultFetchSize,
                              boolean streamOnlyFetchSize,
+                             boolean streamingRequiresTransaction,
                              @Nonnull Function<Throwable, PersistenceException> exceptionTransformer) {
-        super(refFactory, ignore -> statement, bindVarsHandle, affectedType, versionAware, managed, false, defaultFetchSize, streamOnlyFetchSize, exceptionTransformer);
+        super(refFactory, ignore -> statement, bindVarsHandle, affectedType, versionAware, managed, false, defaultFetchSize, streamOnlyFetchSize, streamingRequiresTransaction, exceptionTransformer);
         this.refFactory = refFactory;
         this.statement = statement;
         this.bindVarsHandle = bindVarsHandle;

storm-core/src/main/java/st/orm/core/template/impl/PreparedStatementTemplateImpl.java

@@ -507,7 +507,7 @@ public Query create(@Nonnull TemplateString template) {
                 } catch (SQLException e) {
                     throw new PersistenceException(e);
                 }
-            }, bindVariables == null ? null : bindVariables.getHandle(), sql.affectedType().orElse(null), sql.versionAware(), false, false, dialect.defaultFetchSize(), dialect.streamOnlyFetchSize(), getExceptionTransformer(sql));
+            }, bindVariables == null ? null : bindVariables.getHandle(), sql.affectedType().orElse(null), sql.versionAware(), false, false, dialect.defaultFetchSize(), dialect.streamOnlyFetchSize(), dialect.streamingRequiresTransaction(), getExceptionTransformer(sql));
         } catch (SqlTemplateException e) {
             throw new PersistenceException(e);
         }

storm-core/src/main/java/st/orm/core/template/impl/QueryImpl.java

@@ -71,14 +71,15 @@ class QueryImpl implements Query {
     private final boolean managed;
     private final int defaultFetchSize;
     private final boolean streamOnlyFetchSize;
+    private final boolean streamingRequiresTransaction;
     private final Function<Throwable, PersistenceException> exceptionTransformer;
 
     QueryImpl(@Nonnull RefFactory refFactory,
               @Nonnull Function<Boolean, PreparedStatement> statement,
               @Nullable BindVarsHandle bindVarsHandle,
               boolean versionAware,
               @Nonnull Function<Throwable, PersistenceException> exceptionTransformer) {
-        this(refFactory, statement, bindVarsHandle, null, versionAware, false, false, 0, false, exceptionTransformer);
+        this(refFactory, statement, bindVarsHandle, null, versionAware, false, false, 0, false, false, exceptionTransformer);
     }
 
     QueryImpl(@Nonnull RefFactory refFactory,
@@ -87,7 +88,7 @@ class QueryImpl implements Query {
               boolean versionAware,
               boolean unsafe,
               @Nonnull Function<Throwable, PersistenceException> exceptionTransformer) {
-        this(refFactory, statement, bindVarsHandle, null, versionAware, false, unsafe, 0, false, exceptionTransformer);
+        this(refFactory, statement, bindVarsHandle, null, versionAware, false, unsafe, 0, false, false, exceptionTransformer);
     }
 
     QueryImpl(@Nonnull RefFactory refFactory,
@@ -97,7 +98,7 @@ class QueryImpl implements Query {
               boolean versionAware,
               boolean unsafe,
               @Nonnull Function<Throwable, PersistenceException> exceptionTransformer) {
-        this(refFactory, statement, bindVarsHandle, affectedType, versionAware, false, unsafe, 0, false, exceptionTransformer);
+        this(refFactory, statement, bindVarsHandle, affectedType, versionAware, false, unsafe, 0, false, false, exceptionTransformer);
     }
 
     QueryImpl(@Nonnull RefFactory refFactory,
@@ -108,7 +109,7 @@ class QueryImpl implements Query {
               boolean managed,
               boolean unsafe,
               @Nonnull Function<Throwable, PersistenceException> exceptionTransformer) {
-        this(refFactory, statement, bindVarsHandle, affectedType, versionAware, managed, unsafe, 0, false, exceptionTransformer);
+        this(refFactory, statement, bindVarsHandle, affectedType, versionAware, managed, unsafe, 0, false, false, exceptionTransformer);
     }
 
     QueryImpl(@Nonnull RefFactory refFactory,
@@ -120,6 +121,7 @@ class QueryImpl implements Query {
               boolean unsafe,
               int defaultFetchSize,
               boolean streamOnlyFetchSize,
+              boolean streamingRequiresTransaction,
               @Nonnull Function<Throwable, PersistenceException> exceptionTransformer) {
         this.refFactory = refFactory;
         this.statement = statement;
@@ -130,6 +132,7 @@ class QueryImpl implements Query {
         this.unsafe = unsafe;
         this.defaultFetchSize = defaultFetchSize;
         this.streamOnlyFetchSize = streamOnlyFetchSize;
+        this.streamingRequiresTransaction = streamingRequiresTransaction;
         this.exceptionTransformer = exceptionTransformer;
     }
 
@@ -147,7 +150,7 @@ class QueryImpl implements Query {
      */
     @Override
     public PreparedQuery prepare() {
-        return MonitoredResource.wrap(new PreparedQueryImpl(refFactory, statement.apply(unsafe), bindVarsHandle, affectedType, versionAware, managed, defaultFetchSize, streamOnlyFetchSize, exceptionTransformer));
+        return MonitoredResource.wrap(new PreparedQueryImpl(refFactory, statement.apply(unsafe), bindVarsHandle, affectedType, versionAware, managed, defaultFetchSize, streamOnlyFetchSize, streamingRequiresTransaction, exceptionTransformer));
     }
 
     /**
@@ -158,7 +161,7 @@ public PreparedQuery prepare() {
      */
     @Override
     public Query managed() {
-        return new QueryImpl(refFactory, statement, bindVarsHandle, affectedType, versionAware, true, unsafe, defaultFetchSize, streamOnlyFetchSize, exceptionTransformer);
+        return new QueryImpl(refFactory, statement, bindVarsHandle, affectedType, versionAware, true, unsafe, defaultFetchSize, streamOnlyFetchSize, streamingRequiresTransaction, exceptionTransformer);
     }
 
     /**
@@ -169,11 +172,11 @@ public Query managed() {
      */
     @Override
     public Query unsafe() {
-        return new QueryImpl(refFactory, statement, bindVarsHandle, affectedType, versionAware, managed, true, defaultFetchSize, streamOnlyFetchSize, exceptionTransformer);
+        return new QueryImpl(refFactory, statement, bindVarsHandle, affectedType, versionAware, managed, true, defaultFetchSize, streamOnlyFetchSize, streamingRequiresTransaction, exceptionTransformer);
     }
 
     private QueryImpl withoutFetchSize() {
-        return new QueryImpl(refFactory, statement, bindVarsHandle, affectedType, versionAware, managed, unsafe, 0, false, exceptionTransformer);
+        return new QueryImpl(refFactory, statement, bindVarsHandle, affectedType, versionAware, managed, unsafe, 0, false, false, exceptionTransformer);
     }
 
     private PreparedStatement getStatement() {
@@ -186,6 +189,38 @@ private void applyFetchSize(@Nonnull PreparedStatement statement) throws SQLExce
         }
     }
 
+    /**
+     * Configures the connection for cursor-based streaming when the dialect requires an active transaction.
+     *
+     * <p>If the connection is in auto-commit mode and the dialect indicates that streaming requires a transaction,
+     * auto-commit is disabled to enable cursor-based result batching. The returned {@code Runnable} restores the
+     * connection to its original state when the stream is closed.</p>
+     *
+     * @param statement the prepared statement whose connection to configure.
+     * @return a cleanup action that restores auto-commit, or {@code null} if no configuration was needed.
+     */
+    private @Nullable Runnable configureStreamingTransaction(@Nonnull PreparedStatement statement) {
+        if (streamingRequiresTransaction && defaultFetchSize != 0) {
+            try {
+                var connection = statement.getConnection();
+                if (connection.getAutoCommit()) {
+                    connection.setAutoCommit(false);
+                    return () -> {
+                        try {
+                            connection.commit();
+                            connection.setAutoCommit(true);
+                        } catch (SQLException e) {
+                            throw new PersistenceException(e);
+                        }
+                    };
+                }
+            } catch (SQLException ignore) {
+                // Unable to determine or change auto-commit state; proceed without cursor-based streaming.
+            }
+        }
+        return null;
+    }
+
     protected boolean closeStatement() {
         return true;
     }
@@ -216,6 +251,7 @@ public Stream<Object[]> getResultStream() {
             boolean close = true;
             try {
                 applyFetchSize(statement);
+                Runnable streamingCleanup = configureStreamingTransaction(statement);
                 ResultSet resultSet = statement.executeQuery();
                 try {
                     int columnCount = resultSet.getMetaData().getColumnCount();
@@ -229,7 +265,7 @@ public Stream<Object[]> getResultStream() {
                                 }
                             })
                                     .takeWhile(Objects::nonNull)
-                                    .onClose(() -> close(resultSet, statement)));
+                                    .onClose(() -> close(resultSet, statement, streamingCleanup)));
                 } finally {
                     if (close) {
                         resultSet.close();
@@ -271,6 +307,7 @@ public <T> Stream<T> getResultStream(@Nonnull Class<T> type) {
         try {
             try {
                 applyFetchSize(statement);
+                Runnable streamingCleanup = configureStreamingTransaction(statement);
                 ResultSet resultSet = statement.executeQuery();
                 int columnCount = resultSet.getMetaData().getColumnCount();
                 var mapper = getObjectMapper(columnCount, type, refFactory)
@@ -285,7 +322,7 @@ public <T> Stream<T> getResultStream(@Nonnull Class<T> type) {
                             }
                         })
                                 .takeWhile(Objects::nonNull)
-                                .onClose(() -> close(resultSet, statement)));
+                                .onClose(() -> close(resultSet, statement, streamingCleanup)));
             } finally {
                 if (close && closeStatement()) {
                     statement.close();
@@ -377,9 +414,20 @@ public long getResultCount() {
     }
 
     protected void close(@Nonnull ResultSet resultSet, @Nonnull PreparedStatement statement) {
+        close(resultSet, statement, null);
+    }
+
+    protected void close(@Nonnull ResultSet resultSet, @Nonnull PreparedStatement statement,
+                          @Nullable Runnable streamingCleanup) {
         try {
             try {
-                resultSet.close();
+                try {
+                    resultSet.close();
+                } finally {
+                    if (streamingCleanup != null) {
+                        streamingCleanup.run();
+                    }
+                }
             } finally {
                 if (closeStatement()) {
                     statement.close();

storm-core/src/test/java/st/orm/core/template/impl/FetchSizeTest.java

@@ -64,6 +64,14 @@ private QueryImpl createQueryWithFetchSize(Connection connection,
                                                String sql,
                                                int defaultFetchSize,
                                                boolean streamOnlyFetchSize) {
+        return createQueryWithFetchSize(connection, sql, defaultFetchSize, streamOnlyFetchSize, false);
+    }
+
+    private QueryImpl createQueryWithFetchSize(Connection connection,
+                                               String sql,
+                                               int defaultFetchSize,
+                                               boolean streamOnlyFetchSize,
+                                               boolean streamingRequiresTransaction) {
         return new QueryImpl(
                 DETACHED_REF_FACTORY,
                 unsafe -> {
@@ -80,6 +88,7 @@ private QueryImpl createQueryWithFetchSize(Connection connection,
                 false,
                 defaultFetchSize,
                 streamOnlyFetchSize,
+                streamingRequiresTransaction,
                 e -> new PersistenceException(e)
         );
     }
@@ -242,4 +251,47 @@ public void testPreparePassesFetchSizeToPreparedQuery() throws Exception {
             assertEquals(6, count);
         }
     }
+
+    @Test
+    public void testStreamingRequiresTransaction_disablesAutoCommitDuringStream() throws Exception {
+        try (Connection connection = dataSource.getConnection()) {
+            assertTrue(connection.getAutoCommit());
+            QueryImpl query = createQueryWithFetchSize(connection, "SELECT id FROM city WHERE id = 1", 100, false, true);
+            try (Stream<Object[]> stream = query.getResultStream()) {
+                Object[] row = stream.findFirst().orElseThrow();
+                assertNotNull(row[0]);
+            }
+            // Auto-commit should be restored after the stream is closed.
+            assertTrue(connection.getAutoCommit(), "Auto-commit should be restored after stream close");
+        }
+    }
+
+    @Test
+    public void testStreamingRequiresTransaction_noChangeWhenAlreadyInTransaction() throws Exception {
+        try (Connection connection = dataSource.getConnection()) {
+            connection.setAutoCommit(false);
+            QueryImpl query = createQueryWithFetchSize(connection, "SELECT id FROM city WHERE id = 1", 100, false, true);
+            try (Stream<Object[]> stream = query.getResultStream()) {
+                Object[] row = stream.findFirst().orElseThrow();
+                assertNotNull(row[0]);
+            }
+            // Auto-commit should remain false (unchanged).
+            assertTrue(!connection.getAutoCommit(), "Auto-commit should remain false when already in a transaction");
+            connection.rollback();
+        }
+    }
+
+    @Test
+    public void testStreamingRequiresTransaction_noChangeWhenFlagDisabled() throws Exception {
+        try (Connection connection = dataSource.getConnection()) {
+            assertTrue(connection.getAutoCommit());
+            QueryImpl query = createQueryWithFetchSize(connection, "SELECT id FROM city WHERE id = 1", 100, false, false);
+            try (Stream<Object[]> stream = query.getResultStream()) {
+                Object[] row = stream.findFirst().orElseThrow();
+                assertNotNull(row[0]);
+            }
+            // Auto-commit should remain true (unchanged).
+            assertTrue(connection.getAutoCommit(), "Auto-commit should remain true when flag is disabled");
+        }
+    }
 }