From 082ef56df179991c09f53e7bfe7b71f0af26c30e Mon Sep 17 00:00:00 2001
From: Duarte Nunes <duarte@scylladb.com>
Date: Thu, 29 Dec 2016 11:17:16 +0000
Subject: [PATCH] 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     |  4 ++--
 db/view/view.cc | 11 +++++++++++
 db/view/view.hh | 12 ++++++++++--
 3 files changed, 23 insertions(+), 4 deletions(-)

diff --git a/database.cc b/database.cc
index ab635442b1..26857df63d 100644
--- a/database.cc
+++ b/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();
 }
diff --git a/db/view/view.cc b/db/view/view.cc
index 48bebe0b15..54001d0a81 100644
--- a/db/view/view.cc
+++ b/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
 
diff --git a/db/view/view.hh b/db/view/view.hh
index 7d0c1bd82d..9245a997e3 100644
--- a/db/view/view.hh
+++ b/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);
 };
 
 }