natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

throw out uninteresting rows in anticompaction before collating & reducing. patch by jbellis and Stu Hood for CASSANDRA-607 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/branches/cassandra-0.5@893594 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-12-23 18:56:10 +00:00
parent b3c8e9d364
commit d07e8317f1
3 changed files with 72 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -53,12 +53,12 @@ import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.collections.IteratorUtils;
import org.apache.commons.collections.PredicateUtils;
import org.apache.commons.collections.iterators.CollatingIterator;
import org.apache.commons.collections.iterators.FilterIterator;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterators;
import com.google.common.base.Predicate;
public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
@ -655,7 +655,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
// if we have too many to compact all at once, compact older ones first -- this avoids
// re-compacting files we just created.
Collections.sort(sstables);
boolean major = sstables.size() == ssTables_.size();
filesCompacted += doFileCompaction(sstables.subList(0, Math.min(sstables.size(), maxThreshold)));
}
logger_.debug(filesCompacted + " files compacted");
@ -812,7 +811,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
logger_.debug("Expected bloom filter size : " + expectedBloomFilterSize);
SSTableWriter writer = null;
CompactionIterator ci = new CompactionIterator(sstables, getDefaultGCBefore(), sstables.size() == ssTables_.size());
CompactionIterator ci = new AntiCompactionIterator(sstables, ranges, getDefaultGCBefore(), sstables.size() == ssTables_.size());
Iterator nni = new FilterIterator(ci, PredicateUtils.notNullPredicate());
try
@ -825,17 +824,14 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
while (nni.hasNext())
{
CompactionIterator.CompactedRow row = (CompactionIterator.CompactedRow) nni.next();
if (Range.isTokenInRanges(row.key.token, ranges))
{
if (writer == null)
{
FileUtils.createDirectory(compactionFileLocation);
String newFilename = new File(compactionFileLocation, getTempSSTableFileName()).getAbsolutePath();
writer = new SSTableWriter(newFilename, expectedBloomFilterSize, StorageService.getPartitioner());
}
writer.append(row.key, row.buffer);
totalkeysWritten++;
}
if (writer == null)
{
FileUtils.createDirectory(compactionFileLocation);
String newFilename = new File(compactionFileLocation, getTempSSTableFileName()).getAbsolutePath();
writer = new SSTableWriter(newFilename, expectedBloomFilterSize, StorageService.getPartitioner());
}
writer.append(row.key, row.buffer);
totalkeysWritten++;
}
}
finally
@ -1599,4 +1595,39 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
ssTables_.clearUnsafe();
}
private static class AntiCompactionIterator extends CompactionIterator
{
public AntiCompactionIterator(Collection<SSTableReader> sstables, Collection<Range> ranges, int gcBefore, boolean isMajor)
throws IOException
{
super(getCollatedRangeIterator(sstables, ranges), gcBefore, isMajor);
}
private static Iterator getCollatedRangeIterator(Collection<SSTableReader> sstables, final Collection<Range> ranges)
throws IOException
{
org.apache.commons.collections.Predicate rangesPredicate = new org.apache.commons.collections.Predicate()
{
public boolean evaluate(Object row)
{
return Range.isTokenInRanges(((IteratingRow)row).getKey().token, ranges);
}
};
CollatingIterator iter = FBUtilities.<IteratingRow>getCollatingIterator();
for (SSTableReader sstable : sstables)
{
SSTableScanner scanner = sstable.getScanner(FILE_BUFFER_SIZE);
iter.addIterator(new FilterIterator(scanner, rangesPredicate));
}
return iter;
}
public void close() throws IOException
{
for (Object o : ((CollatingIterator)source).getIterators())
{
((SSTableScanner)((FilterIterator)o).getIterator()).close();
}
}
}
}

25
src/java/org/apache/cassandra/io/CompactionIterator.java
View File

@ -26,12 +26,13 @@ import java.io.IOException;
import java.io.IOError;
import java.util.List;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;
import org.apache.log4j.Logger;
import org.apache.commons.collections.iterators.CollatingIterator;
import org.apache.cassandra.utils.ReducingIterator;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.ColumnFamilyStore;
@ -40,31 +41,29 @@ public class CompactionIterator extends ReducingIterator<IteratingRow, Compactio
{
private static Logger logger = Logger.getLogger(CompactionIterator.class);
private static final int FILE_BUFFER_SIZE = 1024 * 1024;
protected static final int FILE_BUFFER_SIZE = 1024 * 1024;
private final List<IteratingRow> rows = new ArrayList<IteratingRow>();
private final int gcBefore;
private boolean major;
@SuppressWarnings("unchecked")
public CompactionIterator(Iterable<SSTableReader> sstables, int gcBefore, boolean major) throws IOException
{
super(getCollatingIterator(sstables));
this(getCollatingIterator(sstables), gcBefore, major);
}
@SuppressWarnings("unchecked")
protected CompactionIterator(Iterator iter, int gcBefore, boolean major)
{
super(iter);
this.gcBefore = gcBefore;
this.major = major;
}
@SuppressWarnings("unchecked")
private static CollatingIterator getCollatingIterator(Iterable<SSTableReader> sstables) throws IOException
protected static CollatingIterator getCollatingIterator(Iterable<SSTableReader> sstables) throws IOException
{
// CollatingIterator has a bug that causes NPE when you try to use default comparator. :(
CollatingIterator iter = new CollatingIterator(new Comparator()
{
public int compare(Object o1, Object o2)
{
return ((Comparable)o1).compareTo(o2);
}
});
CollatingIterator iter = FBUtilities.<IteratingRow>getCollatingIterator();
for (SSTableReader sstable : sstables)
{
iter.addIterator(sstable.getScanner(FILE_BUFFER_SIZE));

14
src/java/org/apache/cassandra/utils/FBUtilities.java
View File

@ -30,6 +30,8 @@ import java.util.zip.Inflater;
import org.apache.log4j.Logger;
import org.apache.commons.collections.iterators.CollatingIterator;
import org.apache.cassandra.config.DatabaseDescriptor;
public class FBUtilities
@ -304,4 +306,16 @@ public class FBUtilities
throw new IOException("rename failed of " + filename);
}
}
public static <T extends Comparable<T>> CollatingIterator getCollatingIterator()
{
// CollatingIterator will happily NPE if you do not specify a comparator explicitly
return new CollatingIterator(new Comparator<T>()
{
public int compare(T o1, T o2)
{
return o1.compareTo(o2);
}
});
}
}