From de2673759aef6a2088588101da55153e22a6987f Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Thu, 2 Apr 2009 20:48:02 +0000 Subject: [PATCH] rename stageOrderedCompaction -> getCompactionBuckets. clean up code to make the algorithm more clear, and to allow grouping with more than just the previous bucket. add getCompactionBucketsTest git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@761423 13f79535-47bb-0310-9956-ffa450edef68 --- .../cassandra/db/ColumnFamilyStore.java | 75 +++++++++---------- .../cassandra/db/ColumnFamilyStoreTest.java | 65 ++++++++++++++-- 2 files changed, 93 insertions(+), 47 deletions(-) diff --git a/src/org/apache/cassandra/db/ColumnFamilyStore.java b/src/org/apache/cassandra/db/ColumnFamilyStore.java index a7dabf410d..1e59bb32b2 100644 --- a/src/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/org/apache/cassandra/db/ColumnFamilyStore.java @@ -30,6 +30,7 @@ import java.util.Map; import java.util.PriorityQueue; import java.util.Set; import java.util.StringTokenizer; +import java.util.Arrays; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicBoolean; @@ -707,48 +708,47 @@ public class ColumnFamilyStore } /* - * Stage the compactions , compact similar size files. - * This fn figures out the files close enough by size and if they - * are greater than the threshold then compacts. + * Group files of similar size into buckets. */ - Map> stageOrderedCompaction(List files) + static Set> getCompactionBuckets(List files, long min) { - // Sort the files based on the generation ID - Collections.sort(files, new FileNameComparator(FileNameComparator.Ascending)); - Map> buckets = new HashMap>(); - int maxBuckets = 1000; - long averages[] = new long[maxBuckets]; - long min = 50L*1024L*1024L; - Integer i = 0; - for(String file : files) + Map, Long> buckets = new HashMap, Long>(); + for(String fname : files) { - File f = new File(file); + File f = new File(fname); long size = f.length(); - if ( (size > averages[i]/2 && size < 3*averages[i]/2) || ( size < min && averages[i] < min )) - { - averages[i] = (averages[i] + size) / 2 ; - List fileList = buckets.get(i); - if(fileList == null) - { - fileList = new ArrayList(); - buckets.put(i, fileList); - } - fileList.add(file); - } - else + + 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 : new ArrayList>(buckets.keySet())) { - if( i >= maxBuckets ) - break; - i++; - List fileList = new ArrayList(); - buckets.put(i, fileList); - fileList.add(file); - averages[i] = size; + 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(fname); + 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(fname); + buckets.put(bucket, size); } } - return buckets; + + return buckets.keySet(); } - + /* * Break the files into buckets and then compact. */ @@ -759,11 +759,8 @@ public class ColumnFamilyStore try { int count = 0; - Map> buckets = stageOrderedCompaction(files); - Set keySet = buckets.keySet(); - for(Integer key : keySet) - { - List fileList = buckets.get(key); + for(List fileList : getCompactionBuckets(files, 50L*1024L*1024L)) + { Collections.sort( fileList , new FileNameComparator( FileNameComparator.Ascending)); if(fileList.size() >= threshHold_ ) { diff --git a/test/org/apache/cassandra/db/ColumnFamilyStoreTest.java b/test/org/apache/cassandra/db/ColumnFamilyStoreTest.java index ef3dcd4d5c..8ef214ba55 100644 --- a/test/org/apache/cassandra/db/ColumnFamilyStoreTest.java +++ b/test/org/apache/cassandra/db/ColumnFamilyStoreTest.java @@ -4,17 +4,24 @@ import org.apache.cassandra.ServerTest; import org.testng.annotations.Test; import java.io.IOException; +import java.io.File; +import java.io.FileOutputStream; import java.util.Collection; import java.util.List; import java.util.Random; import java.util.Arrays; +import java.util.ArrayList; +import java.util.Set; import java.util.concurrent.FutureTask; import java.util.concurrent.Future; import java.util.concurrent.ExecutionException; -public class ColumnFamilyStoreTest extends ServerTest { +public class ColumnFamilyStoreTest extends ServerTest +{ static byte[] bytes1, bytes2; - static { + + static + { Random random = new Random(); bytes1 = new byte[1024]; bytes2 = new byte[128]; @@ -31,14 +38,14 @@ public class ColumnFamilyStoreTest extends ServerTest { { String key = Integer.toString(i); RowMutation rm; - for ( int j = 0; j < 8; ++j ) + for (int j = 0; j < 8; ++j) { byte[] bytes = j % 2 == 0 ? bytes1 : bytes2; rm = new RowMutation("Table1", key); rm.add("Standard1:" + "Column-" + j, bytes, j); rm.apply(); - for ( int k = 0; k < 4; ++k ) + for (int k = 0; k < 4; ++k) { bytes = (j + k) % 2 == 0 ? bytes1 : bytes2; rm = new RowMutation("Table1", key); @@ -55,7 +62,9 @@ public class ColumnFamilyStoreTest extends ServerTest { // wait for flush to finish Future f = MemtableManager.instance().flusher_.submit(new Runnable() { - public void run() {} + public void run() + { + } }); f.get(); @@ -65,7 +74,7 @@ public class ColumnFamilyStoreTest extends ServerTest { private void validateBytes(Table table) throws ColumnFamilyNotDefinedException, IOException { - for ( int i = 900; i < 1000; ++i ) + for (int i = 900; i < 1000; ++i) { String key = Integer.toString(i); ColumnFamily cf; @@ -99,7 +108,8 @@ public class ColumnFamilyStoreTest extends ServerTest { } @Test - public void testRemove() throws IOException, ColumnFamilyNotDefinedException { + public void testRemove() throws IOException, ColumnFamilyNotDefinedException + { Table table = Table.open("Table1"); ColumnFamilyStore store = table.getColumnFamilyStore("Standard1"); RowMutation rm; @@ -122,7 +132,8 @@ public class ColumnFamilyStoreTest extends ServerTest { } @Test - public void testRemoveSuperColumn() throws IOException, ColumnFamilyNotDefinedException { + public void testRemoveSuperColumn() throws IOException, ColumnFamilyNotDefinedException + { Table table = Table.open("Table1"); ColumnFamilyStore store = table.getColumnFamilyStore("Super1"); RowMutation rm; @@ -152,4 +163,42 @@ public class ColumnFamilyStoreTest extends ServerTest { assert subColumns.iterator().next().timestamp() == 0; assert ColumnFamilyStore.removeDeleted(resolved).getColumnCount() == 0; } + + @Test + public void testGetCompactionBuckets() throws IOException + { + // create files 20 40 60 ... 180 + List small = new ArrayList(); + List med = new ArrayList(); + List all = new ArrayList(); + + String fname; + fname = createFile(20); + small.add(fname); + all.add(fname); + fname = createFile(40); + small.add(fname); + all.add(fname); + + for (int i = 60; i <= 140; i += 20) + { + fname = createFile(i); + med.add(fname); + all.add(fname); + } + + Set> buckets = ColumnFamilyStore.getCompactionBuckets(all, 50); + assert buckets.contains(small); + assert buckets.contains(med); + } + + private String createFile(int nBytes) throws IOException + { + File f = File.createTempFile("bucket_test", ""); + FileOutputStream fos = new FileOutputStream(f); + byte[] bytes = new byte[nBytes]; + fos.write(bytes); + fos.close(); + return f.getAbsolutePath(); + } }