Skip to content
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -56,9 +56,9 @@
import org.apache.hadoop.hbase.io.hfile.HFileContext;
import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
import org.apache.hadoop.hbase.regionserver.BloomType;
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;
Expand Down Expand Up @@ -540,8 +540,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);
}

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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);
Expand Down Expand Up @@ -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);
}

Expand All @@ -226,9 +224,11 @@ 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(),
getStoreContext().getEncryptionContext(), StoreUtils.getChecksumType(conf),
StoreUtils.getBytesPerChecksum(conf), getStoreContext().getBlockSize(), BloomType.NONE,
isCompaction);
}

/**
Expand All @@ -245,10 +245,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);
}
}
Expand All @@ -261,7 +261,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) {
Expand Down Expand Up @@ -352,9 +352,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 {
Expand Down Expand Up @@ -388,7 +390,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) {
Expand Down
Loading