HBASE-25249 Adding HStoreContext

Adding HStoreContext which contains the metadata about the HStore. This
meta data can be used across the HFileWriter/Readers and other HStore
consumers without the need of passing around the complete store and
exposing its internals.

Author: abhishekkhanna01

hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/HFileOutputFormat2.java

@@ -70,6 +70,7 @@
 import org.apache.hadoop.hbase.regionserver.BloomType;
 import org.apache.hadoop.hbase.regionserver.HStore;
 import org.apache.hadoop.hbase.regionserver.StoreFileWriter;
+import org.apache.hadoop.hbase.regionserver.StoreUtils;
 import org.apache.hadoop.hbase.util.BloomFilterUtil;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
@@ -369,8 +370,8 @@ private WriterLength getNewWriter(byte[] tableName, byte[] family, Configuration
         encoding = encoding == null ? datablockEncodingMap.get(tableAndFamily) : encoding;
         encoding = encoding == null ? DataBlockEncoding.NONE : encoding;
         HFileContextBuilder contextBuilder = new HFileContextBuilder().withCompression(compression)
-            .withDataBlockEncoding(encoding).withChecksumType(HStore.getChecksumType(conf))
-            .withBytesPerCheckSum(HStore.getBytesPerChecksum(conf)).withBlockSize(blockSize)
+            .withDataBlockEncoding(encoding).withChecksumType(StoreUtils.getChecksumType(conf))
+            .withBytesPerCheckSum(StoreUtils.getBytesPerChecksum(conf)).withBlockSize(blockSize)
             .withColumnFamily(family).withTableName(tableName);
 
         if (HFile.getFormatVersion(conf) >= HFile.MIN_FORMAT_VERSION_WITH_TAGS) {

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

@@ -59,6 +59,7 @@
 import org.apache.hadoop.hbase.regionserver.HStore;
 import org.apache.hadoop.hbase.regionserver.HStoreFile;
 import org.apache.hadoop.hbase.regionserver.StoreFileWriter;
+import org.apache.hadoop.hbase.regionserver.StoreUtils;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.ChecksumType;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
@@ -540,8 +541,8 @@ public static StoreFileWriter createWriter(Configuration conf, FileSystem fs,
       Compression.Algorithm compression, CacheConfig cacheConfig, Encryption.Context cryptoContext,
       boolean isCompaction) throws IOException {
     return createWriter(conf, fs, family, new Path(basePath, mobFileName.getFileName()),
-      maxKeyCount, compression, cacheConfig, cryptoContext, HStore.getChecksumType(conf),
-      HStore.getBytesPerChecksum(conf), family.getBlocksize(), BloomType.NONE, isCompaction);
+      maxKeyCount, compression, cacheConfig, cryptoContext, StoreUtils.getChecksumType(conf),
+      StoreUtils.getBytesPerChecksum(conf), family.getBlocksize(), BloomType.NONE, isCompaction);
   }
 
   /**

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

@@ -93,7 +93,6 @@ public class HMobStore extends HStore {
   private AtomicLong mobFlushedCellsSize = new AtomicLong();
   private AtomicLong mobScanCellsCount = new AtomicLong();
   private AtomicLong mobScanCellsSize = new AtomicLong();
-  private ColumnFamilyDescriptor family;
   private Map<TableName, List<Path>> map = new ConcurrentHashMap<>();
   private final IdLock keyLock = new IdLock();
   // When we add a MOB reference cell to the HFile, we will add 2 tags along with it
@@ -107,16 +106,15 @@ public class HMobStore extends HStore {
   public HMobStore(final HRegion region, final ColumnFamilyDescriptor family,
       final Configuration confParam, boolean warmup) throws IOException {
     super(region, family, confParam, warmup);
-    this.family = family;
     this.mobFileCache = region.getMobFileCache();
     this.homePath = MobUtils.getMobHome(conf);
     this.mobFamilyPath = MobUtils.getMobFamilyPath(conf, this.getTableName(),
-        family.getNameAsString());
+        getColumnFamilyName());
     List<Path> locations = new ArrayList<>(2);
     locations.add(mobFamilyPath);
     TableName tn = region.getTableDescriptor().getTableName();
     locations.add(HFileArchiveUtil.getStoreArchivePath(conf, tn, MobUtils.getMobRegionInfo(tn)
-        .getEncodedName(), family.getNameAsString()));
+        .getEncodedName(), getColumnFamilyName()));
     map.put(tn, locations);
     List<Tag> tags = new ArrayList<>(2);
     tags.add(MobConstants.MOB_REF_TAG);
@@ -209,7 +207,7 @@ public StoreFileWriter createWriterInTmp(String date, Path basePath, long maxKey
       Compression.Algorithm compression, byte[] startKey,
       boolean isCompaction) throws IOException {
     MobFileName mobFileName = MobFileName.create(startKey, date, UUID.randomUUID()
-        .toString().replaceAll("-", ""),  region.getRegionInfo().getEncodedName());
+        .toString().replaceAll("-", ""),  getHRegion().getRegionInfo().getEncodedName());
     return createWriterInTmp(mobFileName, basePath, maxKeyCount, compression, isCompaction);
   }
 
@@ -226,9 +224,10 @@ public StoreFileWriter createWriterInTmp(String date, Path basePath, long maxKey
   public StoreFileWriter createWriterInTmp(MobFileName mobFileName, Path basePath,
       long maxKeyCount, Compression.Algorithm compression,
       boolean isCompaction) throws IOException {
-    return MobUtils.createWriter(conf, region.getFilesystem(), family,
-      new Path(basePath, mobFileName.getFileName()), maxKeyCount, compression, cacheConf,
-      cryptoContext, checksumType, bytesPerChecksum, blocksize, BloomType.NONE, isCompaction);
+    return MobUtils.createWriter(conf, getFileSystem(), getColumnFamilyDescriptor(),
+      new Path(basePath, mobFileName.getFileName()), maxKeyCount, compression, getCacheConfig(),
+        getEncryptionContext(), StoreUtils.getChecksumType(conf), StoreUtils.getBytesPerChecksum(conf),
+        blocksize, BloomType.NONE, isCompaction);
   }
 
   /**
@@ -245,10 +244,10 @@ public void commitFile(final Path sourceFile, Path targetPath) throws IOExceptio
     validateMobFile(sourceFile);
     LOG.info(" FLUSH Renaming flushed file from {} to {}", sourceFile, dstPath);
     Path parent = dstPath.getParent();
-    if (!region.getFilesystem().exists(parent)) {
-      region.getFilesystem().mkdirs(parent);
+    if (!getFileSystem().exists(parent)) {
+      getFileSystem().mkdirs(parent);
     }
-    if (!region.getFilesystem().rename(sourceFile, dstPath)) {
+    if (!getFileSystem().rename(sourceFile, dstPath)) {
       throw new IOException("Failed rename of " + sourceFile + " to " + dstPath);
     }
   }
@@ -261,7 +260,7 @@ public void commitFile(final Path sourceFile, Path targetPath) throws IOExceptio
   private void validateMobFile(Path path) throws IOException {
     HStoreFile storeFile = null;
     try {
-      storeFile = new HStoreFile(region.getFilesystem(), path, conf, this.cacheConf,
+      storeFile = new HStoreFile(getFileSystem(), path, conf, getCacheConfig(),
           BloomType.NONE, isPrimaryReplicaStore());
       storeFile.initReader();
     } catch (IOException e) {
@@ -352,9 +351,11 @@ public List<Path> getLocations(TableName tableName) throws IOException {
         locations = map.get(tableName);
         if (locations == null) {
           locations = new ArrayList<>(2);
-          locations.add(MobUtils.getMobFamilyPath(conf, tableName, family.getNameAsString()));
+          locations.add(MobUtils.getMobFamilyPath(conf, tableName, getColumnFamilyDescriptor()
+            .getNameAsString()));
           locations.add(HFileArchiveUtil.getStoreArchivePath(conf, tableName,
-            MobUtils.getMobRegionInfo(tableName).getEncodedName(), family.getNameAsString()));
+            MobUtils.getMobRegionInfo(tableName).getEncodedName(), getColumnFamilyDescriptor()
+              .getNameAsString()));
           map.put(tableName, locations);
         }
       } finally {
@@ -388,7 +389,7 @@ private MobCell readCell(List<Path> locations, String fileName, Cell search,
       MobFile file = null;
       Path path = new Path(location, fileName);
       try {
-        file = mobFileCache.openFile(fs, path, cacheConf);
+        file = mobFileCache.openFile(fs, path, getCacheConfig());
         return readPt != -1 ? file.readCell(search, cacheMobBlocks, readPt)
             : file.readCell(search, cacheMobBlocks);
       } catch (IOException e) {