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

Signed-off-by: Wellington Chevreuil <wchevreuil@apache.org>

Author: Apache9

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

@@ -29,7 +29,6 @@
 import java.util.List;
 import java.util.Map.Entry;
 import java.util.Optional;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
@@ -144,17 +143,16 @@ 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.getStoreEngine().createWriter(
+          createParams(fd, shouldDropBehind, major).includeMVCCReadpoint(true).includesTag(true));
+      }
+    };
 
   public DefaultMobStoreCompactor(Configuration conf, HStore store) {
     super(conf, store);

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

@@ -25,7 +25,6 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.Cell;
@@ -127,8 +126,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, store.getColumnFamilyDescriptor().getCompressionType(),
-            false, true, true, false);
+        writer = createWriter(snapshot, true);
         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/CreateStoreFileWriterParams.java

@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.regionserver;
+
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.io.compress.Compression;
+import org.apache.yetus.audience.InterfaceAudience;
+
+@InterfaceAudience.Private
+public final class CreateStoreFileWriterParams {
+
+  private long maxKeyCount;
+
+  private Compression.Algorithm compression;
+
+  private boolean isCompaction;
+
+  private boolean includeMVCCReadpoint;
+
+  private boolean includesTag;
+
+  private boolean shouldDropBehind;
+
+  private long totalCompactedFilesSize = -1;
+
+  private String fileStoragePolicy = HConstants.EMPTY_STRING;
+
+  private CreateStoreFileWriterParams() {
+  }
+
+  public long maxKeyCount() {
+    return maxKeyCount;
+  }
+
+  public CreateStoreFileWriterParams maxKeyCount(long maxKeyCount) {
+    this.maxKeyCount = maxKeyCount;
+    return this;
+  }
+
+  public Compression.Algorithm compression() {
+    return compression;
+  }
+
+  /**
+   * Set the compression algorithm to use
+   */
+  public CreateStoreFileWriterParams compression(Compression.Algorithm compression) {
+    this.compression = compression;
+    return this;
+  }
+
+  public boolean isCompaction() {
+    return isCompaction;
+  }
+
+  /**
+   * Whether we are creating a new file in a compaction
+   */
+  public CreateStoreFileWriterParams isCompaction(boolean isCompaction) {
+    this.isCompaction = isCompaction;
+    return this;
+  }
+
+  public boolean includeMVCCReadpoint() {
+    return includeMVCCReadpoint;
+  }
+
+  /**
+   * Whether to include MVCC or not
+   */
+  public CreateStoreFileWriterParams includeMVCCReadpoint(boolean includeMVCCReadpoint) {
+    this.includeMVCCReadpoint = includeMVCCReadpoint;
+    return this;
+  }
+
+  public boolean includesTag() {
+    return includesTag;
+  }
+
+  /**
+   * Whether to includesTag or not
+   */
+  public CreateStoreFileWriterParams includesTag(boolean includesTag) {
+    this.includesTag = includesTag;
+    return this;
+  }
+
+  public boolean shouldDropBehind() {
+    return shouldDropBehind;
+  }
+
+  public CreateStoreFileWriterParams shouldDropBehind(boolean shouldDropBehind) {
+    this.shouldDropBehind = shouldDropBehind;
+    return this;
+  }
+
+  public long totalCompactedFilesSize() {
+    return totalCompactedFilesSize;
+  }
+
+  public CreateStoreFileWriterParams totalCompactedFilesSize(long totalCompactedFilesSize) {
+    this.totalCompactedFilesSize = totalCompactedFilesSize;
+    return this;
+  }
+
+  public String fileStoragePolicy() {
+    return fileStoragePolicy;
+  }
+
+  public CreateStoreFileWriterParams fileStoragePolicy(String fileStoragePolicy) {
+    this.fileStoragePolicy = fileStoragePolicy;
+    return this;
+  }
+
+  public static CreateStoreFileWriterParams create() {
+    return new CreateStoreFileWriterParams();
+  }
+
+}

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

@@ -19,18 +19,17 @@
 
 import java.io.IOException;
 import java.util.List;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.CellComparator;
-import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequestImpl;
-import org.apache.yetus.audience.InterfaceAudience;
 import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
+import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequestImpl;
 import org.apache.hadoop.hbase.regionserver.compactions.DateTieredCompactionPolicy;
 import org.apache.hadoop.hbase.regionserver.compactions.DateTieredCompactionRequest;
 import org.apache.hadoop.hbase.regionserver.compactions.DateTieredCompactor;
 import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
 import org.apache.hadoop.hbase.security.User;
+import org.apache.yetus.audience.InterfaceAudience;
 
 /**
  * HBASE-15400 This store engine allows us to store data in date tiered layout with exponential

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

@@ -20,7 +20,6 @@
 
 import java.io.IOException;
 import java.util.List;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.CellComparator;
@@ -39,8 +38,8 @@
  * their derivatives.
  */
 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.CONFIG)
-public class DefaultStoreEngine extends StoreEngine<
-  DefaultStoreFlusher, RatioBasedCompactionPolicy, DefaultCompactor, DefaultStoreFileManager> {
+public class DefaultStoreEngine extends StoreEngine<DefaultStoreFlusher,
+  RatioBasedCompactionPolicy, DefaultCompactor, DefaultStoreFileManager> {
 
   public static final String DEFAULT_STORE_FLUSHER_CLASS_KEY =
       "hbase.hstore.defaultengine.storeflusher.class";

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

@@ -21,15 +21,14 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
-
-import org.apache.yetus.audience.InterfaceAudience;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
 import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
 import org.apache.hadoop.util.StringUtils;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Default implementation of StoreFlusher.
@@ -60,9 +59,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,
-            store.getColumnFamilyDescriptor().getCompressionType(), false, true,
-            snapshot.isTagsPresent(), false);
+        writer = createWriter(snapshot, false);
         IOException e = null;
         try {
           performFlush(scanner, writer, throughputController);

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

@@ -28,7 +28,6 @@
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -158,7 +157,7 @@ protected KeyValueScanner createScanner(Scan scan, ScanInfo scanInfo,
   protected StoreEngine<?, ?, ?, ?> createStoreEngine(HStore store, Configuration conf,
       CellComparator cellComparator) throws IOException {
     MobStoreEngine engine = new MobStoreEngine();
-    engine.createComponents(conf, store, cellComparator);
+    engine.createComponentsOnce(conf, store, cellComparator);
     return engine;
   }
 

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

@@ -142,7 +142,7 @@ public Path getRegionDir() {
   //  Temp Helpers
   // ===========================================================================
   /** @return {@link Path} to the region's temp directory, used for file creations */
-  Path getTempDir() {
+  public Path getTempDir() {
     return new Path(getRegionDir(), REGION_TEMP_DIR);
   }
 
@@ -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);