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

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
This commit is contained in:
Jonathan Ellis 2009-04-02 20:48:02 +00:00
parent 8ef8b192dd
commit de2673759a
2 changed files with 93 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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<Integer, List<String>> stageOrderedCompaction(List<String> files)
static Set<List<String>> getCompactionBuckets(List<String> files, long min)
{
// Sort the files based on the generation ID
Collections.sort(files, new FileNameComparator(FileNameComparator.Ascending));
Map<Integer, List<String>> buckets = new HashMap<Integer, List<String>>();
int maxBuckets = 1000;
long averages[] = new long[maxBuckets];
long min = 50L*1024L*1024L;
Integer i = 0;
for(String file : files)
Map<List<String>, Long> buckets = new HashMap<List<String>, 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<String> fileList = buckets.get(i);
if(fileList == null)
{
fileList = new ArrayList<String>();
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<String> bucket : new ArrayList<List<String>>(buckets.keySet()))
{
if( i >= maxBuckets )
break;
i++;
List<String> fileList = new ArrayList<String>();
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<String> bucket = new ArrayList<String>();
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<Integer, List<String>> buckets = stageOrderedCompaction(files);
Set<Integer> keySet = buckets.keySet();
for(Integer key : keySet)
{
List<String> fileList = buckets.get(key);
for(List<String> fileList : getCompactionBuckets(files, 50L*1024L*1024L))
{
Collections.sort( fileList , new FileNameComparator( FileNameComparator.Ascending));
if(fileList.size() >= threshHold_ )
{

65
test/org/apache/cassandra/db/ColumnFamilyStoreTest.java
View File

@ -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<String> small = new ArrayList<String>();
List<String> med = new ArrayList<String>();
List<String> all = new ArrayList<String>();
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<List<String>> 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();
}
}