#1630 - ENH: Improve jdbc batch use when persisting ManyToMany intersection changes

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

@@ -27,6 +27,7 @@ import io.ebeaninternal.server.persist.BatchedSqlException;
 import io.ebeaninternal.server.persist.DeleteMode;
 import io.ebeaninternal.server.persist.Flags;
 import io.ebeaninternal.server.persist.PersistExecute;
+import io.ebeaninternal.server.persist.SaveManyBeans;
 import io.ebeaninternal.server.transaction.BeanPersistIdMap;
 import io.ebeanservice.docstore.api.DocStoreUpdate;
 import io.ebeanservice.docstore.api.DocStoreUpdateContext;
@@ -174,7 +175,7 @@ public final class PersistRequestBean<T> extends PersistRequest implements BeanP
   private int pendingPostUpdateNotify;
 
   /**
-   * Set to true when post execute has occured (so includes batch flush).
+   * Set to true when post execute has occurred (so includes batch flush).
    */
   private boolean postExecute;
 
@@ -183,6 +184,11 @@ public final class PersistRequestBean<T> extends PersistRequest implements BeanP
    */
   private boolean complete;
 
+  /**
+   * Many to many intersection table changes that are held for later batch processing.
+   */
+  private SaveManyBeans saveManyIntersection;
+
   public PersistRequestBean(SpiEbeanServer server, T bean, Object parentBean, BeanManager<T> mgr, SpiTransaction t,
                             PersistExecute persistExecute, PersistRequest.Type type, int flags) {
 
@@ -955,6 +961,13 @@ public final class PersistRequestBean<T> extends PersistRequest implements BeanP
     if (isLogSummary()) {
       logSummary();
     }
+    saveQueuedManyIntersection();
+  }
+
+  private void saveQueuedManyIntersection() {
+    if (saveManyIntersection != null) {
+      saveManyIntersection.saveIntersectionBatch();
+    }
   }
 
   /**
@@ -1119,6 +1132,7 @@ public final class PersistRequestBean<T> extends PersistRequest implements BeanP
       updatedManysOnly = true;
       setNotifyCache();
       addPostCommitListeners();
+      saveQueuedManyIntersection();
     }
     notifyCacheOnComplete();
     postUpdateNotify();
@@ -1454,4 +1468,18 @@ public final class PersistRequestBean<T> extends PersistRequest implements BeanP
   public String getSelectLastInsertedId() {
     return beanDescriptor.getSelectLastInsertedId(publish);
   }
+
+  /**
+   * Return true if the intersection table updates should be queued and batched.
+   */
+  public boolean isQueueManyIntersection() {
+    return !postExecute;
+  }
+
+  /**
+   * The intersection table updates to the batch executed later on postExecute.
+   */
+  public void setManyIntersection(SaveManyBeans saveManyIntersection) {
+    this.saveManyIntersection = saveManyIntersection;
+  }
 }

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

@@ -3,6 +3,7 @@ package io.ebeaninternal.server.core;
 import io.ebeaninternal.api.SpiEbeanServer;
 import io.ebeaninternal.api.SpiSqlUpdate;
 import io.ebeaninternal.api.SpiTransaction;
