Changeset 7123

Timestamp:

Apr 19, 2016, 2:14:44 PM (7 years ago)

Author:

Nicklas Nordborg

Message:

References #2000: Batch API for annotation handling

Re-factored the annotation batcher to be based on the AbstractBatcher class. This helps us hook into the transaction and re-act to commit and rollbacks.

Annotation snapshots are not deleted until the batcher is closed which minimizes the problem with a parallell transaction that re-creates the snapshot from the old data.

Improved logging.

File:

• trunk/src/core/net/sf/basedb/core/AnnotationBatcher.java (modified) (28 diffs)

trunk/src/core/net/sf/basedb/core/AnnotationBatcher.java

@@ -78 +78 @@
 */
 public class AnnotationBatcher 
+  extends AbstractBatcher
 {
-  private final DbControl dc;
+  
+  private static final org.slf4j.Logger log = 
+    org.slf4j.LoggerFactory.getLogger("net.sf.basedb.core.AnnotationBatcher");
+  private final boolean isDebugEnabled = log.isDebugEnabled();
+  
   private final Item itemType;
   
@@ -113 +118 @@
   private final UpdateBatcher updateItemTable;
 
-  // All update batchers (to make it easy to flush all at once)
-  private final List<UpdateBatcher> allBatchers;
+  // All batchers (to make it easy to flush/close all at once)
+  private final List<UpdateBatcher> allUpdateBatchers;
+  private final List<InsertBatcher> allInsertBatchers;
   
   private final UpdateBatcher deleteFromAnnotations;
@@ -127 +133 @@
   private Annotatable currentItem;
   private int currentAnnotationSetId;
+  private boolean deleteSnapshotIfModifed;
   
   // Lists for temporary keeping track of value IDs for
@@ -136 +143 @@
   private Map<Integer, CurrentAnnotationInfo> currentInfo;
   
+  // List of AnnotationSet IDs that should be deleted
+  // from the snapshot manager
+  private final List<Integer> snapshotsToDelete;
+  
   /**
     Create a new batcher instance. 
@@ -150 +161 @@
     }
     
-    this.dc = dc;
     this.itemType = itemType;
     this.batchDate = new Date();
@@ -174 +184 @@
     this.currentInfo = new HashMap<Integer, CurrentAnnotationInfo>();
     
-    this.allBatchers = new ArrayList<UpdateBatcher>();
+    this.allUpdateBatchers = new ArrayList<UpdateBatcher>();
+    this.allInsertBatchers = new ArrayList<InsertBatcher>();
+    this.snapshotsToDelete = new ArrayList<Integer>();
     try
     {
@@ -272 +284 @@
       throw new BaseException(ex);
     }
-  }
+    
+    setDbControl(dc);
+  }
+  
+  
+  /*
+    From the Batcher interface
+    --------------------------
+  */
+  
+  /**
+    Flush the batchers.
+  */
+  @Override
+  public void flush()
+  {
+    System.out.println(new Date() + ": " + numItems + " done; flushing batchers");
+    String currentSql = null;
+    try
+    {
+      for (UpdateBatcher b : allUpdateBatchers)
+      {
+        currentSql = b.getSql();
+        b.flush();
+      }
+    }
+    catch (SQLException ex)
+    {
+      log.error(currentSql, ex);
+      LogUtil.logSQLExceptionChain(log, ex);
+      throw new BaseException(ex);
+    }
+  }
+  // ------------------------
+  
+  /*
+    From the AbstractBatcher class
+    ------------------------------
+  */
+  @Override
+  protected void onBeforeClose()
+  {
+    System.out.println(new Date() + "; done!! " + numItems + " items.");
+    for (UpdateBatcher b : allUpdateBatchers)
+    {
+      b.close();
+    }
+    for (InsertBatcher b : allInsertBatchers)
+    {
+      b.close();
+    }
+    for (Integer id : snapshotsToDelete)
+    {
+      SnapshotManager.removeSnapshot(currentAnnotationSetId);
+    }
+  }
+  // ---------------------------
   
   private InsertBatcher createInsertBatcher(Connection c, String sql, int... types)
@@ -279 +347 @@
     sql = sql.replace('[', dialect.openQuote()).replace(']', dialect.closeQuote());
     InsertBatcher b = new InsertBatcher(c, sql, types);
+    allInsertBatchers.add(b);
     return b;
   }
