mirror of https://github.com/apache/cassandra
add compaction test showing regression. patch by jbellis; reviewed by Todd Lipcon
and Jun Rao for #80. git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@764678 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis
parent
50c1987265
commit
a1ef3b5be5
|
|
@ -74,7 +74,7 @@ public class ColumnFamilyStore
|
|||
private AtomicReference<BinaryMemtable> binaryMemtable_;
|
||||
|
||||
/* SSTables on disk for this column family */
|
||||
private Set<String> ssTables_ = new HashSet<String>();
|
||||
Set<String> ssTables_ = new HashSet<String>();
|
||||
|
||||
/* Modification lock used for protecting reads from compactions. */
|
||||
private ReentrantReadWriteLock lock_ = new ReentrantReadWriteLock(true);
|
||||
|
|
@ -162,6 +162,7 @@ public class ColumnFamilyStore
|
|||
{
|
||||
HintedHandOffManager.instance().submit(this);
|
||||
}
|
||||
// TODO this seems unnecessary -- each memtable flush checks to see if it needs to compact, too
|
||||
MinorCompactionManager.instance().submitPeriodicCompaction(this);
|
||||
}
|
||||
|
||||
|
|
@ -671,7 +672,7 @@ public class ColumnFamilyStore
|
|||
}
|
||||
catch ( Exception ex)
|
||||
{
|
||||
ex.printStackTrace();
|
||||
logger_.warn("corrupt file? or are you just blowing away data files manually out from under me?", ex);
|
||||
try
|
||||
{
|
||||
if (fs != null)
|
||||
|
|
@ -734,7 +735,7 @@ public class ColumnFamilyStore
|
|||
/*
|
||||
* Break the files into buckets and then compact.
|
||||
*/
|
||||
void doCompaction()
|
||||
void doCompaction() throws IOException
|
||||
{
|
||||
isCompacting_.set(true);
|
||||
List<String> files = new ArrayList<String>(ssTables_);
|
||||
|
|
@ -755,17 +756,10 @@ public class ColumnFamilyStore
|
|||
if( count == threshHold_ )
|
||||
break;
|
||||
}
|
||||
try
|
||||
{
|
||||
// For each bucket if it has crossed the threshhold do the compaction
|
||||
// In case of range compaction merge the counting bloom filters also.
|
||||
if( count == threshHold_)
|
||||
doFileCompaction(files, bufSize_);
|
||||
}
|
||||
catch ( Exception ex)
|
||||
{
|
||||
logger_.warn(LogUtil.throwableToString(ex));
|
||||
}
|
||||
// For each bucket if it has crossed the threshhold do the compaction
|
||||
// In case of range compaction merge the counting bloom filters also.
|
||||
if( count == threshHold_)
|
||||
doFileCompaction(files, bufSize_);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1175,7 +1169,7 @@ public class ColumnFamilyStore
|
|||
* to get the latest data.
|
||||
*
|
||||
*/
|
||||
void doFileCompaction(List<String> files, int minBufferSize)
|
||||
void doFileCompaction(List<String> files, int minBufferSize) throws IOException
|
||||
{
|
||||
String newfile = null;
|
||||
long startTime = System.currentTimeMillis();
|
||||
|
|
@ -1183,175 +1177,168 @@ public class ColumnFamilyStore
|
|||
long totalBytesWritten = 0;
|
||||
long totalkeysRead = 0;
|
||||
long totalkeysWritten = 0;
|
||||
try
|
||||
// Calculate the expected compacted filesize
|
||||
long expectedCompactedFileSize = getExpectedCompactedFileSize(files);
|
||||
String compactionFileLocation = DatabaseDescriptor.getCompactionFileLocation(expectedCompactedFileSize);
|
||||
// If the compaction file path is null that means we have no space left for this compaction.
|
||||
if( compactionFileLocation == null )
|
||||
{
|
||||
// Calculate the expected compacted filesize
|
||||
long expectedCompactedFileSize = getExpectedCompactedFileSize(files);
|
||||
String compactionFileLocation = DatabaseDescriptor.getCompactionFileLocation(expectedCompactedFileSize);
|
||||
// If the compaction file path is null that means we have no space left for this compaction.
|
||||
if( compactionFileLocation == null )
|
||||
{
|
||||
String maxFile = getMaxSizeFile( files );
|
||||
files.remove( maxFile );
|
||||
doFileCompaction(files , minBufferSize);
|
||||
return;
|
||||
}
|
||||
PriorityQueue<FileStruct> pq = initializePriorityQueue(files, null, minBufferSize);
|
||||
if (pq.size() > 0)
|
||||
{
|
||||
String mergedFileName = getTempFileName( files );
|
||||
SSTable ssTable = null;
|
||||
String lastkey = null;
|
||||
List<FileStruct> lfs = new ArrayList<FileStruct>();
|
||||
DataOutputBuffer bufOut = new DataOutputBuffer();
|
||||
int expectedBloomFilterSize = SSTable.getApproximateKeyCount(files);
|
||||
expectedBloomFilterSize = (expectedBloomFilterSize > 0) ? expectedBloomFilterSize : SSTable.indexInterval();
|
||||
logger_.debug("Expected bloom filter size : " + expectedBloomFilterSize);
|
||||
/* Create the bloom filter for the compacted file. */
|
||||
BloomFilter compactedBloomFilter = new BloomFilter(expectedBloomFilterSize, 15);
|
||||
List<ColumnFamily> columnFamilies = new ArrayList<ColumnFamily>();
|
||||
|
||||
while (pq.size() > 0 || lfs.size() > 0)
|
||||
{
|
||||
FileStruct fs = null;
|
||||
if (pq.size() > 0)
|
||||
{
|
||||
fs = pq.poll();
|
||||
}
|
||||
if (fs != null
|
||||
&& (lastkey == null || lastkey.equals(fs.getKey())))
|
||||
{
|
||||
// The keys are the same so we need to add this to the
|
||||
// ldfs list
|
||||
lastkey = fs.getKey();
|
||||
lfs.add(fs);
|
||||
}
|
||||
else
|
||||
{
|
||||
Collections.sort(lfs, new FileStructComparator());
|
||||
ColumnFamily columnFamily;
|
||||
bufOut.reset();
|
||||
if(lfs.size() > 1)
|
||||
{
|
||||
for (FileStruct filestruct : lfs)
|
||||
{
|
||||
try
|
||||
{
|
||||
/* read the length although we don't need it */
|
||||
filestruct.getBufIn().readInt();
|
||||
// Skip the Index
|
||||
IndexHelper.skipBloomFilterAndIndex(filestruct.getBufIn());
|
||||
// We want to add only 2 and resolve them right there in order to save on memory footprint
|
||||
if(columnFamilies.size() > 1)
|
||||
{
|
||||
merge(columnFamilies);
|
||||
}
|
||||
// deserialize into column families
|
||||
columnFamilies.add(ColumnFamily.serializer().deserialize(filestruct.getBufIn()));
|
||||
}
|
||||
catch ( Exception ex)
|
||||
{
|
||||
logger_.warn("error in filecompaction", ex);
|
||||
}
|
||||
}
|
||||
// Now after merging all crap append to the sstable
|
||||
columnFamily = resolveAndRemoveDeleted(columnFamilies);
|
||||
columnFamilies.clear();
|
||||
if( columnFamily != null )
|
||||
{
|
||||
/* serialize the cf with column indexes */
|
||||
ColumnFamily.serializerWithIndexes().serialize(columnFamily, bufOut);
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
FileStruct filestruct = lfs.get(0);
|
||||
try
|
||||
{
|
||||
/* read the length although we don't need it */
|
||||
int size = filestruct.getBufIn().readInt();
|
||||
bufOut.write(filestruct.getBufIn(), size);
|
||||
}
|
||||
catch ( Exception ex)
|
||||
{
|
||||
ex.printStackTrace();
|
||||
filestruct.close();
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
if ( ssTable == null )
|
||||
{
|
||||
ssTable = new SSTable(compactionFileLocation, mergedFileName);
|
||||
}
|
||||
ssTable.append(lastkey, bufOut);
|
||||
|
||||
/* Fill the bloom filter with the key */
|
||||
doFill(compactedBloomFilter, lastkey);
|
||||
totalkeysWritten++;
|
||||
for (FileStruct filestruct : lfs)
|
||||
{
|
||||
try
|
||||
{
|
||||
filestruct.advance();
|
||||
if (filestruct.isExhausted())
|
||||
{
|
||||
continue;
|
||||
}
|
||||
pq.add(filestruct);
|
||||
totalkeysRead++;
|
||||
}
|
||||
catch ( Throwable ex )
|
||||
{
|
||||
// Ignore the exception as it might be a corrupted file
|
||||
// in any case we have read as far as possible from it
|
||||
// and it will be deleted after compaction.
|
||||
filestruct.close();
|
||||
}
|
||||
}
|
||||
lfs.clear();
|
||||
lastkey = null;
|
||||
if (fs != null)
|
||||
{
|
||||
/* Add back the fs since we processed the rest of filestructs */
|
||||
pq.add(fs);
|
||||
}
|
||||
}
|
||||
}
|
||||
if ( ssTable != null )
|
||||
{
|
||||
ssTable.closeRename(compactedBloomFilter);
|
||||
newfile = ssTable.getDataFileLocation();
|
||||
}
|
||||
lock_.writeLock().lock();
|
||||
try
|
||||
{
|
||||
for (String file : files)
|
||||
{
|
||||
ssTables_.remove(file);
|
||||
SSTable.removeAssociatedBloomFilter(file);
|
||||
}
|
||||
if ( newfile != null )
|
||||
{
|
||||
ssTables_.add(newfile);
|
||||
logger_.debug("Inserting bloom filter for file " + newfile);
|
||||
SSTable.storeBloomFilter(newfile, compactedBloomFilter);
|
||||
totalBytesWritten = (new File(newfile)).length();
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
lock_.writeLock().unlock();
|
||||
}
|
||||
for (String file : files)
|
||||
{
|
||||
SSTable.delete(file);
|
||||
}
|
||||
}
|
||||
String maxFile = getMaxSizeFile( files );
|
||||
files.remove( maxFile );
|
||||
doFileCompaction(files , minBufferSize);
|
||||
return;
|
||||
}
|
||||
catch ( Exception ex)
|
||||
PriorityQueue<FileStruct> pq = initializePriorityQueue(files, null, minBufferSize);
|
||||
if (pq.size() > 0)
|
||||
{
|
||||
logger_.warn( LogUtil.throwableToString(ex) );
|
||||
String mergedFileName = getTempFileName( files );
|
||||
SSTable ssTable = null;
|
||||
String lastkey = null;
|
||||
List<FileStruct> lfs = new ArrayList<FileStruct>();
|
||||
DataOutputBuffer bufOut = new DataOutputBuffer();
|
||||
int expectedBloomFilterSize = SSTable.getApproximateKeyCount(files);
|
||||
expectedBloomFilterSize = (expectedBloomFilterSize > 0) ? expectedBloomFilterSize : SSTable.indexInterval();
|
||||
logger_.debug("Expected bloom filter size : " + expectedBloomFilterSize);
|
||||
/* Create the bloom filter for the compacted file. */
|
||||
BloomFilter compactedBloomFilter = new BloomFilter(expectedBloomFilterSize, 15);
|
||||
List<ColumnFamily> columnFamilies = new ArrayList<ColumnFamily>();
|
||||
|
||||
while (pq.size() > 0 || lfs.size() > 0)
|
||||
{
|
||||
FileStruct fs = null;
|
||||
if (pq.size() > 0)
|
||||
{
|
||||
fs = pq.poll();
|
||||
}
|
||||
if (fs != null
|
||||
&& (lastkey == null || lastkey.equals(fs.getKey())))
|
||||
{
|
||||
// The keys are the same so we need to add this to the
|
||||
// ldfs list
|
||||
lastkey = fs.getKey();
|
||||
lfs.add(fs);
|
||||
}
|
||||
else
|
||||
{
|
||||
Collections.sort(lfs, new FileStructComparator());
|
||||
ColumnFamily columnFamily;
|
||||
bufOut.reset();
|
||||
if(lfs.size() > 1)
|
||||
{
|
||||
for (FileStruct filestruct : lfs)
|
||||
{
|
||||
try
|
||||
{
|
||||
/* read the length although we don't need it */
|
||||
filestruct.getBufIn().readInt();
|
||||
// Skip the Index
|
||||
IndexHelper.skipBloomFilterAndIndex(filestruct.getBufIn());
|
||||
// We want to add only 2 and resolve them right there in order to save on memory footprint
|
||||
if(columnFamilies.size() > 1)
|
||||
{
|
||||
merge(columnFamilies);
|
||||
}
|
||||
// deserialize into column families
|
||||
columnFamilies.add(ColumnFamily.serializer().deserialize(filestruct.getBufIn()));
|
||||
}
|
||||
catch ( Exception ex)
|
||||
{
|
||||
logger_.warn("error in filecompaction", ex);
|
||||
}
|
||||
}
|
||||
// Now after merging all crap append to the sstable
|
||||
columnFamily = resolveAndRemoveDeleted(columnFamilies);
|
||||
columnFamilies.clear();
|
||||
if( columnFamily != null )
|
||||
{
|
||||
/* serialize the cf with column indexes */
|
||||
ColumnFamily.serializerWithIndexes().serialize(columnFamily, bufOut);
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
FileStruct filestruct = lfs.get(0);
|
||||
try
|
||||
{
|
||||
/* read the length although we don't need it */
|
||||
int size = filestruct.getBufIn().readInt();
|
||||
bufOut.write(filestruct.getBufIn(), size);
|
||||
}
|
||||
catch ( Exception ex)
|
||||
{
|
||||
ex.printStackTrace();
|
||||
filestruct.close();
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
if ( ssTable == null )
|
||||
{
|
||||
ssTable = new SSTable(compactionFileLocation, mergedFileName);
|
||||
}
|
||||
ssTable.append(lastkey, bufOut);
|
||||
|
||||
/* Fill the bloom filter with the key */
|
||||
doFill(compactedBloomFilter, lastkey);
|
||||
totalkeysWritten++;
|
||||
for (FileStruct filestruct : lfs)
|
||||
{
|
||||
try
|
||||
{
|
||||
filestruct.advance();
|
||||
if (filestruct.isExhausted())
|
||||
{
|
||||
continue;
|
||||
}
|
||||
pq.add(filestruct);
|
||||
totalkeysRead++;
|
||||
}
|
||||
catch ( Throwable ex )
|
||||
{
|
||||
// Ignore the exception as it might be a corrupted file
|
||||
// in any case we have read as far as possible from it
|
||||
// and it will be deleted after compaction.
|
||||
filestruct.close();
|
||||
}
|
||||
}
|
||||
lfs.clear();
|
||||
lastkey = null;
|
||||
if (fs != null)
|
||||
{
|
||||
/* Add back the fs since we processed the rest of filestructs */
|
||||
pq.add(fs);
|
||||
}
|
||||
}
|
||||
}
|
||||
if ( ssTable != null )
|
||||
{
|
||||
ssTable.closeRename(compactedBloomFilter);
|
||||
newfile = ssTable.getDataFileLocation();
|
||||
}
|
||||
lock_.writeLock().lock();
|
||||
try
|
||||
{
|
||||
for (String file : files)
|
||||
{
|
||||
ssTables_.remove(file);
|
||||
SSTable.removeAssociatedBloomFilter(file);
|
||||
}
|
||||
if ( newfile != null )
|
||||
{
|
||||
ssTables_.add(newfile);
|
||||
logger_.debug("Inserting bloom filter for file " + newfile);
|
||||
SSTable.storeBloomFilter(newfile, compactedBloomFilter);
|
||||
totalBytesWritten = (new File(newfile)).length();
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
lock_.writeLock().unlock();
|
||||
}
|
||||
for (String file : files)
|
||||
{
|
||||
SSTable.delete(file);
|
||||
}
|
||||
}
|
||||
logger_.debug("Total time taken for compaction ..."
|
||||
+ (System.currentTimeMillis() - startTime));
|
||||
|
|
|
|||
|
|
@ -81,16 +81,16 @@ class MinorCompactionManager implements IComponentShutdown
|
|||
|
||||
public void run()
|
||||
{
|
||||
logger_.debug("Started compaction ..."+columnFamilyStore_.columnFamily_);
|
||||
try
|
||||
{
|
||||
logger_.debug("Started compaction ..."+columnFamilyStore_.columnFamily_);
|
||||
columnFamilyStore_.doCompaction();
|
||||
logger_.debug("Finished compaction ..."+columnFamilyStore_.columnFamily_);
|
||||
columnFamilyStore_.doCompaction();
|
||||
}
|
||||
catch (Throwable th)
|
||||
catch (IOException e)
|
||||
{
|
||||
logger_.error( LogUtil.throwableToString(th) );
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
logger_.debug("Finished compaction ..."+columnFamilyStore_.columnFamily_);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -149,16 +149,9 @@ class MinorCompactionManager implements IComponentShutdown
|
|||
|
||||
public void run()
|
||||
{
|
||||
try
|
||||
{
|
||||
logger_.debug("Started compaction ..."+columnFamilyStore_.columnFamily_);
|
||||
columnFamilyStore_.doCleanupCompaction();
|
||||
logger_.debug("Finished compaction ..."+columnFamilyStore_.columnFamily_);
|
||||
}
|
||||
catch (Throwable th)
|
||||
{
|
||||
logger_.error( LogUtil.throwableToString(th) );
|
||||
}
|
||||
logger_.debug("Started compaction ..."+columnFamilyStore_.columnFamily_);
|
||||
columnFamilyStore_.doCleanupCompaction();
|
||||
logger_.debug("Finished compaction ..."+columnFamilyStore_.columnFamily_);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -181,9 +174,9 @@ class MinorCompactionManager implements IComponentShutdown
|
|||
MinorCompactionManager.intervalInMins_, TimeUnit.MINUTES);
|
||||
}
|
||||
|
||||
public void submit(ColumnFamilyStore columnFamilyStore)
|
||||
public Future submit(ColumnFamilyStore columnFamilyStore)
|
||||
{
|
||||
compactor_.submit(new FileCompactor(columnFamilyStore));
|
||||
return compactor_.submit(new FileCompactor(columnFamilyStore));
|
||||
}
|
||||
|
||||
public void submitCleanup(ColumnFamilyStore columnFamilyStore)
|
||||
|
|
|
|||
|
|
@ -314,4 +314,26 @@ public class ColumnFamilyStoreTest extends ServerTest
|
|||
fos.close();
|
||||
return f.getAbsolutePath();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCompaction() throws IOException, ColumnFamilyNotDefinedException, ExecutionException, InterruptedException
|
||||
{
|
||||
Table table = Table.open("Table1");
|
||||
ColumnFamilyStore store = table.getColumnFamilyStore("Standard1");
|
||||
store.ssTables_.clear(); // TODO integrate this better into test setup/teardown
|
||||
|
||||
for (int j = 0; j < 5; j++) {
|
||||
for (int i = 0; i < 10; i++) {
|
||||
long epoch = System.currentTimeMillis() / 1000;
|
||||
String key = String.format("%s.%s.%s", epoch, 1, i);
|
||||
RowMutation rm = new RowMutation("Table1", key);
|
||||
rm.add("Standard1:A", new byte[0], epoch);
|
||||
rm.apply();
|
||||
}
|
||||
store.forceFlush();
|
||||
waitForFlush();
|
||||
}
|
||||
Future ft = MinorCompactionManager.instance().submit(store);
|
||||
ft.get();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue