Revert "dist: reduce kernel's inclincation to OOM-kill Scylla"

This reverts commit 1ea0af8acb.

Backported by mistake.

.github/ISSUE_TEMPLATE.md

@@ -0,0 +1,9 @@
+## Installation details
+Scylla version (or git commit hash):
+Cluster size:
+OS (RHEL/CentOS/Ubuntu/AWS AMI):
+
+## Hardware details (for performance issues)
+Platform (physical/VM/cloud instance type/docker):
+Hardware: sockets= cores= hyperthreading= memory=
+Disks: (SSD/HDD, count)

.gitmodules

@@ -1,6 +1,6 @@
 [submodule "seastar"]
 	path = seastar
-	url = ../seastar
+	url = ../scylla-seastar
 	ignore = dirty
 [submodule "swagger-ui"]
 	path = swagger-ui

SCYLLA-VERSION-GEN

@@ -1,6 +1,6 @@
 #!/bin/sh
 
-VERSION=666.development
+VERSION=1.1.4
 
 if test -f version
 then

api/api.cc

@@ -61,10 +61,10 @@ future<> set_server_init(http_context& ctx) {
                 new content_replace("html")));
         r.add(GET, url("/ui").remainder("path"), new httpd::directory_handler(ctx.api_dir,
                 new content_replace("html")));
+        rb->set_api_doc(r);
         rb->register_function(r, "system",
                 "The system related API");
         set_system(ctx, r);
-        rb->set_api_doc(r);
     });
 }
 

