Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-15665 Support using different StoreFileComparators for different CompactionPolicies

This commit is contained in:
zhangduo 2016-04-19 11:51:18 +08:00
parent 2b5da6f7a0
commit 22b1061044
7 changed files with 41 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -57,8 +57,9 @@ public class DateTieredStoreEngine extends StoreEngine<DefaultStoreFlusher,
protected void createComponents(Configuration conf, Store store, KVComparator kvComparator)
throws IOException {
this.compactionPolicy = new DateTieredCompactionPolicy(conf, store);
this.storeFileManager = new DefaultStoreFileManager(kvComparator, conf,
compactionPolicy.getConf());
this.storeFileManager =
new DefaultStoreFileManager(kvComparator, StoreFile.Comparators.SEQ_ID_MAX_TIMESTAMP, conf,
compactionPolicy.getConf());
this.storeFlusher = new DefaultStoreFlusher(conf, store);
this.compactor = new DateTieredCompactor(conf, store);
}

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

@ -81,7 +81,9 @@ public class DefaultStoreEngine extends StoreEngine<
} catch (Exception e) {
throw new IOException("Unable to load configured compaction policy '" + className + "'", e);
}
storeFileManager = new DefaultStoreFileManager(kvComparator, conf, compactionPolicy.getConf());
storeFileManager =
new DefaultStoreFileManager(kvComparator, StoreFile.Comparators.SEQ_ID, conf,
compactionPolicy.getConf());
className = conf.get(
DEFAULT_STORE_FLUSHER_CLASS_KEY, DEFAULT_STORE_FLUSHER_CLASS.getName());
try {

24
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultStoreFileManager.java
View File

@ -22,9 +22,14 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import com.google.common.collect.ImmutableCollection;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@ -34,10 +39,6 @@ import org.apache.hadoop.hbase.KeyValue.KVComparator;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.regionserver.compactions.CompactionConfiguration;
import com.google.common.collect.ImmutableCollection;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
/**
* Default implementation of StoreFileManager. Not thread-safe.
*/
@ -48,7 +49,7 @@ class DefaultStoreFileManager implements StoreFileManager {
private final KVComparator kvComparator;
private final CompactionConfiguration comConf;
private final int blockingFileCount;
private final Comparator<StoreFile> storeFileComparator;
/**
* List of store files inside this store. This is an immutable list that
* is atomically replaced when its contents change.
@ -62,9 +63,11 @@ class DefaultStoreFileManager implements StoreFileManager {
*/
private volatile List<StoreFile> compactedfiles = null;
public DefaultStoreFileManager(KVComparator kvComparator, Configuration conf,
public DefaultStoreFileManager(KVComparator kvComparator,
Comparator<StoreFile> storeFileComparator, Configuration conf,
CompactionConfiguration comConf) {
this.kvComparator = kvComparator;
this.storeFileComparator = storeFileComparator;
this.comConf = comConf;
this.blockingFileCount =
conf.getInt(HStore.BLOCKING_STOREFILES_KEY, HStore.DEFAULT_BLOCKING_STOREFILE_COUNT);
@ -210,13 +213,13 @@ class DefaultStoreFileManager implements StoreFileManager {
}
private void sortAndSetStoreFiles(List<StoreFile> storeFiles) {
Collections.sort(storeFiles, StoreFile.Comparators.SEQ_ID);
Collections.sort(storeFiles, storeFileComparator);
storefiles = ImmutableList.copyOf(storeFiles);
}
private List<StoreFile> sortCompactedfiles(List<StoreFile> storefiles) {
// Sorting may not be really needed here for the compacted files?
Collections.sort(storefiles, StoreFile.Comparators.SEQ_ID);
Collections.sort(storefiles, storeFileComparator);
return new ArrayList<StoreFile>(storefiles);
}
@ -229,5 +232,10 @@ class DefaultStoreFileManager implements StoreFileManager {
}
return (double) (storefileCount - minFilesToCompact) / (blockingFileCount - minFilesToCompact);
}
@Override
public Comparator<StoreFile> getStoreFileComparator() {
return storeFileComparator;
}
}

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

@ -1511,7 +1511,8 @@ public class HStore implements Store {
filesToCompact = filesToCompact.subList(count - N, count);
isMajor = (filesToCompact.size() == storeEngine.getStoreFileManager().getStorefileCount());
filesCompacting.addAll(filesToCompact);
Collections.sort(filesCompacting, StoreFile.Comparators.SEQ_ID);
Collections.sort(filesCompacting, storeEngine.getStoreFileManager()
.getStoreFileComparator());
}
} finally {
this.lock.readLock().unlock();
@ -1700,7 +1701,7 @@ public class HStore implements Store {
Preconditions.checkArgument(false, "%s overlaps with %s", filesToAdd, filesCompacting);
}
filesCompacting.addAll(filesToAdd);
Collections.sort(filesCompacting, StoreFile.Comparators.SEQ_ID);
Collections.sort(filesCompacting, storeEngine.getStoreFileManager().getStoreFileComparator());
}
private void removeUnneededFiles() throws IOException {

13
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileManager.java
View File

@ -20,14 +20,15 @@ package org.apache.hadoop.hbase.regionserver;
import java.io.IOException;
import java.util.Collection;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import com.google.common.collect.ImmutableCollection;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.KeyValue;
import com.google.common.collect.ImmutableCollection;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
/**
* Manages the store files and basic metadata about that that determines the logical structure
@ -163,4 +164,10 @@ public interface StoreFileManager {
* @see Store#getCompactionPressure()
*/
double getCompactionPressure();
/**
* @return the comparator used to sort storefiles. Usually, the
* {@link StoreFile#getMaxSequenceId()} is the first priority.
*/
Comparator<StoreFile> getStoreFileComparator();
}

6
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StripeStoreFileManager.java
View File

@ -23,6 +23,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
@ -1048,4 +1049,9 @@ public class StripeStoreFileManager
}
return max;
}
@Override
public Comparator<StoreFile> getStoreFileComparator() {
return StoreFile.Comparators.SEQ_ID;
}
}

2
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/DateTieredCompactionPolicy.java
View File

@ -191,10 +191,8 @@ public class DateTieredCompactionPolicy extends SortedCompactionPolicy {
long now = EnvironmentEdgeManager.currentTime();
long oldestToCompact = getOldestToCompact(comConf.getMaxStoreFileAgeMillis(), now);
// Make sure the store files is sorted by SeqId then maxTimestamp
List<StoreFile> storeFileList = Lists.newArrayList(filterOldStoreFiles(candidateSelection,
oldestToCompact));
Collections.sort(storeFileList, StoreFile.Comparators.SEQ_ID_MAX_TIMESTAMP);
List<Pair<StoreFile, Long>> storefileMaxTimestampPairs =
Lists.newArrayListWithCapacity(Iterables.size(storeFileList));