add CompactionStrategy options for Cassandra storage configuration

titan-cassandra/src/main/java/com/thinkaurelius/titan/diskstorage/cassandra/AbstractCassandraStoreManager.java

@@ -92,6 +92,17 @@ public static Partitioner getPartitioner(String className) {
             "here takes precedence over one set with " + ConfigElement.getPath(REPLICATION_FACTOR),
             ConfigOption.Type.FIXED, String[].class);
 
+    public static final ConfigOption<String> COMPACTION_STRATEGY =
+            new ConfigOption<String>(CASSANDRA_NS, "compaction-strategy-class",
+            "The compaction strategy to use for Titan tables",
+            ConfigOption.Type.FIXED, "org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy");
+
+    public static final ConfigOption<String[]> COMPACTION_OPTIONS =
+            new ConfigOption<String[]>(CASSANDRA_NS, "compaction-strategy-options",
+            "Compaction strategy options.  This list is interpreted as a " +
+            "map.  It must have an even number of elements in [key,val,key,val,...] form.",
+            ConfigOption.Type.FIXED, String[].class);
+
     // Compression
     public static final ConfigOption<Boolean> CF_COMPRESSION =
             new ConfigOption<Boolean>(CASSANDRA_NS, "compression",
@@ -144,6 +155,7 @@ public static Partitioner getPartitioner(String className) {
 
     protected final String keySpaceName;
     protected final Map<String, String> strategyOptions;
+    protected final Map<String, String> compactionOptions;
 
     protected final boolean compressionEnabled;
     protected final int compressionChunkSizeKB;
@@ -189,6 +201,23 @@ public AbstractCassandraStoreManager(Configuration config) {
         } else {
             this.strategyOptions = ImmutableMap.of("replication_factor", String.valueOf(config.get(REPLICATION_FACTOR)));
         }
+
+        if (config.has(COMPACTION_OPTIONS)) {
+            String[] options = config.get(COMPACTION_OPTIONS);
+
+            if (options.length % 2 != 0)
+                throw new IllegalArgumentException(COMPACTION_OPTIONS.getName() + " should have even number of elements.");
+
+            Map<String, String> converted = new HashMap<String, String>(options.length / 2);
+
+            for (int i = 0; i < options.length; i += 2) {
+                converted.put(options[i], options[i + 1]);
+            }
+
+            this.compactionOptions = ImmutableMap.copyOf(converted);
+        } else {
+            this.compactionOptions = ImmutableMap.of();
+        }
     }
 
     public final Partitioner getPartitioner() {

titan-cassandra/src/main/java/com/thinkaurelius/titan/diskstorage/cassandra/astyanax/AstyanaxStoreManager.java

@@ -439,6 +439,8 @@ private void ensureColumnFamilyExists(String name, String comparator) throws Bac
                         cl.makeColumnFamilyDefinition()
                                 .setName(name)
                                 .setKeyspace(keySpaceName)
+                                .setCompactionStrategy(storageConfig.get(COMPACTION_STRATEGY))
+                                .setCompactionStrategyOptions(compactionOptions)
                                 .setComparatorType(comparator);
 
                 ImmutableMap.Builder<String, String> compressionOptions = new ImmutableMap.Builder<String, String>();

titan-cassandra/src/main/java/com/thinkaurelius/titan/diskstorage/cassandra/embedded/CassandraEmbeddedStoreManager.java

@@ -285,6 +285,12 @@ private void ensureColumnFamilyExists(String keyspaceName, String columnfamilyNa
 
         // Column Family not found; create it
         CFMetaData cfm = new CFMetaData(keyspaceName, columnfamilyName, ColumnFamilyType.Standard, CellNames.fromAbstractType(comparator, true));
+        try {
+            cfm.compactionStrategyClass(CFMetaData.createCompactionStrategy(storageConfig.get(COMPACTION_STRATEGY)));
+            cfm.compactionStrategyOptions(compactionOptions);
+        } catch (ConfigurationException e) {
+            throw new PermanentBackendException("Failed to create column family metadata for " + keyspaceName + ":" + columnfamilyName, e);
+        }
 
         // Hard-coded caching settings
         if (columnfamilyName.startsWith(Backend.EDGESTORE_NAME)) {

titan-cassandra/src/main/java/com/thinkaurelius/titan/diskstorage/cassandra/thrift/CassandraThriftStoreManager.java

@@ -540,6 +540,8 @@ private void createColumnFamily(Cassandra.Client client,
         createColumnFamily.setName(cfName);
         createColumnFamily.setKeyspace(ksName);
         createColumnFamily.setComparator_type(comparator);
+        createColumnFamily.setCompaction_strategy(storageConfig.get(COMPACTION_STRATEGY));
+        createColumnFamily.setCompaction_strategy_options(this.compactionOptions);
 
         ImmutableMap.Builder<String, String> compressionOptions = new ImmutableMap.Builder<String, String>();