api/cache_service.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/cache_service.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/collectd.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/collectd.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/column_family.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*
@@ -40,7 +40,7 @@ const utils::UUID& get_uuid(const sstring& name, const database& db) {
     if (pos == sstring::npos) {
         pos  = name.find(":");
         if (pos == sstring::npos) {
-            throw bad_param_exception("Column family name should be in keyspace::column_family format");
+            throw bad_param_exception("Column family name should be in keyspace:column_family format");
         }
         end = pos + 1;
     } else {
@@ -173,6 +173,35 @@ static ratio_holder mean_row_size(column_family& cf) {
     return res;
 }
 
+template <typename T>
+class sum_ratio {
+    uint64_t _n = 0;
+    T _total = 0;
+public:
+    future<> operator()(T value) {
+        if (value > 0) {
+            _total += value;
+            _n++;
+        }
+        return make_ready_future<>();
+    }
+    // Returns average value of all registered ratios.
+    T get() && {
+        return _n ? (_total / _n) : T(0);
+    }
+};
+
+static double get_compression_ratio(column_family& cf) {
+    sum_ratio<double> result;
+    for (auto i : *cf.get_sstables()) {
+        auto compression_ratio = i.second->get_compression_ratio();
+        if (compression_ratio != sstables::metadata_collector::NO_COMPRESSION_RATIO) {
+            result(compression_ratio);
+        }
+    }
+    return std::move(result).get();
+}
+
 void set_column_family(http_context& ctx, routes& r) {
     cf::get_column_family_name.set(r, [&ctx] (const_req req){
         vector<sstring> res;
@@ -719,11 +748,15 @@ void set_column_family(http_context& ctx, routes& r) {
         return std::vector<sstring>();
     });
 
-    cf::get_compression_ratio.set(r, [](const_req) {
-        // FIXME
-        // Currently there are no compression information
-        // so we return 0 as the ratio
-        return 0;
+    cf::get_compression_ratio.set(r, [&ctx](std::unique_ptr<request> req) {
+        auto uuid = get_uuid(req->param["name"], ctx.db.local());
+
+        return ctx.db.map_reduce(sum_ratio<double>(), [uuid](database& db) {
+            column_family& cf = db.find_column_family(uuid);
+            return make_ready_future<double>(get_compression_ratio(cf));
+        }).then([] (const double& result) {
+            return make_ready_future<json::json_return_type>(result);
+        });
     });
 
     cf::get_read_latency_estimated_histogram.set(r, [&ctx](std::unique_ptr<request> req) {

api/column_family.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/commitlog.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/commitlog.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/compaction_manager.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/compaction_manager.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/endpoint_snitch.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/endpoint_snitch.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/failure_detector.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/failure_detector.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/gossiper.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/gossiper.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/hinted_handoff.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/hinted_handoff.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/lsa.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/lsa.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/messaging_service.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/messaging_service.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/storage_proxy.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/storage_proxy.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/storage_service.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*
@@ -382,21 +382,21 @@ void set_storage_service(http_context& ctx, routes& r) {
 
     ss::remove_node.set(r, [](std::unique_ptr<request> req) {
         auto host_id = req->get_query_param("host_id");
-        return service::get_local_storage_service().remove_node(host_id).then([] {
+        return service::get_local_storage_service().removenode(host_id).then([] {
             return make_ready_future<json::json_return_type>(json_void());
         });
     });
 
     ss::get_removal_status.set(r, [](std::unique_ptr<request> req) {
-        //TBD
-        unimplemented();
-        return make_ready_future<json::json_return_type>("");
+        return service::get_local_storage_service().get_removal_status().then([] (auto status) {
+            return make_ready_future<json::json_return_type>(status);
+        });
     });
 
     ss::force_remove_completion.set(r, [](std::unique_ptr<request> req) {
-        //TBD
-        unimplemented();
-        return make_ready_future<json::json_return_type>(json_void());
+        return service::get_local_storage_service().force_remove_completion().then([] {
+            return make_ready_future<json::json_return_type>(json_void());
+        });
     });
 
     ss::set_logging_level.set(r, [](std::unique_ptr<request> req) {
@@ -588,6 +588,8 @@ void set_storage_service(http_context& ctx, routes& r) {
         auto val_str = req->get_query_param("value");
         bool value = (val_str == "True") || (val_str == "true") || (val_str == "1");
         return service::get_local_storage_service().db().invoke_on_all([value] (database& db) {
+            db.set_enable_incremental_backups(value);
+
             // Change both KS and CF, so they are in sync
             for (auto& pair: db.get_keyspaces()) {
                 auto& ks = pair.second;

api/storage_service.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/stream_manager.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/stream_manager.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/system.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

api/system.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

atomic_cell.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2015 Cloudius Systems, Ltd.
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*
@@ -54,9 +54,9 @@ class atomic_cell_or_collection;
  */
 class atomic_cell_type final {
 private:
-    static constexpr int8_t DEAD_FLAGS = 0;
     static constexpr int8_t LIVE_FLAG = 0x01;
     static constexpr int8_t EXPIRY_FLAG = 0x02; // When present, expiry field is present. Set only for live cells
+    static constexpr int8_t REVERT_FLAG = 0x04; // transient flag used to efficiently implement ReversiblyMergeable for atomic cells.
     static constexpr unsigned flags_size = 1;
     static constexpr unsigned timestamp_offset = flags_size;
     static constexpr unsigned timestamp_size = 8;
@@ -67,14 +67,21 @@ private:
     static constexpr unsigned ttl_offset = expiry_offset + expiry_size;
     static constexpr unsigned ttl_size = 4;
 private:
+    static bool is_revert_set(bytes_view cell) {
+        return cell[0] & REVERT_FLAG;
+    }
+    template<typename BytesContainer>
+    static void set_revert(BytesContainer& cell, bool revert) {
+        cell[0] = (cell[0] & ~REVERT_FLAG) | (revert * REVERT_FLAG);
+    }
     static bool is_live(const bytes_view& cell) {
-        return cell[0] != DEAD_FLAGS;
+        return cell[0] & LIVE_FLAG;
     }
     static bool is_live_and_has_ttl(const bytes_view& cell) {
         return cell[0] & EXPIRY_FLAG;
     }
     static bool is_dead(const bytes_view& cell) {
-        return cell[0] == DEAD_FLAGS;
+        return !is_live(cell);
     }
     // Can be called on live and dead cells
     static api::timestamp_type timestamp(const bytes_view& cell) {
@@ -106,7 +113,7 @@ private:
     }
     static managed_bytes make_dead(api::timestamp_type timestamp, gc_clock::time_point deletion_time) {
         managed_bytes b(managed_bytes::initialized_later(), flags_size + timestamp_size + deletion_time_size);
-        b[0] = DEAD_FLAGS;
+        b[0] = 0;
         set_field(b, timestamp_offset, timestamp);
         set_field(b, deletion_time_offset, deletion_time.time_since_epoch().count());
         return b;
@@ -140,8 +147,11 @@ protected:
     ByteContainer _data;
 protected:
     atomic_cell_base(ByteContainer&& data) : _data(std::forward<ByteContainer>(data)) { }
-    atomic_cell_base(const ByteContainer& data) : _data(data) { }
+    friend class atomic_cell_or_collection;
 public:
+    bool is_revert_set() const {
+        return atomic_cell_type::is_revert_set(_data);
+    }
     bool is_live() const {
         return atomic_cell_type::is_live(_data);
     }
@@ -187,10 +197,13 @@ public:
     bytes_view serialize() const {
         return _data;
     }
+    void set_revert(bool revert) {
+        atomic_cell_type::set_revert(_data, revert);
+    }
 };
 
 class atomic_cell_view final : public atomic_cell_base<bytes_view> {
-    atomic_cell_view(bytes_view data) : atomic_cell_base(data) {}
+    atomic_cell_view(bytes_view data) : atomic_cell_base(std::move(data)) {}
 public:
     static atomic_cell_view from_bytes(bytes_view data) { return atomic_cell_view(data); }
 
@@ -198,6 +211,11 @@ public:
     friend std::ostream& operator<<(std::ostream& os, const atomic_cell_view& acv);
 };
 
+class atomic_cell_ref final : public atomic_cell_base<managed_bytes&> {
+public:
+    atomic_cell_ref(managed_bytes& buf) : atomic_cell_base(buf) {}
+};
+
 class atomic_cell final : public atomic_cell_base<managed_bytes> {
     atomic_cell(managed_bytes b) : atomic_cell_base(std::move(b)) {}
 public:

atomic_cell_hash.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2015 Cloudius Systems, Ltd.
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*
@@ -27,16 +27,18 @@
 #include "atomic_cell.hh"
 #include "hashing.hh"
 
-template<typename Hasher>
-void feed_hash(collection_mutation_view cell, Hasher& h, const data_type& type) {
-    auto&& ctype = static_pointer_cast<const collection_type_impl>(type);
-    auto m_view = ctype->deserialize_mutation_form(cell);
-    ::feed_hash(h, m_view.tomb);
-    for (auto&& key_and_value : m_view.cells) {
-        ::feed_hash(h, key_and_value.first);
-        ::feed_hash(h, key_and_value.second);
+template<>
+struct appending_hash<collection_mutation_view> {
+    template<typename Hasher>
+    void operator()(Hasher& h, collection_mutation_view cell) const {
+        auto m_view = collection_type_impl::deserialize_mutation_form(cell);
+        ::feed_hash(h, m_view.tomb);
+        for (auto&& key_and_value : m_view.cells) {
+            ::feed_hash(h, key_and_value.first);
+            ::feed_hash(h, key_and_value.second);
+        }
     }
-}
+};
 
 template<>
 struct appending_hash<atomic_cell_view> {
@@ -55,3 +57,19 @@ struct appending_hash<atomic_cell_view> {
         }
     }
 };
+
+template<>
+struct appending_hash<atomic_cell> {
+    template<typename Hasher>
+    void operator()(Hasher& h, const atomic_cell& cell) const {
+        feed_hash(h, static_cast<atomic_cell_view>(cell));
+    }
+};
+
+template<>
+struct appending_hash<collection_mutation> {
+    template<typename Hasher>
+    void operator()(Hasher& h, const collection_mutation& cm) const {
+        feed_hash(h, static_cast<collection_mutation_view>(cm));
+    }
+};

atomic_cell_or_collection.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2015 Cloudius Systems, Ltd.
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*
@@ -27,6 +27,8 @@
 
 // A variant type that can hold either an atomic_cell, or a serialized collection.
 // Which type is stored is determined by the schema.
+// Has an "empty" state.
+// Objects moved-from are left in an empty state.
 class atomic_cell_or_collection final {
     managed_bytes _data;
 private:
@@ -36,6 +38,7 @@ public:
     atomic_cell_or_collection(atomic_cell ac) : _data(std::move(ac._data)) {}
     static atomic_cell_or_collection from_atomic_cell(atomic_cell data) { return { std::move(data._data) }; }
     atomic_cell_view as_atomic_cell() const { return atomic_cell_view::from_bytes(_data); }
+    atomic_cell_ref as_atomic_cell_ref() { return { _data }; }
     atomic_cell_or_collection(collection_mutation cm) : _data(std::move(cm.data)) {}
     explicit operator bool() const {
         return !_data.empty();

auth/auth.cc

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2016 Cloudius Systems
+ * Copyright (C) 2016 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*
@@ -40,14 +40,19 @@
  */
 #include <seastar/core/sleep.hh>
 
+#include <seastar/core/distributed.hh>
+
 #include "auth.hh"
 #include "authenticator.hh"
+#include "authorizer.hh"
 #include "database.hh"
 #include "cql3/query_processor.hh"
 #include "cql3/statements/cf_statement.hh"
 #include "cql3/statements/create_table_statement.hh"
 #include "db/config.hh"
 #include "service/migration_manager.hh"
+#include "utils/loading_cache.hh"
+#include "utils/hash.hh"
 
 const sstring auth::auth::DEFAULT_SUPERUSER_NAME("cassandra");
 const sstring auth::auth::AUTH_KS("system_auth");
@@ -76,13 +81,10 @@ class auth_migration_listener : public service::migration_listener {
     void on_update_aggregate(const sstring& ks_name, const sstring& aggregate_name) override {}
 
     void on_drop_keyspace(const sstring& ks_name) override {
-        // TODO:
-        //DatabaseDescriptor.getAuthorizer().revokeAll(DataResource.keyspace(ksName));
-
+        auth::authorizer::get().revoke_all(auth::data_resource(ks_name));
     }
     void on_drop_column_family(const sstring& ks_name, const sstring& cf_name) override {
-        // TODO:
-        //DatabaseDescriptor.getAuthorizer().revokeAll(DataResource.columnFamily(ksName, cfName));
+        auth::authorizer::get().revoke_all(auth::data_resource(ks_name, cf_name));
     }
     void on_drop_user_type(const sstring& ks_name, const sstring& type_name) override {}
     void on_drop_function(const sstring& ks_name, const sstring& function_name) override {}
@@ -91,6 +93,64 @@ class auth_migration_listener : public service::migration_listener {
 
 static auth_migration_listener auth_migration;
 
+namespace std {
+template <>
+struct hash<auth::data_resource> {
+    size_t operator()(const auth::data_resource & v) const {
+        return v.hash_value();
+    }
+};
+
+template <>
+struct hash<auth::authenticated_user> {
+    size_t operator()(const auth::authenticated_user & v) const {
+        return utils::tuple_hash()(v.name(), v.is_anonymous());
+    }
+};
+}
+
+class auth::auth::permissions_cache {
+public:
+    typedef utils::loading_cache<std::pair<authenticated_user, data_resource>, permission_set, utils::tuple_hash> cache_type;
+    typedef typename cache_type::key_type key_type;
+
+    permissions_cache()
+                    : permissions_cache(
+                                    cql3::get_local_query_processor().db().local().get_config()) {
+    }
+
+    permissions_cache(const db::config& cfg)
+                    : _cache(cfg.permissions_cache_max_entries(), expiry(cfg),
+                                    std::chrono::milliseconds(
+                                                    cfg.permissions_validity_in_ms()),
+                                    [](const key_type& k) {
+                                        logger.debug("Refreshing permissions for {}", k.first.name());
+                                        return authorizer::get().authorize(::make_shared<authenticated_user>(k.first), k.second);
+                                    }) {
+    }
+
+    static std::chrono::milliseconds expiry(const db::config& cfg) {
+        auto exp = cfg.permissions_update_interval_in_ms();
+        if (exp == 0 || exp == std::numeric_limits<uint32_t>::max()) {
+            exp = cfg.permissions_validity_in_ms();
+        }
+        return std::chrono::milliseconds(exp);
+    }
+
+    future<> stop() {
+        return make_ready_future<>();
+    }
+
+    future<permission_set> get(::shared_ptr<authenticated_user> user, data_resource resource) {
+        return _cache.get(key_type(*user, std::move(resource)));
+    }
+
+private:
+    cache_type _cache;
+};
+
+static distributed<auth::auth::permissions_cache> perm_cache;
+
 /**
  * Poor mans job schedule. For maximum 2 jobs. Sic.
  * Still does nothing more clever than waiting 10 seconds
@@ -163,14 +223,22 @@ bool auth::auth::is_class_type(const sstring& type, const sstring& classname) {
 future<> auth::auth::setup() {
     auto& db = cql3::get_local_query_processor().db().local();
     auto& cfg = db.get_config();
-    auto type = cfg.authenticator();
 
-    if (is_class_type(type, authenticator::ALLOW_ALL_AUTHENTICATOR_NAME)) {
-        return authenticator::setup(type).discard_result(); // just create the object
+    future<> f = perm_cache.start();
+
+    if (is_class_type(cfg.authenticator(),
+                    authenticator::ALLOW_ALL_AUTHENTICATOR_NAME)
+                    && is_class_type(cfg.authorizer(),
+                                    authorizer::ALLOW_ALL_AUTHORIZER_NAME)
+                                    ) {
+        // just create the objects
+        return f.then([&cfg] {
+            return authenticator::setup(cfg.authenticator());
+        }).then([&cfg] {
+            return authorizer::setup(cfg.authorizer());
+        });
     }
 
-    future<> f = make_ready_future();
-
     if (!db.has_keyspace(AUTH_KS)) {
         std::map<sstring, sstring> opts;
         opts["replication_factor"] = "1";
@@ -182,10 +250,10 @@ future<> auth::auth::setup() {
         return setup_table(USERS_CF, sprint("CREATE TABLE %s.%s (%s text, %s boolean, PRIMARY KEY(%s)) WITH gc_grace_seconds=%d",
                                         AUTH_KS, USERS_CF, USER_NAME, SUPER, USER_NAME,
                                         90 * 24 * 60 * 60)); // 3 months.
-    }).then([type] {
-        return authenticator::setup(type).discard_result();
-    }).then([] {
-        // TODO authorizer
+    }).then([&cfg] {
+        return authenticator::setup(cfg.authenticator());
+    }).then([&cfg] {
+        return authorizer::setup(cfg.authorizer());
     }).then([] {
         service::get_local_migration_manager().register_listener(&auth_migration); // again, only one shard...
         // instead of once-timer, just schedule this later
@@ -216,9 +284,15 @@ future<> auth::auth::shutdown() {
     // db-env-shutdown != process shutdown
     return smp::invoke_on_all([] {
         thread_waiters().clear();
+    }).then([] {
+        return perm_cache.stop();
     });
 }
 
+future<auth::permission_set> auth::auth::get_permissions(::shared_ptr<authenticated_user> user, data_resource resource) {
+    return perm_cache.local().get(std::move(user), std::move(resource));
+}
+
 static db::consistency_level consistency_for_user(const sstring& username) {
     if (username == auth::auth::DEFAULT_SUPERUSER_NAME) {
         return db::consistency_level::QUORUM;
@@ -280,9 +354,12 @@ future<> auth::auth::setup_table(const sstring& name, const sstring& cql) {
     ::shared_ptr<cql3::statements::create_table_statement> statement =
                     static_pointer_cast<cql3::statements::create_table_statement>(
                                     parsed->prepare(db)->statement);
-    // Origin sets "Legacy Cf Id" for the new table. We have no need to be
-    // pre-2.1 compatible (afaik), so lets skip a whole lotta hoolaballo
-    return statement->announce_migration(qp.proxy(), false).then([statement](bool) {});
+    auto schema = statement->get_cf_meta_data();
+    auto uuid = generate_legacy_id(schema->ks_name(), schema->cf_name());
+
+    schema_builder b(schema);
+    b.set_uuid(uuid);
+    return service::get_local_migration_manager().announce_new_column_family(b.build(), false);
 }
 
 future<bool> auth::auth::has_existing_users(const sstring& cfname, const sstring& def_user_name, const sstring& name_column) {

auth/auth.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2016 Cloudius Systems
+ * Copyright (C) 2016 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*
@@ -44,13 +44,21 @@
 #include <chrono>
 #include <seastar/core/sstring.hh>
 #include <seastar/core/future.hh>
+#include <seastar/core/shared_ptr.hh>
+
 
 #include "exceptions/exceptions.hh"
+#include "permission.hh"
+#include "data_resource.hh"
 
 namespace auth {
 
+class authenticated_user;
+
 class auth {
 public:
+    class permissions_cache;
+
     static const sstring DEFAULT_SUPERUSER_NAME;
     static const sstring AUTH_KS;
     static const sstring USERS_CF;
@@ -58,12 +66,7 @@ public:
 
     static bool is_class_type(const sstring& type, const sstring& classname);
 
-#if 0
-    public static Set<Permission> getPermissions(AuthenticatedUser user, IResource resource)
-    {
-        return permissionsCache.getPermissions(user, resource);
-    }
-#endif
+    static future<permission_set> get_permissions(::shared_ptr<authenticated_user>, data_resource);
 
     /**
      * Checks if the username is stored in AUTH_KS.USERS_CF.

auth/authenticated_user.cc

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2016 Cloudius Systems
+ * Copyright (C) 2016 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*
@@ -41,6 +41,7 @@
 
 
 #include "authenticated_user.hh"
+#include "auth.hh"
 
 const sstring auth::authenticated_user::ANONYMOUS_USERNAME("anonymous");
 
@@ -52,10 +53,20 @@ auth::authenticated_user::authenticated_user(sstring name)
                 : _name(name), _anon(false)
 {}
 
+auth::authenticated_user::authenticated_user(authenticated_user&&) = default;
+auth::authenticated_user::authenticated_user(const authenticated_user&) = default;
+
 const sstring& auth::authenticated_user::name() const {
     return _anon ? ANONYMOUS_USERNAME : _name;
 }
 
+future<bool> auth::authenticated_user::is_super() const {
+    if (is_anonymous()) {
+        return make_ready_future<bool>(false);
+    }
+    return auth::auth::is_super_user(_name);
+}
+
 bool auth::authenticated_user::operator==(const authenticated_user& v) const {
     return _anon ? v._anon : _name == v._name;
 }

auth/authenticated_user.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2016 Cloudius Systems
+ * Copyright (C) 2016 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*
@@ -42,6 +42,7 @@
 #pragma once
 
 #include <seastar/core/sstring.hh>
+#include <seastar/core/future.hh>
 
 namespace auth {
 
@@ -51,6 +52,8 @@ public:
 
     authenticated_user();
     authenticated_user(sstring name);
+    authenticated_user(authenticated_user&&);
+    authenticated_user(const authenticated_user&);
 
     const sstring& name() const;
 
@@ -60,7 +63,7 @@ public:
      * Im most cased, though not necessarily, a superuser will have Permission.ALL on every resource
      * (depends on IAuthorizer implementation).
      */
-    bool is_super() const;
+    future<bool> is_super() const;
 
     /**
      * If IAuthenticator doesn't require authentication, this method may return true.

auth/authenticator.cc

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2016 Cloudius Systems
+ * Copyright (C) 2016 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*
@@ -49,6 +49,22 @@ const sstring auth::authenticator::USERNAME_KEY("username");
 const sstring auth::authenticator::PASSWORD_KEY("password");
 const sstring auth::authenticator::ALLOW_ALL_AUTHENTICATOR_NAME("org.apache.cassandra.auth.AllowAllAuthenticator");
 
+auth::authenticator::option auth::authenticator::string_to_option(const sstring& name) {
+    if (strcasecmp(name.c_str(), "password") == 0) {
+        return option::PASSWORD;
+    }
+    throw std::invalid_argument(name);
+}
+
+sstring auth::authenticator::option_to_string(option opt) {
+    switch (opt) {
+    case option::PASSWORD:
+        return "PASSWORD";
+    default:
+        throw std::invalid_argument(sprint("Unknown option {}", opt));
+    }
+}
+
 /**
  * Authenticator is assumed to be a fully state-less immutable object (note all the const).
  * We thus store a single instance globally, since it should be safe/ok.
@@ -84,8 +100,9 @@ auth::authenticator::setup(const sstring& type) throw (exceptions::configuration
             future<> drop(sstring username) throw(exceptions::request_validation_exception, exceptions::request_execution_exception) override {
                 return make_ready_future();
             }
-            resource_ids protected_resources() const override {
-                return resource_ids();
+            const resource_ids& protected_resources() const override {
+                static const resource_ids ids;
+                return ids;
             }
             ::shared_ptr<sasl_challenge> new_sasl_challenge() const override {
                 throw std::runtime_error("Should not reach");

auth/authenticator.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2016 Cloudius Systems
+ * Copyright (C) 2016 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*
@@ -79,15 +79,13 @@ public:
         PASSWORD
     };
 
+    static option string_to_option(const sstring&);
+    static sstring option_to_string(option);
+
     using option_set = enum_set<super_enum<option, option::PASSWORD>>;
     using option_map = std::unordered_map<option, boost::any, enum_hash<option>>;
     using credentials_map = std::unordered_map<sstring, sstring>;
 
-    /**
-     * Resource id mappings, i.e. keyspace and/or column families.
-     */
-    using resource_ids = std::set<data_resource>;
-
     /**
      * Setup is called once upon system startup to initialize the IAuthenticator.
      *
@@ -174,7 +172,7 @@ public:
      * @return Keyspaces, column families that will be unmodifiable by users; other resources.
      * @see resource_ids
      */
-    virtual resource_ids protected_resources() const = 0;
+    virtual const resource_ids& protected_resources() const = 0;
 
     class sasl_challenge {
     public:
@@ -194,5 +192,9 @@ public:
     virtual ::shared_ptr<sasl_challenge> new_sasl_challenge() const = 0;
 };
 
+inline std::ostream& operator<<(std::ostream& os, authenticator::option opt) {
+    return os << authenticator::option_to_string(opt);
+}
+
 }
 

auth/authorizer.cc

@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/*
+ * Copyright (C) 2016 ScyllaDB
+ *
+ * Modified by ScyllaDB
+ */
+
+/*
+ * This file is part of Scylla.
+ *
+ * Scylla is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Scylla is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "authorizer.hh"
+#include "authenticated_user.hh"
+#include "default_authorizer.hh"
+#include "auth.hh"
+#include "db/config.hh"
+
+const sstring auth::authorizer::ALLOW_ALL_AUTHORIZER_NAME("org.apache.cassandra.auth.AllowAllAuthorizer");
+
+/**
+ * Authenticator is assumed to be a fully state-less immutable object (note all the const).
+ * We thus store a single instance globally, since it should be safe/ok.
+ */
+static std::unique_ptr<auth::authorizer> global_authorizer;
+
+future<>
+auth::authorizer::setup(const sstring& type) {
+    if (auth::auth::is_class_type(type, ALLOW_ALL_AUTHORIZER_NAME)) {
+        class allow_all_authorizer : public authorizer {
+        public:
+            future<permission_set> authorize(::shared_ptr<authenticated_user>, data_resource) const override {
+                return make_ready_future<permission_set>(permissions::ALL);
+            }
+            future<> grant(::shared_ptr<authenticated_user>, permission_set, data_resource, sstring) override {
+                throw exceptions::invalid_request_exception("GRANT operation is not supported by AllowAllAuthorizer");
+            }
+            future<> revoke(::shared_ptr<authenticated_user>, permission_set, data_resource, sstring) override {
+                throw exceptions::invalid_request_exception("REVOKE operation is not supported by AllowAllAuthorizer");
+            }
+            future<std::vector<permission_details>> list(::shared_ptr<authenticated_user> performer, permission_set, optional<data_resource>, optional<sstring>) const override {
+                throw exceptions::invalid_request_exception("LIST PERMISSIONS operation is not supported by AllowAllAuthorizer");
+            }
+            future<> revoke_all(sstring dropped_user) override {
+                return make_ready_future();
+            }
+            future<> revoke_all(data_resource) override {
+                return make_ready_future();
+            }
+            const resource_ids& protected_resources() override {
+                static const resource_ids ids;
+                return ids;
+            }
+            future<> validate_configuration() const override {
+                return make_ready_future();
+            }
+        };
+
+        global_authorizer = std::make_unique<allow_all_authorizer>();
+    } else if (auth::auth::is_class_type(type, default_authorizer::DEFAULT_AUTHORIZER_NAME)) {
+        auto da = std::make_unique<default_authorizer>();
+        auto f = da->init();
+        return f.then([da = std::move(da)]() mutable {
+            global_authorizer = std::move(da);
+        });
+    } else {
+        throw exceptions::configuration_exception("Invalid authorizer type: " + type);
+    }
+    return make_ready_future();
+}
+
+auth::authorizer& auth::authorizer::get() {
+    assert(global_authorizer);
+    return *global_authorizer;
+}

auth/authorizer.hh

@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/*
+ * Copyright (C) 2016 ScyllaDB
+ *
+ * Modified by ScyllaDB
+ */
+
+/*
+ * This file is part of Scylla.
+ *
+ * Scylla is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Scylla is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include <vector>
+#include <tuple>
+
+#include <experimental/optional>
+#include <seastar/core/future.hh>
+#include <seastar/core/shared_ptr.hh>
+
+#include "permission.hh"
+#include "data_resource.hh"
+
+namespace auth {
+
+class authenticated_user;
+
+struct permission_details {
+    sstring user;
+    data_resource resource;
+    permission_set permissions;
+
+    bool operator<(const permission_details& v) const {
+        return std::tie(user, resource, permissions) < std::tie(v.user, v.resource, v.permissions);
+    }
+};
+
+using std::experimental::optional;
+
+class authorizer {
+public:
+    static const sstring ALLOW_ALL_AUTHORIZER_NAME;
+
+    virtual ~authorizer() {}
+
+    /**
+     * The primary Authorizer method. Returns a set of permissions of a user on a resource.
+     *
+     * @param user Authenticated user requesting authorization.
+     * @param resource Resource for which the authorization is being requested. @see DataResource.
+     * @return Set of permissions of the user on the resource. Should never return empty. Use permission.NONE instead.
+     */
+    virtual future<permission_set> authorize(::shared_ptr<authenticated_user>, data_resource) const = 0;
+
+    /**
+     * Grants a set of permissions on a resource to a user.
+     * The opposite of revoke().
+     *
+     * @param performer User who grants the permissions.
+     * @param permissions Set of permissions to grant.
+     * @param to Grantee of the permissions.
+     * @param resource Resource on which to grant the permissions.
+     *
+     * @throws RequestValidationException
+     * @throws RequestExecutionException
+     */
+    virtual future<> grant(::shared_ptr<authenticated_user> performer, permission_set, data_resource, sstring to) = 0;
+
+    /**
+     * Revokes a set of permissions on a resource from a user.
+     * The opposite of grant().
+     *
+     * @param performer User who revokes the permissions.
+     * @param permissions Set of permissions to revoke.
+     * @param from Revokee of the permissions.
+     * @param resource Resource on which to revoke the permissions.
+     *
+     * @throws RequestValidationException
+     * @throws RequestExecutionException
+     */
+    virtual future<> revoke(::shared_ptr<authenticated_user> performer, permission_set, data_resource, sstring from) = 0;
+
+    /**
+     * Returns a list of permissions on a resource of a user.
+     *
+     * @param performer User who wants to see the permissions.
+     * @param permissions Set of Permission values the user is interested in. The result should only include the matching ones.
+     * @param resource The resource on which permissions are requested. Can be null, in which case permissions on all resources
+     *                 should be returned.
+     * @param of The user whose permissions are requested. Can be null, in which case permissions of every user should be returned.
+     *
+     * @return All of the matching permission that the requesting user is authorized to know about.
+     *
+     * @throws RequestValidationException
+     * @throws RequestExecutionException
+     */
+    virtual future<std::vector<permission_details>> list(::shared_ptr<authenticated_user> performer, permission_set, optional<data_resource>, optional<sstring>) const = 0;
+
+    /**
+     * This method is called before deleting a user with DROP USER query so that a new user with the same
+     * name wouldn't inherit permissions of the deleted user in the future.
+     *
+     * @param droppedUser The user to revoke all permissions from.
+     */
+    virtual future<> revoke_all(sstring dropped_user) = 0;
+
+    /**
+     * This method is called after a resource is removed (i.e. keyspace or a table is dropped).
+     *
+     * @param droppedResource The resource to revoke all permissions on.
+     */
+    virtual future<> revoke_all(data_resource) = 0;
+
+    /**
+     * Set of resources that should be made inaccessible to users and only accessible internally.
+     *
+     * @return Keyspaces, column families that will be unmodifiable by users; other resources.
+     */
+    virtual const resource_ids& protected_resources() = 0;
+
+    /**
+     * Validates configuration of IAuthorizer implementation (if configurable).
+     *
+     * @throws ConfigurationException when there is a configuration error.
+     */
+    virtual future<> validate_configuration() const = 0;
+
+    /**
+     * Setup is called once upon system startup to initialize the IAuthorizer.
+     *
+     * For example, use this method to create any required keyspaces/column families.
+     */
+    static future<> setup(const sstring& type);
+
+    /**
+     * Returns the system authorizer. Must have called setup before calling this.
+     */
+    static authorizer& get();
+};
+
+}

auth/data_resource.cc

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2016 Cloudius Systems
+ * Copyright (C) 2016 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*
@@ -158,7 +158,15 @@ bool auth::data_resource::exists() const {
 }
 
 sstring auth::data_resource::to_string() const {
-    return name();
+    switch (get_level()) {
+        case level::ROOT:
+            return "<all keyspaces>";
+        case level::KEYSPACE:
+            return sprint("<keyspace %s>", _ks);
+        case level::COLUMN_FAMILY:
+        default:
+            return sprint("<table %s.%s>", _ks, _cf);
+    }
 }
 
 bool auth::data_resource::operator==(const data_resource& v) const {
@@ -170,6 +178,6 @@ bool auth::data_resource::operator<(const data_resource& v) const {
 }
 
 std::ostream& auth::operator<<(std::ostream& os, const data_resource& r) {
-    return os << r.name();
+    return os << r.to_string();
 }
 

auth/data_resource.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2016 Cloudius Systems
+ * Copyright (C) 2016 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*
@@ -41,7 +41,9 @@
 
 #pragma once
 
+#include "utils/hash.hh"
 #include <iosfwd>
+#include <set>
 #include <seastar/core/sstring.hh>
 
 namespace auth {
@@ -136,8 +138,17 @@ public:
 
     bool operator==(const data_resource&) const;
     bool operator<(const data_resource&) const;
+
+    size_t hash_value() const {
+        return utils::tuple_hash()(_ks, _cf);
+    }
 };
 
+/**
+ * Resource id mappings, i.e. keyspace and/or column families.
+ */
+using resource_ids = std::set<data_resource>;
+
 std::ostream& operator<<(std::ostream&, const data_resource&);
 
 }

auth/default_authorizer.cc

@@ -0,0 +1,240 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/*
+ * Copyright (C) 2016 ScyllaDB
+ *
+ * Modified by ScyllaDB
+ */
+
+/*
+ * This file is part of Scylla.
+ *
+ * Scylla is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Scylla is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <unistd.h>
+#include <crypt.h>
+#include <random>
+#include <chrono>
+
+#include <seastar/core/reactor.hh>
+
+#include "auth.hh"
+#include "default_authorizer.hh"
+#include "authenticated_user.hh"
+#include "permission.hh"
+#include "cql3/query_processor.hh"
+#include "exceptions/exceptions.hh"
+#include "log.hh"
+
+const sstring auth::default_authorizer::DEFAULT_AUTHORIZER_NAME(
+                "org.apache.cassandra.auth.CassandraAuthorizer");
+
+static const sstring USER_NAME = "username";
+static const sstring RESOURCE_NAME = "resource";
+static const sstring PERMISSIONS_NAME = "permissions";
+static const sstring PERMISSIONS_CF = "permissions";
+
+static logging::logger logger("default_authorizer");
+
+auth::default_authorizer::default_authorizer() {
+}
+auth::default_authorizer::~default_authorizer() {
+}
+
+future<> auth::default_authorizer::init() {
+    sstring create_table = sprint("CREATE TABLE %s.%s ("
+                    "%s text,"
+                    "%s text,"
+                    "%s set<text>,"
+                    "PRIMARY KEY(%s, %s)"
+                    ") WITH gc_grace_seconds=%d", auth::auth::AUTH_KS,
+                    PERMISSIONS_CF, USER_NAME, RESOURCE_NAME, PERMISSIONS_NAME,
+                    USER_NAME, RESOURCE_NAME, 90 * 24 * 60 * 60); // 3 months.
+
+    return auth::setup_table(PERMISSIONS_CF, create_table);
+}
+
+
+future<auth::permission_set> auth::default_authorizer::authorize(
+                ::shared_ptr<authenticated_user> user, data_resource resource) const {
+    return user->is_super().then([this, user, resource = std::move(resource)](bool is_super) {
+        if (is_super) {
+            return make_ready_future<permission_set>(permissions::ALL);
+        }
+
+        /**
+         * TOOD: could create actual data type for permission (translating string<->perm),
+         * but this seems overkill right now. We still must store strings so...
+         */
+        auto& qp = cql3::get_local_query_processor();
+        auto query = sprint("SELECT %s FROM %s.%s WHERE %s = ? AND %s = ?"
+                        , PERMISSIONS_NAME, auth::AUTH_KS, PERMISSIONS_CF, USER_NAME, RESOURCE_NAME);
+        return qp.process(query, db::consistency_level::LOCAL_ONE, {user->name(), resource.name() })
+                        .then_wrapped([=](future<::shared_ptr<cql3::untyped_result_set>> f) {
+            try {
+                auto res = f.get0();
+
+                if (res->empty() || !res->one().has(PERMISSIONS_NAME)) {
+                    return make_ready_future<permission_set>(permissions::NONE);
+                }
+                return make_ready_future<permission_set>(permissions::from_strings(res->one().get_set<sstring>(PERMISSIONS_NAME)));
+            } catch (exceptions::request_execution_exception& e) {
+                logger.warn("CassandraAuthorizer failed to authorize {} for {}", user->name(), resource);
+                return make_ready_future<permission_set>(permissions::NONE);
+            }
+        });
+    });
+}
+
+#include <boost/range.hpp>
+
+future<> auth::default_authorizer::modify(
+                ::shared_ptr<authenticated_user> performer, permission_set set,
+                data_resource resource, sstring user, sstring op) {
+    // TODO: why does this not check super user?
+    auto& qp = cql3::get_local_query_processor();
+    auto query = sprint("UPDATE %s.%s SET %s = %s %s ? WHERE %s = ? AND %s = ?",
+                    auth::AUTH_KS, PERMISSIONS_CF, PERMISSIONS_NAME,
+                    PERMISSIONS_NAME, op, USER_NAME, RESOURCE_NAME);
+    return qp.process(query, db::consistency_level::ONE, {
+                    permissions::to_strings(set), user, resource.name() }).discard_result();
+}
+
+
+future<> auth::default_authorizer::grant(
+                ::shared_ptr<authenticated_user> performer, permission_set set,
+                data_resource resource, sstring to) {
+    return modify(std::move(performer), std::move(set), std::move(resource), std::move(to), "+");
+}
+
+future<> auth::default_authorizer::revoke(
+                ::shared_ptr<authenticated_user> performer, permission_set set,
+                data_resource resource, sstring from) {
+    return modify(std::move(performer), std::move(set), std::move(resource), std::move(from), "-");
+}
+
+future<std::vector<auth::permission_details>> auth::default_authorizer::list(
+                ::shared_ptr<authenticated_user> performer, permission_set set,
+                optional<data_resource> resource, optional<sstring> user) const {
+    return performer->is_super().then([this, performer, set = std::move(set), resource = std::move(resource), user = std::move(user)](bool is_super) {
+        if (!is_super && (!user || performer->name() != *user)) {
+            throw exceptions::unauthorized_exception(sprint("You are not authorized to view %s's permissions", user ? *user : "everyone"));
+        }
+
+        auto query = sprint("SELECT %s, %s, %s FROM %s.%s", USER_NAME, RESOURCE_NAME, PERMISSIONS_NAME, auth::AUTH_KS, PERMISSIONS_CF);
+        auto& qp = cql3::get_local_query_processor();
+
+        // Oh, look, it is a case where it does not pay off to have
+        // parameters to process in an initializer list.
+        future<::shared_ptr<cql3::untyped_result_set>> f = make_ready_future<::shared_ptr<cql3::untyped_result_set>>();
+
+        if (resource && user) {
+            query += sprint(" WHERE %s = ? AND %s = ?", USER_NAME, RESOURCE_NAME);
+            f = qp.process(query, db::consistency_level::ONE, {*user, resource->name()});
+        } else if (resource) {
+            query += sprint(" WHERE %s = ? ALLOW FILTERING", RESOURCE_NAME);
+            f = qp.process(query, db::consistency_level::ONE, {resource->name()});
+        } else if (user) {
+            query += sprint(" WHERE %s = ?", USER_NAME);
+            f = qp.process(query, db::consistency_level::ONE, {*user});
+        } else {
+            f = qp.process(query, db::consistency_level::ONE, {});
+        }
+
+        return f.then([set](::shared_ptr<cql3::untyped_result_set> res) {
+            std::vector<permission_details> result;
+
+            for (auto& row : *res) {
+                if (row.has(PERMISSIONS_NAME)) {
+                    auto username = row.get_as<sstring>(USER_NAME);
+                    auto resource = data_resource::from_name(row.get_as<sstring>(RESOURCE_NAME));
+                    auto ps = permissions::from_strings(row.get_set<sstring>(PERMISSIONS_NAME));
+                    ps = permission_set::from_mask(ps.mask() & set.mask());
+
+                    result.emplace_back(permission_details {username, resource, ps});
+                }
+            }
+            return make_ready_future<std::vector<permission_details>>(std::move(result));
+        });
+    });
+}
+
+future<> auth::default_authorizer::revoke_all(sstring dropped_user) {
+    auto& qp = cql3::get_local_query_processor();
+    auto query = sprint("DELETE FROM %s.%s WHERE %s = ?", auth::AUTH_KS,
+                    PERMISSIONS_CF, USER_NAME);
+    return qp.process(query, db::consistency_level::ONE, { dropped_user }).discard_result().handle_exception(
+                    [dropped_user](auto ep) {
+                        try {
+                            std::rethrow_exception(ep);
+                        } catch (exceptions::request_execution_exception& e) {
+                            logger.warn("CassandraAuthorizer failed to revoke all permissions of {}: {}", dropped_user, e);
+                        }
+                    });
+}
+
+future<> auth::default_authorizer::revoke_all(data_resource resource) {
+    auto& qp = cql3::get_local_query_processor();
+    auto query = sprint("SELECT %s FROM %s.%s WHERE %s = ? ALLOW FILTERING",
+                    USER_NAME, auth::AUTH_KS, PERMISSIONS_CF, RESOURCE_NAME);
+    return qp.process(query, db::consistency_level::LOCAL_ONE, { resource.name() })
+                    .then_wrapped([resource, &qp](future<::shared_ptr<cql3::untyped_result_set>> f) {
+        try {
+            auto res = f.get0();
+            return parallel_for_each(res->begin(), res->end(), [&qp, res, resource](const cql3::untyped_result_set::row& r) {
+                auto query = sprint("DELETE FROM %s.%s WHERE %s = ? AND %s = ?"
+                                , auth::AUTH_KS, PERMISSIONS_CF, USER_NAME, RESOURCE_NAME);
+                return qp.process(query, db::consistency_level::LOCAL_ONE, { r.get_as<sstring>(USER_NAME), resource.name() })
+                                .discard_result().handle_exception([resource](auto ep) {
+                    try {
+                        std::rethrow_exception(ep);
+                    } catch (exceptions::request_execution_exception& e) {
+                        logger.warn("CassandraAuthorizer failed to revoke all permissions on {}: {}", resource, e);
+                    }
+
+                });
+            });
+        } catch (exceptions::request_execution_exception& e) {
+            logger.warn("CassandraAuthorizer failed to revoke all permissions on {}: {}", resource, e);
+            return make_ready_future();
+        }
+    });
+}
+
+
+const auth::resource_ids& auth::default_authorizer::protected_resources() {
+    static const resource_ids ids({ data_resource(auth::AUTH_KS, PERMISSIONS_CF) });
+    return ids;
+}
+
+future<> auth::default_authorizer::validate_configuration() const {
+    return make_ready_future();
+}

auth/default_authorizer.hh

@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/*
+ * Copyright (C) 2016 ScyllaDB
+ *
+ * Modified by ScyllaDB
+ */
+
+/*
+ * This file is part of Scylla.
+ *
+ * Scylla is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Scylla is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include "authorizer.hh"
+
+namespace auth {
+
+class default_authorizer : public authorizer {
+public:
+    static const sstring DEFAULT_AUTHORIZER_NAME;
+
+    default_authorizer();
+    ~default_authorizer();
+
+    future<> init();
+
+    future<permission_set> authorize(::shared_ptr<authenticated_user>, data_resource) const override;
+
+    future<> grant(::shared_ptr<authenticated_user>, permission_set, data_resource, sstring) override;
+
+    future<> revoke(::shared_ptr<authenticated_user>, permission_set, data_resource, sstring) override;
+
+    future<std::vector<permission_details>> list(::shared_ptr<authenticated_user>, permission_set, optional<data_resource>, optional<sstring>) const override;
+
+    future<> revoke_all(sstring) override;
+
+    future<> revoke_all(data_resource) override;
+
+    const resource_ids& protected_resources() override;
+
+    future<> validate_configuration() const override;
+
+private:
+    future<> modify(::shared_ptr<authenticated_user>, permission_set, data_resource, sstring, sstring);
+};
+
+} /* namespace auth */
+

auth/password_authenticator.cc

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2016 Cloudius Systems
+ * Copyright (C) 2016 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*
@@ -281,8 +281,9 @@ future<> auth::password_authenticator::drop(sstring username)
     }
 }
 
-auth::authenticator::resource_ids auth::password_authenticator::protected_resources() const {
-    return { data_resource(auth::AUTH_KS, CREDENTIALS_CF) };
+const auth::resource_ids& auth::password_authenticator::protected_resources() const {
+    static const resource_ids ids({ data_resource(auth::AUTH_KS, CREDENTIALS_CF) });
+    return ids;
 }
 
 ::shared_ptr<auth::authenticator::sasl_challenge> auth::password_authenticator::new_sasl_challenge() const {

auth/password_authenticator.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2016 Cloudius Systems
+ * Copyright (C) 2016 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*
@@ -62,7 +62,7 @@ public:
     future<> create(sstring username, const option_map& options) throw(exceptions::request_validation_exception, exceptions::request_execution_exception) override;
     future<> alter(sstring username, const option_map& options) throw(exceptions::request_validation_exception, exceptions::request_execution_exception) override;
     future<> drop(sstring username) throw(exceptions::request_validation_exception, exceptions::request_execution_exception) override;
-    resource_ids protected_resources() const override;
+    const resource_ids& protected_resources() const override;
     ::shared_ptr<sasl_challenge> new_sasl_challenge() const override;
 
 

auth/permission.cc

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2016 Cloudius Systems
+ * Copyright (C) 2016 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*
@@ -39,11 +39,63 @@
  * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include <unordered_map>
 #include "permission.hh"
 
-const auth::permission_set auth::ALL_DATA = auth::permission_set::of
-                < auth::permission::CREATE, auth::permission::ALTER,
-                auth::permission::DROP, auth::permission::SELECT,
-                auth::permission::MODIFY, auth::permission::AUTHORIZE>();
-const auth::permission_set auth::ALL = auth::ALL_DATA;
-const auth::permission_set auth::NONE;
+const auth::permission_set auth::permissions::ALL_DATA =
+                auth::permission_set::of<auth::permission::CREATE,
+                                auth::permission::ALTER, auth::permission::DROP,
+                                auth::permission::SELECT,
+                                auth::permission::MODIFY,
+                                auth::permission::AUTHORIZE>();
+const auth::permission_set auth::permissions::ALL = auth::permissions::ALL_DATA;
+const auth::permission_set auth::permissions::NONE;
+const auth::permission_set auth::permissions::ALTERATIONS =
+                auth::permission_set::of<auth::permission::CREATE,
+                                auth::permission::ALTER, auth::permission::DROP>();
+
+static const std::unordered_map<sstring, auth::permission> permission_names({
+    { "READ", auth::permission::READ },
+    { "WRITE", auth::permission::WRITE  },
+    { "CREATE", auth::permission::CREATE },
+    { "ALTER", auth::permission::ALTER },
+    { "DROP", auth::permission::DROP },
+    { "SELECT", auth::permission::SELECT  },
+    { "MODIFY", auth::permission::MODIFY   },
+    { "AUTHORIZE", auth::permission::AUTHORIZE },
+});
+
+const sstring& auth::permissions::to_string(permission p) {
+    for (auto& v : permission_names) {
+        if (v.second == p) {
+            return v.first;
+        }
+    }
+    throw std::out_of_range("unknown permission");
+}
+
+auth::permission auth::permissions::from_string(const sstring& s) {
+    return permission_names.at(s);
+}
+
+std::unordered_set<sstring> auth::permissions::to_strings(const permission_set& set) {
+    std::unordered_set<sstring> res;
+    for (auto& v : permission_names) {
+        if (set.contains(v.second)) {
+            res.emplace(v.first);
+        }
+    }
+    return res;
+}
+
+auth::permission_set auth::permissions::from_strings(const std::unordered_set<sstring>& set) {
+    permission_set res = auth::permissions::NONE;
+    for (auto& s : set) {
+        res.set(from_string(s));
+    }
+    return res;
+}
+
+bool auth::operator<(const permission_set& p1, const permission_set& p2) {
+    return p1.mask() < p2.mask();
+}

auth/permission.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2016 Cloudius Systems
+ * Copyright (C) 2016 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*
@@ -41,6 +41,9 @@
 
 #pragma once
 
+#include <unordered_set>
+#include <seastar/core/sstring.hh>
+
 #include "enum_set.hh"
 
 namespace auth {
@@ -74,8 +77,22 @@ typedef enum_set<super_enum<permission,
                 permission::MODIFY,
                 permission::AUTHORIZE>> permission_set;
 
+bool operator<(const permission_set&, const permission_set&);
+
+namespace permissions {
+
 extern const permission_set ALL_DATA;
 extern const permission_set ALL;
 extern const permission_set NONE;
+extern const permission_set ALTERATIONS;
+
+const sstring& to_string(permission);
+permission from_string(const sstring&);
+
+std::unordered_set<sstring> to_strings(const permission_set&);
+permission_set from_strings(const std::unordered_set<sstring>&);
+
+
+}
 
 }

bytes.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2014 Cloudius Systems, Ltd.
+ * Copyright (C) 2014 ScyllaDB
  */
 
 /*

bytes.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2015 Cloudius Systems, Ltd.
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

bytes_ostream.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

caching_options.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2015 Cloudius Systems, Ltd.
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

cartesian_product.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  *
  */
 

checked-file-impl.hh

@@ -0,0 +1,147 @@
+/*
+ * Copyright (C) 2016 ScyllaDB
+ */
+
+/*
+ * This file is part of Scylla.
+ *
+ * Scylla is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Scylla is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include "seastar/core/file.hh"
+#include "disk-error-handler.hh"
+
+class checked_file_impl : public file_impl {
+public:
+
+    checked_file_impl(disk_error_signal_type& s, file f)
+            : _signal(s) , _file(f) {}
+
+    virtual future<size_t> write_dma(uint64_t pos, const void* buffer, size_t len, const io_priority_class& pc) override {
+        return do_io_check(_signal, [&] {
+            return get_file_impl(_file)->write_dma(pos, buffer, len, pc);
+        });
+    }
+
+    virtual future<size_t> write_dma(uint64_t pos, std::vector<iovec> iov, const io_priority_class& pc) override {
+        return do_io_check(_signal, [&] {
+            return get_file_impl(_file)->write_dma(pos, iov, pc);
+        });
+    }
+
+    virtual future<size_t> read_dma(uint64_t pos, void* buffer, size_t len, const io_priority_class& pc) override {
+        return do_io_check(_signal, [&] {
+            return get_file_impl(_file)->read_dma(pos, buffer, len, pc);
+        });
+    }
+
+    virtual future<size_t> read_dma(uint64_t pos, std::vector<iovec> iov, const io_priority_class& pc) override {
+        return do_io_check(_signal, [&] {
+            return get_file_impl(_file)->read_dma(pos, iov, pc);
+        });
+    }
+
+    virtual future<> flush(void) override {
+        return do_io_check(_signal, [&] {
+            return get_file_impl(_file)->flush();
+        });
+    }
+
+    virtual future<struct stat> stat(void) override {
+        return do_io_check(_signal, [&] {
+            return get_file_impl(_file)->stat();
+        });
+    }
+
+    virtual future<> truncate(uint64_t length) override {
+        return do_io_check(_signal, [&] {
+            return get_file_impl(_file)->truncate(length);
+        });
+    }
+
+    virtual future<> discard(uint64_t offset, uint64_t length) override {
+        return do_io_check(_signal, [&] {
+            return get_file_impl(_file)->discard(offset, length);
+        });
+    }
+
+    virtual future<> allocate(uint64_t position, uint64_t length) override {
+        return do_io_check(_signal, [&] {
+            return get_file_impl(_file)->allocate(position, length);
+        });
+    }
+
+    virtual future<uint64_t> size(void) override {
+        return do_io_check(_signal, [&] {
+            return get_file_impl(_file)->size();
+        });
+    }
+
+    virtual future<> close() override {
+        return do_io_check(_signal, [&] {
+            return get_file_impl(_file)->close();
+        });
+    }
+
+    virtual subscription<directory_entry> list_directory(std::function<future<> (directory_entry de)> next) override {
+        return do_io_check(_signal, [&] {
+            return get_file_impl(_file)->list_directory(next);
+        });
+    }
+
+private:
+    disk_error_signal_type &_signal;
+    file _file;
+};
+
+inline file make_checked_file(disk_error_signal_type& signal, file& f)
+{
+    return file(::make_shared<checked_file_impl>(signal, f));
+}
+
+future<file>
+inline open_checked_file_dma(disk_error_signal_type& signal,
+                             sstring name, open_flags flags,
+                             file_open_options options)
+{
+    return do_io_check(signal, [&] {
+        return open_file_dma(name, flags, options).then([&] (file f) {
+            return make_ready_future<file>(make_checked_file(signal, f));
+        });
+    });
+}
+
+future<file>
+inline open_checked_file_dma(disk_error_signal_type& signal,
+                             sstring name, open_flags flags)
+{
+    return do_io_check(signal, [&] {
+        return open_file_dma(name, flags).then([&] (file f) {
+            return make_ready_future<file>(make_checked_file(signal, f));
+        });
+    });
+}
+
+future<file>
+inline open_checked_directory(disk_error_signal_type& signal,
+                              sstring name)
+{
+    return do_io_check(signal, [&] {
+        return engine().open_directory(name).then([&] (file f) {
+            return make_ready_future<file>(make_checked_file(signal, f));
+        });
+    });
+}

combine.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2015 Cloudius Systems, Ltd.
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

compaction_strategy.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

compound.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2015 Cloudius Systems, Ltd.
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

compound_compat.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

compress.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2015 Cloudius Systems, Ltd.
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

conf/scylla.yaml

@@ -106,6 +106,19 @@ write_request_timeout_in_ms: 2000
 # most users should never need to adjust this.
 # phi_convict_threshold: 8
 
+# IEndpointSnitch.  The snitch has two functions:
+# - it teaches Scylla enough about your network topology to route
+#   requests efficiently
+# - it allows Scylla to spread replicas around your cluster to avoid
+#   correlated failures. It does this by grouping machines into
+#   "datacenters" and "racks."  Scylla will do its best not to have
+#   more than one replica on the same "rack" (which may not actually
+#   be a physical location)
+#
+# IF YOU CHANGE THE SNITCH AFTER DATA IS INSERTED INTO THE CLUSTER,
+# YOU MUST RUN A FULL REPAIR, SINCE THE SNITCH AFFECTS WHERE REPLICAS
+# ARE PLACED.
+#
 # Out of the box, Scylla provides
 #  - SimpleSnitch:
 #    Treats Strategy order as proximity. This can improve cache
@@ -179,6 +192,24 @@ api_address: 127.0.0.1
 # Caution should be taken on increasing the size of this threshold as it can lead to node instability.
 batch_size_warn_threshold_in_kb: 5
 
+# Authentication backend, identifying users
+# Out of the box, Scylla provides org.apache.cassandra.auth.{AllowAllAuthenticator,
+# PasswordAuthenticator}.
+#
+# - AllowAllAuthenticator performs no checks - set it to disable authentication.
+# - PasswordAuthenticator relies on username/password pairs to authenticate
+#   users. It keeps usernames and hashed passwords in system_auth.credentials table.
+#   Please increase system_auth keyspace replication factor if you use this authenticator.
+# authenticator: AllowAllAuthenticator
+
+# Authorization backend, implementing IAuthorizer; used to limit access/provide permissions
+# Out of the box, Scylla provides org.apache.cassandra.auth.{AllowAllAuthorizer,
+# CassandraAuthorizer}.
+#
+# - AllowAllAuthorizer allows any action to any user - set it to disable authorization.
+# - CassandraAuthorizer stores permissions in system_auth.permissions table. Please
+#   increase system_auth keyspace replication factor if you use this authorizer.
+# authorizer: AllowAllAuthorizer
 
 ###################################################
 ## Not currently supported, reserved for future use
@@ -216,25 +247,6 @@ batch_size_warn_threshold_in_kb: 5
 # reduced proportionally to the number of nodes in the cluster.
 # batchlog_replay_throttle_in_kb: 1024
 
-# Authentication backend, identifying users
-# Out of the box, Scylla provides org.apache.cassandra.auth.{AllowAllAuthenticator,
-# PasswordAuthenticator}.
-#
-# - AllowAllAuthenticator performs no checks - set it to disable authentication.
-# - PasswordAuthenticator relies on username/password pairs to authenticate
-#   users. It keeps usernames and hashed passwords in system_auth.credentials table.
-#   Please increase system_auth keyspace replication factor if you use this authenticator.
-# authenticator: AllowAllAuthenticator
-
-# Authorization backend, implementing IAuthorizer; used to limit access/provide permissions
-# Out of the box, Scylla provides org.apache.cassandra.auth.{AllowAllAuthorizer,
-# CassandraAuthorizer}.
-#
-# - AllowAllAuthorizer allows any action to any user - set it to disable authorization.
-# - CassandraAuthorizer stores permissions in system_auth.permissions table. Please
-#   increase system_auth keyspace replication factor if you use this authorizer.
-# authorizer: AllowAllAuthorizer
-
 # Validity period for permissions cache (fetching permissions can be an
 # expensive operation depending on the authorizer, CassandraAuthorizer is
 # one example). Defaults to 2000, set to 0 to disable.
@@ -680,58 +692,6 @@ commitlog_total_space_in_mb: -1
 # Default value is 0, which never timeout streams.
 # streaming_socket_timeout_in_ms: 0
 
-
-# endpoint_snitch -- Set this to a class that implements
-# IEndpointSnitch.  The snitch has two functions:
-# - it teaches Scylla enough about your network topology to route
-#   requests efficiently
-# - it allows Scylla to spread replicas around your cluster to avoid
-#   correlated failures. It does this by grouping machines into
-#   "datacenters" and "racks."  Scylla will do its best not to have
-#   more than one replica on the same "rack" (which may not actually
-#   be a physical location)
-#
-# IF YOU CHANGE THE SNITCH AFTER DATA IS INSERTED INTO THE CLUSTER,
-# YOU MUST RUN A FULL REPAIR, SINCE THE SNITCH AFFECTS WHERE REPLICAS
-# ARE PLACED.
-#
-# Out of the box, Scylla provides
-#  - SimpleSnitch:
-#    Treats Strategy order as proximity. This can improve cache
-#    locality when disabling read repair.  Only appropriate for
-#    single-datacenter deployments.
-#  - GossipingPropertyFileSnitch
-#    This should be your go-to snitch for production use.  The rack
-#    and datacenter for the local node are defined in
-#    cassandra-rackdc.properties and propagated to other nodes via
-#    gossip.  If cassandra-topology.properties exists, it is used as a
-#    fallback, allowing migration from the PropertyFileSnitch.
-#  - PropertyFileSnitch:
-#    Proximity is determined by rack and data center, which are
-#    explicitly configured in cassandra-topology.properties.
-#  - Ec2Snitch:
-#    Appropriate for EC2 deployments in a single Region. Loads Region
-#    and Availability Zone information from the EC2 API. The Region is
-#    treated as the datacenter, and the Availability Zone as the rack.
-#    Only private IPs are used, so this will not work across multiple
-#    Regions.
-#  - Ec2MultiRegionSnitch:
-#    Uses public IPs as broadcast_address to allow cross-region
-#    connectivity.  (Thus, you should set seed addresses to the public
-#    IP as well.) You will need to open the storage_port or
-#    ssl_storage_port on the public IP firewall.  (For intra-Region
-#    traffic, Scylla will switch to the private IP after
-#    establishing a connection.)
-#  - RackInferringSnitch:
-#    Proximity is determined by rack and data center, which are
-#    assumed to correspond to the 3rd and 2nd octet of each node's IP
-#    address, respectively.  Unless this happens to match your
-#    deployment conventions, this is best used as an example of
-#    writing a custom Snitch class and is provided in that spirit.
-#
-# You can use a custom Snitch by setting this to the full class name
-# of the snitch, which will be assumed to be on your classpath.
-
 # controls how often to perform the more expensive part of host score
 # calculation
 # dynamic_snitch_update_interval_in_ms: 100 

configure.py

@@ -1,6 +1,6 @@
 #!/usr/bin/python3
 #
-# Copyright 2015 Cloudius Systems
+# Copyright (C) 2015 ScyllaDB
 #
 
 #
@@ -162,6 +162,7 @@ modes = {
 
 scylla_tests = [
     'tests/mutation_test',
+    'tests/schema_registry_test',
     'tests/canonical_mutation_test',
     'tests/range_test',
     'tests/types_test',
@@ -255,6 +256,8 @@ arg_parser.add_argument('--debuginfo', action = 'store', dest = 'debuginfo', typ
                         help = 'Enable(1)/disable(0)compiler debug information generation')
 arg_parser.add_argument('--static-stdc++', dest = 'staticcxx', action = 'store_true',
 			help = 'Link libgcc and libstdc++ statically')
+arg_parser.add_argument('--static-thrift', dest = 'staticthrift', action = 'store_true',
+			help = 'Link libthrift statically')
 arg_parser.add_argument('--tests-debuginfo', action = 'store', dest = 'tests_debuginfo', type = int, default = 0,
                         help = 'Enable(1)/disable(0)compiler debug information generation for tests')
 arg_parser.add_argument('--python', action = 'store', dest = 'python', default = 'python3',
@@ -264,7 +267,6 @@ add_tristate(arg_parser, name = 'xen', dest = 'xen', help = 'Xen support')
 args = arg_parser.parse_args()
 
 defines = []
-scylla_libs = '-llz4 -lsnappy -lz -lboost_thread -lcryptopp -lrt -lyaml-cpp -lboost_date_time'
 
 extra_cxxflags = {}
 
@@ -298,6 +300,7 @@ scylla_core = (['database.cc',
                  'sstables/partition.cc',
                  'sstables/filter.cc',
                  'sstables/compaction.cc',
+                 'sstables/compaction_strategy.cc',
                  'sstables/compaction_manager.cc',
                  'log.cc',
                  'transport/event.cc',
@@ -316,11 +319,14 @@ scylla_core = (['database.cc',
                  'cql3/functions/functions.cc',
                  'cql3/statements/cf_prop_defs.cc',
                  'cql3/statements/cf_statement.cc',
+                 'cql3/statements/authentication_statement.cc',
                  'cql3/statements/create_keyspace_statement.cc',
                  'cql3/statements/create_table_statement.cc',
                  'cql3/statements/create_type_statement.cc',
+                 'cql3/statements/create_user_statement.cc',
                  'cql3/statements/drop_keyspace_statement.cc',
                  'cql3/statements/drop_table_statement.cc',
+                 'cql3/statements/drop_type_statement.cc',
                  'cql3/statements/schema_altering_statement.cc',
                  'cql3/statements/ks_prop_defs.cc',
                  'cql3/statements/modification_statement.cc',
@@ -336,8 +342,18 @@ scylla_core = (['database.cc',
                  'cql3/statements/create_index_statement.cc',
                  'cql3/statements/truncate_statement.cc',
                  'cql3/statements/alter_table_statement.cc',
+                 'cql3/statements/alter_user_statement.cc',
+                 'cql3/statements/drop_user_statement.cc',
+                 'cql3/statements/list_users_statement.cc',
+                 'cql3/statements/authorization_statement.cc',
+                 'cql3/statements/permission_altering_statement.cc',
+                 'cql3/statements/list_permissions_statement.cc',
+                 'cql3/statements/grant_statement.cc',
+                 'cql3/statements/revoke_statement.cc',
+                 'cql3/statements/alter_type_statement.cc',
                  'cql3/update_parameters.cc',
                  'cql3/ut_name.cc',
+                 'cql3/user_options.cc',
                  'thrift/handler.cc',
                  'thrift/server.cc',
                  'thrift/thrift_validation.cc',
@@ -388,6 +404,7 @@ scylla_core = (['database.cc',
                  'utils/file_lock.cc',
                  'utils/dynamic_bitset.cc',
                  'utils/managed_bytes.cc',
+                 'utils/exceptions.cc',
                  'gms/version_generator.cc',
                  'gms/versioned_value.cc',
                  'gms/gossiper.cc',
@@ -423,7 +440,6 @@ scylla_core = (['database.cc',
                  'service/client_state.cc',
                  'service/migration_task.cc',
                  'service/storage_service.cc',
-                 'service/pending_range_calculator_service.cc',
                  'service/load_broadcaster.cc',
                  'service/pager/paging_state.cc',
                  'service/pager/query_pagers.cc',
@@ -449,6 +465,8 @@ scylla_core = (['database.cc',
                  'auth/auth.cc',
                  'auth/authenticated_user.cc',
                  'auth/authenticator.cc',
+                 'auth/authorizer.cc',
+                 'auth/default_authorizer.cc',
                  'auth/data_resource.cc',
                  'auth/password_authenticator.cc',
                  'auth/permission.cc',
@@ -699,7 +717,7 @@ for mode in build_modes:
 seastar_deps = 'practically_anything_can_change_so_lets_run_it_every_time_and_restat.'
 
 args.user_cflags += " " + pkg_config("--cflags", "jsoncpp")
-libs = "-lyaml-cpp -llz4 -lz -lsnappy " + pkg_config("--libs", "jsoncpp") + ' -lboost_filesystem' + ' -lcrypt'
+libs = "-lyaml-cpp -llz4 -lz -lsnappy " + pkg_config("--libs", "jsoncpp") + ' -lboost_filesystem' + ' -lcrypt' + ' -lboost_date_time'
 for pkg in pkgs:
     args.user_cflags += ' ' + pkg_config('--cflags', pkg)
     libs += ' ' + pkg_config('--libs', pkg)
@@ -707,6 +725,10 @@ user_cflags = args.user_cflags
 user_ldflags = args.user_ldflags
 if args.staticcxx:
     user_ldflags += " -static-libgcc -static-libstdc++"
+if args.staticthrift:
+    thrift_libs = "-Wl,-Bstatic -lthrift -Wl,-Bdynamic"
+else:
+    thrift_libs = "-lthrift"
 
 outdir = 'build'
 buildfile = 'build.ninja'
@@ -817,14 +839,14 @@ with open(buildfile, 'w') as f:
                     f.write('build $builddir/{}/{}: {}.{} {} {}\n'.format(mode, binary, tests_link_rule, mode, str.join(' ', objs),
                                                                                      'seastar/build/{}/libseastar.a'.format(mode)))
                     if has_thrift:
-                        f.write('   libs =  -lthrift -lboost_system $libs\n')
+                        f.write('   libs =  {} -lboost_system $libs\n'.format(thrift_libs))
                     f.write('build $builddir/{}/{}_g: link.{} {} {}\n'.format(mode, binary, mode, str.join(' ', objs),
                                                                               'seastar/build/{}/libseastar.a'.format(mode)))
                 else:
                     f.write('build $builddir/{}/{}: link.{} {} {}\n'.format(mode, binary, mode, str.join(' ', objs),
                                                                             'seastar/build/{}/libseastar.a'.format(mode)))
                 if has_thrift:
-                    f.write('   libs =  -lthrift -lboost_system $libs\n')
+                    f.write('   libs =  {} -lboost_system $libs\n'.format(thrift_libs))
             for src in srcs:
                 if src.endswith('.cc'):
                     obj = '$builddir/' + mode + '/' + src.replace('.cc', '.o')
@@ -847,7 +869,8 @@ with open(buildfile, 'w') as f:
         for obj in compiles:
             src = compiles[obj]
             gen_headers = list(ragels.keys())
-            gen_headers += ['seastar/build/{}/http/request_parser.hh'.format(mode)]
+            gen_headers += ['seastar/build/{}/gen/http/request_parser.hh'.format(mode)]
+            gen_headers += ['seastar/build/{}/gen/http/http_response_parser.hh'.format(mode)]
             for th in thrifts:
                 gen_headers += th.headers('$builddir/{}/gen'.format(mode))
             for g in antlr3_grammars:
@@ -879,10 +902,10 @@ with open(buildfile, 'w') as f:
             for cc in grammar.sources('$builddir/{}/gen'.format(mode)):
                 obj = cc.replace('.cpp', '.o')
                 f.write('build {}: cxx.{} {} || {}\n'.format(obj, mode, cc, ' '.join(serializers)))
-        f.write('build seastar/build/{mode}/libseastar.a seastar/build/{mode}/apps/iotune/iotune: ninja {seastar_deps}\n'
+        f.write('build seastar/build/{mode}/libseastar.a seastar/build/{mode}/apps/iotune/iotune seastar/build/{mode}/gen/http/request_parser.hh seastar/build/{mode}/gen/http/http_response_parser.hh: ninja {seastar_deps}\n'
                 .format(**locals()))
         f.write('  subdir = seastar\n')
-        f.write('  target = build/{mode}/libseastar.a build/{mode}/apps/iotune/iotune\n'.format(**locals()))
+        f.write('  target = build/{mode}/libseastar.a build/{mode}/apps/iotune/iotune build/{mode}/gen/http/request_parser.hh build/{mode}/gen/http/http_response_parser.hh\n'.format(**locals()))
         f.write(textwrap.dedent('''\
             build build/{mode}/iotune: copy seastar/build/{mode}/apps/iotune/iotune
             ''').format(**locals()))
@@ -896,10 +919,6 @@ with open(buildfile, 'w') as f:
             command = find -name '*.[chS]' -o -name "*.cc" -o -name "*.hh" | cscope -bq -i-
             description = CSCOPE
         build cscope: cscope
-        rule request_parser_hh
-           command = {ninja} -C seastar build/release/gen/http/request_parser.hh build/debug/gen/http/request_parser.hh
-           description = GEN seastar/http/request_parser.hh
-        build seastar/build/release/http/request_parser.hh seastar/build/debug/http/request_parser.hh: request_parser_hh
         rule clean
             command = rm -rf build
             description = CLEAN

converting_mutation_partition_applier.hh

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2015 Cloudius Systems, Ltd.
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

cql3/Cql.g

@@ -26,6 +26,7 @@ options {
 @parser::namespace{cql3_parser}
 
 @lexer::includes {
+#include "cql3/error_collector.hh"
 #include "cql3/error_listener.hh"
 }
 
@@ -37,6 +38,8 @@ options {
 #include "cql3/statements/create_index_statement.hh"
 #include "cql3/statements/create_table_statement.hh"
 #include "cql3/statements/create_type_statement.hh"
+#include "cql3/statements/drop_type_statement.hh"
+#include "cql3/statements/alter_type_statement.hh"
 #include "cql3/statements/property_definitions.hh"
 #include "cql3/statements/drop_table_statement.hh"
 #include "cql3/statements/truncate_statement.hh"
@@ -46,6 +49,13 @@ options {
 #include "cql3/statements/index_prop_defs.hh"
 #include "cql3/statements/use_statement.hh"
 #include "cql3/statements/batch_statement.hh"
+#include "cql3/statements/create_user_statement.hh"
+#include "cql3/statements/alter_user_statement.hh"
+#include "cql3/statements/drop_user_statement.hh"
+#include "cql3/statements/list_users_statement.hh"
+#include "cql3/statements/grant_statement.hh"
+#include "cql3/statements/revoke_statement.hh"
+#include "cql3/statements/list_permissions_statement.hh"
 #include "cql3/statements/index_target.hh"
 #include "cql3/statements/ks_prop_defs.hh"
 #include "cql3/selection/raw_selector.hh"
@@ -108,10 +118,13 @@ struct uninitialized {
 }
 
 @context {
-    using listener_type = cql3::error_listener<RecognizerType>;
+    using collector_type = cql3::error_collector<ComponentType, ExceptionBaseType::TokenType, ExceptionBaseType>;
+    using listener_type = cql3::error_listener<ComponentType, ExceptionBaseType>;
+
     listener_type* listener;
 
     std::vector<::shared_ptr<cql3::column_identifier>> _bind_variables;
+    std::vector<std::unique_ptr<TokenType>> _missing_tokens;
 
     // Can't use static variable, since it needs to be defined out-of-line
     static const std::unordered_set<sstring>& _reserved_type_names() {
@@ -161,15 +174,26 @@ struct uninitialized {
 
     void displayRecognitionError(ANTLR_UINT8** token_names, ExceptionBaseType* ex)
     {
-        std::stringstream msg;
-        ex->displayRecognitionError(token_names, msg);
-        listener->syntax_error(*this, msg.str());
+        listener->syntax_error(*this, token_names, ex);
     }
 
     void add_recognition_error(const sstring& msg) {
         listener->syntax_error(*this, msg);
     }
 
+    bool is_eof_token(CommonTokenType token) const
+    {
+        return token == CommonTokenType::TOKEN_EOF;
+    }
+
+    std::string token_text(const TokenType* token)
+    {
+        if (!token) {
+            return "";
+        }
+        return token->getText();
+    }
+
     std::map<sstring, sstring> convert_property_map(shared_ptr<cql3::maps::literal> map) {
         if (!map || map->entries.empty()) {
             return std::map<sstring, sstring>{};
@@ -216,6 +240,13 @@ struct uninitialized {
         }
         operations.emplace_back(std::move(key), std::move(update));
     }
+
+    TokenType* getMissingSymbol(IntStreamType* istream, ExceptionBaseType* e,
+                                ANTLR_UINT32 expectedTokenType, BitsetListType* follow) {
+        auto token = BaseType::getMissingSymbol(istream, e, expectedTokenType, follow);
+        _missing_tokens.emplace_back(token);
+        return token;
+    }
 }
 
 @lexer::namespace{cql3_parser}
@@ -233,7 +264,8 @@ struct uninitialized {
 }
 
 @lexer::context {
-    using listener_type = cql3::error_listener<RecognizerType>;
+    using collector_type = cql3::error_collector<ComponentType, ExceptionBaseType::TokenType, ExceptionBaseType>;
+    using listener_type = cql3::error_listener<ComponentType, ExceptionBaseType>;
 
     listener_type* listener;
 
@@ -243,9 +275,20 @@ struct uninitialized {
 
     void displayRecognitionError(ANTLR_UINT8** token_names, ExceptionBaseType* ex)
     {
-        std::stringstream msg;
-        ex->displayRecognitionError(token_names, msg);
-        listener->syntax_error(*this, msg.str());
+        listener->syntax_error(*this, token_names, ex);
+    }
+
+    bool is_eof_token(CommonTokenType token) const
+    {
+        return token == CommonTokenType::TOKEN_EOF;
+    }
+
+    std::string token_text(const TokenType* token) const
+    {
+        if (!token) {
+            return "";
+        }
+        return std::to_string(int(*token));
     }
 }
 
@@ -275,6 +318,7 @@ cqlStatement returns [shared_ptr<parsed_statement> stmt]
     | st14=alterTableStatement         { $stmt = st14; }
 #if 0
     | st15=alterKeyspaceStatement      { $stmt = st15; }
+#endif
     | st16=grantStatement              { $stmt = st16; }
     | st17=revokeStatement             { $stmt = st17; }
     | st18=listPermissionsStatement    { $stmt = st18; }
@@ -282,13 +326,14 @@ cqlStatement returns [shared_ptr<parsed_statement> stmt]
     | st20=alterUserStatement          { $stmt = st20; }
     | st21=dropUserStatement           { $stmt = st21; }
     | st22=listUsersStatement          { $stmt = st22; }
+#if 0
     | st23=createTriggerStatement      { $stmt = st23; }
     | st24=dropTriggerStatement        { $stmt = st24; }
 #endif
     | st25=createTypeStatement         { $stmt = st25; }
-#if 0
     | st26=alterTypeStatement          { $stmt = st26; }
     | st27=dropTypeStatement           { $stmt = st27; }
+#if 0
     | st28=createFunctionStatement     { $stmt = st28; }
     | st29=dropFunctionStatement       { $stmt = st29; }
     | st30=createAggregateStatement    { $stmt = st30; }
@@ -803,26 +848,27 @@ alterTableStatement returns [shared_ptr<alter_table_statement> expr]
     }
     ;
 
-#if 0
 /**
  * ALTER TYPE <name> ALTER <field> TYPE <newtype>;
  * ALTER TYPE <name> ADD <field> <newtype>;
  * ALTER TYPE <name> RENAME <field> TO <newtype> AND ...;
  */
-alterTypeStatement returns [AlterTypeStatement expr]
+alterTypeStatement returns [::shared_ptr<alter_type_statement> expr]
     : K_ALTER K_TYPE name=userTypeName
-          ( K_ALTER f=ident K_TYPE v=comparatorType { $expr = AlterTypeStatement.alter(name, f, v); }
-          | K_ADD   f=ident v=comparatorType        { $expr = AlterTypeStatement.addition(name, f, v); }
+          ( K_ALTER f=ident K_TYPE v=comparatorType { $expr = ::make_shared<alter_type_statement::add_or_alter>(name, false, f, v); }
+          | K_ADD   f=ident v=comparatorType        { $expr = ::make_shared<alter_type_statement::add_or_alter>(name, true, f, v); }
           | K_RENAME
-               { Map<ColumnIdentifier, ColumnIdentifier> renames = new HashMap<ColumnIdentifier, ColumnIdentifier>(); }
-                 id1=ident K_TO toId1=ident { renames.put(id1, toId1); }
-                 ( K_AND idn=ident K_TO toIdn=ident { renames.put(idn, toIdn); } )*
-               { $expr = AlterTypeStatement.renames(name, renames); }
+               { $expr = ::make_shared<alter_type_statement::renames>(name); }
+               renames[{ static_pointer_cast<alter_type_statement::renames>($expr) }]
           )
     ;
-#endif
 
 
+renames[::shared_ptr<alter_type_statement::renames> expr]
+    : fromId=ident K_TO toId=ident { $expr->add_rename(fromId, toId); }
+      ( K_AND renames[$expr] )?
+    ;
+
 /**
  * DROP KEYSPACE [IF EXISTS] <KSP>;
  */
@@ -839,15 +885,15 @@ dropTableStatement returns [::shared_ptr<drop_table_statement> stmt]
     : K_DROP K_COLUMNFAMILY (K_IF K_EXISTS { if_exists = true; } )? cf=columnFamilyName { $stmt = ::make_shared<drop_table_statement>(cf, if_exists); }
     ;
 
-#if 0
 /**
  * DROP TYPE <name>;
  */
-dropTypeStatement returns [DropTypeStatement stmt]
-    @init { boolean ifExists = false; }
-    : K_DROP K_TYPE (K_IF K_EXISTS { ifExists = true; } )? name=userTypeName { $stmt = new DropTypeStatement(name, ifExists); }
+dropTypeStatement returns [::shared_ptr<drop_type_statement> stmt]
+    @init { bool if_exists = false; }
+    : K_DROP K_TYPE (K_IF K_EXISTS { if_exists = true; } )? name=userTypeName { $stmt = ::make_shared<drop_type_statement>(name, if_exists); }
     ;
 
+#if 0
 /**
  * DROP INDEX [IF EXISTS] <INDEX_NAME>
  */
@@ -865,120 +911,118 @@ truncateStatement returns [::shared_ptr<truncate_statement> stmt]
     : K_TRUNCATE (K_COLUMNFAMILY)? cf=columnFamilyName { $stmt = ::make_shared<truncate_statement>(cf); }
     ;
 
-#if 0
 /**
  * GRANT <permission> ON <resource> TO <username>
  */
-grantStatement returns [GrantStatement stmt]
+grantStatement returns [::shared_ptr<grant_statement> stmt]
     : K_GRANT
           permissionOrAll
       K_ON
           resource
       K_TO
           username
-      { $stmt = new GrantStatement($permissionOrAll.perms, $resource.res, $username.text); }
+      { $stmt = ::make_shared<grant_statement>($permissionOrAll.perms, $resource.res, $username.text); } 
     ;
 
 /**
  * REVOKE <permission> ON <resource> FROM <username>
  */
-revokeStatement returns [RevokeStatement stmt]
+revokeStatement returns [::shared_ptr<revoke_statement> stmt]
     : K_REVOKE
           permissionOrAll
       K_ON
           resource
       K_FROM
           username
-      { $stmt = new RevokeStatement($permissionOrAll.perms, $resource.res, $username.text); }
+      { $stmt = ::make_shared<revoke_statement>($permissionOrAll.perms, $resource.res, $username.text); } 
     ;
 
-listPermissionsStatement returns [ListPermissionsStatement stmt]
+listPermissionsStatement returns [::shared_ptr<list_permissions_statement> stmt]
     @init {
-        IResource resource = null;
-        String username = null;
-        boolean recursive = true;
+		std::experimental::optional<auth::data_resource> r;
+		std::experimental::optional<sstring> u;
+		bool recursive = true;
     }
     : K_LIST
           permissionOrAll
-      ( K_ON resource { resource = $resource.res; } )?
-      ( K_OF username { username = $username.text; } )?
+      ( K_ON resource { r = $resource.res; } )?
+      ( K_OF username { u = sstring($username.text); } )?
       ( K_NORECURSIVE { recursive = false; } )?
-      { $stmt = new ListPermissionsStatement($permissionOrAll.perms, resource, username, recursive); }
+      { $stmt = ::make_shared<list_permissions_statement>($permissionOrAll.perms, std::move(r), std::move(u), recursive); } 
     ;
 
-permission returns [Permission perm]
+permission returns [auth::permission perm]
     : p=(K_CREATE | K_ALTER | K_DROP | K_SELECT | K_MODIFY | K_AUTHORIZE)
-    { $perm = Permission.valueOf($p.text.toUpperCase()); }
+    { $perm = auth::permissions::from_string($p.text); }
     ;
 
-permissionOrAll returns [Set<Permission> perms]
-    : K_ALL ( K_PERMISSIONS )?       { $perms = Permission.ALL_DATA; }
-    | p=permission ( K_PERMISSION )? { $perms = EnumSet.of($p.perm); }
+permissionOrAll returns [auth::permission_set perms]
+    : K_ALL ( K_PERMISSIONS )?       { $perms = auth::permissions::ALL_DATA; }
+    | p=permission ( K_PERMISSION )? { $perms = auth::permission_set::from_mask(auth::permission_set::mask_for($p.perm)); }
     ;
 
-resource returns [IResource res]
+resource returns [auth::data_resource res]
     : r=dataResource { $res = $r.res; }
     ;
 
-dataResource returns [DataResource res]
-    : K_ALL K_KEYSPACES { $res = DataResource.root(); }
-    | K_KEYSPACE ks = keyspaceName { $res = DataResource.keyspace($ks.id); }
+dataResource returns [auth::data_resource res]
+    : K_ALL K_KEYSPACES { $res = auth::data_resource(); }
+    | K_KEYSPACE ks = keyspaceName { $res = auth::data_resource($ks.id); }
     | ( K_COLUMNFAMILY )? cf = columnFamilyName
-      { $res = DataResource.columnFamily($cf.name.getKeyspace(), $cf.name.getColumnFamily()); }
+      { $res = auth::data_resource($cf.name->get_keyspace(), $cf.name->get_column_family()); }
     ;
 
 /**
  * CREATE USER [IF NOT EXISTS] <username> [WITH PASSWORD <password>] [SUPERUSER|NOSUPERUSER]
  */
-createUserStatement returns [CreateUserStatement stmt]
+createUserStatement returns [::shared_ptr<create_user_statement> stmt]
     @init {
-        UserOptions opts = new UserOptions();
-        boolean superuser = false;
-        boolean ifNotExists = false;
+    	auto opts = ::make_shared<cql3::user_options>();
+        bool superuser = false;
+        bool ifNotExists = false;
     }
     : K_CREATE K_USER (K_IF K_NOT K_EXISTS { ifNotExists = true; })? username
       ( K_WITH userOptions[opts] )?
       ( K_SUPERUSER { superuser = true; } | K_NOSUPERUSER { superuser = false; } )?
-      { $stmt = new CreateUserStatement($username.text, opts, superuser, ifNotExists); }
+      { $stmt = ::make_shared<create_user_statement>($username.text, std::move(opts), superuser, ifNotExists); }
     ;
 
 /**
  * ALTER USER <username> [WITH PASSWORD <password>] [SUPERUSER|NOSUPERUSER]
  */
-alterUserStatement returns [AlterUserStatement stmt]
+alterUserStatement returns [::shared_ptr<alter_user_statement> stmt]
     @init {
-        UserOptions opts = new UserOptions();
-        Boolean superuser = null;
+    	auto opts = ::make_shared<cql3::user_options>();
+    	std::experimental::optional<bool> superuser;
     }
     : K_ALTER K_USER username
       ( K_WITH userOptions[opts] )?
       ( K_SUPERUSER { superuser = true; } | K_NOSUPERUSER { superuser = false; } )?
-      { $stmt = new AlterUserStatement($username.text, opts, superuser); }
+      { $stmt = ::make_shared<alter_user_statement>($username.text, std::move(opts), std::move(superuser)); }
     ;
 
 /**
  * DROP USER [IF EXISTS] <username>
  */
-dropUserStatement returns [DropUserStatement stmt]
-    @init { boolean ifExists = false; }
-    : K_DROP K_USER (K_IF K_EXISTS { ifExists = true; })? username { $stmt = new DropUserStatement($username.text, ifExists); }
+dropUserStatement returns [::shared_ptr<drop_user_statement> stmt]
+    @init { bool ifExists = false; }
+    : K_DROP K_USER (K_IF K_EXISTS { ifExists = true; })? username { $stmt = ::make_shared<drop_user_statement>($username.text, ifExists); }
     ;
 
 /**
  * LIST USERS
  */
-listUsersStatement returns [ListUsersStatement stmt]
-    : K_LIST K_USERS { $stmt = new ListUsersStatement(); }
+listUsersStatement returns [::shared_ptr<list_users_statement> stmt]
+    : K_LIST K_USERS { $stmt = ::make_shared<list_users_statement>(); }
     ;
 
-userOptions[UserOptions opts]
+userOptions[::shared_ptr<cql3::user_options> opts]
     : userOption[opts]
     ;
 
-userOption[UserOptions opts]
-    : k=K_PASSWORD v=STRING_LITERAL { opts.put($k.text, $v.text); }
+userOption[::shared_ptr<cql3::user_options> opts]
+    : k=K_PASSWORD v=STRING_LITERAL { opts->put($k.text, $v.text); }
     ;
-#endif
 
 /** DEFINITIONS **/
 
@@ -1157,7 +1201,8 @@ columnOperation[operations_type& operations]
 
 columnOperationDifferentiator[operations_type& operations, ::shared_ptr<cql3::column_identifier::raw> key]
     : '=' normalColumnOperation[operations, key]
-    | '[' k=term ']' specializedColumnOperation[operations, key, k]
+    | '[' k=term ']' specializedColumnOperation[operations, key, k, false]
+    | '[' K_SCYLLA_TIMEUUID_LIST_INDEX '(' k=term ')' ']' specializedColumnOperation[operations, key, k, true]
     ;
 
 normalColumnOperation[operations_type& operations, ::shared_ptr<cql3::column_identifier::raw> key]
@@ -1199,11 +1244,12 @@ normalColumnOperation[operations_type& operations, ::shared_ptr<cql3::column_ide
 specializedColumnOperation[std::vector<std::pair<shared_ptr<cql3::column_identifier::raw>,
                                                  shared_ptr<cql3::operation::raw_update>>>& operations,
                            shared_ptr<cql3::column_identifier::raw> key,
-                           shared_ptr<cql3::term::raw> k]
+                           shared_ptr<cql3::term::raw> k,
+                           bool by_uuid]
 
     : '=' t=term
       {
-          add_raw_update(operations, key, make_shared<cql3::operation::set_element>(k, t));
+          add_raw_update(operations, key, make_shared<cql3::operation::set_element>(k, t, by_uuid));
       }
     ;
 
@@ -1383,12 +1429,10 @@ tuple_type returns [shared_ptr<cql3::cql3_type::raw> t]
       '>' { $t = cql3::cql3_type::raw::tuple(std::move(types)); }
     ;
 
-#if 0
 username
     : IDENT
     | STRING_LITERAL
     ;
-#endif
 
 // Basically the same as cident, but we need to exlude existing CQL3 types
 // (which for some reason are not reserved otherwise)
@@ -1567,6 +1611,8 @@ K_OR:          O R;
 K_REPLACE:     R E P L A C E;
 K_DETERMINISTIC: D E T E R M I N I S T I C;
 
+K_SCYLLA_TIMEUUID_LIST_INDEX: S C Y L L A '_' T I M E U U I D '_' L I S T '_' I N D E X;
+
 // Case-insensitive alpha characters
 fragment A: ('a'|'A');
 fragment B: ('b'|'B');
@@ -1612,20 +1658,17 @@ STRING_LITERAL
         setText(txt);
     }
     :
-// FIXME:
-#if 0
       /* pg-style string literal */
       (
-        '\$' '\$'
-        ( /* collect all input until '$$' is reached again */
-          {  (input.size() - input.index() > 1)
-               && !"$$".equals(input.substring(input.index(), input.index() + 1)) }?
-             => c=. { txt.appendCodePoint(c); }
+        '$' '$'
+        (
+          (c=~('$') { txt.push_back(c); })
+          |
+          ('$' (c=~('$') { txt.push_back('$'); txt.push_back(c); }))
         )*
-        '\$' '\$'
+        '$' '$'
       )
       |
-#endif
       /* conventional quoted string literal */
       (
         '\'' (c=~('\'') { txt.push_back(c);} | '\'' '\'' { txt.push_back('\''); })* '\''

cql3/abstract_marker.cc

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/abstract_marker.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/assignment_testable.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2014 Cloudius Systems
+ * Copyright (C) 2014 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/attributes.cc

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/attributes.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/cf_name.cc

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/cf_name.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/column_condition.cc

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/column_condition.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/column_identifier.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

cql3/column_identifier.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/column_specification.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/constants.cc

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/constants.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/cql3_type.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2014 Cloudius Systems, Ltd.
+ * Copyright (C) 2014 ScyllaDB
  */
 
 /*
@@ -148,7 +148,7 @@ public:
         try {
             auto&& ks = db.find_keyspace(_name.get_keyspace());
             try {
-                auto&& type = ks._user_types.get_type(_name.get_user_type_name());
+                auto&& type = ks.metadata()->user_types()->get_type(_name.get_user_type_name());
                 if (!_frozen) {
                     throw exceptions::invalid_request_exception("Non-frozen User-Defined types are not supported, please use frozen<>");
                 }

cql3/cql3_type.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  *
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

cql3/cql_statement.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2014 Cloudius Systems
+ * Copyright (C) 2014 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*
@@ -70,7 +70,7 @@ public:
      *
      * @param state the current client state
      */
-    virtual void check_access(const service::client_state& state) = 0;
+    virtual future<> check_access(const service::client_state& state) = 0;
 
     /**
      * Perform additional validation required by the statment.

cql3/error_collector.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*
@@ -50,8 +50,8 @@ namespace cql3 {
 /**
  * <code>ErrorListener</code> that collect and enhance the errors send by the CQL lexer and parser.
  */
-template<typename Recognizer>
-class error_collector : public error_listener<Recognizer> {
+template<typename RecognizerType, typename TokenType, typename ExceptionBaseType>
+class error_collector : public error_listener<RecognizerType, ExceptionBaseType> {
     /**
      * The offset of the first token of the snippet.
      */
@@ -81,25 +81,19 @@ public:
      */
     error_collector(const sstring_view& query) : _query(query) {}
 
-    virtual void syntax_error(Recognizer& recognizer, const std::vector<sstring>& token_names) override {
-        // FIXME: stub
-        syntax_error(recognizer, "Parsing failed, detailed description construction not implemented yet");
+    virtual void syntax_error(RecognizerType& recognizer, ANTLR_UINT8** token_names, ExceptionBaseType* ex) override {
+        auto hdr = get_error_header(ex);
+        auto msg = get_error_message(recognizer, ex, token_names);
+        std::stringstream result;
+        result << hdr << ' ' << msg;
 #if 0
-        String hdr = recognizer.getErrorHeader(e);
-        String msg = recognizer.getErrorMessage(e, tokenNames);
-
-        StringBuilder builder = new StringBuilder().append(hdr)
-                .append(' ')
-                .append(msg);
-
         if (recognizer instanceof Parser)
             appendQuerySnippet((Parser) recognizer, builder);
-
-        errorMsgs.add(builder.toString());
 #endif
+        _error_msgs.emplace_back(result.str());
     }
 
-    virtual void syntax_error(Recognizer& recognizer, const sstring& msg) override {
+    virtual void syntax_error(RecognizerType& recognizer, const sstring& msg) override {
         _error_msgs.emplace_back(msg);
     }
 
@@ -114,6 +108,60 @@ public:
         }
     }
 
+private:
+    std::string get_error_header(ExceptionBaseType* ex) {
+        std::stringstream result;
+        result << "line " << ex->get_line() << ":" << ex->get_charPositionInLine();
+        return result.str();
+    }
+
+    std::string get_error_message(RecognizerType& recognizer, ExceptionBaseType* ex, ANTLR_UINT8** token_names)
+    {
+        using namespace antlr3;
+        std::stringstream msg;
+        switch (ex->getType()) {
+        case ExceptionType::UNWANTED_TOKEN_EXCEPTION: {
+            msg << "extraneous input " << get_token_error_display(recognizer, ex->get_token());
+            if (token_names != nullptr) {
+                std::string token_name;
+                if (recognizer.is_eof_token(ex->get_expecting())) {
+                    token_name = "EOF";
+                } else {
+                    token_name = reinterpret_cast<const char*>(token_names[ex->get_expecting()]);
+                }
+                msg << " expecting " << token_name;
+            }
+            break;
+        }
+        case ExceptionType::MISSING_TOKEN_EXCEPTION: {
+            std::string token_name;
+            if (token_names == nullptr) {
+                token_name = "(" + std::to_string(ex->get_expecting()) + ")";
+            } else {
+                if (recognizer.is_eof_token(ex->get_expecting())) {
+                    token_name = "EOF";
+                } else {
+                    token_name = reinterpret_cast<const char*>(token_names[ex->get_expecting()]);
+                }
+            }
+            msg << "missing " << token_name << " at " << get_token_error_display(recognizer, ex->get_token());
+            break;
+        }
+        case ExceptionType::NO_VIABLE_ALT_EXCEPTION: {
+            msg << "no viable alternative at input " << get_token_error_display(recognizer, ex->get_token());
+            break;
+        }
+        default:
+            ex->displayRecognitionError(token_names, msg);
+        }
+        return msg.str();
+    }
+
+    std::string get_token_error_display(RecognizerType& recognizer, const TokenType* token)
+    {
+        return "'" + recognizer.token_text(token) + "'";
+    }
+
 #if 0
 
     /**

cql3/error_listener.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*
@@ -41,17 +41,18 @@
 
 #pragma once
 
-#include <vector>
-#include "core/sstring.hh"
+#include <seastar/core/sstring.hh>
+#include <antlr3.hpp>
 
 namespace cql3 {
 
 /**
  * Listener used to collect the syntax errors emitted by the Lexer and Parser.
  */
-template<typename RecognizerType>
+template<typename RecognizerType, typename ExceptionBaseType>
 class error_listener {
 public:
+
     /**
      * Invoked when a syntax error occurs.
      *
@@ -59,7 +60,7 @@ public:
      * @param tokenNames the token names
      * @param e the exception
      */
-    virtual void syntax_error(RecognizerType& recognizer, const std::vector<sstring>& token_names) = 0;
+    virtual void syntax_error(RecognizerType& recognizer, ANTLR_UINT8** token_names, ExceptionBaseType* ex) = 0;
 
     /**
      * Invoked when a syntax error with a specified message occurs.

cql3/functions/abstract_function.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2014 Cloudius Systems
+ * Copyright (C) 2014 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/functions/aggregate_fcts.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2014 Cloudius Systems
+ * Copyright (C) 2014 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/functions/aggregate_function.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  *
- * Copyright 2014 Cloudius Systems
+ * Copyright (C) 2014 ScyllaDB
  */
 
 /*

cql3/functions/bytes_conversion_fcts.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  *
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

cql3/functions/function.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2014 Cloudius Systems
+ * Copyright (C) 2014 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/functions/function_call.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  *
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

cql3/functions/function_name.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2014 Cloudius Systems
+ * Copyright (C) 2014 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*

cql3/functions/functions.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2014 Cloudius Systems, Ltd.
+ * Copyright (C) 2014 ScyllaDB
  */
 
 /*

cql3/functions/functions.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  *
- * Copyright 2015 Cloudius Systems
+ * Copyright (C) 2015 ScyllaDB
  */
 
 /*

cql3/functions/native_aggregate_function.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  *
- * Copyright 2014 Cloudius Systems
+ * Copyright (C) 2014 ScyllaDB
  */
 
 /*

cql3/functions/native_function.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  *
- * Copyright 2014 Cloudius Systems
+ * Copyright (C) 2014 ScyllaDB
  */
 
 /*

cql3/functions/native_scalar_function.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  *
- * Copyright 2014 Cloudius Systems
+ * Copyright (C) 2014 ScyllaDB
  */
 
 /*

cql3/functions/scalar_function.hh

@@ -17,9 +17,9 @@
  */
 
 /*
- * Copyright 2014 Cloudius Systems
+ * Copyright (C) 2014 ScyllaDB
  *
- * Modified by Cloudius Systems
+ * Modified by ScyllaDB
  */
 
 /*