diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index e3008d2b87..8ae2c288a9 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -806,22 +806,23 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean return buckets.keySet(); } - public void doCompaction() throws IOException + public int doCompaction() throws IOException { - doCompaction(COMPACTION_THRESHOLD); + return doCompaction(COMPACTION_THRESHOLD); } /* * Break the files into buckets and then compact. */ - public void doCompaction(int threshold) throws IOException + public int doCompaction(int threshold) throws IOException { isCompacting_.set(true); List files = new ArrayList(ssTables_); + int filesCompacted = 0; try { - int count; - for (List fileList : getCompactionBuckets(files, 50L * 1024L * 1024L)) + Set> buckets = getCompactionBuckets(files, 50L * 1024L * 1024L); + for (List fileList : buckets) { Collections.sort(fileList, new FileNameComparator(FileNameComparator.Ascending)); if (fileList.size() < threshold) @@ -831,14 +832,14 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean // For each bucket if it has crossed the threshhold do the compaction // In case of range compaction merge the counting bloom filters also. files.clear(); - count = 0; + int count = 0; for (String file : fileList) { files.add(file); count++; if (count == threshold) { - doFileCompaction(files, BUFSIZE); + filesCompacted += doFileCompaction(files, BUFSIZE); break; } } @@ -848,6 +849,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean { isCompacting_.set(false); } + return filesCompacted; } void doMajorCompaction(long skip) @@ -1237,7 +1239,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean * to get the latest data. * */ - private void doFileCompaction(List files, int minBufferSize) throws IOException + private int doFileCompaction(List files, int minBufferSize) throws IOException { String compactionFileLocation = DatabaseDescriptor.getCompactionFileLocation(getExpectedCompactedFileSize(files)); // If the compaction file path is null that means we have no space left for this compaction. @@ -1246,8 +1248,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean { String maxFile = getMaxSizeFile( files ); files.remove( maxFile ); - doFileCompaction(files , minBufferSize); - return; + return doFileCompaction(files , minBufferSize); } String newfile = null; @@ -1412,6 +1413,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean String format = "Compacted [%s] to %s. %d/%d bytes for %d/%d keys read/written. Time: %dms."; long dTime = System.currentTimeMillis() - startTime; logger_.info(String.format(format, StringUtils.join(files, ", "), newfile, totalBytesRead, totalBytesWritten, totalkeysRead, totalkeysWritten, dTime)); + return files.size(); } public boolean isSuper() diff --git a/src/java/org/apache/cassandra/db/MinorCompactionManager.java b/src/java/org/apache/cassandra/db/MinorCompactionManager.java index fda2c02614..0e799c536b 100644 --- a/src/java/org/apache/cassandra/db/MinorCompactionManager.java +++ b/src/java/org/apache/cassandra/db/MinorCompactionManager.java @@ -66,7 +66,7 @@ class MinorCompactionManager implements IComponentShutdown return instance_; } - class FileCompactor implements Runnable + class FileCompactor implements Callable { private ColumnFamilyStore columnFamilyStore_; @@ -75,18 +75,21 @@ class MinorCompactionManager implements IComponentShutdown columnFamilyStore_ = columnFamilyStore; } - public void run() + public Integer call() { logger_.debug("Started compaction ..." + columnFamilyStore_.columnFamily_); try { - columnFamilyStore_.doCompaction(); + return columnFamilyStore_.doCompaction(); } catch (IOException e) { throw new RuntimeException(e); } - logger_.debug("Finished compaction ..." + columnFamilyStore_.columnFamily_); + finally + { + logger_.debug("Finished compaction ..." + columnFamilyStore_.columnFamily_); + } } } @@ -164,13 +167,20 @@ class MinorCompactionManager implements IComponentShutdown compactor_.shutdownNow(); } - public void submitPeriodicCompaction(ColumnFamilyStore columnFamilyStore) - { - compactor_.scheduleWithFixedDelay(new FileCompactor(columnFamilyStore), MinorCompactionManager.intervalInMins_, + public void submitPeriodicCompaction(final ColumnFamilyStore columnFamilyStore) + { + Runnable runnable = new Runnable() // having to wrap Callable in Runnable is retarded but that's what the API insists on. + { + public void run() + { + new FileCompactor(columnFamilyStore).call(); + } + }; + compactor_.scheduleWithFixedDelay(runnable, MinorCompactionManager.intervalInMins_, MinorCompactionManager.intervalInMins_, TimeUnit.MINUTES); } - public Future submit(ColumnFamilyStore columnFamilyStore) + public Future submit(ColumnFamilyStore columnFamilyStore) { return compactor_.submit(new FileCompactor(columnFamilyStore)); } diff --git a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java index ed5efb5b3f..e3ddd845ab 100644 --- a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java +++ b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java @@ -3,14 +3,7 @@ package org.apache.cassandra.db; import java.io.File; import java.io.FileOutputStream; import java.io.IOException; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collection; -import java.util.Iterator; -import java.util.List; -import java.util.Random; -import java.util.Set; -import java.util.SortedSet; +import java.util.*; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; @@ -393,23 +386,56 @@ public class ColumnFamilyStoreTest extends ServerTest } @Test - public void testCompaction() throws IOException, ExecutionException, InterruptedException + public void testOneCompaction() throws IOException, ExecutionException, InterruptedException { Table table = Table.open("Table1"); ColumnFamilyStore store = table.getColumnFamilyStore("Standard1"); - 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); + Set inserted = new HashSet(); + for (int j = 0; j < 2; j++) { + String key = "0"; + RowMutation rm = new RowMutation("Table1", key); + rm.add("Standard1:0", new byte[0], j); + rm.apply(); + inserted.add(key); + store.forceBlockingFlush(); + assertEquals(table.getKeyRange("", "", 10000).size(), inserted.size()); + } + store.doCompaction(2); + assertEquals(table.getKeyRange("", "", 10000).size(), inserted.size()); + } + + @Test + public void testCompactions() throws IOException, ExecutionException, InterruptedException + { + // this test does enough rows to force multiple block indexes to be used + Table table = Table.open("Table1"); + ColumnFamilyStore store = table.getColumnFamilyStore("Standard1"); + + final int ROWS_PER_SSTABLE = 10; + Set inserted = new HashSet(); + for (int j = 0; j < (SSTable.indexInterval() * 3) / ROWS_PER_SSTABLE; j++) { + for (int i = 0; i < ROWS_PER_SSTABLE; i++) { + String key = String.valueOf(i % 2); RowMutation rm = new RowMutation("Table1", key); - rm.add("Standard1:A", new byte[0], epoch); + rm.add("Standard1:" + (i / 2), new byte[0], j * ROWS_PER_SSTABLE + i); rm.apply(); + inserted.add(key); } store.forceBlockingFlush(); + assertEquals(table.getKeyRange("", "", 10000).size(), inserted.size()); } - Future ft = MinorCompactionManager.instance().submit(store); - ft.get(); + while (true) + { + Future ft = MinorCompactionManager.instance().submit(store); + if (ft.get() == 0) + break; + } + if (store.getSSTableFilenames().size() > 1) + { + store.doCompaction(store.getSSTableFilenames().size()); + } + assertEquals(table.getKeyRange("", "", 10000).size(), inserted.size()); } @Test