From 8e360f80f4454c1c40edfefdf44b92bfbb9be6f1 Mon Sep 17 00:00:00 2001
From: Jonathan Ellis <jbellis@apache.org>
Date: Tue, 11 Mar 2014 13:00:28 -0500
Subject: [PATCH] Allow cassandra-stress to set compaction strategy options
 patch by Benedict Elliott Smith; reviewed by Russell Spitzer for
 CASSANDRA-6451

---
 CHANGES.txt                                   |   1 +
 .../stress/settings/OptionCompaction.java     |  62 ++++++++++
 .../stress/settings/OptionMulti.java          |  62 +++++++++-
 .../stress/settings/OptionReplication.java    | 112 ++++++------------
 .../stress/settings/OptionSimple.java         |  59 ++++++---
 .../stress/settings/SettingsCommandMixed.java |   2 +-
 .../stress/settings/SettingsSchema.java       |  32 ++---
 7 files changed, 213 insertions(+), 117 deletions(-)
 create mode 100644 tools/stress/src/org/apache/cassandra/stress/settings/OptionCompaction.java

diff --git a/CHANGES.txt b/CHANGES.txt
index 607e2dc63f..06331adeb8 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 2.1.0-beta2
+ * Allow cassandra-stress to set compaction strategy options (CASSANDRA-6451)
  * Add broadcast_rpc_address option to cassandra.yaml (CASSANDRA-5899)
  * Auto reload GossipingPropertyFileSnitch config (CASSANDRA-5897)
  * Fix overflow of memtable_total_space_in_mb (CASSANDRA-6573)
diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/OptionCompaction.java b/tools/stress/src/org/apache/cassandra/stress/settings/OptionCompaction.java
new file mode 100644
index 0000000000..da74e430e8
--- /dev/null
+++ b/tools/stress/src/org/apache/cassandra/stress/settings/OptionCompaction.java
@@ -0,0 +1,62 @@
+package org.apache.cassandra.stress.settings;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import com.google.common.base.Function;
+
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.exceptions.ConfigurationException;
+
+/**
+ * For specifying replication options
+ */
+class OptionCompaction extends OptionMulti
+{
+
+    private final OptionSimple strategy = new OptionSimple("strategy=", new StrategyAdapter(), null, "The compaction strategy to use", false);
+
+    public OptionCompaction()
+    {
+        super("compaction", "Define the compaction strategy and any parameters", true);
+    }
+
+    public String getStrategy()
+    {
+        return strategy.value();
+    }
+
+    public Map<String, String> getOptions()
+    {
+        return extraOptions();
+    }
+
+    protected List<? extends Option> options()
+    {
+        return Arrays.asList(strategy);
+    }
+
+    @Override
+    public boolean happy()
+    {
+        return true;
+    }
+
+    private static final class StrategyAdapter implements Function<String, String>
+    {
+
+        public String apply(String name)
+        {
+            try
+            {
+                CFMetaData.createCompactionStrategy(name);
+            } catch (ConfigurationException e)
+            {
+                throw new IllegalArgumentException("Invalid compaction strategy: " + name);
+            }
+            return name;
+        }
+    }
+
+}
diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/OptionMulti.java b/tools/stress/src/org/apache/cassandra/stress/settings/OptionMulti.java
index 1901587276..7074dc6a69 100644
--- a/tools/stress/src/org/apache/cassandra/stress/settings/OptionMulti.java
+++ b/tools/stress/src/org/apache/cassandra/stress/settings/OptionMulti.java
@@ -22,7 +22,11 @@ package org.apache.cassandra.stress.settings;
 
 
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -39,21 +43,34 @@ abstract class OptionMulti extends Option
         @Override
         public List<? extends Option> options()
         {
-            return OptionMulti.this.options();
+            if (collectAsMap == null)
+                return OptionMulti.this.options();
+
+            List<Option> options = new ArrayList<>(OptionMulti.this.options());
+            options.add(collectAsMap);
+            return options;
         }
     }
 
     protected abstract List<? extends Option> options();
 
