HBASE-27826 Add FSFT implementations for Virtual links and enable them as part of SplitProcedure

Author: Prathyusha Garre

hbase-protocol-shaded/src/main/protobuf/server/region/StoreFileTracker.proto

@@ -25,9 +25,11 @@ option java_generic_services = true;
 option java_generate_equals_and_hash = true;
 option optimize_for = SPEED;
 
+import "server/io/FS.proto";
 message StoreFileEntry {
   required string name = 1;
   required uint64 size = 2;
+  optional Reference reference = 3;
 }
 
 message StoreFileList {

hbase-server/src/main/java/org/apache/hadoop/hbase/io/Reference.java

@@ -56,7 +56,7 @@ public class Reference {
    * For split HStoreFiles, it specifies if the file covers the lower half or the upper half of the
    * key range
    */
-  static enum Range {
+  public static enum Range {
     /** HStoreFile contains upper half of key range */
     top,
     /** HStoreFile contains lower half of key range */

hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/SplitTableRegionProcedure.java

@@ -644,20 +644,46 @@ public void createDaughterRegions(final MasterProcedureEnv env) throws IOExcepti
     HRegionFileSystem regionFs = HRegionFileSystem.openRegionFromFileSystem(
       env.getMasterConfiguration(), fs, tabledir, getParentRegion(), false);
     regionFs.createSplitsDir(daughterOneRI, daughterTwoRI);
+    Pair<List<StoreFileInfo>, List<StoreFileInfo>> expectedReferences =
+      splitStoreFiles(env, regionFs);
+    final ExecutorService threadPool = Executors.newFixedThreadPool(2,
+      new ThreadFactoryBuilder().setNameFormat("RegionCommitter-pool-%d").setDaemon(true)
+        .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
+    threadPool.submit(new Callable<Path>() {
+      @Override
+      public Path call() throws IOException {
+        return regionFs.commitDaughterRegion(daughterOneRI, expectedReferences.getFirst(), env);
+      }
+    });
+    threadPool.submit(new Callable<Path>() {
+      @Override
+      public Path call() throws IOException {
+        return regionFs.commitDaughterRegion(daughterTwoRI, expectedReferences.getSecond(), env);
+      }
+    });
+    // Shutdown the pool
+    threadPool.shutdown();
 
-    Pair<List<Path>, List<Path>> expectedReferences = splitStoreFiles(env, regionFs);
-
-    assertSplitResultFilesCount(fs, expectedReferences.getFirst().size(),
-      regionFs.getSplitsDir(daughterOneRI));
-    regionFs.commitDaughterRegion(daughterOneRI, expectedReferences.getFirst(), env);
-    assertSplitResultFilesCount(fs, expectedReferences.getFirst().size(),
-      new Path(tabledir, daughterOneRI.getEncodedName()));
-
-    assertSplitResultFilesCount(fs, expectedReferences.getSecond().size(),
-      regionFs.getSplitsDir(daughterTwoRI));
-    regionFs.commitDaughterRegion(daughterTwoRI, expectedReferences.getSecond(), env);
-    assertSplitResultFilesCount(fs, expectedReferences.getSecond().size(),
-      new Path(tabledir, daughterTwoRI.getEncodedName()));
+    Configuration conf = env.getMasterConfiguration();
+    // Wait for all the tasks to finish.
+    // When splits ran on the RegionServer, how-long-to-wait-configuration was named
+    // hbase.regionserver.fileSplitTimeout. If set, use its value.
+    long fileSplitTimeout = conf.getLong("hbase.master.fileSplitTimeout",
+      conf.getLong("hbase.regionserver.fileSplitTimeout", 600000));
+    try {
+      boolean stillRunning = !threadPool.awaitTermination(fileSplitTimeout, TimeUnit.MILLISECONDS);
+      if (stillRunning) {
+        threadPool.shutdownNow();
+        // wait for the thread to shutdown completely.
+        while (!threadPool.isTerminated()) {
+          Thread.sleep(50);
+        }
+        throw new IOException(
+          "Took too long to split the" + " files and create the references, aborting split");
+      }
+    } catch (InterruptedException e) {
+      throw (InterruptedIOException) new InterruptedIOException().initCause(e);
+    }
   }
 
   private void deleteDaughterRegions(final MasterProcedureEnv env) throws IOException {
@@ -673,8 +699,8 @@ private void deleteDaughterRegions(final MasterProcedureEnv env) throws IOExcept
    * Create Split directory
    * @param env MasterProcedureEnv
    */
-  private Pair<List<Path>, List<Path>> splitStoreFiles(final MasterProcedureEnv env,
-    final HRegionFileSystem regionFs) throws IOException {
+  private Pair<List<StoreFileInfo>, List<StoreFileInfo>> splitStoreFiles(
+    final MasterProcedureEnv env, final HRegionFileSystem regionFs) throws IOException {
     final Configuration conf = env.getMasterConfiguration();
     TableDescriptor htd = env.getMasterServices().getTableDescriptors().get(getTableName());
     // The following code sets up a thread pool executor with as many slots as
@@ -729,7 +755,8 @@ private Pair<List<Path>, List<Path>> splitStoreFiles(final MasterProcedureEnv en
     final ExecutorService threadPool = Executors.newFixedThreadPool(maxThreads,
       new ThreadFactoryBuilder().setNameFormat("StoreFileSplitter-pool-%d").setDaemon(true)
         .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
-    final List<Future<Pair<Path, Path>>> futures = new ArrayList<Future<Pair<Path, Path>>>(nbFiles);
+    final List<Future<Pair<StoreFileInfo, StoreFileInfo>>> futures =
+      new ArrayList<Future<Pair<StoreFileInfo, StoreFileInfo>>>(nbFiles);
 
     // Split each store file.
     for (Map.Entry<String, Collection<StoreFileInfo>> e : files.entrySet()) {
@@ -776,12 +803,12 @@ private Pair<List<Path>, List<Path>> splitStoreFiles(final MasterProcedureEnv en
       throw (InterruptedIOException) new InterruptedIOException().initCause(e);
     }
 
-    List<Path> daughterA = new ArrayList<>();
-    List<Path> daughterB = new ArrayList<>();
+    List<StoreFileInfo> daughterA = new ArrayList<>();
+    List<StoreFileInfo> daughterB = new ArrayList<>();
     // Look for any exception
-    for (Future<Pair<Path, Path>> future : futures) {
+    for (Future<Pair<StoreFileInfo, StoreFileInfo>> future : futures) {
       try {
-        Pair<Path, Path> p = future.get();
+        Pair<StoreFileInfo, StoreFileInfo> p = future.get();
         if (p.getFirst() != null) {
           daughterA.add(p.getFirst());
         }
@@ -803,6 +830,7 @@ private Pair<List<Path>, List<Path>> splitStoreFiles(final MasterProcedureEnv en
     return new Pair<>(daughterA, daughterB);
   }
 
+  // TODO: update assert to do SFT.load instead of FileSystem listing
   private void assertSplitResultFilesCount(final FileSystem fs,
     final int expectedSplitResultFileCount, Path dir) throws IOException {
     if (expectedSplitResultFileCount != 0) {
@@ -814,8 +842,8 @@ private void assertSplitResultFilesCount(final FileSystem fs,
     }
   }
 
-  private Pair<Path, Path> splitStoreFile(HRegionFileSystem regionFs, TableDescriptor htd,
-    ColumnFamilyDescriptor hcd, HStoreFile sf) throws IOException {
+  private Pair<StoreFileInfo, StoreFileInfo> splitStoreFile(HRegionFileSystem regionFs,
+    TableDescriptor htd, ColumnFamilyDescriptor hcd, HStoreFile sf) throws IOException {
     if (LOG.isDebugEnabled()) {
       LOG.debug("pid=" + getProcId() + " splitting started for store file: " + sf.getPath()
         + " for region: " + getParentRegion().getShortNameToLog());
@@ -831,22 +859,22 @@ private Pair<Path, Path> splitStoreFile(HRegionFileSystem regionFs, TableDescrip
       StoreFileTrackerFactory.create(regionFs.getFileSystem().getConf(), htd, hcd,
         HRegionFileSystem.create(regionFs.getFileSystem().getConf(), regionFs.getFileSystem(),
           regionFs.getTableDir(), daughterTwoRI));
-    final Path path_first = regionFs.splitStoreFile(this.daughterOneRI, familyName, sf, splitRow,
-      false, splitPolicy, daughterOneSft);
-    final Path path_second = regionFs.splitStoreFile(this.daughterTwoRI, familyName, sf, splitRow,
-      true, splitPolicy, daughterTwoSft);
+    final StoreFileInfo sfiFirst = regionFs.splitStoreFile(this.daughterOneRI, familyName, sf,
+      splitRow, false, splitPolicy, daughterOneSft);
+    final StoreFileInfo sfiSecond = regionFs.splitStoreFile(this.daughterTwoRI, familyName, sf,
+      splitRow, true, splitPolicy, daughterTwoSft);
     if (LOG.isDebugEnabled()) {
       LOG.debug("pid=" + getProcId() + " splitting complete for store file: " + sf.getPath()
         + " for region: " + getParentRegion().getShortNameToLog());
     }
-    return new Pair<Path, Path>(path_first, path_second);
+    return new Pair<StoreFileInfo, StoreFileInfo>(sfiFirst, sfiSecond);
   }
 
   /**
    * Utility class used to do the file splitting / reference writing in parallel instead of
    * sequentially.
    */
-  private class StoreFileSplitter implements Callable<Pair<Path, Path>> {
+  private class StoreFileSplitter implements Callable<Pair<StoreFileInfo, StoreFileInfo>> {
     private final HRegionFileSystem regionFs;
     private final ColumnFamilyDescriptor hcd;
     private final HStoreFile sf;
@@ -867,7 +895,7 @@ public StoreFileSplitter(HRegionFileSystem regionFs, TableDescriptor htd,
     }
 
     @Override
-    public Pair<Path, Path> call() throws IOException {
+    public Pair<StoreFileInfo, StoreFileInfo> call() throws IOException {
       return splitStoreFile(regionFs, htd, hcd, sf);
     }
   }

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java

@@ -519,7 +519,7 @@ void cleanupDaughterRegion(final RegionInfo regionInfo) throws IOException {
    * in the filesystem.
    * @param regionInfo daughter {@link org.apache.hadoop.hbase.client.RegionInfo}
    */
-  public Path commitDaughterRegion(final RegionInfo regionInfo, List<Path> allRegionFiles,
+  public Path commitDaughterRegion(final RegionInfo regionInfo, List<StoreFileInfo> allRegionFiles,
     MasterProcedureEnv env) throws IOException {
     Path regionDir = this.getSplitsDir(regionInfo);
     if (fs.exists(regionDir)) {
@@ -534,7 +534,29 @@ public Path commitDaughterRegion(final RegionInfo regionInfo, List<Path> allRegi
     return regionDir;
   }
 
-  private void insertRegionFilesIntoStoreTracker(List<Path> allFiles, MasterProcedureEnv env,
+  private void insertRegionFilesIntoStoreTracker(List<StoreFileInfo> allFiles,
+    MasterProcedureEnv env, HRegionFileSystem regionFs) throws IOException {
+    TableDescriptor tblDesc =
+      env.getMasterServices().getTableDescriptors().get(regionInfo.getTable());
+    // we need to map trackers per store
+    Map<String, StoreFileTracker> trackerMap = new HashMap<>();
+    // we need to map store files per store
+    Map<String, List<StoreFileInfo>> fileInfoMap = new HashMap<>();
+    for (StoreFileInfo sfi : allFiles) {
+      Path file = sfi.getPath();
+      String familyName = file.getParent().getName();
+      trackerMap.computeIfAbsent(familyName, t -> StoreFileTrackerFactory.create(conf, tblDesc,
+        tblDesc.getColumnFamily(Bytes.toBytes(familyName)), regionFs));
+      fileInfoMap.computeIfAbsent(familyName, l -> new ArrayList<>());
+      List<StoreFileInfo> infos = fileInfoMap.get(familyName);
+      infos.add(sfi);
+    }
+    for (Map.Entry<String, StoreFileTracker> entry : trackerMap.entrySet()) {
+      entry.getValue().add(fileInfoMap.get(entry.getKey()));
+    }
+  }
+
+  private void insertRegionfilePathsIntoStoreTracker(List<Path> allFiles, MasterProcedureEnv env,
     HRegionFileSystem regionFs) throws IOException {
     TableDescriptor tblDesc =
       env.getMasterServices().getTableDescriptors().get(regionInfo.getTable());
@@ -591,8 +613,9 @@ public void createSplitsDir(RegionInfo daughterA, RegionInfo daughterB) throws I
    *                    have a reference to a Region.
    * @return Path to created reference.
    */
-  public Path splitStoreFile(RegionInfo hri, String familyName, HStoreFile f, byte[] splitRow,
-    boolean top, RegionSplitPolicy splitPolicy, StoreFileTracker tracker) throws IOException {
+  public StoreFileInfo splitStoreFile(RegionInfo hri, String familyName, HStoreFile f,
+    byte[] splitRow, boolean top, RegionSplitPolicy splitPolicy, StoreFileTracker tracker)
+    throws IOException {
     Path splitDir = new Path(getSplitsDir(hri), familyName);
     // Add the referred-to regions name as a dot separated suffix.
     // See REF_NAME_REGEX regex above. The referred-to regions name is
@@ -604,7 +627,7 @@ public Path splitStoreFile(RegionInfo hri, String familyName, HStoreFile f, byte
     Path p = new Path(splitDir, f.getPath().getName() + "." + parentRegionName);
     if (fs.exists(p)) {
       LOG.warn("Found an already existing split file for {}. Assuming this is a recovery.", p);
-      return p;
+      return tracker.getStoreFileInfo(fs.getFileStatus(p), p, true);
     }
     boolean createLinkFile = false;
     if (splitPolicy == null || !splitPolicy.skipStoreFileRangeCheck(familyName)) {
@@ -662,12 +685,12 @@ public Path splitStoreFile(RegionInfo hri, String familyName, HStoreFile f, byte
           hfileName = m.group(4);
         }
         // must create back reference here
-        tracker.createHFileLink(linkedTable, linkedRegion, hfileName, true);
+        HFileLink hFileLink = tracker.createHFileLink(linkedTable, linkedRegion, hfileName, true);
         Path path =
           new Path(splitDir, HFileLink.createHFileLinkName(linkedTable, linkedRegion, hfileName));
         LOG.info("Created linkFile:" + path.toString() + " for child: " + hri.getEncodedName()
           + ", parent: " + regionInfoForFs.getEncodedName());
-        return path;
+        return new StoreFileInfo(conf, fs, path, hFileLink);
       } catch (IOException e) {
         // if create HFileLink file failed, then just skip the error and create Reference file
         LOG.error("Create link file for " + hfileName + " for child " + hri.getEncodedName()
@@ -678,7 +701,7 @@ public Path splitStoreFile(RegionInfo hri, String familyName, HStoreFile f, byte
     Reference r =
       top ? Reference.createTopReference(splitRow) : Reference.createBottomReference(splitRow);
     tracker.createReference(r, p);
-    return p;
+    return new StoreFileInfo(conf, fs, p, r);
   }
 
   // ===========================================================================
@@ -747,7 +770,8 @@ public void commitMergedRegion(List<Path> allMergedFiles, MasterProcedureEnv env
       Path regionInfoFile = new Path(regionDir, REGION_INFO_FILE);
       byte[] regionInfoContent = getRegionInfoFileContent(regionInfo);
       writeRegionInfoFileContent(conf, fs, regionInfoFile, regionInfoContent);
-      insertRegionFilesIntoStoreTracker(allMergedFiles, env, this);
+      List<StoreFileInfo> infos = new ArrayList<StoreFileInfo>();
+      insertRegionfilePathsIntoStoreTracker(allMergedFiles, env, this);
     }
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileInfo.java

@@ -171,6 +171,17 @@ public StoreFileInfo(final Configuration conf, final FileSystem fs, final FileSt
     this(conf, fs, fileStatus, null, link);
   }
 
+  /**
+   * Create a Store File Info from an HFileLink
+   * @param conf       The {@link Configuration} to use
+   * @param fs         The current file system to use
+   * @param fileStatus The {@link FileStatus} of the file
+   */
+  public StoreFileInfo(final Configuration conf, final FileSystem fs, final Path initiaPath,
+    final HFileLink link) {
+    this(conf, fs, initiaPath, null, link);
+  }
+
   /**
    * Create a Store File Info from an HFileLink
    * @param conf       The {@link Configuration} to use
@@ -183,6 +194,18 @@ public StoreFileInfo(final Configuration conf, final FileSystem fs, final FileSt
     this(conf, fs, fileStatus, reference, null);
   }
 
+  /**
+   * Create a Store File Info from an HFileLink
+   * @param conf       The {@link Configuration} to use
+   * @param fs         The current file system to use
+   * @param fileStatus The {@link FileStatus} of the file
+   * @param reference  The reference instance
+   */
+  public StoreFileInfo(final Configuration conf, final FileSystem fs, final Path initialPath,
+    final Reference reference) {
+    this(conf, fs, initialPath, reference, null);
+  }
+
   /**
    * Create a Store File Info from an HFileLink and a Reference
    * @param conf       The {@link Configuration} to use
@@ -204,6 +227,26 @@ public StoreFileInfo(final Configuration conf, final FileSystem fs, final FileSt
       this.conf.getBoolean(STORE_FILE_READER_NO_READAHEAD, DEFAULT_STORE_FILE_READER_NO_READAHEAD);
   }
 
+  /**
+   * Create a Store File Info from an HFileLink and a Reference
+   * @param conf       The {@link Configuration} to use
+   * @param fs         The current file system to use
+   * @param fileStatus The {@link FileStatus} of the file
+   * @param reference  The reference instance
+   * @param link       The link instance
+   */
+  public StoreFileInfo(final Configuration conf, final FileSystem fs, final Path path,
+    final Reference reference, final HFileLink link) {
+    this.fs = fs;
+    this.conf = conf;
+    this.primaryReplica = false;
+    this.initialPath = path;
+    this.reference = reference;
+    this.link = link;
+    this.noReadahead =
+      this.conf.getBoolean(STORE_FILE_READER_NO_READAHEAD, DEFAULT_STORE_FILE_READER_NO_READAHEAD);
+  }
+
   /**
    * Create a Store File Info from an HFileLink and a Reference
    * @param conf       The {@link Configuration} to use