make row size at which to drop to incremental compaction configurable

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



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

NEWS.txt

@@ -13,6 +13,7 @@ Features
     - DatacenterShardStrategy is ready for use, enabling 
       ConsitencyLevel.DCQUORUM and DCQUORUMSYNC.  See comments in
       `cassandra.yaml.`
+    - row size limit increased from 2GB to 2 billion columns
 
 Configuraton
 ------------
@@ -35,6 +36,7 @@ Configuraton
     - EndPointSnitch was renamed to RackInferringSnitch.  A new SimpleSnitch
       has been added.
     - auto_bootstrap now defaults to true
+    - row_warning_threshold_in_mb replaced with in_memory_compaction_limit_in_mb
 
 JMX
 ---

conf/cassandra.yaml

@@ -97,7 +97,7 @@ flush_data_buffer_size_in_mb: 32
 flush_index_buffer_size_in_mb: 8
 
 column_index_size_in_kb: 64
-row_warning_threshold_in_mb: 512
+in_memory_compaction_limit_in_mb: 128
 
 # commit log
 commitlog_directory: /var/lib/cassandra/commitlog

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

@@ -59,7 +59,7 @@ public class Config {
     
     /* if the size of columns or super-columns are more than this, indexing will kick in */
     public Integer column_index_size_in_kb = 64;
-    public Long row_warning_threshold_in_mb = new Long(512);
+    public Integer in_memory_compaction_limit_in_mb = 256;
     
     public String[] data_file_directories;
     

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

@@ -234,12 +234,6 @@ public class Converter {
                 conf.column_index_size_in_kb = Integer.parseInt(columnIndexSize);
             }
 
-            String rowWarning = xmlUtils.getNodeValue("/Storage/RowWarningThresholdInMB");
-            if (rowWarning != null)
-            {
-                conf.row_warning_threshold_in_mb = Long.parseLong(rowWarning);
-            }
-            
             conf.data_file_directories = xmlUtils.getNodeValues("/Storage/DataFileDirectories/DataFileDirectory");
             
             conf.commitlog_directory = xmlUtils.getNodeValue("/Storage/CommitLogDirectory");

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

@@ -26,8 +26,6 @@ import org.apache.cassandra.db.clock.TimestampReconciler;
 import org.apache.cassandra.db.commitlog.CommitLog;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.db.marshal.BytesType;
-import org.apache.cassandra.db.marshal.TimeUUIDType;
-import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.db.migration.Migration;
 import org.apache.cassandra.dht.IPartitioner;
 import org.apache.cassandra.locator.AbstractReplicationStrategy;
@@ -47,7 +45,6 @@ import org.yaml.snakeyaml.error.YAMLException;
 import java.io.File;
 import java.io.FileFilter;
 import java.io.FileInputStream;
-import java.io.FilenameFilter;
 import java.io.IOError;
 import java.io.IOException;
 import java.io.InputStream;
@@ -272,9 +269,9 @@ public class DatabaseDescriptor
                 throw new ConfigurationException("memtable_operations_in_millions must be a positive double");
             }
             
-            if (conf.row_warning_threshold_in_mb != null && conf.row_warning_threshold_in_mb <= 0)
+            if (conf.in_memory_compaction_limit_in_mb != null && conf.in_memory_compaction_limit_in_mb <= 0)
             {
-                throw new ConfigurationException("row_warning_threshold_in_mb must be a positive integer");
+                throw new ConfigurationException("in_memory_compaction_limit_in_mb must be a positive integer");
             }
             
             /* data file and commit log directories. they get created later, when they're needed. */
@@ -399,7 +396,6 @@ public class DatabaseDescriptor
                     // see if there are other directories present.
                     int dirCount = dataPath.listFiles(new FileFilter()
                     {
-                        @Override
                         public boolean accept(File pathname)
                         {
                             return pathname.isDirectory();
@@ -848,9 +844,9 @@ public class DatabaseDescriptor
             return conf.memtable_flush_writers;
     }
 
-    public static long getRowWarningThreshold()
+    public static long getInMemoryCompactionLimit()
     {
-        return conf.row_warning_threshold_in_mb * 1024 * 1024;
+        return conf.in_memory_compaction_limit_in_mb * 1024 * 1024;
     }
     
     public static String[] getAllDataFileLocations()

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

@@ -354,8 +354,6 @@ public class CompactionManager implements CompactionManagerMBean
                 totalkeysWritten++;
 
                 long rowsize = writer.getFilePointer() - prevpos;
-                if (rowsize > DatabaseDescriptor.getRowWarningThreshold())
-                    logger.warn("Large row " + row.key.key + " in " + cfs.getColumnFamilyName() + " " + rowsize + " bytes");
                 cfs.addToCompactedRowStats(rowsize);
             }
             validator.complete();

src/java/org/apache/cassandra/io/CompactionIterator.java

@@ -28,6 +28,7 @@ import java.util.List;
 import java.util.ArrayList;
 import java.util.Iterator;
 
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.io.sstable.SSTableIdentityIterator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -127,10 +128,13 @@ public class CompactionIterator extends ReducingIterator<SSTableIdentityIterator
         {
             rowSize += row.getDataSize();
         }
-        if (rowSize > 512 * 1024 * 1024)
+
+        if (rowSize > DatabaseDescriptor.getInMemoryCompactionLimit())
+        {
+            logger.info("Compacting large row (" + rowSize + " bytes) incrementally");
             return new LazilyCompactedRow(rows, major, gcBefore);
-        else
-            return new PrecompactedRow(rows, major, gcBefore);
+        }
+        return new PrecompactedRow(rows, major, gcBefore);
     }
 
     public void close() throws IOException