+    public Map<String, String> extraOptions()
+    {
+        return collectAsMap == null ? new HashMap<String, String>() : collectAsMap.options;
+    }
+
     private final String name;
     private final Pattern pattern;
     private final String description;
     private final Delegate delegate = new Delegate();
-    public OptionMulti(String name, String description)
+    private final CollectAsMap collectAsMap;
+
+    public OptionMulti(String name, String description, boolean collectExtraOptionsInMap)
     {
         this.name = name;
         pattern = Pattern.compile(name + "\\((.*)\\)", Pattern.CASE_INSENSITIVE);
         this.description = description;
+        this.collectAsMap = collectExtraOptionsInMap ? new CollectAsMap() : null;
     }
 
     @Override
@@ -70,7 +87,10 @@ abstract class OptionMulti extends Option
                 throw new IllegalArgumentException("Invalid " + name + " specification: " + param);
             last = m.end();
             if (!delegate.accept(m.group()))
+            {
+
                 throw new IllegalArgumentException("Invalid " + name + " specification: " + m.group());
+            }
         }
         return true;
     }
@@ -125,4 +145,42 @@ abstract class OptionMulti extends Option
         return delegate.happy();
     }
 
+    private static final class CollectAsMap extends Option
+    {
+
+        static final String description = "Extra options";
+        Map<String, String> options = new LinkedHashMap<>();
+
+        boolean accept(String param)
+        {
+            String[] args = param.split("=");
+            if (args.length == 2 && args[1].length() > 0 && args[0].length() > 0)
+            {
+                options.put(args[0], args[1]);
+                return true;
+            }
+            return false;
+        }
+
+        boolean happy()
+        {
+            return true;
+        }
+
+        String shortDisplay()
+        {
+            return "[<option 1..N>=?]";
+        }
+
+        String longDisplay()
+        {
+            return GroupedOptions.formatLong(shortDisplay(), description);
+        }
+
+        List<String> multiLineDisplay()
+        {
+            return Collections.emptyList();
+        }
+    };
+
 }
diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/OptionReplication.java b/tools/stress/src/org/apache/cassandra/stress/settings/OptionReplication.java
index 3b8e9d8678..88665ab2e4 100644
--- a/tools/stress/src/org/apache/cassandra/stress/settings/OptionReplication.java
+++ b/tools/stress/src/org/apache/cassandra/stress/settings/OptionReplication.java
@@ -22,86 +22,42 @@ package org.apache.cassandra.stress.settings;
 
 
 import java.util.Arrays;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
+
+import com.google.common.base.Function;
 
 import org.apache.cassandra.locator.AbstractReplicationStrategy;
 
 /**
  * For specifying replication options
  */
