diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index a0e31d08f9..fbc23d248a 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -35,7 +35,6 @@ import org.apache.log4j.Logger; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.dht.IPartitioner; -import org.apache.cassandra.dht.Range; import org.apache.cassandra.io.*; import java.net.InetAddress; import java.util.regex.Matcher; @@ -49,7 +48,6 @@ import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.db.filter.*; import org.apache.cassandra.db.marshal.AbstractType; -import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.ArrayUtils; import org.apache.commons.collections.IteratorUtils; import org.apache.commons.collections.PredicateUtils; @@ -292,31 +290,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean sb.append(newLineSeparator); return sb.toString(); } - - /* - * This method forces a compaction of the SSTables on disk. We wait - * for the process to complete by waiting on a future pointer. - */ - List forceAntiCompaction(Collection ranges, InetAddress target) - { - assert ranges != null; - Future> futurePtr = CompactionManager.instance.submitAnti(ColumnFamilyStore.this, - ranges, target); - - List result; - try - { - /* Waiting for the compaction to complete. */ - result = futurePtr.get(); - if (logger_.isDebugEnabled()) - logger_.debug("Done forcing compaction ..."); - } - catch (Exception ex) - { - throw new RuntimeException(ex); - } - return result; - } /** * @return the name of the column family @@ -501,12 +474,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean */ static ColumnFamily removeDeleted(ColumnFamily cf) { - return removeDeleted(cf, getDefaultGCBefore()); - } - - public static int getDefaultGCBefore() - { - return (int)(System.currentTimeMillis() / 1000) - DatabaseDescriptor.getGcGraceInSeconds(); + return removeDeleted(cf, CompactionManager.getDefaultGCBefore()); } public static ColumnFamily removeDeleted(ColumnFamily cf, int gcBefore) @@ -596,109 +564,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean CompactionManager.instance.submitMinor(this); } - /* - * Group files of similar size into buckets. - */ - static Set> getCompactionBuckets(Iterable files, long min) - { - Map, Long> buckets = new HashMap, Long>(); - for (SSTableReader sstable : files) - { - long size = sstable.length(); - - boolean bFound = false; - // look for a bucket containing similar-sized files: - // group in the same bucket if it's w/in 50% of the average for this bucket, - // or this file and the bucket are all considered "small" (less than `min`) - for (List bucket : buckets.keySet()) - { - long averageSize = buckets.get(bucket); - if ((size > averageSize / 2 && size < 3 * averageSize / 2) - || (size < min && averageSize < min)) - { - // remove and re-add because adding changes the hash - buckets.remove(bucket); - averageSize = (averageSize + size) / 2; - bucket.add(sstable); - buckets.put(bucket, averageSize); - bFound = true; - break; - } - } - // no similar bucket found; put it in a new one - if (!bFound) - { - ArrayList bucket = new ArrayList(); - bucket.add(sstable); - buckets.put(bucket, size); - } - } - - return buckets.keySet(); - } - - /* - * Break the files into buckets and then compact. - */ - int doCompaction(int minThreshold, int maxThreshold) throws IOException - { - int filesCompacted = 0; - if (minThreshold > 0 && maxThreshold > 0) - { - logger_.debug("Checking to see if compaction of " + columnFamily_ + " would be useful"); - for (List sstables : getCompactionBuckets(ssTables_, 50L * 1024L * 1024L)) - { - if (sstables.size() < minThreshold) - { - continue; - } - // 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); - filesCompacted += doFileCompaction(sstables.subList(0, Math.min(sstables.size(), maxThreshold))); - } - logger_.debug(filesCompacted + " files compacted"); - } - else - { - logger_.debug("Compaction is currently disabled."); - } - return filesCompacted; - } - - void doMajorCompaction(long skip) throws IOException - { - doMajorCompactionInternal(skip); - } - - /* - * Compact all the files irrespective of the size. - * skip : is the amount in GB of the files to be skipped - * all files greater than skip GB are skipped for this compaction. - * Except if skip is 0 , in that case this is ignored and all files are taken. - */ - void doMajorCompactionInternal(long skip) throws IOException - { - Collection sstables; - if (skip > 0) - { - sstables = new ArrayList(); - for (SSTableReader sstable : ssTables_) - { - if (sstable.length() < skip * 1024L * 1024L * 1024L) - { - sstables.add(sstable); - } - } - } - else - { - sstables = ssTables_.getSSTables(); - } - - doFileCompaction(sstables); - } - /* * Add up all the files sizes this is the worst case file * size for compaction of all the list of files given. @@ -732,277 +597,41 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean return maxFile; } - List doAntiCompaction(Collection ranges, InetAddress target) throws IOException - { - return doFileAntiCompaction(ssTables_.getSSTables(), ranges, target); - } - void forceCleanup() { CompactionManager.instance.submitCleanup(ColumnFamilyStore.this); } - /** - * This function goes over each file and removes the keys that the node is not responsible for - * and only keeps keys that this node is responsible for. - * - * @throws IOException - */ - void doCleanupCompaction() throws IOException + public Table getTable() { - for (SSTableReader sstable : ssTables_) - { - doCleanup(sstable); - } - gcAfterRpcTimeout(); - } - - /** - * cleans up one particular file by removing keys that this node is not responsible for. - * @throws IOException - */ - /* TODO: Take care of the comments later. */ - void doCleanup(SSTableReader sstable) throws IOException - { - assert sstable != null; - List sstables = doFileAntiCompaction(Arrays.asList(sstable), StorageService.instance().getLocalRanges(), null); - if (!sstables.isEmpty()) - { - assert sstables.size() == 1; - addSSTable(sstables.get(0)); - } - if (logger_.isDebugEnabled()) - logger_.debug("Original file : " + sstable + " of size " + sstable.length()); - ssTables_.markCompacted(Arrays.asList(sstable)); - } - - /** - * This function is used to do the anti compaction process , it spits out the file which has keys that belong to a given range - * If the target is not specified it spits out the file as a compacted file with the unecessary ranges wiped out. - * - * @param sstables - * @param ranges - * @param target - * @return - * @throws IOException - */ - List doFileAntiCompaction(Collection sstables, final Collection ranges, InetAddress target) throws IOException - { - logger_.info("AntiCompacting [" + StringUtils.join(sstables, ",") + "]"); - // Calculate the expected compacted filesize - long expectedRangeFileSize = getExpectedCompactedFileSize(sstables) / 2; - String compactionFileLocation = DatabaseDescriptor.getDataFileLocationForTable(table_, expectedRangeFileSize); - if (compactionFileLocation == null) - { - throw new UnsupportedOperationException("disk full"); - } - if (target != null) - { - // compacting for streaming: send to subdirectory - compactionFileLocation = compactionFileLocation + File.separator + DatabaseDescriptor.STREAMING_SUBDIR; - } - List results = new ArrayList(); - - long startTime = System.currentTimeMillis(); - long totalkeysWritten = 0; - - int expectedBloomFilterSize = Math.max(SSTableReader.indexInterval(), (int)(SSTableReader.getApproximateKeyCount(sstables) / 2)); - if (logger_.isDebugEnabled()) - logger_.debug("Expected bloom filter size : " + expectedBloomFilterSize); - - SSTableWriter writer = null; - CompactionIterator ci = new AntiCompactionIterator(sstables, ranges, getDefaultGCBefore(), sstables.size() == ssTables_.size()); - Iterator nni = new FilterIterator(ci, PredicateUtils.notNullPredicate()); - try { - if (!nni.hasNext()) - { - return results; - } - - while (nni.hasNext()) - { - CompactionIterator.CompactedRow row = nni.next(); - 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++; - } + return Table.open(table_); } - finally + catch (IOException e) { - ci.close(); + throw new RuntimeException(e); } - - if (writer != null) - { - results.add(writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table_))); - String format = "AntiCompacted to %s. %d/%d bytes for %d keys. Time: %dms."; - long dTime = System.currentTimeMillis() - startTime; - logger_.info(String.format(format, writer.getFilename(), getTotalBytes(sstables), results.get(0).length(), totalkeysWritten, dTime)); - } - - return results; } - private int doFileCompaction(Collection sstables) throws IOException + void markCompacted(Collection sstables) throws IOException { - return doFileCompaction(sstables, getDefaultGCBefore()); - } - - /* - * This function does the actual compaction for files. - * It maintains a priority queue of with the first key from each file - * and then removes the top of the queue and adds it to the SStable and - * repeats this process while reading the next from each file until its - * done with all the files . The SStable to which the keys are written - * represents the new compacted file. Before writing if there are keys - * that occur in multiple files and are the same then a resolution is done - * to get the latest data. - * - * The collection of sstables passed may be empty (but not null); even if - * it is not empty, it may compact down to nothing if all rows are deleted. - */ - int doFileCompaction(Collection sstables, int gcBefore) throws IOException - { - if (DatabaseDescriptor.isSnapshotBeforeCompaction()) - Table.open(table_).snapshot("compact-" + columnFamily_); - logger_.info("Compacting [" + StringUtils.join(sstables, ",") + "]"); - String compactionFileLocation = DatabaseDescriptor.getDataFileLocationForTable(table_, getExpectedCompactedFileSize(sstables)); - // If the compaction file path is null that means we have no space left for this compaction. - // try again w/o the largest one. - if (compactionFileLocation == null) - { - SSTableReader maxFile = getMaxSizeFile(sstables); - List smallerSSTables = new ArrayList(sstables); - smallerSSTables.remove(maxFile); - return doFileCompaction(smallerSSTables, gcBefore); - } - - // new sstables from flush can be added during a compaction, but only the compaction can remove them, - // so in our single-threaded compaction world this is a valid way of determining if we're compacting - // all the sstables (that existed when we started) - boolean major = sstables.size() == ssTables_.size(); - - long startTime = System.currentTimeMillis(); - long totalkeysWritten = 0; - - // TODO the int cast here is potentially buggy - int expectedBloomFilterSize = Math.max(SSTableReader.indexInterval(), (int)SSTableReader.getApproximateKeyCount(sstables)); - if (logger_.isDebugEnabled()) - logger_.debug("Expected bloom filter size : " + expectedBloomFilterSize); - - SSTableWriter writer; - CompactionIterator ci = new CompactionIterator(sstables, gcBefore, major); // retain a handle so we can call close() - Iterator nni = new FilterIterator(ci, PredicateUtils.notNullPredicate()); - - try - { - if (!nni.hasNext()) - { - // don't mark compacted in the finally block, since if there _is_ nondeleted data, - // we need to sync it (via closeAndOpen) first, so there is no period during which - // a crash could cause data loss. - ssTables_.markCompacted(sstables); - return 0; - } - - String newFilename = new File(compactionFileLocation, getTempSSTableFileName()).getAbsolutePath(); - writer = new SSTableWriter(newFilename, expectedBloomFilterSize, StorageService.getPartitioner()); - - // validate the CF as we iterate over it - AntiEntropyService.IValidator validator = AntiEntropyService.instance().getValidator(table_, columnFamily_, null, major); - validator.prepare(); - while (nni.hasNext()) - { - CompactionIterator.CompactedRow row = nni.next(); - writer.append(row.key, row.buffer); - validator.add(row); - totalkeysWritten++; - } - validator.complete(); - } - finally - { - ci.close(); - } - - SSTableReader ssTable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table_)); - ssTables_.add(ssTable); ssTables_.markCompacted(sstables); - gcAfterRpcTimeout(); - CompactionManager.instance.submitMinor(ColumnFamilyStore.this); - - String format = "Compacted to %s. %d/%d bytes for %d keys. Time: %dms."; - long dTime = System.currentTimeMillis() - startTime; - logger_.info(String.format(format, writer.getFilename(), getTotalBytes(sstables), ssTable.length(), totalkeysWritten, dTime)); - return sstables.size(); } - /** - * perform a GC to clean out obsolete sstables, sleeping rpc timeout first so that most in-progress ops can complete - * (thus, no longer reference the sstables in question) - */ - private void gcAfterRpcTimeout() + boolean isCompleteSSTables(Collection sstables) { - new Thread(new Runnable() - { - public void run() - { - try - { - Thread.sleep(DatabaseDescriptor.getRpcTimeout()); - } - catch (InterruptedException e) - { - throw new AssertionError(e); - } - System.gc(); - } - }).start(); + return ssTables_.getSSTables().equals(new HashSet(sstables)); } - /** - * Performs a readonly "compaction" of all sstables in order to validate complete rows, - * but without writing the merge result - */ - void doReadonlyCompaction(InetAddress initiator) throws IOException + void replaceCompactedSSTables(Collection sstables, Iterable replacements) + throws IOException { - Collection sstables = ssTables_.getSSTables(); - CompactionIterator ci = new CompactionIterator(sstables, getDefaultGCBefore(), true); - try + for (SSTableReader sstable : replacements) { - Iterator nni = new FilterIterator(ci, PredicateUtils.notNullPredicate()); - - // validate the CF as we iterate over it - AntiEntropyService.IValidator validator = AntiEntropyService.instance().getValidator(table_, columnFamily_, initiator, true); - validator.prepare(); - while (nni.hasNext()) - { - CompactionIterator.CompactedRow row = nni.next(); - validator.add(row); - } - validator.complete(); + ssTables_.add(sstable); } - finally - { - ci.close(); - } - } - - private long getTotalBytes(Iterable sstables) - { - long sum = 0; - for (SSTableReader sstable : sstables) - { - sum += sstable.length(); - } - return sum; + ssTables_.markCompacted(sstables); } public static List getUnflushedMemtables(String cfName) @@ -1169,7 +798,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean public ColumnFamily getColumnFamily(QueryFilter filter) throws IOException { - return getColumnFamily(filter, getDefaultGCBefore()); + return getColumnFamily(filter, CompactionManager.getDefaultGCBefore()); } /** @@ -1594,40 +1223,4 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean memtable_.clearUnsafe(); 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/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java index 0fb591df33..5ecacb1472 100644 --- a/src/java/org/apache/cassandra/db/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/CompactionManager.java @@ -19,9 +19,9 @@ package org.apache.cassandra.db; import java.io.IOException; +import java.io.File; import java.lang.management.ManagementFactory; -import java.util.List; -import java.util.Collection; +import java.util.*; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; @@ -31,9 +31,22 @@ import org.apache.log4j.Logger; import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor; import org.apache.cassandra.dht.Range; -import org.apache.cassandra.io.SSTableReader; +import org.apache.cassandra.io.*; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.service.AntiEntropyService; +import org.apache.cassandra.utils.FileUtils; +import org.apache.cassandra.utils.FBUtilities; + import java.net.InetAddress; +import org.apache.commons.lang.StringUtils; +import org.apache.commons.collections.iterators.FilterIterator; +import org.apache.commons.collections.iterators.CollatingIterator; +import org.apache.commons.collections.PredicateUtils; + +import com.sun.corba.se.impl.logging.POASystemException; + public class CompactionManager implements CompactionManagerMBean { public static final String MBEAN_OBJECT_NAME = "org.apache.cassandra.db:type=CompactionManager"; @@ -64,18 +77,34 @@ public class CompactionManager implements CompactionManagerMBean * It's okay to over-call (within reason) since the compactions are single-threaded, * and if a call is unnecessary, it will just be no-oped in the bucketing phase. */ - public Future submitMinor(final ColumnFamilyStore columnFamilyStore) - { - return submitMinor(columnFamilyStore, minimumCompactionThreshold, maximumCompactionThreshold); - } - - Future submitMinor(final ColumnFamilyStore cfStore, final int minThreshold, final int maxThreshold) + public Future submitMinor(final ColumnFamilyStore cfs) { Callable callable = new Callable() { public Integer call() throws IOException { - return cfStore.doCompaction(minThreshold, maxThreshold); + int filesCompacted = 0; + if (minimumCompactionThreshold > 0 && maximumCompactionThreshold > 0) + { + logger.debug("Checking to see if compaction of " + cfs.columnFamily_ + " would be useful"); + for (List sstables : getCompactionBuckets(cfs.getSSTables(), 50L * 1024L * 1024L)) + { + if (sstables.size() < minimumCompactionThreshold) + { + continue; + } + // 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); + filesCompacted += doCompaction(cfs, sstables.subList(0, Math.min(sstables.size(), maximumCompactionThreshold)), getDefaultGCBefore()); + } + logger.debug(filesCompacted + " files compacted"); + } + else + { + logger.debug("Compaction is currently disabled."); + } + return filesCompacted; } }; return compactor_.submit(callable); @@ -87,32 +116,54 @@ public class CompactionManager implements CompactionManagerMBean { public Object call() throws IOException { - cfStore.doCleanupCompaction(); + doCleanupCompaction(cfStore); return this; } }; return compactor_.submit(runnable); } - public Future> submitAnti(final ColumnFamilyStore cfStore, final Collection ranges, final InetAddress target) + public Future> submitAnticompaction(final ColumnFamilyStore cfStore, final Collection ranges, final InetAddress target) { Callable> callable = new Callable>() { public List call() throws IOException { - return cfStore.doAntiCompaction(ranges, target); + return doAntiCompaction(cfStore, cfStore.getSSTables(), ranges, target); } }; return compactor_.submit(callable); } - public Future submitMajor(final ColumnFamilyStore cfStore, final long skip) + public Future submitMajor(final ColumnFamilyStore cfStore) + { + return submitMajor(cfStore, 0, getDefaultGCBefore()); + } + + public Future submitMajor(final ColumnFamilyStore cfStore, final long skip, final int gcBefore) { Callable callable = new Callable() { public Object call() throws IOException { - cfStore.doMajorCompaction(skip); + Collection sstables; + if (skip > 0) + { + sstables = new ArrayList(); + for (SSTableReader sstable : cfStore.getSSTables()) + { + if (sstable.length() < skip * 1024L * 1024L * 1024L) + { + sstables.add(sstable); + } + } + } + else + { + sstables = cfStore.getSSTables(); + } + + doCompaction(cfStore, sstables, gcBefore); return this; } }; @@ -125,7 +176,7 @@ public class CompactionManager implements CompactionManagerMBean { public Object call() throws IOException { - cfStore.doReadonlyCompaction(initiator); + doReadonlyCompaction(cfStore, initiator); return this; } }; @@ -164,9 +215,318 @@ public class CompactionManager implements CompactionManagerMBean maximumCompactionThreshold = threshold; } - public void disableCompactions() + public void disableAutoCompaction() { minimumCompactionThreshold = 0; maximumCompactionThreshold = 0; } + + /** + * For internal use and testing only. The rest of the system should go through the submit* methods, + * which are properly serialized. + */ + int doCompaction(ColumnFamilyStore cfs, Collection sstables, int gcBefore) throws IOException + { + // The collection of sstables passed may be empty (but not null); even if + // it is not empty, it may compact down to nothing if all rows are deleted. + Table table = cfs.getTable(); + if (DatabaseDescriptor.isSnapshotBeforeCompaction()) + table.snapshot("compact-" + cfs.columnFamily_); + logger.info("Compacting [" + StringUtils.join(sstables, ",") + "]"); + String compactionFileLocation = table.getDataFileLocation(cfs.getExpectedCompactedFileSize(sstables)); + // If the compaction file path is null that means we have no space left for this compaction. + // try again w/o the largest one. + if (compactionFileLocation == null) + { + SSTableReader maxFile = cfs.getMaxSizeFile(sstables); + List smallerSSTables = new ArrayList(sstables); + smallerSSTables.remove(maxFile); + return doCompaction(cfs, smallerSSTables, gcBefore); + } + + // new sstables from flush can be added during a compaction, but only the compaction can remove them, + // so in our single-threaded compaction world this is a valid way of determining if we're compacting + // all the sstables (that existed when we started) + boolean major = cfs.isCompleteSSTables(sstables); + + long startTime = System.currentTimeMillis(); + long totalkeysWritten = 0; + + // TODO the int cast here is potentially buggy + int expectedBloomFilterSize = Math.max(SSTableReader.indexInterval(), (int)SSTableReader.getApproximateKeyCount(sstables)); + if (logger.isDebugEnabled()) + logger.debug("Expected bloom filter size : " + expectedBloomFilterSize); + + SSTableWriter writer; + CompactionIterator ci = new CompactionIterator(sstables, gcBefore, major); // retain a handle so we can call close() + Iterator nni = new FilterIterator(ci, PredicateUtils.notNullPredicate()); + + try + { + if (!nni.hasNext()) + { + // don't mark compacted in the finally block, since if there _is_ nondeleted data, + // we need to sync it (via closeAndOpen) first, so there is no period during which + // a crash could cause data loss. + cfs.markCompacted(sstables); + return 0; + } + + String newFilename = new File(compactionFileLocation, cfs.getTempSSTableFileName()).getAbsolutePath(); + writer = new SSTableWriter(newFilename, expectedBloomFilterSize, StorageService.getPartitioner()); + + // validate the CF as we iterate over it + AntiEntropyService.IValidator validator = AntiEntropyService.instance().getValidator(table.name, cfs.getColumnFamilyName(), null, major); + validator.prepare(); + while (nni.hasNext()) + { + CompactionIterator.CompactedRow row = nni.next(); + writer.append(row.key, row.buffer); + validator.add(row); + totalkeysWritten++; + } + validator.complete(); + } + finally + { + ci.close(); + } + + SSTableReader ssTable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table.name)); + cfs.replaceCompactedSSTables(sstables, Arrays.asList(ssTable)); + gcAfterRpcTimeout(); + instance.submitMinor(cfs); + + String format = "Compacted to %s. %d/%d bytes for %d keys. Time: %dms."; + long dTime = System.currentTimeMillis() - startTime; + logger.info(String.format(format, writer.getFilename(), SSTable.getTotalBytes(sstables), ssTable.length(), totalkeysWritten, dTime)); + return sstables.size(); + } + + /** + * This function is used to do the anti compaction process , it spits out the file which has keys that belong to a given range + * If the target is not specified it spits out the file as a compacted file with the unecessary ranges wiped out. + * + * @param cfs + * @param sstables + * @param ranges + * @param target + * @return + * @throws java.io.IOException + */ + private List doAntiCompaction(ColumnFamilyStore cfs, Collection sstables, Collection ranges, InetAddress target) + throws IOException + { + Table table = cfs.getTable(); + logger.info("AntiCompacting [" + StringUtils.join(sstables, ",") + "]"); + // Calculate the expected compacted filesize + long expectedRangeFileSize = cfs.getExpectedCompactedFileSize(sstables) / 2; + String compactionFileLocation = DatabaseDescriptor.getDataFileLocationForTable(table.name, expectedRangeFileSize); + if (compactionFileLocation == null) + { + throw new UnsupportedOperationException("disk full"); + } + if (target != null) + { + // compacting for streaming: send to subdirectory + compactionFileLocation = compactionFileLocation + File.separator + DatabaseDescriptor.STREAMING_SUBDIR; + } + List results = new ArrayList(); + + long startTime = System.currentTimeMillis(); + long totalkeysWritten = 0; + + int expectedBloomFilterSize = Math.max(SSTableReader.indexInterval(), (int)(SSTableReader.getApproximateKeyCount(sstables) / 2)); + if (logger.isDebugEnabled()) + logger.debug("Expected bloom filter size : " + expectedBloomFilterSize); + + SSTableWriter writer = null; + CompactionIterator ci = new AntiCompactionIterator(sstables, ranges, getDefaultGCBefore(), cfs.isCompleteSSTables(sstables)); + Iterator nni = new FilterIterator(ci, PredicateUtils.notNullPredicate()); + + try + { + if (!nni.hasNext()) + { + return results; + } + + while (nni.hasNext()) + { + CompactionIterator.CompactedRow row = nni.next(); + if (writer == null) + { + FileUtils.createDirectory(compactionFileLocation); + String newFilename = new File(compactionFileLocation, cfs.getTempSSTableFileName()).getAbsolutePath(); + writer = new SSTableWriter(newFilename, expectedBloomFilterSize, StorageService.getPartitioner()); + } + writer.append(row.key, row.buffer); + totalkeysWritten++; + } + } + finally + { + ci.close(); + } + + if (writer != null) + { + results.add(writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table.name))); + String format = "AntiCompacted to %s. %d/%d bytes for %d keys. Time: %dms."; + long dTime = System.currentTimeMillis() - startTime; + logger.info(String.format(format, writer.getFilename(), SSTable.getTotalBytes(sstables), results.get(0).length(), totalkeysWritten, dTime)); + } + + return results; + } + + /** + * This function goes over each file and removes the keys that the node is not responsible for + * and only keeps keys that this node is responsible for. + * + * @throws IOException + */ + private void doCleanupCompaction(ColumnFamilyStore cfs) throws IOException + { + Collection originalSSTables = cfs.getSSTables(); + List sstables = doAntiCompaction(cfs, originalSSTables, StorageService.instance().getLocalRanges(), null); + if (!sstables.isEmpty()) + { + cfs.replaceCompactedSSTables(originalSSTables, sstables); + } + CompactionManager.gcAfterRpcTimeout(); + } + + /** + * Performs a readonly "compaction" of all sstables in order to validate complete rows, + * but without writing the merge result + */ + private void doReadonlyCompaction(ColumnFamilyStore cfs, InetAddress initiator) throws IOException + { + Collection sstables = cfs.getSSTables(); + CompactionIterator ci = new CompactionIterator(sstables, getDefaultGCBefore(), true); + try + { + Iterator nni = new FilterIterator(ci, PredicateUtils.notNullPredicate()); + + // validate the CF as we iterate over it + AntiEntropyService.IValidator validator = AntiEntropyService.instance().getValidator(cfs.getTable().name, cfs.getColumnFamilyName(), initiator, true); + validator.prepare(); + while (nni.hasNext()) + { + CompactionIterator.CompactedRow row = nni.next(); + validator.add(row); + } + validator.complete(); + } + finally + { + ci.close(); + } + } + + /** + * perform a GC to clean out obsolete sstables, sleeping rpc timeout first so that most in-progress ops can complete + * (thus, no longer reference the sstables in question) + */ + static void gcAfterRpcTimeout() + { + new Thread(new Runnable() + { + public void run() + { + try + { + Thread.sleep(DatabaseDescriptor.getRpcTimeout()); + } + catch (InterruptedException e) + { + throw new AssertionError(e); + } + System.gc(); + } + }).start(); + } + + /* + * Group files of similar size into buckets. + */ + static Set> getCompactionBuckets(Iterable files, long min) + { + Map, Long> buckets = new HashMap, Long>(); + for (SSTableReader sstable : files) + { + long size = sstable.length(); + + boolean bFound = false; + // look for a bucket containing similar-sized files: + // group in the same bucket if it's w/in 50% of the average for this bucket, + // or this file and the bucket are all considered "small" (less than `min`) + for (List bucket : buckets.keySet()) + { + long averageSize = buckets.get(bucket); + if ((size > averageSize / 2 && size < 3 * averageSize / 2) + || (size < min && averageSize < min)) + { + // remove and re-add because adding changes the hash + buckets.remove(bucket); + averageSize = (averageSize + size) / 2; + bucket.add(sstable); + buckets.put(bucket, averageSize); + bFound = true; + break; + } + } + // no similar bucket found; put it in a new one + if (!bFound) + { + ArrayList bucket = new ArrayList(); + bucket.add(sstable); + buckets.put(bucket, size); + } + } + + return buckets.keySet(); + } + + public static int getDefaultGCBefore() + { + return (int)(System.currentTimeMillis() / 1000) - DatabaseDescriptor.getGcGraceInSeconds(); + } + + 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/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java index 6d72047c9f..ccd3e57087 100644 --- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java +++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java @@ -24,6 +24,7 @@ import java.util.TimerTask; import java.util.Arrays; import java.util.concurrent.TimeoutException; import java.util.concurrent.ExecutorService; +import java.util.concurrent.ExecutionException; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import java.io.IOException; @@ -187,7 +188,14 @@ public class HintedHandOffManager } } hintStore.forceFlush(); - hintStore.doMajorCompaction(0); + try + { + CompactionManager.instance.submitMajor(hintStore).get(); + } + catch (Exception e) + { + throw new RuntimeException(e); + } if (logger_.isDebugEnabled()) logger_.debug("Finished deliverAllHints"); diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index 46c023baf2..c94650a9ce 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -25,6 +25,7 @@ import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.concurrent.Future; +import java.util.concurrent.ExecutionException; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.dht.Range; @@ -298,7 +299,14 @@ public class Table continue; ColumnFamilyStore cfStore = columnFamilyStores.get( columnFamily ); - allResults.addAll(cfStore.forceAntiCompaction(ranges, target)); + try + { + allResults.addAll(CompactionManager.instance.submitAnticompaction(cfStore, ranges, target).get()); + } + catch (Exception e) + { + throw new RuntimeException(e); + } } return allResults; } @@ -314,7 +322,7 @@ public class Table { ColumnFamilyStore cfStore = columnFamilyStores.get( columnFamily ); if ( cfStore != null ) - CompactionManager.instance.submitMajor(cfStore, 0); + CompactionManager.instance.submitMajor(cfStore); } } @@ -474,6 +482,11 @@ public class Table return applicationColumnFamilies; } + public String getDataFileLocation(long expectedCompactedFileSize) + { + return DatabaseDescriptor.getDataFileLocationForTable(name, expectedCompactedFileSize); + } + public static String getSnapshotPath(String dataDirPath, String tableName, String snapshotName) { return dataDirPath + File.separator + tableName + File.separator + SNAPSHOT_SUBDIR_NAME + File.separator + snapshotName; diff --git a/src/java/org/apache/cassandra/io/SSTable.java b/src/java/org/apache/cassandra/io/SSTable.java index ab836795fb..e5b0690a4f 100644 --- a/src/java/org/apache/cassandra/io/SSTable.java +++ b/src/java/org/apache/cassandra/io/SSTable.java @@ -161,6 +161,16 @@ public abstract class SSTable logger.info("Deleted " + path); } + public static long getTotalBytes(Iterable sstables) + { + long sum = 0; + for (SSTableReader sstable : sstables) + { + sum += sstable.length(); + } + return sum; + } + /** * This is a simple container for the index Key and its corresponding position * in the data file. Binary search is performed on a list of these objects diff --git a/src/java/org/apache/cassandra/service/AntiEntropyService.java b/src/java/org/apache/cassandra/service/AntiEntropyService.java index a39992f31c..45d7aae5d5 100644 --- a/src/java/org/apache/cassandra/service/AntiEntropyService.java +++ b/src/java/org/apache/cassandra/service/AntiEntropyService.java @@ -652,7 +652,7 @@ public class AntiEntropyService try { List ranges = new ArrayList(differences); - List sstables = CompactionManager.instance.submitAnti(cfstore, ranges, remote).get(); + List sstables = CompactionManager.instance.submitAnticompaction(cfstore, ranges, remote).get(); Streaming.transferSSTables(remote, sstables, cf.left); } catch(Exception e) diff --git a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java index e1f92c2a96..0e0dbf951f 100644 --- a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java +++ b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java @@ -18,13 +18,10 @@ */ package org.apache.cassandra.db; -import java.io.File; -import java.io.FileOutputStream; import java.io.IOException; import java.util.*; import java.util.concurrent.ExecutionException; -import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.ArrayUtils; import static org.junit.Assert.assertNull; import org.junit.Test; @@ -119,7 +116,7 @@ public class ColumnFamilyStoreTest extends CleanupHelper Range r = new Range(partitioner.getToken("0"), partitioner.getToken("zzzzzzz")); ranges.add(r); - List fileList = store.forceAntiCompaction(ranges, InetAddress.getByName("127.0.0.1")); + List fileList = CompactionManager.instance.submitAnticompaction(store, ranges, InetAddress.getByName("127.0.0.1")).get(); assert fileList.size() >= 1; } diff --git a/test/unit/org/apache/cassandra/db/CompactionsTest.java b/test/unit/org/apache/cassandra/db/CompactionsTest.java index a62c60567f..1fa6b301ad 100644 --- a/test/unit/org/apache/cassandra/db/CompactionsTest.java +++ b/test/unit/org/apache/cassandra/db/CompactionsTest.java @@ -45,6 +45,8 @@ public class CompactionsTest extends CleanupHelper @Test public void testCompactions() throws IOException, ExecutionException, InterruptedException { + CompactionManager.instance.disableAutoCompaction(); + // this test does enough rows to force multiple block indexes to be used Table table = Table.open(TABLE1); ColumnFamilyStore store = table.getColumnFamilyStore("Standard1"); @@ -70,7 +72,7 @@ public class CompactionsTest extends CleanupHelper } if (store.getSSTables().size() > 1) { - store.doCompaction(2, store.getSSTables().size()); + CompactionManager.instance.submitMajor(store).get(); } assertEquals(inserted.size(), table.getColumnFamilyStore("Standard1").getKeyRange("", "", 10000).keys.size()); } @@ -78,7 +80,7 @@ public class CompactionsTest extends CleanupHelper @Test public void testCompactionPurge() throws IOException, ExecutionException, InterruptedException { - CompactionManager.instance.disableCompactions(); + CompactionManager.instance.disableAutoCompaction(); Table table = Table.open(TABLE1); String cfName = "Standard1"; @@ -117,12 +119,12 @@ public class CompactionsTest extends CleanupHelper rm.add(new QueryPath(cfName, null, "0".getBytes()), new byte[0], 0); rm.apply(); store.forceBlockingFlush(); - store.doFileCompaction(sstablesIncomplete, Integer.MAX_VALUE); + CompactionManager.instance.doCompaction(store, sstablesIncomplete, CompactionManager.getDefaultGCBefore()); ColumnFamily cf = table.getColumnFamilyStore(cfName).getColumnFamily(new IdentityQueryFilter(key, new QueryPath(cfName))); assert cf.getColumnCount() == 10; // major compact and test that all columns but the resurrected one is completely gone - store.doFileCompaction(store.getSSTables(), Integer.MAX_VALUE); + CompactionManager.instance.submitMajor(store, 0, Integer.MAX_VALUE).get(); cf = table.getColumnFamilyStore(cfName).getColumnFamily(new IdentityQueryFilter(key, new QueryPath(cfName))); assert cf.getColumnCount() == 1; assert cf.getColumn(String.valueOf(5).getBytes()) != null; @@ -131,7 +133,7 @@ public class CompactionsTest extends CleanupHelper @Test public void testCompactionPurgeOneFile() throws IOException, ExecutionException, InterruptedException { - CompactionManager.instance.disableCompactions(); + CompactionManager.instance.disableAutoCompaction(); Table table = Table.open(TABLE1); String cfName = "Standard2"; @@ -160,7 +162,7 @@ public class CompactionsTest extends CleanupHelper assert store.getSSTables().size() == 1 : store.getSSTables(); // inserts & deletes were in the same memtable -> only deletes in sstable // compact and test that the row is completely gone - store.doFileCompaction(store.getSSTables(), Integer.MAX_VALUE); + CompactionManager.instance.submitMajor(store, 0, Integer.MAX_VALUE).get(); assert store.getSSTables().isEmpty(); ColumnFamily cf = table.getColumnFamilyStore(cfName).getColumnFamily(new IdentityQueryFilter(key, new QueryPath(cfName))); assert cf == null : cf; @@ -169,6 +171,8 @@ public class CompactionsTest extends CleanupHelper @Test public void testCompactionReadonly() throws IOException, ExecutionException, InterruptedException { + CompactionManager.instance.disableAutoCompaction(); + Table table = Table.open(TABLE2); ColumnFamilyStore store = table.getColumnFamilyStore("Standard1"); @@ -188,7 +192,7 @@ public class CompactionsTest extends CleanupHelper // perform readonly compaction and confirm that no sstables changed ArrayList oldsstables = new ArrayList(store.getSSTables()); - store.doReadonlyCompaction(LOCAL); + CompactionManager.instance.submitReadonly(store, LOCAL).get(); assertEquals(oldsstables, new ArrayList(store.getSSTables())); assertEquals(inserted.size(), table.getColumnFamilyStore("Standard1").getKeyRange("", "", 10000).keys.size()); } diff --git a/test/unit/org/apache/cassandra/db/OneCompactionTest.java b/test/unit/org/apache/cassandra/db/OneCompactionTest.java index 3725707d8f..b53e0fbe5d 100644 --- a/test/unit/org/apache/cassandra/db/OneCompactionTest.java +++ b/test/unit/org/apache/cassandra/db/OneCompactionTest.java @@ -33,6 +33,8 @@ public class OneCompactionTest { private void testCompaction(String columnFamilyName, int insertsPerTable) throws IOException, ExecutionException, InterruptedException { + CompactionManager.instance.disableAutoCompaction(); + Table table = Table.open("Keyspace1"); ColumnFamilyStore store = table.getColumnFamilyStore(columnFamilyName); @@ -46,8 +48,7 @@ public class OneCompactionTest store.forceBlockingFlush(); assertEquals(inserted.size(), table.getColumnFamilyStore(columnFamilyName).getKeyRange("", "", 10000).keys.size()); } - Future ft = CompactionManager.instance.submitMinor(store, 2, 32); - ft.get(); + CompactionManager.instance.submitMajor(store).get(); assertEquals(1, store.getSSTables().size()); assertEquals(table.getColumnFamilyStore(columnFamilyName).getKeyRange("", "", 10000).keys.size(), inserted.size()); } diff --git a/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java b/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java index d719eacaad..57677cd733 100644 --- a/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java +++ b/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java @@ -37,7 +37,7 @@ public class RecoveryManager2Test extends CleanupHelper @Test public void testWithFlush() throws IOException, ExecutionException, InterruptedException { - CompactionManager.instance.disableCompactions(); + CompactionManager.instance.disableAutoCompaction(); for (int i = 0; i < 100; i++) { diff --git a/test/unit/org/apache/cassandra/db/RemoveSuperColumnTest.java b/test/unit/org/apache/cassandra/db/RemoveSuperColumnTest.java index 3f6a8337e5..e51167d834 100644 --- a/test/unit/org/apache/cassandra/db/RemoveSuperColumnTest.java +++ b/test/unit/org/apache/cassandra/db/RemoveSuperColumnTest.java @@ -59,8 +59,7 @@ public class RemoveSuperColumnTest store.forceBlockingFlush(); validateRemoveTwoSources(); - Future ft = CompactionManager.instance.submitMinor(store, 2, 32); - ft.get(); + CompactionManager.instance.submitMajor(store).get(); assertEquals(1, store.getSSTables().size()); validateRemoveCompacted(); } @@ -144,8 +143,7 @@ public class RemoveSuperColumnTest store.forceBlockingFlush(); validateRemoveWithNewData(); - Future ft = CompactionManager.instance.submitMinor(store, 2, 32); - ft.get(); + CompactionManager.instance.submitMajor(store).get(); assertEquals(1, store.getSSTables().size()); validateRemoveWithNewData(); } diff --git a/test/unit/org/apache/cassandra/db/TableTest.java b/test/unit/org/apache/cassandra/db/TableTest.java index 316375dbf2..5d3db60848 100644 --- a/test/unit/org/apache/cassandra/db/TableTest.java +++ b/test/unit/org/apache/cassandra/db/TableTest.java @@ -359,7 +359,7 @@ public class TableTest extends CleanupHelper // compact so we have a big row with more than the minimum index count if (cfStore.getSSTables().size() > 1) { - cfStore.doCompaction(2, cfStore.getSSTables().size()); + CompactionManager.instance.submitMajor(cfStore).get(); } SSTableReader sstable = cfStore.getSSTables().iterator().next(); DecoratedKey decKey = sstable.getPartitioner().decorateKey(key);