HBASE-26064 Introduce a StoreFileTracker to abstract the store file tracking logic

Author: Apache9

hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreCompactor.java

@@ -143,17 +143,17 @@ public InternalScanner createScanner(ScanInfo scanInfo, List<StoreFileScanner> s
   };
 
   private final CellSinkFactory<StoreFileWriter> writerFactory =
-      new CellSinkFactory<StoreFileWriter>() {
-        @Override
-        public StoreFileWriter createWriter(InternalScanner scanner,
-            org.apache.hadoop.hbase.regionserver.compactions.Compactor.FileDetails fd,
-            boolean shouldDropBehind, boolean major) throws IOException {
-          // make this writer with tags always because of possible new cells with tags.
-          return store.createWriterInTmp(fd.maxKeyCount,
-            major ? majorCompactionCompression : minorCompactionCompression,
-            true, true, true, shouldDropBehind);
-        }
-      };
+    new CellSinkFactory<StoreFileWriter>() {
+      @Override
+      public StoreFileWriter createWriter(InternalScanner scanner,
+        org.apache.hadoop.hbase.regionserver.compactions.Compactor.FileDetails fd,
+        boolean shouldDropBehind, boolean major) throws IOException {
+        // make this writer with tags always because of possible new cells with tags.
+        return store.createWriter(fd.maxKeyCount,
+          major ? majorCompactionCompression : minorCompactionCompression, true, true, true,
+          shouldDropBehind);
+      }
+    };
 
   public DefaultMobStoreCompactor(Configuration conf, HStore store) {
     super(conf, store);

hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreFlusher.java

@@ -127,8 +127,8 @@ public List<Path> flushSnapshot(MemStoreSnapshot snapshot, long cacheFlushId,
       synchronized (flushLock) {
         status.setStatus("Flushing " + store + ": creating writer");
         // Write the map out to the disk
-        writer = store.createWriterInTmp(cellsCount, store.getColumnFamilyDescriptor().getCompressionType(),
-            false, true, true, false);
+        writer = store.createWriter(cellsCount,
+          store.getColumnFamilyDescriptor().getCompressionType(), false, true, true, false);
         IOException e = null;
         try {
           // It's a mob store, flush the cells in a mob way. This is the difference of flushing

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

@@ -60,7 +60,7 @@ public List<Path> flushSnapshot(MemStoreSnapshot snapshot, long cacheFlushId,
       synchronized (flushLock) {
         status.setStatus("Flushing " + store + ": creating writer");
         // Write the map out to the disk
-        writer = store.createWriterInTmp(cellsCount,
+        writer = store.createWriter(cellsCount,
             store.getColumnFamilyDescriptor().getCompressionType(), false, true,
             snapshot.isTagsPresent(), false);
         IOException e = null;

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

@@ -237,11 +237,7 @@ public String getStoragePolicyName(String familyName) {
    * @param familyName Column Family Name
    * @return a set of {@link StoreFileInfo} for the specified family.
    */
-  public Collection<StoreFileInfo> getStoreFiles(final byte[] familyName) throws IOException {
-    return getStoreFiles(Bytes.toString(familyName));
-  }
-
-  public Collection<StoreFileInfo> getStoreFiles(final String familyName) throws IOException {
+  public List<StoreFileInfo> getStoreFiles(final String familyName) throws IOException {
     return getStoreFiles(familyName, true);
   }
 
@@ -251,7 +247,7 @@ public Collection<StoreFileInfo> getStoreFiles(final String familyName) throws I
    * @param familyName Column Family Name
    * @return a set of {@link StoreFileInfo} for the specified family.
    */
-  public Collection<StoreFileInfo> getStoreFiles(final String familyName, final boolean validate)
+  public List<StoreFileInfo> getStoreFiles(final String familyName, final boolean validate)
       throws IOException {
     Path familyDir = getStoreDir(familyName);
     FileStatus[] files = CommonFSUtils.listStatus(this.fs, familyDir);

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

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hbase.regionserver;
 
+import com.google.errorprone.annotations.RestrictedApi;
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.net.InetSocketAddress;
@@ -66,6 +67,7 @@
 import org.apache.hadoop.hbase.backup.FailedArchiveException;
 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
 import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.client.RegionReplicaUtil;
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.conf.ConfigurationManager;
 import org.apache.hadoop.hbase.conf.PropagatingConfigurationObserver;
@@ -89,6 +91,8 @@
 import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequestImpl;
 import org.apache.hadoop.hbase.regionserver.compactions.OffPeakHours;
 import org.apache.hadoop.hbase.regionserver.querymatcher.ScanQueryMatcher;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTracker;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
 import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
 import org.apache.hadoop.hbase.regionserver.wal.WALUtil;
 import org.apache.hadoop.hbase.security.EncryptionUtil;
@@ -210,6 +214,8 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
 
   private HFileDataBlockEncoder dataBlockEncoder;
 
+  private final StoreFileTracker storeFileTracker;
+
   final StoreEngine<?, ?, ?, ?> storeEngine;
 
   private static final AtomicBoolean offPeakCompactionTracker = new AtomicBoolean();
@@ -290,6 +296,7 @@ protected HStore(final HRegion region, final ColumnFamilyDescriptor family,
     }
 
     this.storeEngine = createStoreEngine(this, this.conf, region.getCellComparator());
+    this.storeFileTracker = StoreFileTrackerFactory.create(storeContext);
     List<HStoreFile> hStoreFiles = loadStoreFiles(warmup);
     // Move the storeSize calculation out of loadStoreFiles() method, because the secondary read
     // replica's refreshStoreFiles() will also use loadStoreFiles() to refresh its store files and
@@ -337,6 +344,7 @@ private StoreContext initializeStoreContext(ColumnFamilyDescriptor family) throw
       .withFamilyStoreDirectoryPath(region.getRegionFileSystem()
         .getStoreDir(family.getNameAsString()))
       .withRegionCoprocessorHost(region.getCoprocessorHost())
+      .withPrimary(RegionReplicaUtil.isDefaultReplica(region.getRegionInfo()))
       .build();
   }
 
@@ -518,7 +526,7 @@ void setDataBlockEncoderInTest(HFileDataBlockEncoder blockEncoder) {
    * from the given directory.
    */
   private List<HStoreFile> loadStoreFiles(boolean warmup) throws IOException {
-    Collection<StoreFileInfo> files = getRegionFileSystem().getStoreFiles(getColumnFamilyName());
+    Collection<StoreFileInfo> files = storeFileTracker.loadStoreFiles();
     return openStoreFiles(files, warmup);
   }
 
@@ -607,7 +615,7 @@ private List<HStoreFile> openStoreFiles(Collection<StoreFileInfo> files, boolean
 
   @Override
   public void refreshStoreFiles() throws IOException {
-    Collection<StoreFileInfo> newFiles = getRegionFileSystem().getStoreFiles(getColumnFamilyName());
+    Collection<StoreFileInfo> newFiles = storeFileTracker.loadStoreFiles();
     refreshStoreFilesInternal(newFiles);
   }
 
@@ -674,7 +682,7 @@ private void refreshStoreFilesInternal(Collection<StoreFileInfo> newFiles) throw
     List<HStoreFile> openedFiles = openStoreFiles(toBeAddedFiles, false);
 
     // propogate the file changes to the underlying store file manager
-    replaceStoreFiles(toBeRemovedStoreFiles, openedFiles); //won't throw an exception
+    replaceStoreFiles(toBeRemovedStoreFiles, openedFiles, false); //won't throw an exception
 
     // Advance the memstore read point to be at least the new store files seqIds so that
     // readers might pick it up. This assumes that the store is not getting any writes (otherwise
@@ -872,8 +880,7 @@ public Path bulkLoadHFile(byte[] family, String srcPathStr, Path dstPath) throws
     HStoreFile sf = createStoreFileAndReader(dstPath);
     bulkLoadHFile(sf);
 
-    LOG.info("Successfully loaded {} into {} (new location: {})",
-        srcPath, this, dstPath);
+    LOG.info("Successfully loaded {} into {} (new location: {})", srcPath, this, dstPath);
 
     return dstPath;
   }
@@ -888,6 +895,7 @@ private void bulkLoadHFile(HStoreFile sf) throws IOException {
     this.storeSize.addAndGet(r.length());
     this.totalUncompressedBytes.addAndGet(r.getTotalUncompressedBytes());
 
+    storeFileTracker.addNewStoreFiles(Lists.newArrayList(sf.getFileInfo()));
     // Append the new storefile into the list
     this.lock.writeLock().lock();
     try {
@@ -1071,6 +1079,7 @@ public HStoreFile tryCommitRecoveredHFile(Path path) throws IOException {
     this.storeSize.addAndGet(r.length());
     this.totalUncompressedBytes.addAndGet(r.getTotalUncompressedBytes());
 
+    storeFileTracker.addNewStoreFiles(Lists.newArrayList(sf.getFileInfo()));
     this.lock.writeLock().lock();
     try {
       this.storeEngine.getStoreFileManager().insertNewFiles(Lists.newArrayList(sf));
@@ -1088,9 +1097,20 @@ public HStoreFile tryCommitRecoveredHFile(Path path) throws IOException {
    * @return store file created.
    */
   private HStoreFile commitFile(Path path, long logCacheFlushId, MonitoredTask status)
-      throws IOException {
-    // Write-out finished successfully, move into the right spot
-    Path dstPath = getRegionFileSystem().commitStoreFile(getColumnFamilyName(), path);
+    throws IOException {
+    HRegionFileSystem hfs = getRegionFileSystem();
+    Path storeDir = hfs.getStoreDir(getColumnFamilyName());
+    Path dstPath;
+    // Do not use StoreFileTracker.requireWritingToTmpDirFirst here, as we have lots of places to
+    // commit a store file other than flush and compaction, where the store file could be in any
+    // places...
+    if (path.getParent() != null && path.getParent().equals(storeDir)) {
+      // already in the right place, skip renmaing
+      dstPath = path;
+    } else {
+      // Write-out finished successfully, move into the right spot
+      dstPath = hfs.commitStoreFile(getColumnFamilyName(), path);
+    }
 
     status.setStatus("Flushing " + this + ": reopening flushed file");
     HStoreFile sf = createStoreFileAndReader(dstPath);
@@ -1100,30 +1120,33 @@ private HStoreFile commitFile(Path path, long logCacheFlushId, MonitoredTask sta
     this.totalUncompressedBytes.addAndGet(r.getTotalUncompressedBytes());
 
     if (LOG.isInfoEnabled()) {
-      LOG.info("Added " + sf + ", entries=" + r.getEntries() +
-        ", sequenceid=" + logCacheFlushId +
+      LOG.info("Added " + sf + ", entries=" + r.getEntries() + ", sequenceid=" + logCacheFlushId +
         ", filesize=" + TraditionalBinaryPrefix.long2String(r.length(), "", 1));
     }
     return sf;
   }
 
-  public StoreFileWriter createWriterInTmp(long maxKeyCount, Compression.Algorithm compression,
+  public StoreFileWriter createWriter(long maxKeyCount, Compression.Algorithm compression,
     boolean isCompaction, boolean includeMVCCReadpoint, boolean includesTag,
     boolean shouldDropBehind) throws IOException {
-    return createWriterInTmp(maxKeyCount, compression, isCompaction, includeMVCCReadpoint,
+    return createWriter(maxKeyCount, compression, isCompaction, includeMVCCReadpoint,
       includesTag, shouldDropBehind, -1, HConstants.EMPTY_STRING);
   }
 
   /**
+   * Create a writer for writing new store files.
+   * <p/>
+   * Whether or not creating in tmp dir depends on
+   * {@link StoreFileTracker#requireWritingToTmpDirFirst()}
    * @param compression Compression algorithm to use
    * @param isCompaction whether we are creating a new file in a compaction
    * @param includeMVCCReadpoint - whether to include MVCC or not
    * @param includesTag - includesTag or not
-   * @return Writer for a new StoreFile in the tmp dir.
+   * @return Writer for a new StoreFile
    */
   // TODO : allow the Writer factory to create Writers of ShipperListener type only in case of
   // compaction
-  public StoreFileWriter createWriterInTmp(long maxKeyCount, Compression.Algorithm compression,
+  public StoreFileWriter createWriter(long maxKeyCount, Compression.Algorithm compression,
       boolean isCompaction, boolean includeMVCCReadpoint, boolean includesTag,
       boolean shouldDropBehind, long totalCompactedFilesSize, String fileStoragePolicy)
         throws IOException {
@@ -1171,10 +1194,15 @@ public StoreFileWriter createWriterInTmp(long maxKeyCount, Compression.Algorithm
     Encryption.Context encryptionContext = storeContext.getEncryptionContext();
     HFileContext hFileContext = createFileContext(compression, includeMVCCReadpoint, includesTag,
       encryptionContext);
-    Path familyTempDir = new Path(getRegionFileSystem().getTempDir(), getColumnFamilyName());
+    Path outputDir;
+    if (storeFileTracker.requireWritingToTmpDirFirst()) {
+      outputDir= new Path(getRegionFileSystem().getTempDir(), getColumnFamilyName());
+    } else {
+      throw new UnsupportedOperationException("not supported yet");
+    }
     StoreFileWriter.Builder builder =
       new StoreFileWriter.Builder(conf, writerCacheConf, getFileSystem())
-        .withOutputDir(familyTempDir)
+        .withOutputDir(outputDir)
         .withBloomType(storeContext.getBloomFilterType())
         .withMaxKeyCount(maxKeyCount)
         .withFavoredNodes(storeContext.getFavoredNodes())
@@ -1220,6 +1248,7 @@ private long getTotalSize(Collection<HStoreFile> sfs) {
    * @return Whether compaction is required.
    */
   private boolean updateStorefiles(List<HStoreFile> sfs, long snapshotId) throws IOException {
+    storeFileTracker.addNewStoreFiles(StoreUtils.toStoreFileInfo(sfs));
     this.lock.writeLock().lock();
     try {
       this.storeEngine.getStoreFileManager().insertNewFiles(sfs);
@@ -1502,8 +1531,7 @@ protected List<HStoreFile> doCompaction(CompactionRequestImpl cr,
     // Do the steps necessary to complete the compaction.
     setStoragePolicyFromFileName(newFiles);
     List<HStoreFile> sfs = moveCompactedFilesIntoPlace(cr, newFiles, user);
-    writeCompactionWalRecord(filesToCompact, sfs);
-    replaceStoreFiles(filesToCompact, sfs);
+    replaceStoreFiles(filesToCompact, sfs, true);
     if (cr.isMajor()) {
       majorCompactedCellsCount.addAndGet(getCompactionProgress().getTotalCompactingKVs());
       majorCompactedCellsSize.addAndGet(getCompactionProgress().totalCompactedSize);
@@ -1557,7 +1585,8 @@ private List<HStoreFile> moveCompactedFilesIntoPlace(CompactionRequestImpl cr,
     return sfs;
   }
 
-  // Package-visible for tests
+  @RestrictedApi(explanation = "Should only be called in TestHStore", link = "",
+    allowedOnPath = ".*/(HStore|TestCompaction).java")
   HStoreFile moveFileIntoPlace(Path newFile) throws IOException {
     validateStoreFile(newFile);
     // Move the file into the right spot
@@ -1590,8 +1619,15 @@ private void writeCompactionWalRecord(Collection<HStoreFile> filesCompacted,
         this.region.getRegionInfo(), compactionDescriptor, this.region.getMVCC());
   }
 
-  void replaceStoreFiles(Collection<HStoreFile> compactedFiles, Collection<HStoreFile> result)
-      throws IOException {
+  @RestrictedApi(explanation = "Should only be called in TestHStore", link = "",
+    allowedOnPath = ".*/(HStore|TestHStore).java")
+  void replaceStoreFiles(Collection<HStoreFile> compactedFiles, Collection<HStoreFile> result,
+    boolean writeCompactionMarker) throws IOException {
+    storeFileTracker.addCompactionResults(StoreUtils.toStoreFileInfo(compactedFiles),
+      StoreUtils.toStoreFileInfo(result));
+    if (writeCompactionMarker) {
+      writeCompactionWalRecord(compactedFiles, result);
+    }
     this.lock.writeLock().lock();
     try {
       this.storeEngine.getStoreFileManager().addCompactionResults(compactedFiles, result);
@@ -1604,8 +1640,8 @@ void replaceStoreFiles(Collection<HStoreFile> compactedFiles, Collection<HStoreF
       RegionServerServices rsServices = region.getRegionServerServices();
       if (rsServices != null && rsServices.getRegionServerSpaceQuotaManager() != null) {
         updateSpaceQuotaAfterFileReplacement(
-            rsServices.getRegionServerSpaceQuotaManager().getRegionSizeStore(), getRegionInfo(),
-            compactedFiles, result);
+          rsServices.getRegionServerSpaceQuotaManager().getRegionSizeStore(), getRegionInfo(),
+          compactedFiles, result);
       }
     } finally {
       this.lock.writeLock().unlock();
@@ -1739,7 +1775,7 @@ public void replayCompactionMarker(CompactionDescriptor compaction, boolean pick
     if (!inputStoreFiles.isEmpty() || !outputStoreFiles.isEmpty()) {
       LOG.info("Replaying compaction marker, replacing input files: " +
           inputStoreFiles + " with output files : " + outputStoreFiles);
-      this.replaceStoreFiles(inputStoreFiles, outputStoreFiles);
+      this.replaceStoreFiles(inputStoreFiles, outputStoreFiles, false);
       this.refreshStoreSizeAndTotalBytes();
     }
   }
@@ -1921,8 +1957,7 @@ private void removeUnneededFiles() throws IOException {
     }
 
     Collection<HStoreFile> newFiles = Collections.emptyList(); // No new files.
-    writeCompactionWalRecord(delSfs, newFiles);
-    replaceStoreFiles(delSfs, newFiles);
+    replaceStoreFiles(delSfs, newFiles, true);
     refreshStoreSizeAndTotalBytes();
     LOG.info("Completed removal of " + delSfs.size() + " unnecessary (expired) file(s) in "
         + this + "; total size is "

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

@@ -49,6 +49,7 @@ public final class StoreContext implements HeapSize {
   private final ColumnFamilyDescriptor family;
   private final Path familyStoreDirectoryPath;
   private final RegionCoprocessorHost coprocessorHost;
+  private final boolean primary;
 
   private StoreContext(Builder builder) {
     this.blockSize = builder.blockSize;
@@ -62,6 +63,7 @@ private StoreContext(Builder builder) {
     this.family = builder.family;
     this.familyStoreDirectoryPath = builder.familyStoreDirectoryPath;
     this.coprocessorHost = builder.coprocessorHost;
+    this.primary = builder.primary;
   }
 
   public int getBlockSize() {
@@ -108,6 +110,10 @@ public RegionCoprocessorHost getCoprocessorHost() {
     return coprocessorHost;
   }
 
+  public boolean isPrimary() {
+    return primary;
+  }
+
   public static Builder getBuilder() {
     return new Builder();
   }
@@ -129,6 +135,7 @@ public static class Builder {
     private ColumnFamilyDescriptor family;
     private Path familyStoreDirectoryPath;
     private RegionCoprocessorHost coprocessorHost;
+    private boolean primary;
 
     public Builder withBlockSize(int blockSize) {
       this.blockSize = blockSize;
@@ -186,9 +193,13 @@ public Builder withRegionCoprocessorHost(RegionCoprocessorHost coprocessorHost)
       return this;
     }
 
+    public Builder withPrimary(boolean primary) {
+      this.primary = primary;
+      return this;
+    }
+
     public StoreContext build() {
       return new StoreContext(this);
     }
   }
-
 }