From a1ef3b5be5be5c35f5f2620cf3b62499f87baa98 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Tue, 14 Apr 2009 05:12:42 +0000 Subject: [PATCH] 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 --- .../cassandra/db/ColumnFamilyStore.java | 349 +++++++++--------- .../cassandra/db/MinorCompactionManager.java | 27 +- .../cassandra/db/ColumnFamilyStoreTest.java | 22 ++ 3 files changed, 200 insertions(+), 198 deletions(-) diff --git a/src/org/apache/cassandra/db/ColumnFamilyStore.java b/src/org/apache/cassandra/db/ColumnFamilyStore.java index a1f286650c..cacefd25b9 100644 --- a/src/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/org/apache/cassandra/db/ColumnFamilyStore.java @@ -74,7 +74,7 @@ public class ColumnFamilyStore private AtomicReference binaryMemtable_; /* SSTables on disk for this column family */ - private Set ssTables_ = new HashSet(); + Set ssTables_ = new HashSet(); /* 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 files = new ArrayList(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 files, int minBufferSize) + void doFileCompaction(List 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 pq = initializePriorityQueue(files, null, minBufferSize); - if (pq.size() > 0) - { - String mergedFileName = getTempFileName( files ); - SSTable ssTable = null; - String lastkey = null; - List lfs = new ArrayList(); - 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 columnFamilies = new ArrayList(); - - 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 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 lfs = new ArrayList(); + 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 columnFamilies = new ArrayList(); + + 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)); diff --git a/src/org/apache/cassandra/db/MinorCompactionManager.java b/src/org/apache/cassandra/db/MinorCompactionManager.java index a424bea7fa..ef69e5c153 100644 --- a/src/org/apache/cassandra/db/MinorCompactionManager.java +++ b/src/org/apache/cassandra/db/MinorCompactionManager.java @@ -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) diff --git a/test/org/apache/cassandra/db/ColumnFamilyStoreTest.java b/test/org/apache/cassandra/db/ColumnFamilyStoreTest.java index 5089f45ef2..25db5e5783 100644 --- a/test/org/apache/cassandra/db/ColumnFamilyStoreTest.java +++ b/test/org/apache/cassandra/db/ColumnFamilyStoreTest.java @@ -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(); + } }