diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 0342df6f70..5459fcf1cb 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -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 sstables, Collection ranges, int gcBefore, boolean isMajor) + throws IOException + { + super(getCollatedRangeIterator(sstables, ranges), gcBefore, isMajor); + } + + private static Iterator getCollatedRangeIterator(Collection sstables, final Collection 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.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(); + } + } + } } diff --git a/src/java/org/apache/cassandra/io/CompactionIterator.java b/src/java/org/apache/cassandra/io/CompactionIterator.java index b3124eee61..c085b63f8f 100644 --- a/src/java/org/apache/cassandra/io/CompactionIterator.java +++ b/src/java/org/apache/cassandra/io/CompactionIterator.java @@ -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 rows = new ArrayList(); private final int gcBefore; private boolean major; - @SuppressWarnings("unchecked") public CompactionIterator(Iterable 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 sstables) throws IOException + protected static CollatingIterator getCollatingIterator(Iterable 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.getCollatingIterator(); for (SSTableReader sstable : sstables) { iter.addIterator(sstable.getScanner(FILE_BUFFER_SIZE)); diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java index 2b93d477f0..c57a7acdeb 100644 --- a/src/java/org/apache/cassandra/utils/FBUtilities.java +++ b/src/java/org/apache/cassandra/utils/FBUtilities.java @@ -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 > CollatingIterator getCollatingIterator() + { + // CollatingIterator will happily NPE if you do not specify a comparator explicitly + return new CollatingIterator(new Comparator() + { + public int compare(T o1, T o2) + { + return o1.compareTo(o2); + } + }); + } }