@@ -287 +356 @@
     sql = sql.replace('[', dialect.openQuote()).replace(']', dialect.closeQuote());
     UpdateBatcher b = new UpdateBatcher(c, sql, types);
-    allBatchers.add(b);
+    allUpdateBatchers.add(b);
     return b;
   }
   
-  /**
-    Flush the batchers.
-  */
-  private void flushBatchers()
-  {
-    try
-    {
-      for (UpdateBatcher b : allBatchers)
-      {
-        b.flush();
-      }
-    }
-    catch (SQLException ex)
-    {
-      throw new BaseException(ex);
-    }
-  }
   
   /**
@@ -318 +370 @@
   public void addAnnotationTypes(Collection<AnnotationType> types)
   {
+    if (isClosed()) throw new ConnectionClosedException();
     if (currentItem != null)
     {
@@ -327 +380 @@
     {
       if (currentInfo.containsKey(at.getId())) continue; // Already added
+      
+      log.debug("Adding annotation type: " + at);
       
       at.checkPermission(Permission.USE);
@@ -354 +409 @@
   public void setCurrentItem(Annotatable item)
   {
+    if (isClosed()) throw new ConnectionClosedException();
     if (currentInfo.size() == 0)
     {
@@ -365 +421 @@
     item.checkPermission(Permission.WRITE);
     
-    if (numItems > 0 && numItems % 100 == 0)
-    {
-      System.out.println(new Date() + ": " + numItems + " done; flushing batchers");
-      flushBatchers();
+    log.debug("Setting current item: " + item);
+    
+    if (numItems > 0 && numItems % getBatchSize() == 0)
+    {
+      flush();
     }
     
@@ -375 +432 @@
     this.currentItem = item;
     this.currentAnnotationSetId = 0;
+    this.deleteSnapshotIfModifed = false;
 
     // Read current values
@@ -381 +439 @@
       currentAnnotationSetId = ((AnnotatableData)((BasicItem)item).getData()).getAnnotationSet().getId();
       loadAnnotationInfo.setInteger("annotationSet", currentAnnotationSetId);
+      deleteSnapshotIfModifed = true;
+      
+      if (isDebugEnabled)
+      {
+        log.debug("Loading annotations for current item: " + item);
+      }
       
       // Maps value_id -> Annotation info
       Map<Integer, CurrentAnnotationInfo> tmp = new HashMap<Integer, CurrentAnnotationInfo>();
       List<Object[]> list = HibernateUtil.loadList(Object[].class, loadAnnotationInfo, null);
+
+      if (isDebugEnabled)
+      {
+        log.debug("Found " + list.size() + " annotations for current item: " + item);
+      }
       
       for (Object[] data : list)
@@ -394 +463 @@
           // This annotation type is not among the specified annotation types
           continue; // with the next annotation
+        }
+        
+        if (isDebugEnabled)
+        {
+          log.trace("Found annotation: AnnotationType.id=" + info.annotationTypeId);
         }
         
@@ -419 +493 @@
         if (valueIds[typeIndex].isEmpty()) continue; // No annotation with this value type
         
+        if (isDebugEnabled)
+        {
+          log.debug("Loading " + t.name() + " values: " + valueIds[typeIndex]);
+        }
+        
         org.hibernate.Query query = loadAnnotationValues[typeIndex];
         query.setParameterList("listOfIds", valueIds[typeIndex], Type.INT.getTypeWrapper().getHibernateType());
       
         list = HibernateUtil.loadList(Object[].class, query, null);
+        if (isDebugEnabled)
+        {
+          log.debug("Found " + list.size() + " values");
+        }
         for (Object[] data : list)
         {
           Integer valueId = (Integer)data[0];
           tmp.get(valueId).values.add(data[1]);
+          if (isDebugEnabled) log.trace(valueId + ": "+data[1]);
         }
         
@@ -479 +563 @@
       throw new IllegalStateException("Not allowed to call this method with the same annotation more than once: " + annotationType);
     }
+    
+    if (isDebugEnabled)
+    {
+      log.trace("Setting values for annotation type " + annotationType + ": " + values);
+    }
+    
     info.hasBeenUsed = true;
     info.itemId = currentItem.getId();
@@ -489 +579 @@
         if (infoIsForCurrentItem)
         {
+          if (isDebugEnabled) 
+          {
+            log.trace("Deleting " + annotationType + " from current item " + currentItem);
+          }
+          
           // Delete the annotation
           changeType = Change.DELETED;
@@ -540 +635 @@
           if (!EqualsHelper.invariantEquals(convertedValues, info.values))
           {
+            if (isDebugEnabled) 
+            {
+              log.trace("Updating " + annotationType + " for current item " + currentItem);
+            }
+            
             // Update
             changeType = Change.UPDATED;
@@ -555 +655 @@
             updateAnnotations.addToBatch(info.version+1, unit != null ? unit.getId() : info.unitId, batchDate, info.annotationId, info.version);
           }
+          else
+          {
+            if (isDebugEnabled)
+            {
+              log.trace("No change " + annotationType + " for current item " + currentItem);
+            }
+          }
         }
         else
         {
+          if (isDebugEnabled) 
+          {
+            log.trace("Creating " + annotationType + " for current item " + currentItem);
+          }
+          
           // Create a new annotation
           changeType = Change.ADDED;
@@ -589 +701 @@
     }
     
-    if (changeType != Change.NO_CHANGE) 
-    {
-      SnapshotManager.removeSnapshot(currentAnnotationSetId);
+    // Store the annotation set ID so that we can tell
+    // the SnapshotManager to clear the cache when this
+    // batcher is closed
+    if (changeType != Change.NO_CHANGE && deleteSnapshotIfModifed) 
+    {
+      deleteSnapshotIfModifed = false;
+      snapshotsToDelete.add(currentAnnotationSetId);
     }
     
     return changeType;
-  }
-  
-  public void close()
-  {
-    System.out.println(new Date() + "; done!! " + numItems + " items.");
-    flushBatchers();
   }
   
@@ -690 +800 @@
       throws SQLException
     {
+      log.debug(sql);
       this.sql = sql;
       this.statement = c.prepareStatement(sql, new String[] { "id" } );
@@ -695 +806 @@
     }
     
+    /**
+      The SQL statement this batcher is executing.
+    */
+    String getSql()
+    {
+      return sql;
+    }
+
     /**
       Execute an insert SQL statement with the given parameter
@@ -721 +840 @@
       return keys.getInt(1);
     }
+    
+    /**
+      Close the batcher.
+    */
+    void close()
+    {
+      try
+      {
+        statement.close();
+      }
+      catch (SQLException ex)
+      {
+        log.error(ex.getMessage(), ex);
+        LogUtil.logSQLExceptionChain(log, ex);
+      }
+    }
+
   }
   
@@ -741 +877 @@
       throws SQLException
     {
+      log.debug(sql);
       this.sql = sql;
       this.statement = c.prepareStatement(sql);
@@ -747 +884 @@
     
     /**
+      The SQL statement this batcher is executing.
+    */
+    String getSql()
+    {
+      return sql;
+    }
+    
+    /**
       Add a new entry to the batch. The number of values must match the number of 
       parameter types. 
@@ -780 +925 @@
       }
     }
+
+    /**
+      Close the batcher.
+    */
+    void close()
+    {
+      try
+      {
+        statement.close();
+      }
+      catch (SQLException ex)
+      {
+        log.error(ex.getMessage(), ex);
+        LogUtil.logSQLExceptionChain(log, ex);
+      }
+    }
   }