#1845 - Refactor for Persisting ElementCollection to limit max JDBC batch size

src/main/java/io/ebeaninternal/server/core/PersistRequest.java

@@ -66,6 +66,11 @@ public abstract class PersistRequest extends BeanRequest implements BatchPostExe
     }
   }
 
+  @Override
+  public boolean isFlushQueue() {
+    return false;
+  }
+
   /**
    * Execute a the request or queue/batch it for later execution.
    */

src/main/java/io/ebeaninternal/server/core/PersistRequestUpdateSql.java

@@ -32,6 +32,7 @@ public final class PersistRequestUpdateSql extends PersistRequest {
   private final boolean forceNoBatch;
 
   private boolean batchThisRequest;
+  private boolean flushQueue;
 
   public PersistRequestUpdateSql(SpiEbeanServer server, SpiSqlUpdate sqlUpdate,
                                  SpiTransaction t, PersistExecute persistExecute, boolean forceNoBatch) {
@@ -50,7 +51,7 @@ public final class PersistRequestUpdateSql extends PersistRequest {
 
   @Override
   public void profile(long offset, int flushCount) {
-    profileBase(EVT_UPDATESQL, offset, (short)0, flushCount);
+    profileBase(EVT_UPDATESQL, offset, (short) 0, flushCount);
   }
 
   /**
@@ -77,9 +78,15 @@ public final class PersistRequestUpdateSql extends PersistRequest {
     if (control == null) {
       control = persistExecute.createBatchControl(transaction);
     }
+    flushQueue = true;
     control.addToFlushQueue(this, early);
   }
 
+  @Override
+  public boolean isFlushQueue() {
+    return flushQueue;
+  }
+
   @Override
   public int executeNow() {
     return persistExecute.executeSqlUpdate(this);

src/main/java/io/ebeaninternal/server/persist/BatchPostExecute.java

@@ -10,6 +10,12 @@ package io.ebeaninternal.server.persist;
  */
 public interface BatchPostExecute {
 
+  /**
+   * Return true if this is a queued sql update for element collection or intersection table.
+   * In this case we can executeBatch on the PreparedStatement.
+   */
+  boolean isFlushQueue();
+
   /**
    * Check that the rowCount is correct for this execute. This is for
    * performing concurrency checking in batch execution.

src/main/java/io/ebeaninternal/server/persist/BatchedPstmt.java

@@ -1,7 +1,7 @@
 package io.ebeaninternal.server.persist;
 
-import io.ebeaninternal.api.SpiTransaction;
 import io.ebeaninternal.api.SpiProfileTransactionEvent;
+import io.ebeaninternal.api.SpiTransaction;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -74,12 +74,31 @@ public class BatchedPstmt implements SpiProfileTransactionEvent {
   }
 
   /**
-   * Return the statement.
+   * Return the statement adding the postExecute task.
    */
-  public PreparedStatement getStatement() {
+  public PreparedStatement getStatement(BatchPostExecute postExecute) throws SQLException {
+    if (postExecute.isFlushQueue() && list.size() >= 20) {
+      flushStatementBatch();
+    }
+    list.add(postExecute);
     return pstmt;
   }
 
+  /**
+   * Flush this PreparedStatement using executeBatch() as this was queued element collection
+   * or intersection table sql (and otherwise it can be unlimited size).
+   */
+  private void flushStatementBatch() throws SQLException {
+    final int[] rows = pstmt.executeBatch();
+    if (rows.length != list.size()) {
+      throw new IllegalStateException("Invalid state? rows:" + rows.length + " != " + list.size());
+    }
+    for (BatchPostExecute item : list) {
+      item.postExecute();
+    }
+    list.clear();
+  }
+
   /**
    * Add the BatchPostExecute to the list for post execute processing.
    */

src/main/java/io/ebeaninternal/server/persist/BatchedPstmtHolder.java

@@ -31,30 +31,28 @@ public class BatchedPstmtHolder {
    */
   private int maxSize;
 
-  public BatchedPstmtHolder() {
+  BatchedPstmtHolder() {
   }
 
   /**
    * Return the PreparedStatement if it has already been used in this Batch.
    * This will return null if no matching PreparedStatement is found.
    */
-  public PreparedStatement getStmt(String stmtKey, BatchPostExecute postExecute) {
-    BatchedPstmt batchedPstmt = getBatchedPstmt(stmtKey, postExecute);
-    return (batchedPstmt == null) ? null : batchedPstmt.getStatement();
+  PreparedStatement getStmt(String stmtKey, BatchPostExecute postExecute) throws SQLException {
+    BatchedPstmt batchedPstmt = getBatchedPstmt(stmtKey);
+    return (batchedPstmt == null) ? null : batchedPstmt.getStatement(postExecute);
   }
 
   /**
    * Return the BatchedPstmt that holds the batched statement.
    */
-  public BatchedPstmt getBatchedPstmt(String stmtKey, BatchPostExecute postExecute) {
+  public BatchedPstmt getBatchedPstmt(String stmtKey) {
 
     BatchedPstmt bs = stmtMap.get(stmtKey);
     if (bs == null) {
       // the PreparedStatement has need been created
       return null;
     }
-    // add the post execute processing for this bean/row
-    bs.add(postExecute);
 
     // maintain a max batch size for any given batched stmt.
     // Used to determine when to flush.

src/main/java/io/ebeaninternal/server/persist/ExeCallableSql.java

@@ -74,7 +74,7 @@ class ExeCallableSql {
 
     CallableStatement cstmt;
     if (batchThisRequest) {
-      cstmt = pstmtFactory.getCstmt(t, logSql, sql, request);
+      cstmt = pstmtFactory.getCstmtBatch(t, logSql, sql, request);
     } else {
       if (logSql) {
         t.logSql(TrimLogSql.trim(sql));

src/main/java/io/ebeaninternal/server/persist/ExeOrmUpdate.java

@@ -92,13 +92,11 @@ class ExeOrmUpdate {
 
     ormUpdate.setGeneratedSql(sql);
 
-    boolean logSql = request.isLogSql();
-
     PreparedStatement pstmt;
     if (batchThisRequest) {
-      pstmt = pstmtFactory.getPstmt(t, logSql, sql, request);
+      pstmt = pstmtFactory.getPstmtBatch(t, sql, request);
     } else {
-      if (logSql) {
+      if (t.isLogSql()) {
         t.logSql(sql);
       }
       pstmt = pstmtFactory.getPstmt(t, sql, false);

src/main/java/io/ebeaninternal/server/persist/ExeUpdateSql.java

@@ -92,10 +92,11 @@ class ExeUpdateSql {
     // process named parameters if required
     String sql = updateSql.getBaseSql();
     sql = BindParamsParser.parse(bindParams, sql);
+    parseUpdate(sql, request);
 
     PreparedStatement pstmt;
     if (batchThisRequest) {
-      pstmt = pstmtFactory.getPstmt(t, request.isLogSql(), sql, request);
+      pstmt = pstmtFactory.getPstmtBatch(t, sql, request);
     } else {
       pstmt = pstmtFactory.getPstmt(t, sql, request.isGetGeneratedKeys());
     }
@@ -112,8 +113,6 @@ class ExeUpdateSql {
     request.setBindLog(bindLog);
     updateSql.setGeneratedSql(sql);
 
-    // derive the statement type (for TransactionEvent)
-    parseUpdate(sql, request);
     if (batchThisRequest) {
       request.logSqlBatchBind();
     }

src/main/java/io/ebeaninternal/server/persist/PstmtFactory.java

@@ -23,7 +23,7 @@ class PstmtFactory {
   /**
    * Get a callable statement without any batching.
    */
-  public CallableStatement getCstmt(SpiTransaction t, String sql) throws SQLException {
+  CallableStatement getCstmt(SpiTransaction t, String sql) throws SQLException {
     Connection conn = t.getInternalConnection();
     return conn.prepareCall(sql);
   }
@@ -31,7 +31,7 @@ class PstmtFactory {
   /**
    * Get a prepared statement without any batching.
    */
-  public PreparedStatement getPstmt(SpiTransaction t, String sql, boolean getGeneratedKeys) throws SQLException {
+  PreparedStatement getPstmt(SpiTransaction t, String sql, boolean getGeneratedKeys) throws SQLException {
     Connection conn = t.getInternalConnection();
     if (getGeneratedKeys) {
       return conn.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS);
@@ -43,21 +43,20 @@ class PstmtFactory {
   /**
    * Return a prepared statement taking into account batch requirements.
    */
-  public PreparedStatement getPstmt(SpiTransaction t, boolean logSql, String sql, BatchPostExecute batchExe) throws SQLException {
+  PreparedStatement getPstmtBatch(SpiTransaction t, String sql, BatchPostExecute batchExe) throws SQLException {
 
     BatchedPstmtHolder batch = t.getBatchControl().getPstmtHolder();
-    PreparedStatement stmt = batch.getStmt(sql, batchExe);
-
-    if (stmt != null) {
-      return stmt;
+    BatchedPstmt existingStmt = batch.getBatchedPstmt(sql);
+    if (existingStmt != null) {
+      return existingStmt.getStatement(batchExe);
     }
 
-    if (logSql) {
+    if (t.isLogSql()) {
       t.logSql(TrimLogSql.trim(sql));
     }
 
     Connection conn = t.getInternalConnection();
-    stmt = conn.prepareStatement(sql);
+    PreparedStatement stmt = conn.prepareStatement(sql);
 
     BatchedPstmt bs = new BatchedPstmt(stmt, false, sql, t);
     batch.addStmt(bs, batchExe);
@@ -67,7 +66,7 @@ class PstmtFactory {
   /**
    * Return a callable statement taking into account batch requirements.
    */
-  public CallableStatement getCstmt(SpiTransaction t, boolean logSql, String sql, BatchPostExecute batchExe) throws SQLException {
+  CallableStatement getCstmtBatch(SpiTransaction t, boolean logSql, String sql, BatchPostExecute batchExe) throws SQLException {
 
     BatchedPstmtHolder batch = t.getBatchControl().getPstmtHolder();
     CallableStatement stmt = (CallableStatement) batch.getStmt(sql, batchExe);

src/main/java/io/ebeaninternal/server/persist/dml/DeleteHandler.java

@@ -35,7 +35,7 @@ public class DeleteHandler extends DmlHandler {
 
     PreparedStatement pstmt;
     if (persistRequest.isBatched()) {
-      pstmt = getPstmt(t, sql, persistRequest, false);
+      pstmt = getPstmtBatch(t, sql, persistRequest, false);
     } else {
       pstmt = getPstmt(t, sql, false);
     }

src/main/java/io/ebeaninternal/server/persist/dml/DmlHandler.java

@@ -250,13 +250,13 @@ public abstract class DmlHandler implements PersistHandler, BindableRequest {
   /**
    * Return a prepared statement taking into account batch requirements.
    */
-  PreparedStatement getPstmt(SpiTransaction t, String sql, PersistRequestBean<?> request, boolean genKeys) throws SQLException {
+  PreparedStatement getPstmtBatch(SpiTransaction t, String sql, PersistRequestBean<?> request, boolean genKeys) throws SQLException {
 
     BatchedPstmtHolder batch = t.getBatchControl().getPstmtHolder();
-    batchedPstmt = batch.getBatchedPstmt(sql, request);
+    batchedPstmt = batch.getBatchedPstmt(sql);
     if (batchedPstmt != null) {
       batchedStatus = BATCHED;
-      return batchedPstmt.getStatement();
+      return batchedPstmt.getStatement(request);
     }
 
     batchedStatus = BATCHED_FIRST;

src/main/java/io/ebeaninternal/server/persist/dml/InsertHandler.java

@@ -91,7 +91,7 @@ public class InsertHandler extends DmlHandler {
 
     PreparedStatement pstmt;
     if (persistRequest.isBatched()) {
-      pstmt = getPstmt(t, sql, persistRequest, useGeneratedKeys);
+      pstmt = getPstmtBatch(t, sql, persistRequest, useGeneratedKeys);
     } else {
       pstmt = getPstmt(t, sql, useGeneratedKeys);
     }

src/main/java/io/ebeaninternal/server/persist/dml/UpdateHandler.java

@@ -46,7 +46,7 @@ public class UpdateHandler extends DmlHandler {
 
     PreparedStatement pstmt;
     if (persistRequest.isBatched()) {
-      pstmt = getPstmt(t, sql, persistRequest, false);
+      pstmt = getPstmtBatch(t, sql, persistRequest, false);
     } else {
       pstmt = getPstmt(t, sql, false);
     }