db: more const correctness for column_family and component types

Ensure that read-side accessors are const.  This is important in preparation
for multiple memtables (and later, sstables) since a read-side
mutation_partition may be a temporary object coming from multiple memtables
(and sstables) while a write-side mutation_partition is guaranteed to belong
to a single memtable (and thus, not be temporary).

Since writers will want non-const mutation_partitions to write to, they won't
be able to use the read-side accessors by accident.

database.cc

@@ -32,20 +32,20 @@ column_family::column_family(schema_ptr schema)
 column_family::~column_family() {
 }
 
-mutation_partition*
-column_family::find_partition(const dht::decorated_key& key) {
+const mutation_partition*
+column_family::find_partition(const dht::decorated_key& key) const {
     auto i = partitions.find(key);
     return i == partitions.end() ? nullptr : &i->second;
 }
 
-mutation_partition*
-column_family::find_partition_slow(const partition_key& key) {
+const mutation_partition*
+column_family::find_partition_slow(const partition_key& key) const {
     return find_partition(dht::global_partitioner().decorate_key(*_schema, key));
 }
 
-row*
-column_family::find_row(const dht::decorated_key& partition_key, const clustering_key& clustering_key) {
-    mutation_partition* p = find_partition(partition_key);
+const row*
+column_family::find_row(const dht::decorated_key& partition_key, const clustering_key& clustering_key) const {
+    const mutation_partition* p = find_partition(partition_key);
     if (!p) {
         return nullptr;
     }
@@ -461,7 +461,7 @@ merge_column(const column_definition& def,
 }
 
 future<lw_shared_ptr<query::result>>
-column_family::query(const query::read_command& cmd) {
+column_family::query(const query::read_command& cmd) const {
     query::result::builder builder(cmd.slice);
 
     uint32_t limit = cmd.row_limit;

database.hh

@@ -59,15 +59,15 @@ struct column_family {
     ~column_family();
     mutation_partition& find_or_create_partition(const dht::decorated_key& key);
     mutation_partition& find_or_create_partition_slow(const partition_key& key);
-    mutation_partition* find_partition(const dht::decorated_key& key);
-    mutation_partition* find_partition_slow(const partition_key& key);
+    const mutation_partition* find_partition(const dht::decorated_key& key) const;
+    const mutation_partition* find_partition_slow(const partition_key& key) const;
     row& find_or_create_row_slow(const partition_key& partition_key, const clustering_key& clustering_key);
-    row* find_row(const dht::decorated_key& partition_key, const clustering_key& clustering_key);
+    const row* find_row(const dht::decorated_key& partition_key, const clustering_key& clustering_key) const;
     schema_ptr _schema;
     std::map<dht::decorated_key, mutation_partition, dht::decorated_key::less_comparator> partitions;
     void apply(const mutation& m);
     // Returns at most "cmd.limit" rows
-    future<lw_shared_ptr<query::result>> query(const query::read_command& cmd);
+    future<lw_shared_ptr<query::result>> query(const query::read_command& cmd) const;
 
     future<> populate(sstring datadir);
 private:

mutation.cc

@@ -59,8 +59,8 @@ void mutation::set_cell(const exploded_clustering_prefix& prefix, const column_d
 }
 
 std::experimental::optional<atomic_cell_or_collection>
-mutation::get_cell(const clustering_key& rkey, const column_definition& def) {
-    auto find_cell = [&def] (row& r) {
+mutation::get_cell(const clustering_key& rkey, const column_definition& def) const {
+    auto find_cell = [&def] (const row& r) {
         auto i = r.find(def.id);
         if (i == r.end()) {
             return std::experimental::optional<atomic_cell_or_collection>{};

mutation.hh

@@ -27,7 +27,7 @@ public:
     void set_clustered_cell(const clustering_key& key, const column_definition& def, atomic_cell_or_collection value);
     void set_cell(const exploded_clustering_prefix& prefix, const bytes& name, const boost::any& value, api::timestamp_type timestamp, ttl_opt ttl = {});
     void set_cell(const exploded_clustering_prefix& prefix, const column_definition& def, atomic_cell_or_collection value);
-    std::experimental::optional<atomic_cell_or_collection> get_cell(const clustering_key& rkey, const column_definition& def);
+    std::experimental::optional<atomic_cell_or_collection> get_cell(const clustering_key& rkey, const column_definition& def) const;
     const partition_key& key() const { return _dk._key; };
     const dht::decorated_key& decorated_key() const { return _dk; };
     const dht::token token() const { return _dk._token; }

mutation_partition.cc

@@ -127,8 +127,8 @@ mutation_partition::apply_delete(schema_ptr schema, clustering_key&& key, tombst
     }
 }
 
-rows_entry*
-mutation_partition::find_entry(schema_ptr schema, const clustering_key_prefix& key) {
+const rows_entry*
+mutation_partition::find_entry(schema_ptr schema, const clustering_key_prefix& key) const {
     auto i = _rows.find(key, rows_entry::key_comparator(clustering_key::less_compare_with_prefix(*schema)));
     if (i == _rows.end()) {
         return nullptr;
@@ -136,8 +136,8 @@ mutation_partition::find_entry(schema_ptr schema, const clustering_key_prefix& k
     return &*i;
 }
 
-row*
-mutation_partition::find_row(const clustering_key& key) {
+const row*
+mutation_partition::find_row(const clustering_key& key) const {
     auto i = _rows.find(key);
     if (i == _rows.end()) {
         return nullptr;

mutation_partition.hh

@@ -179,8 +179,8 @@ public:
     const rows_type& clustered_rows() const { return _rows; }
     const row& static_row() const { return _static_row; }
     deletable_row& clustered_row(const clustering_key& key);
-    row* find_row(const clustering_key& key);
-    rows_entry* find_entry(schema_ptr schema, const clustering_key_prefix& key);
+    const row* find_row(const clustering_key& key) const;
+    const rows_entry* find_entry(schema_ptr schema, const clustering_key_prefix& key) const;
     tombstone range_tombstone_for_row(const schema& schema, const clustering_key& key) const;
     tombstone tombstone_for_row(const schema& schema, const clustering_key& key) const;
     tombstone tombstone_for_row(const schema& schema, const rows_entry& e) const;

thrift/handler.cc

@@ -133,7 +133,7 @@ public:
                 throw unimplemented_exception();
             } else if (predicate.__isset.slice_range) {
                 auto&& range = predicate.slice_range;
-                row* rw = cf.find_row(dk, clustering_key::make_empty(*cf._schema));
+                const row* rw = cf.find_row(dk, clustering_key::make_empty(*cf._schema));
                 if (rw) {
                     auto beg = cf._schema->regular_begin();
                     if (!range.start.empty()) {