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/HFileLink.java

@@ -81,7 +81,7 @@ public class HFileLink extends FileLink {
    * The pattern should be used for hfile and reference links that can be found in
    * /hbase/table/region/family/
    */
-  private static final Pattern REF_OR_HFILE_LINK_PATTERN =
+  public static final Pattern REF_OR_HFILE_LINK_PATTERN =
     Pattern.compile(String.format("^(?:(%s)(?:=))?(%s)=(%s)-(.+)$", TableName.VALID_NAMESPACE_REGEX,
       TableName.VALID_TABLE_QUALIFIER_REGEX, RegionInfoBuilder.ENCODED_REGION_NAME_REGEX));
 

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/MergeTableRegionsProcedure.java

@@ -605,7 +605,7 @@ private void createMergedRegion(final MasterProcedureEnv env) throws IOException
     final MasterFileSystem mfs = env.getMasterServices().getMasterFileSystem();
     final Path tableDir = CommonFSUtils.getTableDir(mfs.getRootDir(), regionsToMerge[0].getTable());
     final FileSystem fs = mfs.getFileSystem();
-    List<Path> mergedFiles = new ArrayList<>();
+    List<StoreFileInfo> mergedFiles = new ArrayList<StoreFileInfo>();
     HRegionFileSystem mergeRegionFs = HRegionFileSystem
       .createRegionOnFileSystem(env.getMasterConfiguration(), fs, tableDir, mergedRegion);
 
@@ -622,11 +622,11 @@ private void createMergedRegion(final MasterProcedureEnv env) throws IOException
       .setState(State.MERGING_NEW);
   }
 
