Merge branch '3.1'

Author: dlmarion

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

@@ -792,7 +792,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

@@ -570,7 +570,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

@@ -25,6 +25,7 @@
 import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.TIME;
 import static org.apache.accumulo.core.metadata.schema.TabletMetadata.LocationType.CURRENT;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
@@ -63,6 +64,7 @@
 import org.apache.accumulo.core.tabletserver.thrift.TabletServerClientService;
 import org.apache.accumulo.core.trace.TraceUtil;
 import org.apache.accumulo.core.util.PeekingIterator;
+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 +84,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 +103,7 @@ public LoadFiles(BulkInfo bulkInfo) {
 
   @Override
   public long isReady(FateId fateId, Manager manager) throws Exception {
+    log.info("Starting bulk import for {} (tid = {})", bulkInfo.sourceDir, fateId);
     if (manager.onlineTabletServers().isEmpty()) {
       log.warn("There are no tablet server to process bulkDir import, waiting (fateId = " + fateId
           + ")");
@@ -104,7 +114,19 @@ public long isReady(FateId fateId, 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, fateId);
+
+      Loader loader = new Loader();
+
+      List<ColumnType> fetchCols = new ArrayList<>(List.of(PREV_ROW, LOCATION, LOADED, TIME));
+      if (loader.pauseLimit > 0) {
+        fetchCols.add(FILES);
+      }
+
+      TabletsMetadataFactory tmf = (startRow) -> TabletsMetadata.builder(manager.getContext())
+          .forTable(bulkInfo.tableId).overlapping(startRow, null).checkConsistency()
+          .fetch(fetchCols.toArray(new ColumnType[0])).build();
+
+      return loadFiles(loader, bulkInfo, bulkDir, lmi, tmf, manager, fateId);
     }
   }
 
@@ -113,7 +135,8 @@ public Repo<Manager> call(final FateId fateId, final Manager manager) {
     return new RefreshTablets(bulkInfo);
   }
 
-  private static class Loader {
+  // visible for testing
+  public static class Loader {
     protected Path bulkDir;
     protected Manager manager;
     protected FateId fateId;
@@ -344,14 +367,26 @@ long finish() {
     }
   }
 
+  /**
+   * 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, FateId fateId) throws Exception {
+  // visible for testing
+  static long loadFiles(Loader loader, BulkInfo bulkInfo, Path bulkDir,
+      LoadMappingIterator loadMapIter, TabletsMetadataFactory factory, Manager manager,
+      FateId fateId) throws Exception {
     PeekingIterator<Map.Entry<KeyExtent,Bulk.Files>> lmi = new PeekingIterator<>(loadMapIter);
     Map.Entry<KeyExtent,Bulk.Files> loadMapEntry = lmi.peek();
 
@@ -360,39 +395,41 @@ private long loadFiles(TableId tableId, Path bulkDir, LoadMappingIterator loadMa
     String fmtTid = fateId.getTxUUIDStr();
     log.trace("{}: Starting bulk load at row: {}", fmtTid, startRow);
 
-    Loader loader = new Loader();
-    long t1;
-    loader.start(bulkDir, manager, tableId, fateId, bulkInfo.setTime);
+    loader.start(bulkDir, manager, bulkInfo.tableId, fateId, bulkInfo.setTime);
 
-    List<ColumnType> fetchCols = new ArrayList<>(List.of(PREV_ROW, LOCATION, LOADED, TIME));
-    if (loader.pauseLimit > 0) {
-      fetchCols.add(FILES);
-    }
-
-    try (TabletsMetadata tabletsMetadata =
-        TabletsMetadata.builder(manager.getContext()).forTable(tableId).overlapping(startRow, null)
-            .checkConsistency().fetch(fetchCols.toArray(new ColumnType[0])).build()) {
+    ImportTimingStats importTimingStats = new ImportTimingStats();
+    Timer timer = Timer.startNew();
+    try (TabletsMetadata tabletsMetadata = factory.newTabletsMetadata(startRow)) {
 
       // The tablet iterator and load mapping iterator are both iterating over data that is sorted
       // in the same way. The two iterators are each independently advanced to find common points in
       // the sorted data.
-      var tabletIter = tabletsMetadata.iterator();
-
-      t1 = System.currentTimeMillis();
+      Iterator<TabletMetadata> tabletIter = tabletsMetadata.iterator();
       while (lmi.hasNext()) {
         loadMapEntry = lmi.next();
         List<TabletMetadata> tablets =
-            findOverlappingTablets(fmtTid, loadMapEntry.getKey(), tabletIter);
+            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, fateId, 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);
     }
@@ -406,8 +443,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;
 
@@ -419,12 +457,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);
@@ -445,6 +489,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(