-class OptionReplication extends Option
+class OptionReplication extends OptionMulti
 {
 
-    private static final Pattern FULL = Pattern.compile("replication\\((.*)\\)", Pattern.CASE_INSENSITIVE);
-    private static final Pattern OPTION = Pattern.compile("([^,=]+)=([^,]+)", Pattern.CASE_INSENSITIVE);
+    private final OptionSimple strategy = new OptionSimple("strategy=", new StrategyAdapter(), "org.apache.cassandra.locator.SimpleStrategy", "The replication strategy to use", false);
+    private final OptionSimple factor = new OptionSimple("factor=", "[0-9]+", "1", "The number of replicas", false);
 
-    private String strategy = "org.apache.cassandra.locator.SimpleStrategy";
-    private Map<String, String> options = new HashMap<>();
+    public OptionReplication()
+    {
+        super("replication", "Define the replication strategy and any parameters", true);
+    }
 
     public String getStrategy()
     {
-        return strategy;
+        return strategy.value();
     }
 
     public Map<String, String> getOptions()
     {
-        if (!options.containsKey("replication_factor") && strategy.endsWith("SimpleStrategy"))
-            options.put("replication_factor", "1");
+        Map<String, String> options = extraOptions();
+        options.put("replication_factor", factor.value());
         return options;
     }
 
-
-    @Override
-    public boolean accept(String param)
+    protected List<? extends Option> options()
     {
-        Matcher m = FULL.matcher(param);
-        if (!m.matches())
-            return false;
-        String args = m.group(1);
-        m = OPTION.matcher(args);
-        int last = -1;
-        while (m.find())
-        {
-            if (m.start() != last + 1)
-                throw new IllegalArgumentException("Invalid replication specification: " + param);
-            last = m.end();
-            String key = m.group(1).toLowerCase();
-            sw: switch(key)
-            {
-                case "factor":
-                    try
-                    {
-                        Integer.parseInt(m.group(2));
-                    } catch (NumberFormatException e)
-                    {
-                        throw new IllegalArgumentException("Invalid replication factor: " + param);
-                    }
-                    options.put("replication_factor", m.group(2));
-                    break;
-                case "strategy":
-                    for (String name : new String[] { m.group(2), "org.apache.cassandra.locator." + m.group(2) })
-                    {
-                        try
-                        {
-                            Class<?> clazz = Class.forName(name);
-                            if (!AbstractReplicationStrategy.class.isAssignableFrom(clazz))
-                                throw new RuntimeException();
-                            strategy = name;
-                            break sw;
-                        } catch (Exception _)
-                        {
-                        }
-                    }
-                    throw new IllegalArgumentException("Invalid replication strategy: " + param);
-                default:
-
-            }
-        }
-        return true;
+        return Arrays.asList(strategy, factor);
     }
 
     @Override
@@ -110,26 +66,28 @@ class OptionReplication extends Option
         return true;
     }
 
-    @Override
-    public String shortDisplay()
+    private static final class StrategyAdapter implements Function<String, String>
     {
-        return "replication(?)";
-    }
-
-    @Override
-    public String longDisplay()
-    {
-        return "replication(factor=?,strategy=?,<option1>=?,...)";
-    }
-
-    @Override
-    public List<String> multiLineDisplay()
-    {
-        return Arrays.asList(
-                GroupedOptions.formatMultiLine("factor=?","The replication factor to use (default 1)"),
-                GroupedOptions.formatMultiLine("strategy=?","The replication strategy to use (default SimpleStrategy)"),
-                GroupedOptions.formatMultiLine("option=?","Arbitrary replication strategy options")
-        );
+        public String apply(String name)
+        {
+            String strategy = null;
+            for (String fullname : new String[] { name, "org.apache.cassandra.locator." + name })
+            {
+                try
+                {
+                    Class<?> clazz = Class.forName(fullname);
+                    if (!AbstractReplicationStrategy.class.isAssignableFrom(clazz))
+                        throw new RuntimeException();
+                    strategy = fullname;
+                    break;
+                } catch (Exception _)
+                {
+                }
+            }
+            if (strategy == null)
+                throw new IllegalArgumentException("Invalid replication strategy: " + name);
+            return strategy;
+        }
     }
 
 }
diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/OptionSimple.java b/tools/stress/src/org/apache/cassandra/stress/settings/OptionSimple.java
index 4c80dba959..2a9738a84c 100644
--- a/tools/stress/src/org/apache/cassandra/stress/settings/OptionSimple.java
+++ b/tools/stress/src/org/apache/cassandra/stress/settings/OptionSimple.java
@@ -25,35 +25,58 @@ import java.util.Collections;
 import java.util.List;
 import java.util.regex.Pattern;
 
