column_family: Add schema setters

There is one current schema for given column_family. Entries in
memtables and cache can be at any of the previous schemas, but they're
always upgraded to current schema on access.

database.cc

@@ -2237,3 +2237,15 @@ std::ostream& operator<<(std::ostream& os, const keyspace_metadata& m) {
     os << "}";
     return os;
 }
+
+void column_family::set_schema(schema_ptr s) {
+    dblog.debug("Changing schema version of {}.{} ({}) from {} to {}",
+                _schema->ks_name(), _schema->cf_name(), _schema->id(), _schema->version(), s->version());
+
+    for (auto& m : *_memtables) {
+        m->set_schema(s);
+    }
+
+    _cache.set_schema(s);
+    _schema = std::move(s);
+}

database.hh

@@ -228,6 +228,7 @@ public:
     column_family(column_family&&) = delete; // 'this' is being captured during construction
     ~column_family();
     schema_ptr schema() const { return _schema; }
+    void set_schema(schema_ptr);
     db::commitlog* commitlog() { return _commitlog; }
     future<const_mutation_partition_ptr> find_partition(schema_ptr, const dht::decorated_key& key) const;
     future<const_mutation_partition_ptr> find_partition_slow(schema_ptr, const partition_key& key) const;

memtable.cc

@@ -295,3 +295,7 @@ void memtable::upgrade_entry(partition_entry& e) {
         });
     }
 }
+
+void memtable::set_schema(schema_ptr new_schema) noexcept {
+    _schema = std::move(new_schema);
+}

memtable.hh

@@ -114,6 +114,7 @@ public:
     explicit memtable(schema_ptr schema, logalloc::region_group* dirty_memory_region_group = nullptr);
     ~memtable();
     schema_ptr schema() const { return _schema; }
+    void set_schema(schema_ptr) noexcept;
     future<> apply(memtable&);
     // Applies mutation to this memtable.
     // The mutation is upgraded to current schema.

row_cache.cc

@@ -549,6 +549,10 @@ cache_entry::cache_entry(cache_entry&& o) noexcept
     }
 }
 
+void row_cache::set_schema(schema_ptr new_schema) noexcept {
+    _schema = std::move(new_schema);
+}
+
 mutation cache_entry::read(const schema_ptr& s) {
     auto m = mutation(_schema, _key, _p);
     if (_schema != s) {

row_cache.hh

@@ -242,6 +242,7 @@ public:
         return _tracker;
     }
 
+    void set_schema(schema_ptr) noexcept;
     const schema_ptr& schema() const;
 
     friend class just_cache_scanning_reader;