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

clean up compactions api and move into CompactionManager. patch by jbellis; reviewed by eevans for CASSANDRA-599 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@893600 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-12-23 19:18:05 +00:00
parent 278f77eafd
commit 453c8171d9
12 changed files with 445 additions and 461 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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<SSTableReader> forceAntiCompaction(Collection<Range> ranges, InetAddress target)
{
assert ranges != null;
Future<List<SSTableReader>> futurePtr = CompactionManager.instance.submitAnti(ColumnFamilyStore.this,
ranges, target);
List<SSTableReader> 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<List<SSTableReader>> getCompactionBuckets(Iterable<SSTableReader> files, long min)
{
Map<List<SSTableReader>, Long> buckets = new HashMap<List<SSTableReader>, 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<SSTableReader> 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<SSTableReader> bucket = new ArrayList<SSTableReader>();
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<SSTableReader> 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<SSTableReader> sstables;
if (skip > 0)
{
sstables = new ArrayList<SSTableReader>();
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<SSTableReader> doAntiCompaction(Collection<Range> 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<SSTableReader> 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<SSTableReader> doFileAntiCompaction(Collection<SSTableReader> sstables, final Collection<Range> 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<SSTableReader> results = new ArrayList<SSTableReader>();
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<CompactionIterator.CompactedRow> 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<SSTableReader> sstables) throws IOException
void markCompacted(Collection<SSTableReader> 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<SSTableReader> 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<SSTableReader> smallerSSTables = new ArrayList<SSTableReader>(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<CompactionIterator.CompactedRow> 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<SSTableReader> 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<SSTableReader>(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<SSTableReader> sstables, Iterable<SSTableReader> replacements)
throws IOException
{
Collection<SSTableReader> sstables = ssTables_.getSSTables();
CompactionIterator ci = new CompactionIterator(sstables, getDefaultGCBefore(), true);
try
for (SSTableReader sstable : replacements)
{
Iterator<CompactionIterator.CompactedRow> 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<SSTableReader> sstables)
{
long sum = 0;
for (SSTableReader sstable : sstables)
{
sum += sstable.length();
}
return sum;
ssTables_.markCompacted(sstables);
}
public static List<Memtable> 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<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();
}
}
}
}

394
src/java/org/apache/cassandra/db/CompactionManager.java
View File

@ -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<Integer> submitMinor(final ColumnFamilyStore columnFamilyStore)
{
return submitMinor(columnFamilyStore, minimumCompactionThreshold, maximumCompactionThreshold);
}
Future<Integer> submitMinor(final ColumnFamilyStore cfStore, final int minThreshold, final int maxThreshold)
public Future<Integer> submitMinor(final ColumnFamilyStore cfs)
{
Callable<Integer> callable = new Callable<Integer>()
{
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<SSTableReader> 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<List<SSTableReader>> submitAnti(final ColumnFamilyStore cfStore, final Collection<Range> ranges, final InetAddress target)
public Future<List<SSTableReader>> submitAnticompaction(final ColumnFamilyStore cfStore, final Collection<Range> ranges, final InetAddress target)
{
Callable<List<SSTableReader>> callable = new Callable<List<SSTableReader>>()
{
public List<SSTableReader> 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<Object> callable = new Callable<Object>()
{
public Object call() throws IOException
{
cfStore.doMajorCompaction(skip);
Collection<SSTableReader> sstables;
if (skip > 0)
{
sstables = new ArrayList<SSTableReader>();
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<SSTableReader> 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<SSTableReader> smallerSSTables = new ArrayList<SSTableReader>(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<CompactionIterator.CompactedRow> 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<SSTableReader> doAntiCompaction(ColumnFamilyStore cfs, Collection<SSTableReader> sstables, Collection<Range> 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<SSTableReader> results = new ArrayList<SSTableReader>();
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<CompactionIterator.CompactedRow> 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<SSTableReader> originalSSTables = cfs.getSSTables();
List<SSTableReader> 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<SSTableReader> sstables = cfs.getSSTables();
CompactionIterator ci = new CompactionIterator(sstables, getDefaultGCBefore(), true);
try
{
Iterator<CompactionIterator.CompactedRow> 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<List<SSTableReader>> getCompactionBuckets(Iterable<SSTableReader> files, long min)
{
Map<List<SSTableReader>, Long> buckets = new HashMap<List<SSTableReader>, 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<SSTableReader> 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<SSTableReader> bucket = new ArrayList<SSTableReader>();
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<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();
}
}
}
}

10
src/java/org/apache/cassandra/db/HintedHandOffManager.java
View File

@ -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");

17
src/java/org/apache/cassandra/db/Table.java
View File

@ -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;

10
src/java/org/apache/cassandra/io/SSTable.java
View File

@ -161,6 +161,16 @@ public abstract class SSTable
logger.info("Deleted " + path);
}
public static long getTotalBytes(Iterable<SSTableReader> 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

2
src/java/org/apache/cassandra/service/AntiEntropyService.java
View File

@ -652,7 +652,7 @@ public class AntiEntropyService
try
{
List<Range> ranges = new ArrayList<Range>(differences);
List<SSTableReader> sstables = CompactionManager.instance.submitAnti(cfstore, ranges, remote).get();
List<SSTableReader> sstables = CompactionManager.instance.submitAnticompaction(cfstore, ranges, remote).get();
Streaming.transferSSTables(remote, sstables, cf.left);
}
catch(Exception e)

5
test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
View File

@ -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<SSTableReader> fileList = store.forceAntiCompaction(ranges, InetAddress.getByName("127.0.0.1"));
List<SSTableReader> fileList = CompactionManager.instance.submitAnticompaction(store, ranges, InetAddress.getByName("127.0.0.1")).get();
assert fileList.size() >= 1;
}

18
test/unit/org/apache/cassandra/db/CompactionsTest.java
View File

@ -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<SSTableReader> oldsstables = new ArrayList<SSTableReader>(store.getSSTables());
store.doReadonlyCompaction(LOCAL);
CompactionManager.instance.submitReadonly(store, LOCAL).get();
assertEquals(oldsstables, new ArrayList<SSTableReader>(store.getSSTables()));
assertEquals(inserted.size(), table.getColumnFamilyStore("Standard1").getKeyRange("", "", 10000).keys.size());
}

5
test/unit/org/apache/cassandra/db/OneCompactionTest.java
View File

@ -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<Integer> 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());
}

2
test/unit/org/apache/cassandra/db/RecoveryManager2Test.java
View File

@ -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++)
{

6
test/unit/org/apache/cassandra/db/RemoveSuperColumnTest.java
View File

@ -59,8 +59,7 @@ public class RemoveSuperColumnTest
store.forceBlockingFlush();
validateRemoveTwoSources();
Future<Integer> 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<Integer> ft = CompactionManager.instance.submitMinor(store, 2, 32);
ft.get();
CompactionManager.instance.submitMajor(store).get();
assertEquals(1, store.getSSTables().size());
validateRemoveWithNewData();
}

2
test/unit/org/apache/cassandra/db/TableTest.java
View File

@ -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);