Use autocommit false with findIterate for Postgres

ebean-api/src/main/java/io/ebean/config/dbplatform/DatabasePlatform.java

@@ -167,6 +167,7 @@ public class DatabasePlatform {
    * findIterate() and findVisit().
    */
   protected boolean forwardOnlyHintOnFindIterate;
+  protected boolean autoCommitFalseOnFindIterate;
 
   /**
    * If set then use the CONCUR_UPDATABLE hint when creating ResultSets.
@@ -528,6 +529,10 @@ public void setForwardOnlyHintOnFindIterate(boolean forwardOnlyHintOnFindIterate
     this.forwardOnlyHintOnFindIterate = forwardOnlyHintOnFindIterate;
   }
 
+  public boolean autoCommitFalseOnFindIterate() {
+    return autoCommitFalseOnFindIterate;
+  }
+
   /**
    * Return true if the ResultSet CONCUR_UPDATABLE Hint should be used on
    * createNativeSqlTree() PreparedStatements.
@@ -780,4 +785,5 @@ public String inlineSqlComment(String label) {
   public String inlineSqlHint(String hint) {
     return "/*+ " + hint + " */ ";
   }
+
 }

ebean-core/src/main/java/io/ebeaninternal/api/SpiTransaction.java

@@ -362,4 +362,14 @@ default int depth() {
    * Set the transaction to be inactive via external transaction manager.
    */
   void deactivateExternal();
+
+  /**
+   * Set autocommit to false for a findIterate query.
+   * <p>
+   * This is done for specific platforms that need it, in order to make
+   * use cursors to stream a large or unbounded query result to the client.
+   */
+  default void setAutoCommitOnFindIterate() {
+    throw new UnsupportedOperationException();
+  }
 }

ebean-core/src/main/java/io/ebeaninternal/server/core/AbstractSqlQueryRequest.java

@@ -1,7 +1,6 @@
 package io.ebeaninternal.server.core;
 
 import io.ebean.CancelableQuery;
-import io.ebean.Transaction;
 import io.ebean.util.JdbcClose;
 import io.ebeaninternal.api.*;
 import io.ebeaninternal.server.persist.Binder;
@@ -98,6 +97,13 @@ public String getBindLog() {
   public void setDefaultFetchBuffer(int fetchSize) {
     query.setDefaultFetchBuffer(fetchSize);
   }
+
+  public void setAutoCommitOnFindIterate() {
+    if (createdTransaction) {
+      transaction.setAutoCommitOnFindIterate();
+    }
+  }
+
   /**
    * Close the underlying resources.
    */

ebean-core/src/main/java/io/ebeaninternal/server/core/InternalConfiguration.java

@@ -301,12 +301,12 @@ AutoTuneService createAutoTuneService(SpiEbeanServer server) {
   }
 
   DtoQueryEngine createDtoQueryEngine() {
-    return new DtoQueryEngine(binder, config.getJdbcFetchSizeFindEach(), config.getJdbcFetchSizeFindList());
+    return new DtoQueryEngine(binder, config.getJdbcFetchSizeFindEach(), config.getJdbcFetchSizeFindList(), databasePlatform.autoCommitFalseOnFindIterate());
   }
 
   RelationalQueryEngine createRelationalQueryEngine() {
     return new DefaultRelationalQueryEngine(binder, config.getDatabaseBooleanTrue(), config.getPlatformConfig().getDbUuid().useBinaryOptimized(),
-      config.getJdbcFetchSizeFindEach(), config.getJdbcFetchSizeFindList());
+      config.getJdbcFetchSizeFindEach(), config.getJdbcFetchSizeFindList(), databasePlatform.autoCommitFalseOnFindIterate());
   }
 
   OrmQueryEngine createOrmQueryEngine() {

ebean-core/src/main/java/io/ebeaninternal/server/core/OrmQueryRequest.java

@@ -786,4 +786,10 @@ public void unmodifiableFreeze(EntityBean bean) {
       beanDescriptor.freeze(bean);
     }
   }
+
+  public void setAutoCommitOnFindIterate() {
+    if (createdTransaction) {
+      transaction.setAutoCommitOnFindIterate();
+    }
+  }
 }

ebean-core/src/main/java/io/ebeaninternal/server/query/CQueryEngine.java