-  private List<Path> mergeStoreFiles(MasterProcedureEnv env, HRegionFileSystem regionFs,
+  private List<StoreFileInfo> mergeStoreFiles(MasterProcedureEnv env, HRegionFileSystem regionFs,
     HRegionFileSystem mergeRegionFs, RegionInfo mergedRegion) throws IOException {
     final TableDescriptor htd =
       env.getMasterServices().getTableDescriptors().get(mergedRegion.getTable());
-    List<Path> mergedFiles = new ArrayList<>();
+    List<StoreFileInfo> mergedFiles = new ArrayList<StoreFileInfo>();
     for (ColumnFamilyDescriptor hcd : htd.getColumnFamilies()) {
       String family = hcd.getNameAsString();
       StoreFileTracker tracker =
@@ -643,7 +643,7 @@ private List<Path> mergeStoreFiles(MasterProcedureEnv env, HRegionFileSystem reg
           // is running in a regionserver's Store context, or we might not be able
           // to read the hfiles.
           storeFileInfo.setConf(storeConfiguration);
-          Path refFile = mergeRegionFs.mergeStoreFile(regionFs.getRegionInfo(), family,
+          StoreFileInfo refFile = mergeRegionFs.mergeStoreFile(regionFs.getRegionInfo(), family,
             new HStoreFile(storeFileInfo, hcd.getBloomFilterType(), CacheConfig.DISABLED), tracker);
           mergedFiles.add(refFile);
         }

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

@@ -71,7 +71,6 @@
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
-import org.apache.hadoop.hbase.util.FSUtils;
 import org.apache.hadoop.hbase.util.Pair;
 import org.apache.hadoop.hbase.util.Threads;
 import org.apache.hadoop.hbase.wal.WALSplitUtil;
@@ -660,20 +659,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 {
@@ -689,8 +714,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
@@ -745,7 +770,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()) {
@@ -792,12 +818,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());
         }
@@ -819,19 +845,8 @@ private Pair<List<Path>, List<Path>> splitStoreFiles(final MasterProcedureEnv en
     return new Pair<>(daughterA, daughterB);
   }
 
-  private void assertSplitResultFilesCount(final FileSystem fs,
-    final int expectedSplitResultFileCount, Path dir) throws IOException {
-    if (expectedSplitResultFileCount != 0) {
-      int resultFileCount = FSUtils.getRegionReferenceAndLinkFileCount(fs, dir);
-      if (expectedSplitResultFileCount != resultFileCount) {
-        throw new IOException("Failing split. Didn't have expected reference and HFileLink files"
-          + ", expected=" + expectedSplitResultFileCount + ", actual=" + resultFileCount);
-      }
-    }
-  }
-
-  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());
@@ -847,22 +862,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;
@@ -883,7 +898,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/HMobStore.java

@@ -102,6 +102,7 @@ public class HMobStore extends HStore {
   // table, we need to find the original mob files by this table name. For details please see
   // cloning snapshot for mob files.
   private final byte[] refCellTags;
+  private StoreFileTracker mobStoreSFT = null;
 
   public HMobStore(final HRegion region, final ColumnFamilyDescriptor family,
     final Configuration confParam, boolean warmup) throws IOException {

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

@@ -496,7 +496,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)) {
@@ -511,21 +511,43 @@ public Path commitDaughterRegion(final RegionInfo regionInfo, List<Path> allRegi
     return regionDir;
   }
 
-  private void insertRegionFilesIntoStoreTracker(List<Path> allFiles, MasterProcedureEnv env,
-    HRegionFileSystem regionFs) throws IOException {
+  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 (Path file : allFiles) {
+    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(trackerMap.get(familyName).getStoreFileInfo(file, true));
+      infos.add(sfi);
+    }
+    for (Map.Entry<String, StoreFileTracker> entry : trackerMap.entrySet()) {
+      entry.getValue().add(fileInfoMap.get(entry.getKey()));
+    }
+  }
+
+  private void insertRegionfilePathsIntoStoreTracker(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 file : allFiles) {
+      String familyName = file.getPath().getParent().getName();
+      trackerMap.computeIfAbsent(familyName, t -> StoreFileTrackerFactory.create(conf, tblDesc,
+        tblDesc.getColumnFamily(familyName.getBytes()), regionFs));
+      fileInfoMap.computeIfAbsent(familyName, l -> new ArrayList<>());
+      List<StoreFileInfo> infos = fileInfoMap.get(familyName);
+      infos.add(file);
     }
     for (Map.Entry<String, StoreFileTracker> entry : trackerMap.entrySet()) {
       entry.getValue().add(fileInfoMap.get(entry.getKey()));
@@ -568,8 +590,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
@@ -581,7 +604,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)) {
@@ -639,12 +662,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()
@@ -655,7 +678,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);
   }
 
   // ===========================================================================
@@ -696,7 +719,7 @@ static boolean mkdirs(FileSystem fs, Configuration conf, Path dir) throws IOExce
    * @return Path to created reference.
    * @throws IOException if the merge write fails.
    */
-  public Path mergeStoreFile(RegionInfo mergingRegion, String familyName, HStoreFile f,
+  public StoreFileInfo mergeStoreFile(RegionInfo mergingRegion, String familyName, HStoreFile f,
     StoreFileTracker tracker) throws IOException {
     Path referenceDir = new Path(getMergesDir(regionInfoForFs), familyName);
     // A whole reference to the store file.
@@ -710,21 +733,22 @@ public Path mergeStoreFile(RegionInfo mergingRegion, String familyName, HStoreFi
     // suffix and into the new region location (under same family).
     Path p = new Path(referenceDir, f.getPath().getName() + "." + mergingRegionName);
     tracker.createReference(r, p);
-    return p;
+    StoreFileInfo storeFileInfo = new StoreFileInfo(conf, fs, p, r);
+    return storeFileInfo;
   }
 
   /**
    * Commit a merged region, making it ready for use.
    */
-  public void commitMergedRegion(List<Path> allMergedFiles, MasterProcedureEnv env)
+  public void commitMergedRegion(List<StoreFileInfo> allMergedFiles, MasterProcedureEnv env)
     throws IOException {
     Path regionDir = getMergesDir(regionInfoForFs);
     if (regionDir != null && fs.exists(regionDir)) {
       // Write HRI to a file in case we need to recover hbase:meta
       Path regionInfoFile = new Path(regionDir, REGION_INFO_FILE);
       byte[] regionInfoContent = getRegionInfoFileContent(regionInfo);
       writeRegionInfoFileContent(conf, fs, regionInfoFile, regionInfoContent);
-      insertRegionFilesIntoStoreTracker(allMergedFiles, env, this);
+      insertRegionfilePathsIntoStoreTracker(allMergedFiles, env, this);
     }
   }