HBASE-7935 make policy and compactor in default store engine separately pluggable (for things like tier-based, and default policy experiments with permutations)
git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1453586 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
6ae6e5559b
commit
e67d82a2a0
|
@ -31,6 +31,7 @@ import org.apache.hadoop.hbase.regionserver.compactions.CompactionPolicy;
|
||||||
import org.apache.hadoop.hbase.regionserver.compactions.Compactor;
|
import org.apache.hadoop.hbase.regionserver.compactions.Compactor;
|
||||||
import org.apache.hadoop.hbase.regionserver.compactions.DefaultCompactionPolicy;
|
import org.apache.hadoop.hbase.regionserver.compactions.DefaultCompactionPolicy;
|
||||||
import org.apache.hadoop.hbase.regionserver.compactions.DefaultCompactor;
|
import org.apache.hadoop.hbase.regionserver.compactions.DefaultCompactor;
|
||||||
|
import org.apache.hadoop.hbase.util.ReflectionUtils;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Default StoreEngine creates the default compactor, policy, and store file manager, or
|
* Default StoreEngine creates the default compactor, policy, and store file manager, or
|
||||||
|
@ -40,21 +41,40 @@ import org.apache.hadoop.hbase.regionserver.compactions.DefaultCompactor;
|
||||||
public class DefaultStoreEngine extends StoreEngine<
|
public class DefaultStoreEngine extends StoreEngine<
|
||||||
DefaultCompactionPolicy, DefaultCompactor, DefaultStoreFileManager> {
|
DefaultCompactionPolicy, DefaultCompactor, DefaultStoreFileManager> {
|
||||||
|
|
||||||
public DefaultStoreEngine(Configuration conf, Store store, KVComparator comparator) {
|
public static final String DEFAULT_COMPACTOR_CLASS_KEY =
|
||||||
super(conf, store, comparator);
|
"hbase.hstore.defaultengine.compactor.class";
|
||||||
|
public static final String DEFAULT_COMPACTION_POLICY_CLASS_KEY =
|
||||||
|
"hbase.hstore.defaultengine.compactionpolicy.class";
|
||||||
|
|
||||||
|
private static final Class<? extends DefaultCompactor>
|
||||||
|
DEFAULT_COMPACTOR_CLASS = DefaultCompactor.class;
|
||||||
|
private static final Class<? extends DefaultCompactionPolicy>
|
||||||
|
DEFAULT_COMPACTION_POLICY_CLASS = DefaultCompactionPolicy.class;
|
||||||
|
|
||||||
|
@Override
|
||||||
|
protected void createComponents(
|
||||||
|
Configuration conf, Store store, KVComparator kvComparator) throws IOException {
|
||||||
|
storeFileManager = new DefaultStoreFileManager(kvComparator, conf);
|
||||||
|
String className = conf.get(DEFAULT_COMPACTOR_CLASS_KEY, DEFAULT_COMPACTOR_CLASS.getName());
|
||||||
|
try {
|
||||||
|
compactor = ReflectionUtils.instantiateWithCustomCtor(className,
|
||||||
|
new Class[] { Configuration.class, Store.class }, new Object[] { conf, store });
|
||||||
|
} catch (Exception e) {
|
||||||
|
throw new IOException("Unable to load configured compactor '" + className + "'", e);
|
||||||
|
}
|
||||||
|
className = conf.get(
|
||||||
|
DEFAULT_COMPACTION_POLICY_CLASS_KEY, DEFAULT_COMPACTION_POLICY_CLASS.getName());
|
||||||
|
try {
|
||||||
|
compactionPolicy = ReflectionUtils.instantiateWithCustomCtor(className,
|
||||||
|
new Class[] { Configuration.class, StoreConfigInformation.class },
|
||||||
|
new Object[] { conf, store });
|
||||||
|
} catch (Exception e) {
|
||||||
|
throw new IOException("Unable to load configured compaction policy '" + className + "'", e);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected void createComponents() {
|
public CompactionContext createCompaction() {
|
||||||
storeFileManager = new DefaultStoreFileManager(this.comparator, this.conf);
|
|
||||||
|
|
||||||
// TODO: compactor and policy may be separately pluggable, but must derive from default ones.
|
|
||||||
compactor = new DefaultCompactor(this.conf, this.store);
|
|
||||||
compactionPolicy = new DefaultCompactionPolicy(this.conf, this.store/*as StoreConfigInfo*/);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
protected CompactionContext createCompactionContext() {
|
|
||||||
return new DefaultCompactionContext();
|
return new DefaultCompactionContext();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -113,7 +113,6 @@ public class HStore implements Store {
|
||||||
// This stores directory in the filesystem.
|
// This stores directory in the filesystem.
|
||||||
private final HRegion region;
|
private final HRegion region;
|
||||||
private final HColumnDescriptor family;
|
private final HColumnDescriptor family;
|
||||||
final CompactionPolicy compactionPolicy;
|
|
||||||
private final HRegionFileSystem fs;
|
private final HRegionFileSystem fs;
|
||||||
private final Configuration conf;
|
private final Configuration conf;
|
||||||
private final CacheConfig cacheConf;
|
private final CacheConfig cacheConf;
|
||||||
|
@ -129,7 +128,6 @@ public class HStore implements Store {
|
||||||
|
|
||||||
private ScanInfo scanInfo;
|
private ScanInfo scanInfo;
|
||||||
|
|
||||||
private final StoreFileManager storeFileManager;
|
|
||||||
final List<StoreFile> filesCompacting = Lists.newArrayList();
|
final List<StoreFile> filesCompacting = Lists.newArrayList();
|
||||||
|
|
||||||
// All access must be synchronized.
|
// All access must be synchronized.
|
||||||
|
@ -210,12 +208,8 @@ public class HStore implements Store {
|
||||||
"hbase.hstore.close.check.interval", 10*1000*1000 /* 10 MB */);
|
"hbase.hstore.close.check.interval", 10*1000*1000 /* 10 MB */);
|
||||||
}
|
}
|
||||||
|
|
||||||
storeEngine = StoreEngine.create(this, this.conf, this.comparator);
|
this.storeEngine = StoreEngine.create(this, this.conf, this.comparator);
|
||||||
// Copy some things to local fields for convenience.
|
this.storeEngine.getStoreFileManager().loadFiles(loadStoreFiles());
|
||||||
this.storeFileManager = storeEngine.getStoreFileManager();
|
|
||||||
this.compactionPolicy = storeEngine.getCompactionPolicy();
|
|
||||||
|
|
||||||
this.storeFileManager.loadFiles(loadStoreFiles());
|
|
||||||
|
|
||||||
// Initialize checksum type from name. The names are CRC32, CRC32C, etc.
|
// Initialize checksum type from name. The names are CRC32, CRC32C, etc.
|
||||||
this.checksumType = getChecksumType(conf);
|
this.checksumType = getChecksumType(conf);
|
||||||
|
@ -479,7 +473,7 @@ public class HStore implements Store {
|
||||||
*/
|
*/
|
||||||
@Override
|
@Override
|
||||||
public Collection<StoreFile> getStorefiles() {
|
public Collection<StoreFile> getStorefiles() {
|
||||||
return this.storeFileManager.getStorefiles();
|
return this.storeEngine.getStoreFileManager().getStorefiles();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -560,7 +554,7 @@ public class HStore implements Store {
|
||||||
// Append the new storefile into the list
|
// Append the new storefile into the list
|
||||||
this.lock.writeLock().lock();
|
this.lock.writeLock().lock();
|
||||||
try {
|
try {
|
||||||
this.storeFileManager.insertNewFile(sf);
|
this.storeEngine.getStoreFileManager().insertNewFile(sf);
|
||||||
} finally {
|
} finally {
|
||||||
// We need the lock, as long as we are updating the storeFiles
|
// We need the lock, as long as we are updating the storeFiles
|
||||||
// or changing the memstore. Let us release it before calling
|
// or changing the memstore. Let us release it before calling
|
||||||
|
@ -579,7 +573,7 @@ public class HStore implements Store {
|
||||||
this.lock.writeLock().lock();
|
this.lock.writeLock().lock();
|
||||||
try {
|
try {
|
||||||
// Clear so metrics doesn't find them.
|
// Clear so metrics doesn't find them.
|
||||||
ImmutableCollection<StoreFile> result = storeFileManager.clearFiles();
|
ImmutableCollection<StoreFile> result = storeEngine.getStoreFileManager().clearFiles();
|
||||||
|
|
||||||
if (!result.isEmpty()) {
|
if (!result.isEmpty()) {
|
||||||
// initialize the thread pool for closing store files in parallel.
|
// initialize the thread pool for closing store files in parallel.
|
||||||
|
@ -879,7 +873,7 @@ public class HStore implements Store {
|
||||||
throws IOException {
|
throws IOException {
|
||||||
this.lock.writeLock().lock();
|
this.lock.writeLock().lock();
|
||||||
try {
|
try {
|
||||||
this.storeFileManager.insertNewFile(sf);
|
this.storeEngine.getStoreFileManager().insertNewFile(sf);
|
||||||
this.memstore.clearSnapshot(set);
|
this.memstore.clearSnapshot(set);
|
||||||
} finally {
|
} finally {
|
||||||
// We need the lock, as long as we are updating the storeFiles
|
// We need the lock, as long as we are updating the storeFiles
|
||||||
|
@ -919,7 +913,8 @@ public class HStore implements Store {
|
||||||
List<KeyValueScanner> memStoreScanners;
|
List<KeyValueScanner> memStoreScanners;
|
||||||
this.lock.readLock().lock();
|
this.lock.readLock().lock();
|
||||||
try {
|
try {
|
||||||
storeFilesToScan = this.storeFileManager.getFilesForScanOrGet(isGet, startRow, stopRow);
|
storeFilesToScan =
|
||||||
|
this.storeEngine.getStoreFileManager().getFilesForScanOrGet(isGet, startRow, stopRow);
|
||||||
memStoreScanners = this.memstore.getScanners();
|
memStoreScanners = this.memstore.getScanners();
|
||||||
} finally {
|
} finally {
|
||||||
this.lock.readLock().unlock();
|
this.lock.readLock().unlock();
|
||||||
|
@ -1083,7 +1078,7 @@ public class HStore implements Store {
|
||||||
this.lock.readLock().lock();
|
this.lock.readLock().lock();
|
||||||
try {
|
try {
|
||||||
synchronized (filesCompacting) {
|
synchronized (filesCompacting) {
|
||||||
filesToCompact = Lists.newArrayList(storeFileManager.getStorefiles());
|
filesToCompact = Lists.newArrayList(storeEngine.getStoreFileManager().getStorefiles());
|
||||||
if (!filesCompacting.isEmpty()) {
|
if (!filesCompacting.isEmpty()) {
|
||||||
// exclude all files older than the newest file we're currently
|
// exclude all files older than the newest file we're currently
|
||||||
// compacting. this allows us to preserve contiguity (HBASE-2856)
|
// compacting. this allows us to preserve contiguity (HBASE-2856)
|
||||||
|
@ -1098,7 +1093,7 @@ public class HStore implements Store {
|
||||||
}
|
}
|
||||||
|
|
||||||
filesToCompact = filesToCompact.subList(count - N, count);
|
filesToCompact = filesToCompact.subList(count - N, count);
|
||||||
isMajor = (filesToCompact.size() == storeFileManager.getStorefileCount());
|
isMajor = (filesToCompact.size() == storeEngine.getStoreFileManager().getStorefileCount());
|
||||||
filesCompacting.addAll(filesToCompact);
|
filesCompacting.addAll(filesToCompact);
|
||||||
Collections.sort(filesCompacting, StoreFile.Comparators.SEQ_ID);
|
Collections.sort(filesCompacting, StoreFile.Comparators.SEQ_ID);
|
||||||
}
|
}
|
||||||
|
@ -1129,7 +1124,7 @@ public class HStore implements Store {
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public boolean hasReferences() {
|
public boolean hasReferences() {
|
||||||
return StoreUtils.hasReferences(this.storeFileManager.getStorefiles());
|
return StoreUtils.hasReferences(this.storeEngine.getStoreFileManager().getStorefiles());
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -1139,14 +1134,15 @@ public class HStore implements Store {
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public boolean isMajorCompaction() throws IOException {
|
public boolean isMajorCompaction() throws IOException {
|
||||||
for (StoreFile sf : this.storeFileManager.getStorefiles()) {
|
for (StoreFile sf : this.storeEngine.getStoreFileManager().getStorefiles()) {
|
||||||
// TODO: what are these reader checks all over the place?
|
// TODO: what are these reader checks all over the place?
|
||||||
if (sf.getReader() == null) {
|
if (sf.getReader() == null) {
|
||||||
LOG.debug("StoreFile " + sf + " has null Reader");
|
LOG.debug("StoreFile " + sf + " has null Reader");
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return compactionPolicy.isMajorCompaction(this.storeFileManager.getStorefiles());
|
return storeEngine.getCompactionPolicy().isMajorCompaction(
|
||||||
|
this.storeEngine.getStoreFileManager().getStorefiles());
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -1307,7 +1303,7 @@ public class HStore implements Store {
|
||||||
// delete old store files until we have sent out notification of
|
// delete old store files until we have sent out notification of
|
||||||
// change in case old files are still being accessed by outstanding
|
// change in case old files are still being accessed by outstanding
|
||||||
// scanners.
|
// scanners.
|
||||||
this.storeFileManager.addCompactionResults(compactedFiles, result);
|
this.storeEngine.getStoreFileManager().addCompactionResults(compactedFiles, result);
|
||||||
filesCompacting.removeAll(compactedFiles); // safe bc: lock.writeLock()
|
filesCompacting.removeAll(compactedFiles); // safe bc: lock.writeLock()
|
||||||
} finally {
|
} finally {
|
||||||
// We need the lock, as long as we are updating the storeFiles
|
// We need the lock, as long as we are updating the storeFiles
|
||||||
|
@ -1336,7 +1332,7 @@ public class HStore implements Store {
|
||||||
// 4. Compute new store size
|
// 4. Compute new store size
|
||||||
this.storeSize = 0L;
|
this.storeSize = 0L;
|
||||||
this.totalUncompressedBytes = 0L;
|
this.totalUncompressedBytes = 0L;
|
||||||
for (StoreFile hsf : this.storeFileManager.getStorefiles()) {
|
for (StoreFile hsf : this.storeEngine.getStoreFileManager().getStorefiles()) {
|
||||||
StoreFile.Reader r = hsf.getReader();
|
StoreFile.Reader r = hsf.getReader();
|
||||||
if (r == null) {
|
if (r == null) {
|
||||||
LOG.warn("StoreFile " + hsf + " has a null Reader");
|
LOG.warn("StoreFile " + hsf + " has a null Reader");
|
||||||
|
@ -1384,16 +1380,16 @@ public class HStore implements Store {
|
||||||
this.memstore.getRowKeyAtOrBefore(state);
|
this.memstore.getRowKeyAtOrBefore(state);
|
||||||
// Check if match, if we got a candidate on the asked for 'kv' row.
|
// Check if match, if we got a candidate on the asked for 'kv' row.
|
||||||
// Process each relevant store file. Run through from newest to oldest.
|
// Process each relevant store file. Run through from newest to oldest.
|
||||||
Iterator<StoreFile> sfIterator =
|
Iterator<StoreFile> sfIterator = this.storeEngine.getStoreFileManager()
|
||||||
this.storeFileManager.getCandidateFilesForRowKeyBefore(state.getTargetKey());
|
.getCandidateFilesForRowKeyBefore(state.getTargetKey());
|
||||||
while (sfIterator.hasNext()) {
|
while (sfIterator.hasNext()) {
|
||||||
StoreFile sf = sfIterator.next();
|
StoreFile sf = sfIterator.next();
|
||||||
sfIterator.remove(); // Remove sf from iterator.
|
sfIterator.remove(); // Remove sf from iterator.
|
||||||
boolean haveNewCandidate = rowAtOrBeforeFromStoreFile(sf, state);
|
boolean haveNewCandidate = rowAtOrBeforeFromStoreFile(sf, state);
|
||||||
if (haveNewCandidate) {
|
if (haveNewCandidate) {
|
||||||
// TODO: we may have an optimization here which stops the search if we find exact match.
|
// TODO: we may have an optimization here which stops the search if we find exact match.
|
||||||
sfIterator = this.storeFileManager.updateCandidateFilesForRowKeyBefore(sfIterator,
|
sfIterator = this.storeEngine.getStoreFileManager().updateCandidateFilesForRowKeyBefore(
|
||||||
state.getTargetKey(), state.getCandidate());
|
sfIterator, state.getTargetKey(), state.getCandidate());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return state.getCandidate();
|
return state.getCandidate();
|
||||||
|
@ -1537,7 +1533,7 @@ public class HStore implements Store {
|
||||||
assert false : "getSplitPoint() called on a region that can't split!";
|
assert false : "getSplitPoint() called on a region that can't split!";
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
return this.storeFileManager.getSplitPoint();
|
return this.storeEngine.getStoreFileManager().getSplitPoint();
|
||||||
} catch(IOException e) {
|
} catch(IOException e) {
|
||||||
LOG.warn("Failed getting store size for " + this, e);
|
LOG.warn("Failed getting store size for " + this, e);
|
||||||
} finally {
|
} finally {
|
||||||
|
@ -1594,7 +1590,7 @@ public class HStore implements Store {
|
||||||
@Override
|
@Override
|
||||||
// TODO: why is there this and also getNumberOfStorefiles?! Remove one.
|
// TODO: why is there this and also getNumberOfStorefiles?! Remove one.
|
||||||
public int getStorefilesCount() {
|
public int getStorefilesCount() {
|
||||||
return this.storeFileManager.getStorefileCount();
|
return this.storeEngine.getStoreFileManager().getStorefileCount();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -1605,7 +1601,7 @@ public class HStore implements Store {
|
||||||
@Override
|
@Override
|
||||||
public long getStorefilesSize() {
|
public long getStorefilesSize() {
|
||||||
long size = 0;
|
long size = 0;
|
||||||
for (StoreFile s: this.storeFileManager.getStorefiles()) {
|
for (StoreFile s: this.storeEngine.getStoreFileManager().getStorefiles()) {
|
||||||
StoreFile.Reader r = s.getReader();
|
StoreFile.Reader r = s.getReader();
|
||||||
if (r == null) {
|
if (r == null) {
|
||||||
LOG.warn("StoreFile " + s + " has a null Reader");
|
LOG.warn("StoreFile " + s + " has a null Reader");
|
||||||
|
@ -1619,7 +1615,7 @@ public class HStore implements Store {
|
||||||
@Override
|
@Override
|
||||||
public long getStorefilesIndexSize() {
|
public long getStorefilesIndexSize() {
|
||||||
long size = 0;
|
long size = 0;
|
||||||
for (StoreFile s: this.storeFileManager.getStorefiles()) {
|
for (StoreFile s: this.storeEngine.getStoreFileManager().getStorefiles()) {
|
||||||
StoreFile.Reader r = s.getReader();
|
StoreFile.Reader r = s.getReader();
|
||||||
if (r == null) {
|
if (r == null) {
|
||||||
LOG.warn("StoreFile " + s + " has a null Reader");
|
LOG.warn("StoreFile " + s + " has a null Reader");
|
||||||
|
@ -1633,7 +1629,7 @@ public class HStore implements Store {
|
||||||
@Override
|
@Override
|
||||||
public long getTotalStaticIndexSize() {
|
public long getTotalStaticIndexSize() {
|
||||||
long size = 0;
|
long size = 0;
|
||||||
for (StoreFile s : this.storeFileManager.getStorefiles()) {
|
for (StoreFile s : this.storeEngine.getStoreFileManager().getStorefiles()) {
|
||||||
size += s.getReader().getUncompressedDataIndexSize();
|
size += s.getReader().getUncompressedDataIndexSize();
|
||||||
}
|
}
|
||||||
return size;
|
return size;
|
||||||
|
@ -1642,7 +1638,7 @@ public class HStore implements Store {
|
||||||
@Override
|
@Override
|
||||||
public long getTotalStaticBloomSize() {
|
public long getTotalStaticBloomSize() {
|
||||||
long size = 0;
|
long size = 0;
|
||||||
for (StoreFile s : this.storeFileManager.getStorefiles()) {
|
for (StoreFile s : this.storeEngine.getStoreFileManager().getStorefiles()) {
|
||||||
StoreFile.Reader r = s.getReader();
|
StoreFile.Reader r = s.getReader();
|
||||||
size += r.getTotalBloomSize();
|
size += r.getTotalBloomSize();
|
||||||
}
|
}
|
||||||
|
@ -1656,12 +1652,12 @@ public class HStore implements Store {
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public int getCompactPriority() {
|
public int getCompactPriority() {
|
||||||
return this.storeFileManager.getStoreCompactionPriority();
|
return this.storeEngine.getStoreFileManager().getStoreCompactionPriority();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public boolean throttleCompaction(long compactionSize) {
|
public boolean throttleCompaction(long compactionSize) {
|
||||||
return compactionPolicy.throttleCompaction(compactionSize);
|
return storeEngine.getCompactionPolicy().throttleCompaction(compactionSize);
|
||||||
}
|
}
|
||||||
|
|
||||||
public HRegion getHRegion() {
|
public HRegion getHRegion() {
|
||||||
|
@ -1780,7 +1776,8 @@ public class HStore implements Store {
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public boolean needsCompaction() {
|
public boolean needsCompaction() {
|
||||||
return compactionPolicy.needsCompaction(this.storeFileManager.getStorefiles(), filesCompacting);
|
return storeEngine.getCompactionPolicy().needsCompaction(
|
||||||
|
this.storeEngine.getStoreFileManager().getStorefiles(), filesCompacting);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -1789,7 +1786,7 @@ public class HStore implements Store {
|
||||||
}
|
}
|
||||||
|
|
||||||
public static final long FIXED_OVERHEAD =
|
public static final long FIXED_OVERHEAD =
|
||||||
ClassSize.align((19 * ClassSize.REFERENCE) + (4 * Bytes.SIZEOF_LONG)
|
ClassSize.align((17 * ClassSize.REFERENCE) + (4 * Bytes.SIZEOF_LONG)
|
||||||
+ (2 * Bytes.SIZEOF_INT) + Bytes.SIZEOF_BOOLEAN);
|
+ (2 * Bytes.SIZEOF_INT) + Bytes.SIZEOF_BOOLEAN);
|
||||||
|
|
||||||
public static final long DEEP_OVERHEAD = ClassSize.align(FIXED_OVERHEAD
|
public static final long DEEP_OVERHEAD = ClassSize.align(FIXED_OVERHEAD
|
||||||
|
|
|
@ -39,14 +39,9 @@ import org.apache.hadoop.hbase.util.ReflectionUtils;
|
||||||
@InterfaceAudience.Private
|
@InterfaceAudience.Private
|
||||||
public abstract class StoreEngine<
|
public abstract class StoreEngine<
|
||||||
CP extends CompactionPolicy, C extends Compactor, SFM extends StoreFileManager> {
|
CP extends CompactionPolicy, C extends Compactor, SFM extends StoreFileManager> {
|
||||||
protected final Store store;
|
|
||||||
protected final Configuration conf;
|
|
||||||
protected final KVComparator comparator;
|
|
||||||
|
|
||||||
protected CP compactionPolicy;
|
protected CP compactionPolicy;
|
||||||
protected C compactor;
|
protected C compactor;
|
||||||
protected SFM storeFileManager;
|
protected SFM storeFileManager;
|
||||||
private boolean isInitialized = false;
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* The name of the configuration parameter that specifies the class of
|
* The name of the configuration parameter that specifies the class of
|
||||||
|
@ -61,7 +56,6 @@ public abstract class StoreEngine<
|
||||||
* @return Compaction policy to use.
|
* @return Compaction policy to use.
|
||||||
*/
|
*/
|
||||||
public CompactionPolicy getCompactionPolicy() {
|
public CompactionPolicy getCompactionPolicy() {
|
||||||
createComponentsOnce();
|
|
||||||
return this.compactionPolicy;
|
return this.compactionPolicy;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -69,7 +63,6 @@ public abstract class StoreEngine<
|
||||||
* @return Compactor to use.
|
* @return Compactor to use.
|
||||||
*/
|
*/
|
||||||
public Compactor getCompactor() {
|
public Compactor getCompactor() {
|
||||||
createComponentsOnce();
|
|
||||||
return this.compactor;
|
return this.compactor;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -77,33 +70,27 @@ public abstract class StoreEngine<
|
||||||
* @return Store file manager to use.
|
* @return Store file manager to use.
|
||||||
*/
|
*/
|
||||||
public StoreFileManager getStoreFileManager() {
|
public StoreFileManager getStoreFileManager() {
|
||||||
createComponentsOnce();
|
|
||||||
return this.storeFileManager;
|
return this.storeFileManager;
|
||||||
}
|
}
|
||||||
|
|
||||||
protected StoreEngine(Configuration conf, Store store, KVComparator comparator) {
|
/**
|
||||||
this.store = store;
|
* Creates an instance of a compaction context specific to this engine.
|
||||||
this.conf = conf;
|
* Doesn't actually select or start a compaction. See CompactionContext class comment.
|
||||||
this.comparator = comparator;
|
* @return New CompactionContext object.
|
||||||
}
|
*/
|
||||||
|
public abstract CompactionContext createCompaction() throws IOException;
|
||||||
public CompactionContext createCompaction() {
|
|
||||||
createComponentsOnce();
|
|
||||||
return this.createCompactionContext();
|
|
||||||
}
|
|
||||||
|
|
||||||
protected abstract CompactionContext createCompactionContext();
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Create the StoreEngine's components.
|
* Create the StoreEngine's components.
|
||||||
*/
|
*/
|
||||||
protected abstract void createComponents();
|
protected abstract void createComponents(
|
||||||
|
Configuration conf, Store store, KVComparator kvComparator) throws IOException;
|
||||||
|
|
||||||
private void createComponentsOnce() {
|
private void createComponentsOnce(
|
||||||
if (isInitialized) return;
|
Configuration conf, Store store, KVComparator kvComparator) throws IOException {
|
||||||
createComponents();
|
assert compactor == null && compactionPolicy == null && storeFileManager == null;
|
||||||
|
createComponents(conf, store, kvComparator);
|
||||||
assert compactor != null && compactionPolicy != null && storeFileManager != null;
|
assert compactor != null && compactionPolicy != null && storeFileManager != null;
|
||||||
isInitialized = true;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -114,13 +101,14 @@ public abstract class StoreEngine<
|
||||||
* @param kvComparator KVComparator for storeFileManager.
|
* @param kvComparator KVComparator for storeFileManager.
|
||||||
* @return StoreEngine to use.
|
* @return StoreEngine to use.
|
||||||
*/
|
*/
|
||||||
public static StoreEngine create(Store store, Configuration conf, KVComparator kvComparator)
|
public static StoreEngine<?, ?, ?> create(
|
||||||
throws IOException {
|
Store store, Configuration conf, KVComparator kvComparator) throws IOException {
|
||||||
String className = conf.get(STORE_ENGINE_CLASS_KEY, DEFAULT_STORE_ENGINE_CLASS.getName());
|
String className = conf.get(STORE_ENGINE_CLASS_KEY, DEFAULT_STORE_ENGINE_CLASS.getName());
|
||||||
try {
|
try {
|
||||||
return ReflectionUtils.instantiateWithCustomCtor(className,
|
StoreEngine<?,?,?> se = ReflectionUtils.instantiateWithCustomCtor(
|
||||||
new Class[] { Configuration.class, Store.class, KVComparator.class },
|
className, new Class[] { }, new Object[] { });
|
||||||
new Object[] { conf, store, kvComparator });
|
se.createComponentsOnce(conf, store, kvComparator);
|
||||||
|
return se;
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
throw new IOException("Unable to load configured store engine '" + className + "'", e);
|
throw new IOException("Unable to load configured store engine '" + className + "'", e);
|
||||||
}
|
}
|
||||||
|
|
|
@ -27,7 +27,8 @@ import org.apache.hadoop.hbase.regionserver.StoreFile;
|
||||||
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* This class holds all "physical" details necessary to run a compaction.
|
* This class holds all "physical" details necessary to run a compaction,
|
||||||
|
* and abstracts away the details specific to a particular compaction.
|
||||||
* It also has compaction request with all the logical details.
|
* It also has compaction request with all the logical details.
|
||||||
* Hence, this class is basically the compaction.
|
* Hence, this class is basically the compaction.
|
||||||
*/
|
*/
|
||||||
|
|
|
@ -309,7 +309,7 @@ public class TestCompaction extends HBaseTestCase {
|
||||||
conf.setFloat("hbase.hregion.majorcompaction.jitter", jitterPct);
|
conf.setFloat("hbase.hregion.majorcompaction.jitter", jitterPct);
|
||||||
|
|
||||||
HStore s = ((HStore) r.getStore(COLUMN_FAMILY));
|
HStore s = ((HStore) r.getStore(COLUMN_FAMILY));
|
||||||
s.compactionPolicy.setConf(conf);
|
s.storeEngine.getCompactionPolicy().setConf(conf);
|
||||||
try {
|
try {
|
||||||
createStoreFile(r);
|
createStoreFile(r);
|
||||||
createStoreFile(r);
|
createStoreFile(r);
|
||||||
|
@ -321,7 +321,7 @@ public class TestCompaction extends HBaseTestCase {
|
||||||
assertEquals(2, s.getStorefilesCount());
|
assertEquals(2, s.getStorefilesCount());
|
||||||
|
|
||||||
// ensure that major compaction time is deterministic
|
// ensure that major compaction time is deterministic
|
||||||
DefaultCompactionPolicy c = (DefaultCompactionPolicy)s.compactionPolicy;
|
DefaultCompactionPolicy c = (DefaultCompactionPolicy)s.storeEngine.getCompactionPolicy();
|
||||||
Collection<StoreFile> storeFiles = s.getStorefiles();
|
Collection<StoreFile> storeFiles = s.getStorefiles();
|
||||||
long mcTime = c.getNextMajorCompactTime(storeFiles);
|
long mcTime = c.getNextMajorCompactTime(storeFiles);
|
||||||
for (int i = 0; i < 10; ++i) {
|
for (int i = 0; i < 10; ++i) {
|
||||||
|
|
|
@ -234,8 +234,8 @@ public class TestDefaultCompactSelection extends TestCase {
|
||||||
throws IOException {
|
throws IOException {
|
||||||
store.forceMajor = forcemajor;
|
store.forceMajor = forcemajor;
|
||||||
//Test Default compactions
|
//Test Default compactions
|
||||||
CompactionRequest result = ((DefaultCompactionPolicy)store.compactionPolicy).selectCompaction(
|
CompactionRequest result = ((DefaultCompactionPolicy)store.storeEngine.getCompactionPolicy())
|
||||||
candidates, new ArrayList<StoreFile>(), false, isOffPeak, forcemajor);
|
.selectCompaction(candidates, new ArrayList<StoreFile>(), false, isOffPeak, forcemajor);
|
||||||
List<StoreFile> actual = new ArrayList<StoreFile>(result.getFiles());
|
List<StoreFile> actual = new ArrayList<StoreFile>(result.getFiles());
|
||||||
if (isOffPeak && !forcemajor) {
|
if (isOffPeak && !forcemajor) {
|
||||||
assertTrue(result.isOffPeak());
|
assertTrue(result.isOffPeak());
|
||||||
|
@ -288,7 +288,7 @@ public class TestDefaultCompactSelection extends TestCase {
|
||||||
compactEquals(sfCreate(100,50,23,12,12), true, 23, 12, 12);
|
compactEquals(sfCreate(100,50,23,12,12), true, 23, 12, 12);
|
||||||
conf.setLong(HConstants.MAJOR_COMPACTION_PERIOD, 1);
|
conf.setLong(HConstants.MAJOR_COMPACTION_PERIOD, 1);
|
||||||
conf.setFloat("hbase.hregion.majorcompaction.jitter", 0);
|
conf.setFloat("hbase.hregion.majorcompaction.jitter", 0);
|
||||||
store.compactionPolicy.setConf(conf);
|
store.storeEngine.getCompactionPolicy().setConf(conf);
|
||||||
try {
|
try {
|
||||||
// trigger an aged major compaction
|
// trigger an aged major compaction
|
||||||
compactEquals(sfCreate(50,25,12,12), 50, 25, 12, 12);
|
compactEquals(sfCreate(50,25,12,12), 50, 25, 12, 12);
|
||||||
|
@ -321,7 +321,7 @@ public class TestDefaultCompactSelection extends TestCase {
|
||||||
*/
|
*/
|
||||||
// set an off-peak compaction threshold
|
// set an off-peak compaction threshold
|
||||||
this.conf.setFloat("hbase.hstore.compaction.ratio.offpeak", 5.0F);
|
this.conf.setFloat("hbase.hstore.compaction.ratio.offpeak", 5.0F);
|
||||||
store.compactionPolicy.setConf(this.conf);
|
store.storeEngine.getCompactionPolicy().setConf(this.conf);
|
||||||
// Test with and without the flag.
|
// Test with and without the flag.
|
||||||
compactEquals(sfCreate(999, 50, 12, 12, 1), false, true, 50, 12, 12, 1);
|
compactEquals(sfCreate(999, 50, 12, 12, 1), false, true, 50, 12, 12, 1);
|
||||||
compactEquals(sfCreate(999, 50, 12, 12, 1), 12, 12, 1);
|
compactEquals(sfCreate(999, 50, 12, 12, 1), 12, 12, 1);
|
||||||
|
|
|
@ -0,0 +1,59 @@
|
||||||
|
/*
|
||||||
|
*
|
||||||
|
* 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.conf.Configuration;
|
||||||
|
import org.apache.hadoop.hbase.HBaseConfiguration;
|
||||||
|
import org.apache.hadoop.hbase.KeyValue.KVComparator;
|
||||||
|
import org.apache.hadoop.hbase.SmallTests;
|
||||||
|
import org.apache.hadoop.hbase.regionserver.compactions.DefaultCompactionPolicy;
|
||||||
|
import org.apache.hadoop.hbase.regionserver.compactions.DefaultCompactor;
|
||||||
|
import org.junit.Assert;
|
||||||
|
import org.junit.Test;
|
||||||
|
import org.junit.experimental.categories.Category;
|
||||||
|
import org.mockito.Mockito;
|
||||||
|
|
||||||
|
@Category(SmallTests.class)
|
||||||
|
public class TestDefaultStoreEngine {
|
||||||
|
public static class DummyCompactor extends DefaultCompactor {
|
||||||
|
public DummyCompactor(Configuration conf, Store store) {
|
||||||
|
super(conf, store);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public static class DummyCompactionPolicy extends DefaultCompactionPolicy {
|
||||||
|
public DummyCompactionPolicy(Configuration conf, StoreConfigInformation storeConfigInfo) {
|
||||||
|
super(conf, storeConfigInfo);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testCustomPolicyAndCompactor() throws Exception {
|
||||||
|
Configuration conf = HBaseConfiguration.create();
|
||||||
|
conf.set(DefaultStoreEngine.DEFAULT_COMPACTOR_CLASS_KEY, DummyCompactor.class.getName());
|
||||||
|
conf.set(DefaultStoreEngine.DEFAULT_COMPACTION_POLICY_CLASS_KEY,
|
||||||
|
DummyCompactionPolicy.class.getName());
|
||||||
|
Store mockStore = Mockito.mock(Store.class);
|
||||||
|
StoreEngine<?, ?, ?> se = StoreEngine.create(mockStore, conf, new KVComparator());
|
||||||
|
Assert.assertTrue(se instanceof DefaultStoreEngine);
|
||||||
|
Assert.assertTrue(se.getCompactionPolicy() instanceof DummyCompactionPolicy);
|
||||||
|
Assert.assertTrue(se.getCompactor() instanceof DummyCompactor);
|
||||||
|
}
|
||||||
|
}
|
|
@ -49,6 +49,7 @@ import org.apache.hadoop.hbase.HColumnDescriptor;
|
||||||
import org.apache.hadoop.hbase.HRegionInfo;
|
import org.apache.hadoop.hbase.HRegionInfo;
|
||||||
import org.apache.hadoop.hbase.HTableDescriptor;
|
import org.apache.hadoop.hbase.HTableDescriptor;
|
||||||
import org.apache.hadoop.hbase.KeyValue;
|
import org.apache.hadoop.hbase.KeyValue;
|
||||||
|
import org.apache.hadoop.hbase.KeyValue.KVComparator;
|
||||||
import org.apache.hadoop.hbase.MediumTests;
|
import org.apache.hadoop.hbase.MediumTests;
|
||||||
import org.apache.hadoop.hbase.client.Get;
|
import org.apache.hadoop.hbase.client.Get;
|
||||||
import org.apache.hadoop.hbase.io.compress.Compression;
|
import org.apache.hadoop.hbase.io.compress.Compression;
|
||||||
|
@ -59,6 +60,7 @@ import org.apache.hadoop.hbase.monitoring.MonitoredTask;
|
||||||
import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
|
import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
|
||||||
import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
|
import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
|
||||||
import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
|
import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
|
||||||
|
import org.apache.hadoop.hbase.regionserver.compactions.DefaultCompactor;
|
||||||
import org.apache.hadoop.hbase.regionserver.wal.HLog;
|
import org.apache.hadoop.hbase.regionserver.wal.HLog;
|
||||||
import org.apache.hadoop.hbase.regionserver.wal.HLogFactory;
|
import org.apache.hadoop.hbase.regionserver.wal.HLogFactory;
|
||||||
import org.apache.hadoop.hbase.security.User;
|
import org.apache.hadoop.hbase.security.User;
|
||||||
|
@ -853,5 +855,22 @@ public class TestStore extends TestCase {
|
||||||
assertTrue(store.throttleCompaction(anyValue + 1));
|
assertTrue(store.throttleCompaction(anyValue + 1));
|
||||||
assertFalse(store.throttleCompaction(anyValue));
|
assertFalse(store.throttleCompaction(anyValue));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public static class DummyStoreEngine extends DefaultStoreEngine {
|
||||||
|
public static DefaultCompactor lastCreatedCompactor = null;
|
||||||
|
@Override
|
||||||
|
protected void createComponents(
|
||||||
|
Configuration conf, Store store, KVComparator comparator) throws IOException {
|
||||||
|
super.createComponents(conf, store, comparator);
|
||||||
|
lastCreatedCompactor = this.compactor;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public void testStoreUsesSearchEngineOverride() throws Exception {
|
||||||
|
Configuration conf = HBaseConfiguration.create();
|
||||||
|
conf.set(StoreEngine.STORE_ENGINE_CLASS_KEY, DummyStoreEngine.class.getName());
|
||||||
|
init(this.getName(), conf);
|
||||||
|
assertEquals(DummyStoreEngine.lastCreatedCompactor, this.store.storeEngine.getCompactor());
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue