create_view_statement: Require MV feature

This patch adds the MATERIALIZED_VIEWS_FEATURE to the set of cluster
features and requires its presence to allow creating a view. This
ensures view schemas can be safely propagated across nodes.

Signed-off-by: Duarte Nunes <duarte@scylladb.com>

cql3/statements/create_view_statement.cc

@@ -52,6 +52,7 @@
 #include "service/storage_proxy.hh"
 #include "validation.hh"
 #include "db/config.hh"
+#include "service/storage_service.hh"
 
 
 namespace cql3 {
@@ -75,6 +76,9 @@ create_view_statement::create_view_statement(
     , _if_not_exists{if_not_exists}
 {
     service::get_local_storage_proxy().get_db().local().get_config().check_experimental("Creating materialized views");
+    if (!service::get_local_storage_service().cluster_supports_materialized_views()) {
+        throw exceptions::invalid_request_exception("Can't create materialized views until the whole cluster has been upgraded");
+    }
     // TODO: probably need to create a "statement_restrictions" like select does
     // based on the select_clause, base_name and where_clause; However need to
     // pass for_view=true.

service/storage_service.cc

@@ -81,6 +81,7 @@ static logging::logger logger("storage_service");
 
 static const sstring RANGE_TOMBSTONES_FEATURE = "RANGE_TOMBSTONES";
 static const sstring LARGE_PARTITIONS_FEATURE = "LARGE_PARTITIONS";
+static const sstring MATERIALIZED_VIEWS_FEATURE = "MATERIALIZED_VIEWS";
 
 distributed<storage_service> _the_storage_service;
 
@@ -120,6 +121,7 @@ sstring storage_service::get_config_supported_features() {
     std::vector<sstring> features = {
         RANGE_TOMBSTONES_FEATURE,
         LARGE_PARTITIONS_FEATURE,
+        MATERIALIZED_VIEWS_FEATURE,
     };
     return join(",", features);
 }
@@ -1326,6 +1328,7 @@ future<> storage_service::init_server(int delay) {
         get_storage_service().invoke_on_all([] (auto& ss) {
             ss._range_tombstones_feature = gms::feature(RANGE_TOMBSTONES_FEATURE);
             ss._large_partitions_feature = gms::feature(LARGE_PARTITIONS_FEATURE);
+            ss._materialized_views_feature = gms::feature(MATERIALIZED_VIEWS_FEATURE);
         }).get();
     });
 }

service/storage_service.hh

@@ -260,6 +260,7 @@ private:
 
     gms::feature _range_tombstones_feature;
     gms::feature _large_partitions_feature;
+    gms::feature _materialized_views_feature;
 
 public:
     void finish_bootstrapping() {
@@ -2210,6 +2211,10 @@ public:
     bool cluster_supports_large_partitions() const {
         return bool(_large_partitions_feature);
     }
+
+    bool cluster_supports_materialized_views() const {
+        return bool(_materialized_views_feature);
+    }
 };
 
 inline future<> init_storage_service(distributed<database>& db) {