From b55b26d790bee95fa53078e85d8aa5dc86fc6193 Mon Sep 17 00:00:00 2001
From: Jonathan Ellis <jbellis@apache.org>
Date: Wed, 17 Feb 2010 23:22:23 +0000
Subject: [PATCH] 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
---
 CHANGES.txt                                   |  4 ++
 conf/storage-conf.xml                         |  2 +-
 .../apache/cassandra/db/BinaryMemtable.java   | 16 ++---
 .../cassandra/db/ColumnFamilyStore.java       | 16 ++---
 .../org/apache/cassandra/db/IFlushable.java   |  5 --
 .../org/apache/cassandra/db/Memtable.java     | 68 ++++++-------------
 6 files changed, 40 insertions(+), 71 deletions(-)

diff --git a/CHANGES.txt b/CHANGES.txt
index 6d6d0945e3..526d600562 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -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
diff --git a/conf/storage-conf.xml b/conf/storage-conf.xml
index 324babbc63..f08b5e0b21 100644
--- a/conf/storage-conf.xml
+++ b/conf/storage-conf.xml
@@ -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
diff --git a/src/java/org/apache/cassandra/db/BinaryMemtable.java b/src/java/org/apache/cassandra/db/BinaryMemtable.java
index 1ca76579ad..b6d6c33f99 100644
--- a/src/java/org/apache/cassandra/db/BinaryMemtable.java
+++ b/src/java/org/apache/cassandra/db/BinaryMemtable.java
@@ -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
 {
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index c120e9dd66..1db1def96e 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -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
diff --git a/src/java/org/apache/cassandra/db/IFlushable.java b/src/java/org/apache/cassandra/db/IFlushable.java
index a038a83515..71da1cd57b 100644
--- a/src/java/org/apache/cassandra/db/IFlushable.java
+++ b/src/java/org/apache/cassandra/db/IFlushable.java
@@ -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);
diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java
index 14f9ec0cc4..f4d0499f96 100644
--- a/src/java/org/apache/cassandra/db/Memtable.java
+++ b/src/java/org/apache/cassandra/db/Memtable.java
@@ -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()