schema: Support compaction enabled attribute

Fixes #2547

Signed-off-by: Duarte Nunes <duarte@scylladb.com>
Message-Id: <20170721132206.3037-1-duarte@scylladb.com>
(cherry picked from commit 7eecda3a61)

cql3/statements/cf_prop_defs.cc

@@ -41,6 +41,8 @@
 
 #include "cql3/statements/cf_prop_defs.hh"
 
+#include <boost/algorithm/string/predicate.hpp>
+
 namespace cql3 {
 
 namespace statements {
@@ -65,6 +67,8 @@ const sstring cf_prop_defs::KW_CRC_CHECK_CHANCE = "crc_check_chance";
 
 const sstring cf_prop_defs::COMPACTION_STRATEGY_CLASS_KEY = "class";
 
+const sstring cf_prop_defs::COMPACTION_ENABLED_KEY = "enabled";
+
 void cf_prop_defs::validate() {
     // Skip validation if the comapction strategy class is already set as it means we've alreayd
     // prepared (and redoing it would set strategyClass back to null, which we don't want)
@@ -188,6 +192,13 @@ void cf_prop_defs::apply_to_builder(schema_builder& builder) {
     builder.set_min_compaction_threshold(min_compaction_threshold);
     builder.set_max_compaction_threshold(max_compaction_threshold);
 
+    if (has_property(KW_COMPACTION)) {
+        if (get_compaction_options().count(COMPACTION_ENABLED_KEY)) {
+            auto enabled = boost::algorithm::iequals(get_compaction_options().at(COMPACTION_ENABLED_KEY), "true");
+            builder.set_compaction_enabled(enabled);
+        }
+    }
+
     builder.set_default_time_to_live(gc_clock::duration(get_int(KW_DEFAULT_TIME_TO_LIVE, DEFAULT_DEFAULT_TIME_TO_LIVE)));
 
     if (has_property(KW_SPECULATIVE_RETRY)) {

cql3/statements/cf_prop_defs.hh

@@ -73,6 +73,7 @@ public:
     static const sstring KW_CRC_CHECK_CHANCE;
 
     static const sstring COMPACTION_STRATEGY_CLASS_KEY;
+    static const sstring COMPACTION_ENABLED_KEY;
 
     // FIXME: In origin the following consts are in CFMetaData.
     static constexpr int32_t DEFAULT_DEFAULT_TIME_TO_LIVE = 0;

db/schema_tables.cc

@@ -64,6 +64,7 @@
 #include "db/config.hh"
 #include "md5_hasher.hh"
 
+#include <boost/algorithm/string/predicate.hpp>
 #include <boost/range/algorithm/copy.hpp>
 #include <boost/range/adaptor/map.hpp>
 #include <boost/range/join.hpp>
@@ -1418,7 +1419,7 @@ static void add_table_params_to_mutations(mutation& m, const clustering_key& cke
 
     {
         auto map = table->compaction_strategy_options();
-        map["class"] = sstables::compaction_strategy::name(table->compaction_strategy());
+        map["class"] = sstables::compaction_strategy::name(table->configured_compaction_strategy());
         store_map(m, ckey, "compaction", timestamp, map);
     }
 
@@ -1815,7 +1816,7 @@ static void prepare_builder_from_table_row(schema_builder& builder, const query:
             builder.set_min_compaction_threshold(std::stoi(map["min_threshold"]));
         }
         if (map.count("enabled")) {
-            // TODO: enable/disable?
+            builder.set_compaction_enabled(boost::algorithm::iequals(map["enabled"], "true"));
         }
 
         builder.set_compaction_strategy_options(map);

schema.cc

@@ -444,6 +444,7 @@ bool operator==(const schema& x, const schema& y)
         && x._raw._speculative_retry == y._raw._speculative_retry
         && x._raw._compaction_strategy == y._raw._compaction_strategy
         && x._raw._compaction_strategy_options == y._raw._compaction_strategy_options
+        && x._raw._compaction_enabled == y._raw._compaction_enabled
         && x._raw._caching_options == y._raw._caching_options
         && x._raw._dropped_columns == y._raw._dropped_columns
         && x._raw._collections == y._raw._collections
@@ -562,11 +563,10 @@ std::ostream& operator<<(std::ostream& os, const schema& s) {
     os << ",compactionStrategyOptions={";
     n = 0;
     for (auto& p : s._raw._compaction_strategy_options) {
-        if (n++ != 0) {
-            os << ", ";
-        }
         os << p.first << "=" << p.second;
+        os << ", ";
     }
+    os << "enabled=" << std::boolalpha << s._raw._compaction_enabled;
     os << "}";
     os << ",compressionParameters={";
     n = 0;

schema.hh

@@ -457,6 +457,7 @@ private:
         // we will use by default - when we have the choice.
         sstables::compaction_strategy_type _compaction_strategy = sstables::compaction_strategy_type::size_tiered;
         std::map<sstring, sstring> _compaction_strategy_options;
+        bool _compaction_enabled = true;
         caching_options _caching_options;
         table_schema_version _version;
         std::unordered_map<sstring, dropped_column> _dropped_columns;
@@ -602,14 +603,22 @@ public:
         return _raw._memtable_flush_period;
     }
 
-    sstables::compaction_strategy_type compaction_strategy() const {
+    sstables::compaction_strategy_type configured_compaction_strategy() const {
         return _raw._compaction_strategy;
     }
 
+    sstables::compaction_strategy_type compaction_strategy() const {
+        return _raw._compaction_enabled ? _raw._compaction_strategy : sstables::compaction_strategy_type::null;
+    }
+
     const std::map<sstring, sstring>& compaction_strategy_options() const {
         return _raw._compaction_strategy_options;
     }
 
+    bool compaction_enabled() const {
+        return _raw._compaction_enabled;
+    }
+
     const ::speculative_retry& speculative_retry() const {
         return _raw._speculative_retry;
     }

schema_builder.hh

@@ -132,6 +132,15 @@ public:
         return _raw._max_compaction_threshold;
     }
 
+    schema_builder& set_compaction_enabled(bool enabled) {
+        _raw._compaction_enabled = enabled;
+        return *this;
+    }
+
+    bool compaction_enabled() const {
+        return _raw._compaction_enabled;
+    }
+
     schema_builder& set_min_index_interval(int32_t t) {
         _raw._min_index_interval = t;
         return *this;