From f41a5e554da329766d24d5d8769edb422e088ca4 Mon Sep 17 00:00:00 2001
From: Duarte Nunes <duarte@scylladb.com>
Date: Thu, 9 Mar 2017 15:05:36 +0100
Subject: [PATCH] view_info: Store base regular col in the view's PK as
 column_id

This patch stores the base_non_pk_column_in_view column as column_id,
which is more convenient, and it also stores a two-level optional to
encode both lazy initialization and the absence of such a column.

Signed-off-by: Duarte Nunes <duarte@scylladb.com>
---
 db/view/view.cc | 29 ++++++++++++++---------------
 view_info.hh    |  5 +++--
 2 files changed, 17 insertions(+), 17 deletions(-)

diff --git a/db/view/view.cc b/db/view/view.cc
index 3f3d58f20d..d351028ee7 100644
--- a/db/view/view.cc
+++ b/db/view/view.cc
@@ -56,7 +56,6 @@ static logging::logger logger("view");
 view_info::view_info(const schema& schema, const raw_view_info& raw_view_info)
         : _schema(schema)
         , _raw(raw_view_info)
-        , _base_non_pk_column_in_view_pk(nullptr)
 { }
 
 cql3::statements::select_statement& view_info::select_statement() const {
@@ -97,17 +96,18 @@ const column_definition* view_info::view_column(const schema& base, column_id ba
     return _schema.get_column_definition(base.regular_column_at(base_id).name());
 }
 
-const column_definition* view_info::base_non_pk_column_in_view_pk(const schema& base) const {
+stdx::optional<column_id> view_info::base_non_pk_column_in_view_pk(const schema& base) const {
     if (!_base_non_pk_column_in_view_pk) {
-        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;
+        _base_non_pk_column_in_view_pk.emplace(stdx::nullopt);
+        for (auto&& view_col : boost::range::join(_schema.partition_key_columns(), _schema.clustering_key_columns())) {
+            auto* base_col = base.get_column_definition(view_col.name());
+            if (!base_col->is_primary_key()) {
+                _base_non_pk_column_in_view_pk.emplace(base_col->id);
                 break;
             }
         }
     }
-    return _base_non_pk_column_in_view_pk;
+    return *_base_non_pk_column_in_view_pk;
 }
 
 namespace db {
@@ -234,10 +234,10 @@ row_marker view_updates::compute_row_marker(const clustering_row& base_row) cons
      */
 
     auto marker = base_row.marker();
-    auto* col = _view_info.base_non_pk_column_in_view_pk(*_base);
-    if (col) {
+    auto col_id = _view_info.base_non_pk_column_in_view_pk(*_base);
+    if (col_id) {
         // Note: multi-cell columns can't be part of the primary key.
-        auto cell = base_row.cells().cell_at(col->id).as_atomic_cell();
+        auto cell = base_row.cells().cell_at(*col_id).as_atomic_cell();
         auto timestamp = std::max(marker.timestamp(), cell.timestamp());
         return cell.is_live_and_has_ttl() ? row_marker(timestamp, cell.ttl(), cell.expiry()) : row_marker(timestamp);
     }
@@ -402,8 +402,8 @@ void view_updates::generate_update(
         return;
     }
 
-    auto* col = _view_info.base_non_pk_column_in_view_pk(*_base);
-    if (!col) {
+    auto col_id = _view_info.base_non_pk_column_in_view_pk(*_base);
+    if (!col_id) {
         // The view key is necessarily the same pre and post update.
         if (existing && !existing->empty()) {
             if (update.empty()) {
@@ -417,10 +417,9 @@ void view_updates::generate_update(
         return;
     }
 
-    auto col_id = col->id;
-    auto* after = update.cells().find_cell(col_id);
+    auto* after = update.cells().find_cell(*col_id);
     if (existing) {
-        auto* before = existing->cells().find_cell(col_id);
+        auto* before = existing->cells().find_cell(*col_id);
         if (before) {
             if (after) {
                 // Note: multi-cell columns can't be part of the primary key.
diff --git a/view_info.hh b/view_info.hh
index 9df5e233fb..b3a832b9c3 100644
--- a/view_info.hh
+++ b/view_info.hh
@@ -33,7 +33,8 @@ class view_info 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;
-    mutable const column_definition* _base_non_pk_column_in_view_pk;
+    // Lazily initializes the column id of a regular base table included in the view's PK, if any.
+    mutable stdx::optional<stdx::optional<column_id>> _base_non_pk_column_in_view_pk;
 public:
     view_info(const schema& schema, const raw_view_info& raw_view_info);
 
@@ -61,7 +62,7 @@ public:
     const query::partition_slice& partition_slice() const;
     const dht::partition_range_vector& partition_ranges() const;
     const column_definition* view_column(const schema& base, column_id base_id) const;
-    const column_definition* base_non_pk_column_in_view_pk(const schema& base) const;
+    stdx::optional<column_id> base_non_pk_column_in_view_pk(const schema& base) const;
 
     friend bool operator==(const view_info& x, const view_info& y) {
         return x._raw == y._raw;