Merge branch '2.1' into 3.1

Author: dlmarion

core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java

@@ -518,7 +518,7 @@ public static <E extends Entry<Key,Value>> TabletMetadata convertRow(Iterator<E>
   }
 
   @VisibleForTesting
-  static TabletMetadata create(String id, String prevEndRow, String endRow) {
+  public static TabletMetadata create(String id, String prevEndRow, String endRow) {
     final var tmBuilder = new Builder();
     tmBuilder.table(TableId.of(id));
     tmBuilder.sawPrevEndRow(true);

core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletsMetadata.java

@@ -619,7 +619,8 @@ private TabletsMetadata(TabletMetadata tm) {
     this.tablets = Collections.singleton(tm);
   }
 
-  private TabletsMetadata(AutoCloseable closeable, Iterable<TabletMetadata> tmi) {
+  // visible for testing
+  public TabletsMetadata(AutoCloseable closeable, Iterable<TabletMetadata> tmi) {
     this.closeable = closeable;
     this.tablets = tmi;
   }

server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/LoadFiles.java

@@ -23,6 +23,7 @@
 import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.LOCATION;
 import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.PREV_ROW;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.HashMap;
@@ -41,7 +42,6 @@
 import org.apache.accumulo.core.clientImpl.bulk.LoadMappingIterator;
 import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.data.Mutation;
-import org.apache.accumulo.core.data.TableId;
 import org.apache.accumulo.core.dataImpl.KeyExtent;
 import org.apache.accumulo.core.dataImpl.thrift.TKeyExtent;
 import org.apache.accumulo.core.fate.FateTxId;
@@ -64,6 +64,7 @@
 import org.apache.accumulo.core.util.MapCounter;
 import org.apache.accumulo.core.util.PeekingIterator;
 import org.apache.accumulo.core.util.TextUtil;
+import org.apache.accumulo.core.util.Timer;
 import org.apache.accumulo.manager.Manager;
 import org.apache.accumulo.manager.tableOps.ManagerRepo;
 import org.apache.accumulo.server.fs.VolumeManager;
@@ -82,6 +83,13 @@
  */
 class LoadFiles extends ManagerRepo {
 
+  // visible for testing
+  interface TabletsMetadataFactory {
+
+    TabletsMetadata newTabletsMetadata(Text startRow);
+
+  }
+
   private static final long serialVersionUID = 1L;
 
   private static final Logger log = LoggerFactory.getLogger(LoadFiles.class);
@@ -94,6 +102,7 @@ public LoadFiles(BulkInfo bulkInfo) {
 
   @Override
   public long isReady(long tid, Manager manager) throws Exception {
+    log.info("Starting bulk import for {} (tid = {})", bulkInfo.sourceDir, FateTxId.formatTid(tid));
     if (manager.onlineTabletServers().isEmpty()) {
       log.warn("There are no tablet server to process bulkDir import, waiting (tid = "
           + FateTxId.formatTid(tid) + ")");
@@ -104,7 +113,19 @@ public long isReady(long tid, Manager manager) throws Exception {
     manager.updateBulkImportStatus(bulkInfo.sourceDir, BulkImportState.LOADING);
     try (LoadMappingIterator lmi =
         BulkSerialize.getUpdatedLoadMapping(bulkDir.toString(), bulkInfo.tableId, fs::open)) {
-      return loadFiles(bulkInfo.tableId, bulkDir, lmi, manager, tid);
+
+      Loader loader;
+      if (bulkInfo.tableState == TableState.ONLINE) {
+        loader = new OnlineLoader();
+      } else {
+        loader = new OfflineLoader();
+      }
+
+      TabletsMetadataFactory tmf = (startRow) -> TabletsMetadata.builder(manager.getContext())
+          .forTable(bulkInfo.tableId).overlapping(startRow, null).checkConsistency()
+          .fetch(PREV_ROW, LOCATION, LOADED).build();
+
+      return loadFiles(loader, bulkInfo, bulkDir, lmi, tmf, manager, tid);
     }
   }
 
@@ -117,7 +138,8 @@ public Repo<Manager> call(final long tid, final Manager manager) {
     }
   }
 
-  private abstract static class Loader {
+  // visible for testing
+  public abstract static class Loader {
     protected Path bulkDir;
     protected Manager manager;
     protected long tid;
@@ -315,14 +337,26 @@ long finish() throws Exception {
     }
   }
 
+  /**
+   * Stats for the loadFiles method. Helps track wasted time and iterations.
+   */
+  static class ImportTimingStats {
+    Duration totalWastedTime = Duration.ZERO;
+    long wastedIterations = 0;
+    long tabletCount = 0;
+    long callCount = 0;
+  }
+
   /**
    * Make asynchronous load calls to each overlapping Tablet in the bulk mapping. Return a sleep
    * time to isReady based on a factor of the TabletServer with the most Tablets. This method will
    * scan the metadata table getting Tablet range and location information. It will return 0 when
    * all files have been loaded.
    */
-  private long loadFiles(TableId tableId, Path bulkDir, LoadMappingIterator loadMapIter,
-      Manager manager, long tid) throws Exception {
+  // visible for testing
+  static long loadFiles(Loader loader, BulkInfo bulkInfo, Path bulkDir,
+      LoadMappingIterator loadMapIter, TabletsMetadataFactory factory, Manager manager, long tid)
+      throws Exception {
     PeekingIterator<Map.Entry<KeyExtent,Bulk.Files>> lmi = new PeekingIterator<>(loadMapIter);
     Map.Entry<KeyExtent,Bulk.Files> loadMapEntry = lmi.peek();
 
@@ -331,33 +365,38 @@ private long loadFiles(TableId tableId, Path bulkDir, LoadMappingIterator loadMa
     String fmtTid = FateTxId.formatTid(tid);
     log.trace("{}: Starting bulk load at row: {}", fmtTid, startRow);
 
-    Loader loader;
-    if (bulkInfo.tableState == TableState.ONLINE) {
-      loader = new OnlineLoader();
-    } else {
-      loader = new OfflineLoader();
-    }
-    long t1;
     loader.start(bulkDir, manager, tid, bulkInfo.setTime);
-    try (TabletsMetadata tabletsMetadata =
-        TabletsMetadata.builder(manager.getContext()).forTable(tableId).overlapping(startRow, null)
-            .checkConsistency().fetch(PREV_ROW, LOCATION, LOADED).build()) {
 
-      t1 = System.currentTimeMillis();
+    ImportTimingStats importTimingStats = new ImportTimingStats();
+    Timer timer = Timer.startNew();
+    try (TabletsMetadata tabletsMetadata = factory.newTabletsMetadata(startRow)) {
+
+      Iterator<TabletMetadata> tabletIter = tabletsMetadata.iterator();
       while (lmi.hasNext()) {
         loadMapEntry = lmi.next();
         List<TabletMetadata> tablets =
-            findOverlappingTablets(fmtTid, loadMapEntry.getKey(), tabletsMetadata.iterator());
+            findOverlappingTablets(fmtTid, loadMapEntry.getKey(), tabletIter, importTimingStats);
         loader.load(tablets, loadMapEntry.getValue());
       }
     }
+    Duration totalProcessingTime = timer.elapsed();
 
     log.trace("{}: Completed Finding Overlapping Tablets", fmtTid);
 
+    if (importTimingStats.callCount > 0) {
+      log.debug(
+          "Bulk import stats for {} (tid = {}): processed {} tablets in {} calls which took {}ms ({} nanos). Skipped {} iterations which took {}ms ({} nanos) or {}% of the processing time.",
+          bulkInfo.sourceDir, FateTxId.formatTid(tid), importTimingStats.tabletCount,
+          importTimingStats.callCount, totalProcessingTime.toMillis(),
+          totalProcessingTime.toNanos(), importTimingStats.wastedIterations,
+          importTimingStats.totalWastedTime.toMillis(), importTimingStats.totalWastedTime.toNanos(),
+          (importTimingStats.totalWastedTime.toNanos() * 100) / totalProcessingTime.toNanos());
+    }
+
     long sleepTime = loader.finish();
     if (sleepTime > 0) {
       log.trace("{}: Tablet Max Sleep is {}", fmtTid, sleepTime);
-      long scanTime = Math.min(System.currentTimeMillis() - t1, 30_000);
+      long scanTime = Math.min(totalProcessingTime.toMillis(), 30_000);
       log.trace("{}: Scan time is {}", fmtTid, scanTime);
       sleepTime = Math.max(sleepTime, scanTime * 2);
     }
@@ -371,8 +410,9 @@ private long loadFiles(TableId tableId, Path bulkDir, LoadMappingIterator loadMa
   /**
    * Find all the tablets within the provided bulk load mapping range.
    */
-  private List<TabletMetadata> findOverlappingTablets(String fmtTid, KeyExtent loadRange,
-      Iterator<TabletMetadata> tabletIter) {
+  // visible for testing
+  static List<TabletMetadata> findOverlappingTablets(String fmtTid, KeyExtent loadRange,
+      Iterator<TabletMetadata> tabletIter, ImportTimingStats importTimingStats) {
 
     TabletMetadata currTablet = null;
 
@@ -384,12 +424,18 @@ private List<TabletMetadata> findOverlappingTablets(String fmtTid, KeyExtent loa
 
       int cmp;
 
+      long wastedIterations = 0;
+      Timer timer = Timer.startNew();
+
       // skip tablets until we find the prevEndRow of loadRange
       while ((cmp = PREV_COMP.compare(currTablet.getPrevEndRow(), loadRange.prevEndRow())) < 0) {
+        wastedIterations++;
         log.trace("{}: Skipping tablet: {}", fmtTid, currTablet.getExtent());
         currTablet = tabletIter.next();
       }
 
+      Duration wastedTime = timer.elapsed();
+
       if (cmp != 0) {
         throw new IllegalStateException(
             "Unexpected prev end row " + currTablet.getExtent() + " " + loadRange);
@@ -410,6 +456,11 @@ private List<TabletMetadata> findOverlappingTablets(String fmtTid, KeyExtent loa
         throw new IllegalStateException("Unexpected end row " + currTablet + " " + loadRange);
       }
 
+      importTimingStats.wastedIterations += wastedIterations;
+      importTimingStats.totalWastedTime = importTimingStats.totalWastedTime.plus(wastedTime);
+      importTimingStats.tabletCount += tablets.size();
+      importTimingStats.callCount++;
+
       return tablets;
     } catch (NoSuchElementException e) {
       NoSuchElementException ne2 = new NoSuchElementException(