@@ -37,12 +37,14 @@ public final class CQueryEngine {
   private final CQueryBuilder queryBuilder;
   private final CQueryHistorySupport historySupport;
   private final DatabasePlatform dbPlatform;
+  private final boolean autoCommitFalseOnFindIterate;
 
   public CQueryEngine(DatabaseBuilder.Settings config, DatabasePlatform dbPlatform, Binder binder, Map<String, String> asOfTableMapping, Map<String, String> draftTableMap) {
     this.dbPlatform = dbPlatform;
     this.defaultFetchSizeFindEach = config.getJdbcFetchSizeFindEach();
     this.defaultFetchSizeFindList = config.getJdbcFetchSizeFindList();
     this.forwardOnlyHintOnFindIterate = dbPlatform.forwardOnlyHintOnFindIterate();
+    this.autoCommitFalseOnFindIterate = dbPlatform.autoCommitFalseOnFindIterate();
     this.historySupport = new CQueryHistorySupport(dbPlatform.historySupport(), asOfTableMapping, config.getAsOfSysPeriod());
     this.queryBuilder = new CQueryBuilder(config, dbPlatform, binder, historySupport, new CQueryDraftSupport(draftTableMap));
   }
@@ -181,6 +183,9 @@ public <T> QueryIterator<T> findIterate(OrmQueryRequest<T> request) {
       if (defaultFetchSizeFindEach > 0) {
         request.setDefaultFetchBuffer(defaultFetchSizeFindEach);
       }
+      if (autoCommitFalseOnFindIterate) {
+        request.setAutoCommitOnFindIterate();
+      }
       if (!cquery.prepareBindExecuteQueryForwardOnly(forwardOnlyHintOnFindIterate)) {
         // query has been cancelled already
         return null;

ebean-core/src/main/java/io/ebeaninternal/server/query/DefaultRelationalQueryEngine.java

@@ -15,6 +15,8 @@
 import io.ebeaninternal.server.persist.Binder;
 
 import jakarta.persistence.PersistenceException;
+
+import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Consumer;
@@ -28,15 +30,18 @@ public final class DefaultRelationalQueryEngine implements RelationalQueryEngine
   private final Binder binder;
   private final String dbTrueValue;
   private final boolean binaryOptimizedUUID;
+  private final boolean autoCommitFalseOnFindIterate;
   private final TimedMetricMap timedMetricMap;
   private final int defaultFetchSizeFindEach;
   private final int defaultFetchSizeFindList;
 
   public DefaultRelationalQueryEngine(Binder binder, String dbTrueValue, boolean binaryOptimizedUUID,
-                                      int defaultFetchSizeFindEach, int defaultFetchSizeFindList) {
+                                      int defaultFetchSizeFindEach, int defaultFetchSizeFindList,
+                                      boolean autoCommitFalseOnFindIterate) {
     this.binder = binder;
     this.dbTrueValue = dbTrueValue == null ? "true" : dbTrueValue;
     this.binaryOptimizedUUID = binaryOptimizedUUID;
+    this.autoCommitFalseOnFindIterate = autoCommitFalseOnFindIterate;
     this.timedMetricMap = MetricFactory.get().createTimedMetricMap("sql.query.");
     this.defaultFetchSizeFindEach = defaultFetchSizeFindEach;
     this.defaultFetchSizeFindList = defaultFetchSizeFindList;
@@ -61,13 +66,20 @@ private String errMsg(String msg, String sql) {
     return "Query threw SQLException:" + msg + " Query was:" + sql;
   }
 
+  private <T> void prepareForIterate(RelationalQueryRequest request) throws SQLException {
+    if (defaultFetchSizeFindEach > 0) {
+      request.setDefaultFetchBuffer(defaultFetchSizeFindEach);
+    }
+    if (autoCommitFalseOnFindIterate) {
+      request.setAutoCommitOnFindIterate();
+    }
+    request.executeSql(binder, SpiQuery.Type.ITERATE);
+  }
+
   @Override
   public void findEach(RelationalQueryRequest request, RowConsumer consumer) {
     try {
-      if (defaultFetchSizeFindEach > 0) {
-        request.setDefaultFetchBuffer(defaultFetchSizeFindEach);
-      }
-      request.executeSql(binder, SpiQuery.Type.ITERATE);
+      prepareForIterate(request);
       request.mapEach(consumer);
       request.logSummary();
 
@@ -82,10 +94,7 @@ public void findEach(RelationalQueryRequest request, RowConsumer consumer) {
   @Override
   public <T> void findEach(RelationalQueryRequest request, RowReader<T> reader, Predicate<T> consumer) {
     try {
-      if (defaultFetchSizeFindEach > 0) {
-        request.setDefaultFetchBuffer(defaultFetchSizeFindEach);
-      }
-      request.executeSql(binder, SpiQuery.Type.ITERATE);
+      prepareForIterate(request);
       while (request.next()) {
         if (!consumer.test(reader.read())) {
           break;

ebean-core/src/main/java/io/ebeaninternal/server/query/DtoQueryEngine.java

@@ -17,11 +17,13 @@ public final class DtoQueryEngine {
   private final Binder binder;
   private final int defaultFetchSizeFindEach;
   private final int defaultFetchSizeFindList;
+  private final boolean autoCommitFalseOnFindIterate;
 
-  public DtoQueryEngine(Binder binder, int defaultFetchSizeFindEach, int defaultFetchSizeFindList) {
+  public DtoQueryEngine(Binder binder, int defaultFetchSizeFindEach, int defaultFetchSizeFindList, boolean autoCommitFalseOnFindIterate) {
     this.binder = binder;
     this.defaultFetchSizeFindEach = defaultFetchSizeFindEach;
     this.defaultFetchSizeFindList = defaultFetchSizeFindList;
+    this.autoCommitFalseOnFindIterate = autoCommitFalseOnFindIterate;
   }
 
   public <T> List<T> findList(DtoQueryRequest<T> request) {
@@ -43,12 +45,19 @@ public <T> List<T> findList(DtoQueryRequest<T> request) {
     }
   }
 
+  private <T> void prepareForIterate(DtoQueryRequest<T> request) throws SQLException {
+    if (defaultFetchSizeFindEach > 0) {
+      request.setDefaultFetchBuffer(defaultFetchSizeFindEach);
+    }
+    if (autoCommitFalseOnFindIterate) {
+      request.setAutoCommitOnFindIterate();
+    }
+    request.executeSql(binder, SpiQuery.Type.ITERATE);
+  }
+
   public <T> QueryIterator<T> findIterate(DtoQueryRequest<T> request) {
     try {
-      if (defaultFetchSizeFindEach > 0) {
-        request.setDefaultFetchBuffer(defaultFetchSizeFindEach);
-      }
-      request.executeSql(binder, SpiQuery.Type.ITERATE);
+      prepareForIterate(request);
       return new DtoQueryIterator<>(request);
     } catch (SQLException e) {
       throw new PersistenceException(errMsg(e.getMessage(), request.getSql()), e);
@@ -57,10 +66,7 @@ public <T> QueryIterator<T> findIterate(DtoQueryRequest<T> request) {
 
   public <T> void findEach(DtoQueryRequest<T> request, Consumer<T> consumer) {
     try {
-      if (defaultFetchSizeFindEach > 0) {
-        request.setDefaultFetchBuffer(defaultFetchSizeFindEach);
-      }
-      request.executeSql(binder, SpiQuery.Type.ITERATE);
+      prepareForIterate(request);
       while (request.next()) {
         consumer.accept(request.readNextBean());
       }
@@ -73,11 +79,8 @@ public <T> void findEach(DtoQueryRequest<T> request, Consumer<T> consumer) {
 
   public <T> void findEach(DtoQueryRequest<T> request, int batchSize, Consumer<List<T>> consumer) {
     try {
+      prepareForIterate(request);
       List<T> buffer = new ArrayList<>();
-      if (defaultFetchSizeFindEach > 0) {
-        request.setDefaultFetchBuffer(defaultFetchSizeFindEach);
-      }
-      request.executeSql(binder, SpiQuery.Type.ITERATE);
       while (request.next()) {
         buffer.add(request.readNextBean());
         if (buffer.size() >= batchSize) {
@@ -98,10 +101,7 @@ public <T> void findEach(DtoQueryRequest<T> request, int batchSize, Consumer<Lis
 
   public <T> void findEachWhile(DtoQueryRequest<T> request, Predicate<T> consumer) {
     try {
-      if (defaultFetchSizeFindEach > 0) {
-        request.setDefaultFetchBuffer(defaultFetchSizeFindEach);
-      }
-      request.executeSql(binder, SpiQuery.Type.ITERATE);
+      prepareForIterate(request);
       while (request.next()) {
         if (!consumer.test(request.readNextBean())) {
           break;

ebean-core/src/main/java/io/ebeaninternal/server/transaction/ImplicitReadOnlyTransaction.java

@@ -36,7 +36,7 @@ final class ImplicitReadOnlyTransaction implements SpiTransaction, TxnProfileEve
   /**
    * Set false when using autoCommit (as a performance optimisation for the read-only case).
    */
-  private final boolean useCommit;
+  private boolean useCommit;
   private final TransactionManager manager;
   private final SpiTxnLogger logger;
   private final boolean logSql;
@@ -550,6 +550,16 @@ public void setRollbackOnly() {
     throw new IllegalStateException(notExpectedMessage);
   }
 
+  @Override
+  public void setAutoCommitOnFindIterate() {
+    try {
+      connection.setAutoCommit(false);
+      useCommit = true;
+    } catch (SQLException e) {
+      throw new PersistenceException(e);
+    }
+  }
+
   @Override
   public void rollbackAndContinue() {
     // do nothing

ebean-core/src/main/java/io/ebeaninternal/server/transaction/TransactionFactory.java

@@ -34,7 +34,7 @@ abstract class TransactionFactory {
   /**
    * Set the Transaction Isolation level if required.
    */
-  final SpiTransaction setIsolationLevel(SpiTransaction t, boolean explicit, int isolationLevel) {
+  final SpiTransaction setIsolationLevel(SpiTransaction t, int isolationLevel) {
     if (isolationLevel > -1) {
       Connection connection = t.internalConnection();
       try {

ebean-core/src/main/java/io/ebeaninternal/server/transaction/TransactionFactoryBasic.java

@@ -39,8 +39,8 @@ public final SpiTransaction createTransaction(boolean explicit, int isolationLev
     Connection connection = null;
     try {
       connection = dataSource.getConnection();
-      SpiTransaction t = create(explicit, connection);
-      return setIsolationLevel(t, explicit, isolationLevel);
+      SpiTransaction t = manager.createTransaction(explicit, connection);
+      return setIsolationLevel(t, isolationLevel);
     } catch (PersistenceException ex) {
       JdbcClose.close(connection);
       throw ex;
@@ -49,8 +49,4 @@ public final SpiTransaction createTransaction(boolean explicit, int isolationLev
     }
   }
 
-  private SpiTransaction create(boolean explicit, Connection c) {
-    return manager.createTransaction(explicit, c);
-  }
-
 }

ebean-core/src/main/java/io/ebeaninternal/server/transaction/TransactionFactoryTenant.java

@@ -25,26 +25,32 @@ class TransactionFactoryTenant extends TransactionFactory {
 
   @Override
   public SpiTransaction createReadOnlyTransaction(Object tenantId, boolean useMaster) {
-    return create(false, tenantId);
+    Connection connection = null;
+    try {
+      if (tenantId == null) {
+        tenantId = tenantProvider.currentId();
+      }
+      connection = dataSourceSupplier.readOnlyConnection(tenantId, useMaster);
+      return new ImplicitReadOnlyTransaction(manager, connection, tenantId);
+
+    } catch (PersistenceException ex) {
+      JdbcClose.close(connection);
+      throw ex;
+
+    } catch (SQLException ex) {
+      throw new PersistenceException(ex);
+    }
   }
 
   @Override
   public final SpiTransaction createTransaction(boolean explicit, int isolationLevel) {
-    SpiTransaction t = create(explicit, null);
-    return setIsolationLevel(t, explicit, isolationLevel);
-  }
-
-  private SpiTransaction create(boolean explicit, Object tenantId) {
     Connection connection = null;
     try {
-      if (tenantId == null) {
-        // tenantId not set (by lazy loading) so get current tenantId
-        tenantId = tenantProvider.currentId();
-      }
+      Object tenantId = tenantProvider.currentId();
       connection = dataSourceSupplier.connection(tenantId);
       SpiTransaction transaction = manager.createTransaction(explicit, connection);
       transaction.setTenantId(tenantId);
-      return transaction;
+      return setIsolationLevel(transaction, isolationLevel);
 
     } catch (PersistenceException ex) {
       JdbcClose.close(connection);

ebean-test/src/test/java/main/StartPostgres.java

@@ -5,7 +5,7 @@
 public class StartPostgres {
 
   public static void main(String[] args) {
-    PostgresContainer.builder("15")
+    PostgresContainer.builder("17")
       .dbName("unit")
       //.port(6432)
       //.user("unit")