+import com.google.common.base.Function;
+
 /**
  * For parsing a simple (sub)option for a command/major option
  */
 class OptionSimple extends Option
 {
 
-    final String displayPrefix;
-    final Pattern matchPrefix;
-    final String defaultValue;
-    final Pattern pattern;
-    final String description;
-    final boolean required;
-    String value;
+    private final String displayPrefix;
+    private final Pattern matchPrefix;
+    private final String defaultValue;
+    private final Function<String, String> valueAdapter;
+    private final String description;
+    private final boolean required;
+    private String value;
+
+    private static final class ValueMatcher implements Function<String, String>
+    {
+        final Pattern pattern;
+        private ValueMatcher(Pattern pattern)
+        {
+            this.pattern = pattern;
+        }
+        public String apply(String s)
+        {
+            if (!pattern.matcher(s).matches())
+                throw new IllegalArgumentException("Invalid value " + s + "; must match pattern " + pattern);
+            return s;
+        }
+    }
 
     public OptionSimple(String prefix, String valuePattern, String defaultValue, String description, boolean required)
     {
-        this.displayPrefix = prefix;
-        this.matchPrefix = Pattern.compile(Pattern.quote(prefix), Pattern.CASE_INSENSITIVE);
-        this.pattern = Pattern.compile(valuePattern, Pattern.CASE_INSENSITIVE);
-        this.defaultValue = defaultValue;
-        this.description = description;
-        this.required = required;
+        this(prefix, Pattern.compile(Pattern.quote(prefix), Pattern.CASE_INSENSITIVE),
+             Pattern.compile(valuePattern, Pattern.CASE_INSENSITIVE), defaultValue, description, required);
+    }
+
+    public OptionSimple(String prefix, Function<String, String> valueAdapter, String defaultValue, String description, boolean required)
+    {
+        this(prefix, Pattern.compile(Pattern.quote(prefix), Pattern.CASE_INSENSITIVE), valueAdapter, defaultValue, description, required);
     }
 
     public OptionSimple(String displayPrefix, Pattern matchPrefix, Pattern valuePattern, String defaultValue, String description, boolean required)
+    {
+        this(displayPrefix, matchPrefix, new ValueMatcher(valuePattern), defaultValue, description, required);
+    }
+
+    public OptionSimple(String displayPrefix, Pattern matchPrefix, Function<String, String> valueAdapter, String defaultValue, String description, boolean required)
     {
         this.displayPrefix = displayPrefix;
         this.matchPrefix = matchPrefix;
-        this.pattern = valuePattern;
+        this.valueAdapter = valueAdapter;
         this.defaultValue = defaultValue;
         this.description = description;
         this.required = required;
@@ -81,9 +104,8 @@ class OptionSimple extends Option
             if (value != null)
                 throw new IllegalArgumentException("Suboption " + displayPrefix + " has been specified more than once");
             String v = param.substring(displayPrefix.length());
-            if (!pattern.matcher(v).matches())
-                throw new IllegalArgumentException("Invalid option " + param + "; must match pattern " + pattern);
-            value = v;
+            value = valueAdapter.apply(v);
+            assert value != null;
             return true;
         }
         return false;
@@ -114,7 +136,8 @@ class OptionSimple extends Option
 
     public String longDisplay()
     {
-        if (description.equals("") && defaultValue == null && pattern.pattern().equals(""))
+        if (description.equals("") && defaultValue == null
+            && (valueAdapter instanceof ValueMatcher && ((ValueMatcher) valueAdapter).pattern.pattern().equals("")))
             return null;
         StringBuilder sb = new StringBuilder();
         sb.append(displayPrefix);
diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandMixed.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandMixed.java
index 289dd305ec..bce786a265 100644
--- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandMixed.java
+++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandMixed.java
@@ -111,7 +111,7 @@ public class SettingsCommandMixed extends SettingsCommandMulti
 
         public Probabilities()
         {
-            super("ratio", "Specify the ratios for operations to perform; e.g. (reads=2,writes=1) will perform 2 reads for each write");
+            super("ratio", "Specify the ratios for operations to perform; e.g. (reads=2,writes=1) will perform 2 reads for each write", false);
             OptionSimple[] ratios = new OptionSimple[Command.values().length];
             List<OptionSimple> grouping = new ArrayList<>();
             for (Command command : Command.values())
diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java
index e9b6d55593..5b5333c21f 100644
--- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java
+++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java
@@ -43,6 +43,7 @@ public class SettingsSchema implements Serializable
     private final boolean replicateOnWrite;
     private final String compression;
     private final String compactionStrategy;
+    private final Map<String, String> compactionStrategyOptions;
     public final String keyspace;
     public final String columnFamily;
 
@@ -56,26 +57,12 @@ public class SettingsSchema implements Serializable
         else
             indexType = null;
         compression = options.compression.value();
-        compactionStrategy = options.compactionStrategy.value();
-        if (compactionStrategy != null)
-        {
-            try
-            {
-                CFMetaData.createCompactionStrategy(compactionStrategy);
-            } catch (ConfigurationException e)
-            {
-                throw new IllegalArgumentException("Invalid compaction strategy: " + compactionStrategy);
-            }
-        }
+        compactionStrategy = options.compaction.getStrategy();
+        compactionStrategyOptions = options.compaction.getOptions();
         keyspace = options.keyspace.value();
         columnFamily = options.columnFamily.value();
     }
 
-    private void createKeyspacesCql3(StressSettings settings)
-    {
-//        settings.getJavaDriverClient().execute("create table Standard1")
-    }
-
     public void createKeySpaces(StressSettings settings)
     {
         createKeySpacesThrift(settings);
@@ -152,9 +139,16 @@ public class SettingsSchema implements Serializable
             superCfDef.setCompaction_strategy(compactionStrategy);
             counterCfDef.setCompaction_strategy(compactionStrategy);
             counterSuperCfDef.setCompaction_strategy(compactionStrategy);
+            if (!compactionStrategyOptions.isEmpty())
+            {
+                standardCfDef.setCompaction_strategy_options(compactionStrategyOptions);
+                superCfDef.setCompaction_strategy_options(compactionStrategyOptions);
+                counterCfDef.setCompaction_strategy_options(compactionStrategyOptions);
+                counterSuperCfDef.setCompaction_strategy_options(compactionStrategyOptions);
+            }
         }
 
-        ksdef.setCf_defs(new ArrayList<CfDef>(Arrays.asList(standardCfDef, superCfDef, counterCfDef, counterSuperCfDef)));
+        ksdef.setCf_defs(new ArrayList<>(Arrays.asList(standardCfDef, superCfDef, counterCfDef, counterSuperCfDef)));
 
         Cassandra.Client client = settings.getRawThriftClient(false);
 
@@ -205,17 +199,17 @@ public class SettingsSchema implements Serializable
     private static final class Options extends GroupedOptions
     {
         final OptionReplication replication = new OptionReplication();
+        final OptionCompaction compaction = new OptionCompaction();
         final OptionSimple index = new OptionSimple("index=", "KEYS|CUSTOM|COMPOSITES", null, "Type of index to create on needed column families (KEYS)", false);
         final OptionSimple keyspace = new OptionSimple("keyspace=", ".*", "Keyspace1", "The keyspace name to use", false);
         final OptionSimple columnFamily = new OptionSimple("columnfamily=", ".*", "Standard1", "The column family name to use", false);
-        final OptionSimple compactionStrategy = new OptionSimple("compaction=", ".*", null, "The compaction strategy to use", false);
         final OptionSimple noReplicateOnWrite = new OptionSimple("no-replicate-on-write", "", null, "Set replicate_on_write to false for counters. Only counter add with CL=ONE will work", false);
         final OptionSimple compression = new OptionSimple("compression=", ".*", null, "Specify the compression to use for sstable, default:no compression", false);
 
         @Override
         public List<? extends Option> options()
         {
-            return Arrays.asList(replication, index, keyspace, columnFamily, compactionStrategy, noReplicateOnWrite, compression);
+            return Arrays.asList(replication, index, keyspace, columnFamily, compaction, noReplicateOnWrite, compression);
         }
     }