view: Store pk view column that's non-pk in the base

To help calculate the view mutations from a base update, we store in
the view class the column that's part of the view's primary key but
not part of the base's, if such column exists.

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

database.cc

@@ -3631,9 +3631,9 @@ void column_family::update_view_schemas() {
 }
 
 void column_family::add_or_update_view(view_ptr v) {
-    auto e = _views.emplace(v->cf_name(), make_lw_shared<db::view::view>(v));
+    auto e = _views.emplace(v->cf_name(), make_lw_shared<db::view::view>(v, *schema()));
     if (!e.second) {
-        e.first->second->update(v);
+        e.first->second->update(v, *schema());
     }
     update_view_schemas();
 }

db/view/view.cc

@@ -134,6 +134,17 @@ bool view::matches_view_filter(const ::schema& base, const partition_key& key, c
                     });
 }
 
+void view::set_base_non_pk_column_in_view_pk(const ::schema& base) {
+    for (auto&& base_col : base.regular_columns()) {
+        auto view_col = _schema->get_column_definition(base_col.name());
+        if (view_col && view_col->is_primary_key()) {
+            _base_non_pk_column_in_view_pk = view_col;
+            return;
+        }
+    }
+    _base_non_pk_column_in_view_pk = nullptr;
+}
+
 } // namespace view
 } // namespace db
 

db/view/view.hh

@@ -42,20 +42,27 @@ class view final {
     mutable shared_ptr<cql3::statements::select_statement> _select_statement;
     mutable stdx::optional<query::partition_slice> _partition_slice;
     mutable stdx::optional<dht::partition_range_vector> _partition_ranges;
+    const column_definition* _base_non_pk_column_in_view_pk;
 public:
-    explicit view(view_ptr schema)
+    explicit view(view_ptr schema, const ::schema& base)
             : _schema(std::move(schema)) {
+        set_base_non_pk_column_in_view_pk(base);
     }
 
     view_ptr schema() const {
         return _schema;
     }
 
-    void update(view_ptr new_schema) {
+    void update(view_ptr new_schema, const ::schema& base) {
         _schema = new_schema;
         _select_statement = nullptr;
         _partition_slice = { };
         _partition_ranges = { };
+        set_base_non_pk_column_in_view_pk(base);
+    }
+
+    const column_definition* base_non_pk_column_in_view_pk() const {
+        return _base_non_pk_column_in_view_pk;
     }
 
     /**
@@ -104,6 +111,7 @@ private:
     const query::partition_slice& partition_slice() const;
     const dht::partition_range_vector& partition_ranges() const;
     bool clustering_prefix_matches(const ::schema& base, const partition_key& key, const clustering_key_prefix& ck) const;
+    void set_base_non_pk_column_in_view_pk(const ::schema& base);
 };
 
 }