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

use a sorted map for memtable contents to make range queries not have to sort every time 

patch by jbellis; reviewed by Stu Hood for CASSANDRA-799

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@911223 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-02-17 23:22:23 +00:00
parent a4e9b14d21
commit b55b26d790
6 changed files with 40 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
CHANGES.txt
View File

@ -25,6 +25,10 @@ dev
get_string_list_property
* jmx interface for tracking operation mode and streams in general.
(CASSANDRA-709)
* keep memtables in sorted order to improve range query performance
(CASSANDRA-799)
* use while loop instead of recursion when trimming sstables compaction list
to avoid blowing stack in pathological cases (CASSANDRA-804)
0.5.1

2
conf/storage-conf.xml
View File

@ -308,7 +308,7 @@
~ ColumnFamily before flushing to disk. This is also a per-memtable
~ setting. Use with MemtableThroughputInMB to tune memory usage.
-->
<MemtableOperationsInMillions>0.1</MemtableOperationsInMillions>
<MemtableOperationsInMillions>0.3</MemtableOperationsInMillions>
<!--
~ The maximum time to leave a dirty memtable unflushed.
~ (While any affected columnfamilies have unflushed data from a

16
src/java/org/apache/cassandra/db/BinaryMemtable.java
View File

@ -19,26 +19,22 @@
package org.apache.cassandra.db;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.cassandra.io.SSTableWriter;
import org.apache.cassandra.io.SSTableReader;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.log4j.Logger;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.SSTableReader;
import org.apache.cassandra.io.SSTableWriter;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.WrappedRunnable;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
import org.apache.cassandra.dht.IPartitioner;
public class BinaryMemtable implements IFlushable
{

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

@ -90,14 +90,14 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
Runtime.getRuntime().availableProcessors(),
Integer.MAX_VALUE,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(1 + 2 * Runtime.getRuntime().availableProcessors()),
new LinkedBlockingQueue<Runnable>(1 + Runtime.getRuntime().availableProcessors()),
new NamedThreadFactory("FLUSH-SORTER-POOL"));
private static ExecutorService flushWriter_
= new JMXEnabledThreadPoolExecutor(DatabaseDescriptor.getAllDataFileLocations().length,
= new JMXEnabledThreadPoolExecutor(1,
DatabaseDescriptor.getAllDataFileLocations().length,
Integer.MAX_VALUE,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
new LinkedBlockingQueue<Runnable>(1 + 2 * DatabaseDescriptor.getAllDataFileLocations().length),
new NamedThreadFactory("FLUSH-WRITER-POOL"));
private static ExecutorService commitLogUpdater_ = new JMXEnabledThreadPoolExecutor("MEMTABLE-POST-FLUSHER");
@ -641,7 +641,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
* flushing thread finishes sorting, which will almost always be longer than any of the flushSorter threads proper
* (since, by definition, it started last).
*/
Condition submitFlush(final IFlushable flushable)
Condition submitFlush(IFlushable flushable)
{
logger_.info("Enqueuing flush of " + flushable);
final Condition condition = new SimpleCondition();
@ -692,12 +692,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
}
public Iterator<DecoratedKey> memtableKeyIterator() throws ExecutionException, InterruptedException
public Iterator<DecoratedKey> memtableKeyIterator(DecoratedKey startWith) throws ExecutionException, InterruptedException
{
Table.flusherLock.readLock().lock();
try
{
return memtable_.getKeyIterator();
return memtable_.getKeyIterator(startWith);
}
finally
{
@ -928,11 +928,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
};
// current memtable keys. have to go through the CFS api for locking.
iterators.add(Iterators.filter(memtableKeyIterator(), p));
iterators.add(Iterators.filter(memtableKeyIterator(startWith), p));
// historical memtables
for (Memtable memtable : ColumnFamilyStore.getUnflushedMemtables(columnFamily_))
{
iterators.add(Iterators.filter(memtable.getKeyIterator(), p));
iterators.add(Iterators.filter(memtable.getKeyIterator(startWith), p));
}
// sstables

5
src/java/org/apache/cassandra/db/IFlushable.java
View File

@ -21,14 +21,9 @@ package org.apache.cassandra.db;
*/
import java.io.IOException;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.locks.Condition;
import org.apache.cassandra.io.SSTableReader;
public interface IFlushable
{
public void flushAndSignal(Condition condition, ExecutorService sorter, ExecutorService writer);

68
src/java/org/apache/cassandra/db/Memtable.java
View File

@ -20,25 +20,24 @@ package org.apache.cassandra.db;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import org.apache.log4j.Logger;
import org.apache.commons.lang.ArrayUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.SSTableReader;
import org.apache.cassandra.io.SSTableWriter;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.DestructivePQIterator;
import org.apache.cassandra.db.filter.*;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.SSTableReader;
import org.apache.cassandra.io.SSTableWriter;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.WrappedRunnable;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
import org.apache.log4j.Logger;
public class Memtable implements Comparable<Memtable>, IFlushable
{
@ -55,7 +54,7 @@ public class Memtable implements Comparable<Memtable>, IFlushable
private final String table;
private final String columnfamilyName;
private final long creationTime;
private final NonBlockingHashMap<DecoratedKey, ColumnFamily> columnFamilies = new NonBlockingHashMap<DecoratedKey, ColumnFamily>();
private final ConcurrentNavigableMap<DecoratedKey, ColumnFamily> columnFamilies = new ConcurrentSkipListMap<DecoratedKey, ColumnFamily>();
private final IPartitioner partitioner = StorageService.getPartitioner();
Memtable(String table, String cfName)
@ -144,30 +143,21 @@ public class Memtable implements Comparable<Memtable>, IFlushable
return builder.toString();
}
private List<DecoratedKey> getSortedKeys()
{
logger.info("Sorting " + this);
// sort keys in the order they would be in when decorated
ArrayList<DecoratedKey> orderedKeys = new ArrayList<DecoratedKey>(columnFamilies.keySet());
Collections.sort(orderedKeys);
return orderedKeys;
}
private SSTableReader writeSortedContents(List<DecoratedKey> sortedKeys) throws IOException
private SSTableReader writeSortedContents() throws IOException
{
logger.info("Writing " + this);
ColumnFamilyStore cfStore = Table.open(table).getColumnFamilyStore(columnfamilyName);
SSTableWriter writer = new SSTableWriter(cfStore.getFlushPath(), columnFamilies.size(), StorageService.getPartitioner());
DataOutputBuffer buffer = new DataOutputBuffer();
for (DecoratedKey key : sortedKeys)
for (Map.Entry<DecoratedKey, ColumnFamily> entry : columnFamilies.entrySet())
{
buffer.reset();
ColumnFamily columnFamily = columnFamilies.get(key);
/* serialize the cf with column indexes */
ColumnFamily.serializer().serializeWithIndexes(columnFamily, buffer);
ColumnFamily.serializer().serializeWithIndexes(entry.getValue(), buffer);
/* Now write the key and value to disk */
writer.append(key, buffer);
writer.append(entry.getKey(), buffer);
}
SSTableReader ssTable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table, columnfamilyName));
@ -178,21 +168,14 @@ public class Memtable implements Comparable<Memtable>, IFlushable
public void flushAndSignal(final Condition condition, ExecutorService sorter, final ExecutorService writer)
{
ColumnFamilyStore.getMemtablesPendingFlushNotNull(columnfamilyName).add(this); // it's ok for the MT to briefly be both active and pendingFlush
sorter.submit(new Runnable()
writer.submit(new WrappedRunnable()
{
public void run()
public void runMayThrow() throws IOException
{
final List<DecoratedKey> sortedKeys = getSortedKeys();
writer.submit(new WrappedRunnable()
{
public void runMayThrow() throws IOException
{
ColumnFamilyStore cfs = Table.open(table).getColumnFamilyStore(columnfamilyName);
cfs.addSSTable(writeSortedContents(sortedKeys));
ColumnFamilyStore.getMemtablesPendingFlushNotNull(columnfamilyName).remove(Memtable.this);
condition.signalAll();
}
});
ColumnFamilyStore cfs = Table.open(table).getColumnFamilyStore(columnfamilyName);
cfs.addSSTable(writeSortedContents());
ColumnFamilyStore.getMemtablesPendingFlushNotNull(columnfamilyName).remove(Memtable.this);
condition.signalAll();
}
});
}
@ -202,18 +185,9 @@ public class Memtable implements Comparable<Memtable>, IFlushable
return "Memtable(" + columnfamilyName + ")@" + hashCode();
}
public Iterator<DecoratedKey> getKeyIterator()
public Iterator<DecoratedKey> getKeyIterator(DecoratedKey startWith)
{
// even though we are using NBHM, it is okay to use size() twice here, since size() will never decrease
// w/in a single memtable's lifetime
if (columnFamilies.size() == 0)
{
// cannot create a PQ of size zero (wtf?)
return Arrays.asList(new DecoratedKey[0]).iterator();
}
PriorityQueue<DecoratedKey> pq = new PriorityQueue<DecoratedKey>(columnFamilies.size());
pq.addAll(columnFamilies.keySet());
return new DestructivePQIterator<DecoratedKey>(pq);
return columnFamilies.navigableKeySet().tailSet(startWith).iterator();
}
public boolean isClean()