merge from 0.4 branch

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@821137 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -5,6 +5,7 @@
  * Fix for serializing a row that only contains tombstones
    (CASSANDRA-458)
  * Fix for discarding unneeded commitlog segments (CASSANDRA-459)
+ * Add SnapshotBeforeCompaction configuration option (CASSANDRA-426)
 
 
 0.4.0

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -128,6 +128,8 @@ public class DatabaseDescriptor
     private static double commitLogSyncBatchMS_;
     private static int commitLogSyncPeriodMS_;
 
+    private static boolean snapshotBeforeCompaction_;
+
     static
     {
         try
@@ -305,18 +307,31 @@ public class DatabaseDescriptor
             String framedRaw = xmlUtils.getNodeValue("/Storage/ThriftFramedTransport");
             if (framedRaw != null)
             {
-                if (framedRaw.compareToIgnoreCase("true") == 0 || 
-                        framedRaw.compareToIgnoreCase("false") == 0)
+                if (framedRaw.equalsIgnoreCase("true") || framedRaw.equalsIgnoreCase("false"))
                 {
                     thriftFramed_ = Boolean.valueOf(framedRaw);
                 }
                 else
                 {
-                    throw new ConfigurationException("Unrecognized value " + 
-                            "for ThriftFramedTransport.  Use 'true' or 'false'."); 
+                    throw new ConfigurationException("Unrecognized value for ThriftFramedTransport.  Use 'true' or 'false'.");
+                }
+            }
+
+            /* snapshot-before-compaction.  defaults to false */
+            String sbc = xmlUtils.getNodeValue("/Storage/SnapshotBeforeCompaction");
+            if (sbc != null)
+            {
+                if (sbc.equalsIgnoreCase("true") || sbc.equalsIgnoreCase("false"))
+                {
+                    if (logger_.isDebugEnabled())
+                        logger_.debug("setting snapshotBeforeCompaction to " + sbc);
+                    snapshotBeforeCompaction_ = Boolean.valueOf(sbc);
+                }
+                else
+                {
+                    throw new ConfigurationException("Unrecognized value for SnapshotBeforeCompaction.  Use 'true' or 'false'.");
                 }
             }
-            
 
             /* Number of days to keep the memtable around w/o flushing */
             String lifetime = xmlUtils.getNodeValue("/Storage/MemtableFlushAfterMinutes");
@@ -973,4 +988,9 @@ public class DatabaseDescriptor
     {
         return bmtThreshold_;
     }
+
+    public static boolean isSnapshotBeforeCompaction()
+    {
+        return snapshotBeforeCompaction_;
+    }
 }

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -417,14 +417,9 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
     void forceBlockingFlush() throws IOException, ExecutionException, InterruptedException
     {
-        Memtable oldMemtable = getMemtableThreadSafe();
         Future<?> future = forceFlush();
         if (future != null)
             future.get();
-        /* this assert is not threadsafe -- the memtable could have been clean when forceFlush
-           checked it, but dirty now thanks to another thread.  But as long as we are only
-           calling this from single-threaded test code it is useful to have as a sanity check. */
-        assert oldMemtable.isFlushed() || oldMemtable.isClean(); 
     }
 
     public void forceFlushBinary()
@@ -833,6 +828,8 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
     */
     private int doFileCompaction(Collection<SSTableReader> sstables) throws IOException
     {
+        if (DatabaseDescriptor.isSnapshotBeforeCompaction())
+            Table.open(table_).snapshot("compact-" + columnFamily_);
         logger_.info("Compacting [" + StringUtils.join(sstables, ",") + "]");
         String compactionFileLocation = DatabaseDescriptor.getDataFileLocationForTable(table_, getExpectedCompactedFileSize(sstables));
         // If the compaction file path is null that means we have no space left for this compaction.
@@ -1285,6 +1282,19 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
      */
     public void snapshot(String snapshotName) throws IOException
     {
+        try
+        {
+            forceBlockingFlush();
+        }
+        catch (ExecutionException e)
+        {
+            throw new RuntimeException(e);
+        }
+        catch (InterruptedException e)
+        {
+            throw new AssertionError(e);
+        }
+
         for (SSTableReader ssTable : ssTables_)
         {
             // mkdir

src/java/org/apache/cassandra/service/ThriftValidation.java

@@ -95,7 +95,7 @@ public class ThriftValidation
         else
         {
             if (column_path.super_column == null)
-                throw new InvalidRequestException("column parameter is not optional for super CF " + column_path.column_family);
+                throw new InvalidRequestException("supercolumn parameter is not optional for super CF " + column_path.column_family);
         }
         if (column_path.column != null)
         {