+import io.ebeaninternal.server.persist.BatchControl;
 import io.ebeaninternal.server.persist.PersistExecute;
 
 /**
@@ -58,6 +59,25 @@ public final class PersistRequestUpdateSql extends PersistRequest {
     return executeOrQueue();
   }
 
+  /**
+   * Execute using jdbc batch.
+   */
+  public void executeAddBatch() {
+    this.addBatch = true;
+    persistExecute.executeSqlUpdate(this);
+  }
+
+  /**
+   * Add this request to BatchControl to flush later.
+   */
+  public void addToFlushQueue(boolean early) {
+    BatchControl control = transaction.getBatchControl();
+    if (control == null) {
+      control = persistExecute.createBatchControl(transaction);
+    }
+    control.addToFlushQueue(this, early);
+  }
+
   @Override
   public int executeNow() {
     return persistExecute.executeSqlUpdate(this);

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

@@ -103,6 +103,11 @@ public interface Persister {
    */
   void visitMetrics(MetricVisitor visitor);
 
+  /**
+   * Execute or queue the update.
+   */
+  void executeOrQueue(SpiSqlUpdate update, SpiTransaction t, boolean queue);
+
   /**
    * Add the statement to JDBC batch for later execution via executeBatch.
    */

src/main/java/io/ebeaninternal/server/deploy/IntersectionRow.java

@@ -1,8 +1,8 @@
 package io.ebeaninternal.server.deploy;
 
-import io.ebean.SqlUpdate;
 import io.ebeaninternal.api.BindParams;
 import io.ebeaninternal.api.SpiEbeanServer;
+import io.ebeaninternal.api.SpiSqlUpdate;
 import io.ebeaninternal.server.core.DefaultSqlUpdate;
 import io.ebeaninternal.server.expression.DefaultExpressionRequest;
 import io.ebeaninternal.server.expression.IdInExpression;
@@ -45,7 +45,7 @@ public class IntersectionRow {
     values.put(key, value);
   }
 
-  public SqlUpdate createInsert(SpiEbeanServer server) {
+  public SpiSqlUpdate createInsert(SpiEbeanServer server) {
 
     BindParams bindParams = new BindParams();
 
@@ -73,7 +73,7 @@ public class IntersectionRow {
     return new DefaultSqlUpdate(server, sb.toString(), bindParams);
   }
 
-  public SqlUpdate createDelete(SpiEbeanServer server, DeleteMode deleteMode) {
+  public SpiSqlUpdate createDelete(SpiEbeanServer server, DeleteMode deleteMode) {
 
     BindParams bindParams = new BindParams();
 
@@ -108,7 +108,7 @@ public class IntersectionRow {
     return new DefaultSqlUpdate(server, sb.toString(), bindParams);
   }
 
-  public SqlUpdate createDeleteChildren(SpiEbeanServer server) {
+  public SpiSqlUpdate createDeleteChildren(SpiEbeanServer server) {
 
     BindParams bindParams = new BindParams();
 

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

@@ -3,6 +3,7 @@ package io.ebeaninternal.server.persist;
 import io.ebeaninternal.api.SpiTransaction;
 import io.ebeaninternal.server.core.PersistRequest;
 import io.ebeaninternal.server.core.PersistRequestBean;
+import io.ebeaninternal.server.core.PersistRequestUpdateSql;
 import io.ebeaninternal.server.deploy.BeanDescriptor;
 import io.ebeaninternal.server.deploy.BeanPropertyAssocOne;
 
@@ -10,6 +11,7 @@ import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
+import java.util.List;
 
 /**
  * Controls the batch ordering of persist requests.
@@ -68,6 +70,9 @@ public final class BatchControl {
    */
   private int bufferMax;
 
+  private Queue earlyQueue;
+  private Queue lateQueue;
+
   /**
    * Create for a given transaction, PersistExecute, default size and getGeneratedKeys.
    */
@@ -224,14 +229,14 @@ public final class BatchControl {
    * Flush without resetting the topOrder (maintains the depth info).
    */
   public void flush() throws BatchedSqlException {
-    flush(false);
+    flushBuffer(false);
   }
 
   /**
    * Flush with a reset the topOrder (fully empty the batch).
    */
   public void flushReset() throws BatchedSqlException {
-    flush(true);
+    flushBuffer(true);
   }
 
   /**
@@ -243,10 +248,22 @@ public final class BatchControl {
     maxDepth = 0;
   }
 
+  private void flushBuffer(boolean resetTop) throws BatchedSqlException {
+    flushInternal(resetTop);
+    flushQueue(earlyQueue);
+    flushQueue(lateQueue);
+  }
+
+  private void flushQueue(Queue queue) throws BatchedSqlException {
+    if (queue != null && queue.flush() && !pstmtHolder.isEmpty()) {
+      flushPstmtHolder();
+    }
+  }
+
   /**
    * execute all the requests currently queued or batched.
    */
-  private void flush(boolean resetTop) throws BatchedSqlException {
+  private void flushInternal(boolean resetTop) throws BatchedSqlException {
 
     try {
       bufferMax = 0;
@@ -367,4 +384,43 @@ public final class BatchControl {
   public int[] execute(String key, boolean getGeneratedKeys) throws SQLException {
     return pstmtHolder.execute(key, getGeneratedKeys);
   }
+
+  /**
+   * Add a SqlUpdate request to execute after flush.
+   */
+  public void addToFlushQueue(PersistRequestUpdateSql request, boolean early) {
+    if (early) {
+      // add it to the early queue
+      if (earlyQueue == null) {
+        earlyQueue = new Queue();
+      }
+      earlyQueue.add(request);
+    } else {
+      // add it to the late queue
+      if (lateQueue == null) {
+        lateQueue = new Queue();
+      }
+      lateQueue.add(request);
+    }
+  }
+
+  private static class Queue {
+
+    private final List<PersistRequestUpdateSql> queue = new ArrayList<>();
+
+    boolean flush() {
+      if (queue.isEmpty()) {
+        return false;
+      }
+      for (PersistRequestUpdateSql request : queue) {
+        request.executeAddBatch();
+      }
+      queue.clear();
+      return true;
+    }
+
+    void add(PersistRequestUpdateSql request) {
+      queue.add(request);
+    }
+  }
 }

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

@@ -141,12 +141,24 @@ public final class DefaultPersister implements Persister {
     }
   }
 
+  @Override
+  public void executeOrQueue(SpiSqlUpdate update, SpiTransaction t, boolean queue) {
+    if (queue) {
+      addToFlushQueue(update, t, false);
+    } else {
+      executeSqlUpdate(update, t);
+    }
+  }
+
+  private void addToFlushQueue(SpiSqlUpdate update, SpiTransaction t, boolean early) {
+    new PersistRequestUpdateSql(server, update, t, persistExecute).addToFlushQueue(early);
+  }
+
   /**
    * Execute the updateSql.
    */
   @Override
   public int executeSqlUpdate(SqlUpdate updSql, Transaction t) {
-
     return executeOrQueue(new PersistRequestUpdateSql(server, (SpiSqlUpdate) updSql, (SpiTransaction) t, persistExecute));
   }
 
@@ -958,12 +970,19 @@ public final class DefaultPersister implements Persister {
     }
   }
 
-  void deleteAssocManyIntersection(EntityBean bean, BeanPropertyAssocMany<?> many, Transaction t, boolean publish) {
+  void deleteManyIntersection(EntityBean bean, BeanPropertyAssocMany<?> many, SpiTransaction t, boolean publish, boolean queue) {
 
-    // delete all intersection rows for this bean
+    SpiSqlUpdate sqlDelete = deleteAllIntersection(bean, many, publish);
+    if (queue) {
+      addToFlushQueue(sqlDelete, t, true);
+    } else {
+      executeSqlUpdate(sqlDelete, t);
+    }
+  }
+
+  private SpiSqlUpdate deleteAllIntersection(EntityBean bean, BeanPropertyAssocMany<?> many, boolean publish) {
     IntersectionRow intRow = many.buildManyToManyDeleteChildren(bean, publish);
-    SqlUpdate sqlDelete = intRow.createDeleteChildren(server);
-    executeSqlUpdate(sqlDelete, t);
+    return intRow.createDeleteChildren(server);
   }
 
   /**
@@ -1012,7 +1031,7 @@ public final class DefaultPersister implements Persister {
       if (many.hasJoinTable()) {
         if (deleteMode.isHard()) {
           // delete associated rows from intersection table (but not during soft delete)
-          deleteAssocManyIntersection(parentBean, many, t, request.isPublish());
+          deleteManyIntersection(parentBean, many, t, request.isPublish(), false);
         }
       } else {
 

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

@@ -1,9 +1,9 @@
 package io.ebeaninternal.server.persist;
 
-import io.ebean.SqlUpdate;
 import io.ebean.bean.BeanCollection;
 import io.ebean.bean.EntityBean;
 import io.ebean.bean.EntityBeanIntercept;
+import io.ebeaninternal.api.SpiSqlUpdate;
 import io.ebeaninternal.server.core.PersistRequest;
 import io.ebeaninternal.server.core.PersistRequestBean;
 import io.ebeaninternal.server.deploy.BeanCollectionUtil;
@@ -24,7 +24,7 @@ import java.util.Set;
 /**
  * Saves the details for a OneToMany or ManyToMany relationship (entity beans).
  */
-class SaveManyBeans extends SaveManyBase {
+public class SaveManyBeans extends SaveManyBase {
 
   private static final Logger log = LoggerFactory.getLogger(SaveManyBeans.class);
 
@@ -244,14 +244,29 @@ class SaveManyBeans extends SaveManyBase {
     if (value == null) {
       return;
     }
+    if (request.isQueueManyIntersection()) {
+      // queue/delay until bean persist request is flushed
+      this.deleteMissing = deleteMissingChildren;
+      request.setManyIntersection(this);
+    } else {
+      saveAssocManyIntersection(deleteMissingChildren, false);
+    }
+  }
+
+  /**
+   * Push intersection table changes onto batch flush queue.
+   */
+  public void saveIntersectionBatch() {
+    saveAssocManyIntersection(deleteMissing, true);
+  }
+
+  private void saveAssocManyIntersection(boolean deleteMissingChildren, boolean queue) {
 
-    //SpiTransaction t = saveManyPropRequest.getTransaction();
     boolean vanillaCollection = !(value instanceof BeanCollection<?>);
-
     if (vanillaCollection || deleteMissingChildren) {
       // delete all intersection rows and then treat all
       // beans in the collection as additions
-      persister.deleteAssocManyIntersection(parentBean, many, transaction, publish);
+      persister.deleteManyIntersection(parentBean, many, transaction, publish, queue);
     }
 
     Collection<?> deletions = null;
@@ -287,10 +302,6 @@ class SaveManyBeans extends SaveManyBase {
     transaction.depth(+1);
 
     if (additions != null && !additions.isEmpty()) {
-      // ensure any cascade batch has been flushed prior
-      // to inserting into the intersection table
-      transaction.flushBatch();
-
       for (Object other : additions) {
         EntityBean otherBean = (EntityBean) other;
         // the object from the 'other' side of the ManyToMany
@@ -309,24 +320,20 @@ class SaveManyBeans extends SaveManyBase {
           } else {
             // build a intersection row for 'insert'
             IntersectionRow intRow = many.buildManyToManyMapBean(parentBean, otherBean, publish);
-            SqlUpdate sqlInsert = intRow.createInsert(server);
-            persister.executeSqlUpdate(sqlInsert, transaction);
+            SpiSqlUpdate sqlInsert = intRow.createInsert(server);
+            persister.executeOrQueue(sqlInsert, transaction, queue);
           }
         }
       }
     }
     if (deletions != null && !deletions.isEmpty()) {
-      // ensure any cascade batch has been flushed prior
-      // to inserting into the intersection table
-      transaction.flushBatch();
-
       for (Object other : deletions) {
         EntityBean otherDelete = (EntityBean) other;
         // the object from the 'other' side of the ManyToMany
         // build a intersection row for 'delete'
         IntersectionRow intRow = many.buildManyToManyMapBean(parentBean, otherDelete, publish);
-        SqlUpdate sqlDelete = intRow.createDelete(server, DeleteMode.HARD);
-        persister.executeSqlUpdate(sqlDelete, transaction);
+        SpiSqlUpdate sqlDelete = intRow.createDelete(server, DeleteMode.HARD);
+        persister.executeOrQueue(sqlDelete, transaction, queue);
       }
     }
 

src/test/java/org/tests/o2m/jointable/TestOneToManyJoinTable.java

@@ -47,12 +47,11 @@ public class TestOneToManyJoinTable extends BaseTestCase {
       assertThat(sql.get(1)).contains("insert into troop_monkey (troop_pid, monkey_mid) values (?, ?)");
     }
 
-    int intersectionRows = Ebean.createSqlQuery("select count(*) as total from troop_monkey where troop_pid = ?")
+    long intersectionRows = Ebean.createSqlQuery("select count(*) as total from troop_monkey where troop_pid = ?")
       .setParameter(1, troop.getPid())
-      .findOne()
-      .getInteger("total");
+      .findSingleLong();
 
-    assertThat(intersectionRows).isEqualTo(2);
+    assertThat(intersectionRows).isEqualTo(2L);
 
     LoggedSqlCollector.current();
     JtTroop fetchTroop = Ebean.find(JtTroop.class)

src/test/java/org/tests/o2m/jointable/TestOneToManyJoinTableInheritance.java

@@ -34,14 +34,15 @@ public class TestOneToManyJoinTableInheritance extends BaseTestCase {
 
     List<String> sql = LoggedSqlCollector.current();
 
-    assertThat(sql).hasSize(11);
+    assertThat(sql).hasSize(8);
     assertThat(sql.get(0)).contains("insert into class_super ");
-    assertThat(sql.get(2)).contains("insert into monkey ");
-    assertThat(sql.get(2)).contains("insert into monkey ");
-    assertThat(sql.get(5)).contains("insert into class_super_monkey ");
-    assertThat(sql.get(6)).contains("insert into class_super ");
-    assertThat(sql.get(8)).contains("insert into monkey ");
-    assertThat(sql.get(10)).contains("insert into class_super_monkey ");
+    assertThat(sql.get(1)).contains("-- bind(ClassA)");
+    assertThat(sql.get(2)).contains("-- bind(ClassB)");
+    assertThat(sql.get(3)).contains("insert into monkey ");
+    assertThat(sql.get(4)).contains("-- bind(Sim");
+    assertThat(sql.get(5)).contains("-- bind(Tim");
+    assertThat(sql.get(6)).contains("-- bind(Uim");
+    assertThat(sql.get(7)).contains("insert into class_super_monkey (class_super_sid, monkey_mid) values (?, ?)");
 
     ClassA dbA = Ebean.find(ClassA.class, 1);
     ClassB dbB = Ebean.find(ClassB.class, 2);

src/test/java/org/tests/sp/TestManyToManySaveTwice.java

@@ -1,43 +1,107 @@
 package org.tests.sp;
 
 import io.ebean.BaseTestCase;
+import io.ebean.DB;
 import io.ebean.Ebean;
+import io.ebean.Transaction;
+import org.junit.Test;
 import org.tests.sp.model.car.Car;
 import org.tests.sp.model.car.Wheel;
-import org.junit.Assert;
-import org.junit.Test;
 
 import java.util.LinkedList;
 import java.util.List;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
 public class TestManyToManySaveTwice extends BaseTestCase {
 
+  @Test
+  public void insertBatch() {
+
+    delete();
+
+    Wheel w0 = new Wheel("wx0");
+    Wheel w1 = new Wheel("wx1");
+
+    List<Wheel> wheels = new LinkedList<>();
+    wheels.add(w0);
+    wheels.add(w1);
+
+    DB.saveAll(wheels);
+
+    Car c0 = new Car("cx0");
+    c0.getWheels().add(w0);
+    c0.getWheels().add(w1);
+
+    Car c1 = new Car("cx1");
+    c1.getWheels().add(w1);
+
+    try (Transaction transaction = DB.beginTransaction()) {
+      transaction.setBatchMode(true);
+
+      DB.save(c0);
+      DB.save(c1);
+
+      transaction.commit();
+    }
+  }
+
+
+  @Test
+  public void insertBasic() {
+
+    delete();
+
+    Wheel w0 = new Wheel("wx0");
+    Wheel w1 = new Wheel("wx1");
+
+    List<Wheel> wheels = new LinkedList<>();
+    wheels.add(w0);
+    wheels.add(w1);
+
+    DB.saveAll(wheels);
+
+    Car c0 = new Car("cx0");
+    c0.getWheels().add(w0);
+    c0.getWheels().add(w1);
+
+    Car c1 = new Car("cx1");
+    c1.getWheels().add(w1);
+
+    DB.save(c0);
+    DB.save(c1);
+  }
+
   @Test
   public void testInsertCarTwice() {
 
-    Ebean.createSqlUpdate("delete from sp_car_car_wheels").execute();
-    Ebean.createSqlUpdate("delete from sp_car_wheel").execute();
-    Ebean.createSqlUpdate("delete from sp_car_car").execute();
+    delete();
 
     List<Wheel> wheels = new LinkedList<>();
-    wheels.add(new Wheel());
-    wheels.add(new Wheel());
-    wheels.add(new Wheel());
-    wheels.add(new Wheel());
+    wheels.add(new Wheel("w0"));
+    wheels.add(new Wheel("w1"));
+    wheels.add(new Wheel("w2"));
+    wheels.add(new Wheel("w3"));
 
-    Car c = new Car();
+    Car c = new Car("c1");
     c.setWheels(wheels);
 
-    Ebean.save(c); // NOTE 1ST SAVE
+    DB.save(c); // NOTE 1ST SAVE
+    assertNotNull(c.getId());
 
-    Assert.assertFalse("No ID assigned!", c.getId() == null);
-
-    Ebean.save(c); // NOTE 2ND SAVE
+    DB.save(c); // NOTE 2ND SAVE
 
     List<Car> allCars = Ebean.find(Car.class).findList();
-    Assert.assertEquals("Inserted 1 car, received more/less!", 1, allCars.size());
+    assertEquals(1, allCars.size());
 
     List<Wheel> allWheels = Ebean.find(Wheel.class).findList();
-    Assert.assertEquals("Inserted 4 wheels, received more/less!", 4, allWheels.size());
+    assertEquals(4, allWheels.size());
+  }
+
+  private void delete() {
+    DB.sqlUpdate("delete from sp_car_car_wheels").execute();
+    DB.sqlUpdate("delete from sp_car_wheel").execute();
+    DB.sqlUpdate("delete from sp_car_car").execute();
   }
 }

src/test/java/org/tests/sp/model/car/Car.java

@@ -16,10 +16,20 @@ public class Car extends IdEntity {
 
   private static final long serialVersionUID = 2579148859565507940L;
 
+  private final String name;
+
   @ManyToMany(cascade = CascadeType.ALL)
   @JoinTable(name = "sp_car_car_wheels", joinColumns = {@JoinColumn(name = "car")}, inverseJoinColumns = {@JoinColumn(name = "wheel")})
   private List<Wheel> wheels;
 
+  public Car(String name) {
+    this.name = name;
+  }
+
+  public String getName() {
+    return name;
+  }
+
   public List<Wheel> getWheels() {
     return wheels;
   }

src/test/java/org/tests/sp/model/car/Wheel.java

@@ -11,4 +11,13 @@ public class Wheel extends IdEntity {
 
   private static final long serialVersionUID = 2399600193947163469L;
 
+  private String name;
+
+  public Wheel(String name) {
+    this.name = name;
+  }
+
+  public String getName() {
+    return name;
+  }
 }