mutation_partition: Regular base column in view determines row liveness

When views contain a primary key column that is not part of the base
table primary key, that column determines whether the row is live or
not. We need to ensure that when that cell is dead, and thus the
derived row marker, either by normal deletion of by TTL, so is the
rest of the row.

This patch introduces the idea of shawdowing row marker. We map the
status of the regular base column in the view's PK to the view row's
marker. If this marker is dead, so is that cell in the base table, and
so should the view row become. To enforce that, a view row's dead
marker shadows the whole row if that view includes a base regular
column in its PK.

Fixes #3360

Signed-off-by: Duarte Nunes <duarte@scylladb.com>
(cherry picked from commit 67dac67c46)

db/view/view.cc

@@ -578,12 +578,12 @@ void view_update_builder::generate_update(clustering_row&& update, stdx::optiona
     // We allow existing to be disengaged, which we treat the same as an empty row.
     if (existing) {
         existing->marker().compact_and_expire(tombstone(), _now, always_gc, gc_before);
-        existing->cells().compact_and_expire(*_schema, column_kind::regular_column, row_tombstone(), _now, always_gc, gc_before);
+        existing->cells().compact_and_expire(*_schema, column_kind::regular_column, row_tombstone(), _now, always_gc, gc_before, existing->marker());
         update.apply(*_schema, *existing);
     }
 
     update.marker().compact_and_expire(tombstone(), _now, always_gc, gc_before);
-    update.cells().compact_and_expire(*_schema, column_kind::regular_column, row_tombstone(), _now, always_gc, gc_before);
+    update.cells().compact_and_expire(*_schema, column_kind::regular_column, row_tombstone(), _now, always_gc, gc_before, update.marker());
 
     for (auto&& v : _view_updates) {
         v.generate_update(_key, update, existing, _now);

mutation_compactor.hh

@@ -215,7 +215,7 @@ public:
         }
         t.apply(current_tombstone);
         bool is_live = cr.marker().compact_and_expire(t.tomb(), _query_time, _can_gc, _gc_before);
-        is_live |= cr.cells().compact_and_expire(_schema, column_kind::regular_column, t, _query_time, _can_gc, _gc_before);
+        is_live |= cr.cells().compact_and_expire(_schema, column_kind::regular_column, t, _query_time, _can_gc, _gc_before, cr.marker());
         if (only_live() && is_live) {
             partition_is_not_empty(consumer);
             auto stop = consumer.consume(std::move(cr), t, true);

mutation_partition.cc

@@ -35,6 +35,7 @@
 #include "intrusive_set_external_comparator.hh"
 #include "counters.hh"
 #include "row_cache.hh"
+#include "view_info.hh"
 #include <seastar/core/execution_stage.hh>
 
 template<bool reversed>
@@ -1264,8 +1265,8 @@ uint32_t mutation_partition::do_compact(const schema& s,
         deletable_row& row = e.row();
         row_tombstone tomb = tombstone_for_row(s, e);
 
-        bool is_live = row.cells().compact_and_expire(s, column_kind::regular_column, tomb, query_time, can_gc, gc_before);
-        is_live |= row.marker().compact_and_expire(tomb.tomb(), query_time, can_gc, gc_before);
+        bool is_live = row.marker().compact_and_expire(tomb.tomb(), query_time, can_gc, gc_before);
+        is_live |= row.cells().compact_and_expire(s, column_kind::regular_column, tomb, query_time, can_gc, gc_before, row.marker());
 
         if (should_purge_row_tombstone(row.deleted_at())) {
             row.remove_tombstone();
@@ -1552,9 +1553,30 @@ void row::apply_monotonically(const schema& s, column_kind kind, row&& other) {
     });
 }
 
-bool row::compact_and_expire(const schema& s, column_kind kind, row_tombstone tomb, gc_clock::time_point query_time,
-    can_gc_fn& can_gc, gc_clock::time_point gc_before)
+// When views contain a primary key column that is not part of the base table primary key,
+// that column determines whether the row is live or not. We need to ensure that when that
+// cell is dead, and thus the derived row marker, either by normal deletion of by TTL, so
+// is the rest of the row. To ensure that none of the regular columns keep the row alive,
+// we erase the live cells according to the shadowable_tombstone rules.
+static bool dead_marker_shadows_row(const schema& s, column_kind kind, const row_marker& marker) {
+    return s.is_view()
+            && s.view_info()->base_non_pk_column_in_view_pk()
+            && !marker.is_live()
+            && kind == column_kind::regular_column; // not applicable to static rows
+}
+
+bool row::compact_and_expire(
+        const schema& s,
+        column_kind kind,
+        row_tombstone tomb,
+        gc_clock::time_point query_time,
+        can_gc_fn& can_gc,
+        gc_clock::time_point gc_before,
+        const row_marker& marker)
 {
+    if (dead_marker_shadows_row(s, kind, marker)) {
+        tomb.apply(shadowable_tombstone(api::max_timestamp, gc_clock::time_point::max()), row_marker());
+    }
     bool any_live = false;
     remove_if([&] (column_id id, atomic_cell_or_collection& c) {
         bool erase = false;
@@ -1596,6 +1618,17 @@ bool row::compact_and_expire(const schema& s, column_kind kind, row_tombstone to
     return any_live;
 }
 
+bool row::compact_and_expire(
+        const schema& s,
+        column_kind kind,
+        row_tombstone tomb,
+        gc_clock::time_point query_time,
+        can_gc_fn& can_gc,
+        gc_clock::time_point gc_before) {
+    row_marker m;
+    return compact_and_expire(s, kind, tomb, query_time, can_gc, gc_before, m);
+}
+
 deletable_row deletable_row::difference(const schema& s, column_kind kind, const deletable_row& other) const
 {
     deletable_row dr;

mutation_partition.hh

@@ -314,8 +314,22 @@ public:
     // Expires cells based on query_time. Expires tombstones based on gc_before
     // and max_purgeable. Removes cells covered by tomb.
     // Returns true iff there are any live cells left.
-    bool compact_and_expire(const schema& s, column_kind kind, row_tombstone tomb, gc_clock::time_point query_time,
-        can_gc_fn&, gc_clock::time_point gc_before);
+    bool compact_and_expire(
+            const schema& s,
+            column_kind kind,
+            row_tombstone tomb,
+            gc_clock::time_point query_time,
+            can_gc_fn&,
+            gc_clock::time_point gc_before,
+            const row_marker& marker);
+
+    bool compact_and_expire(
+            const schema& s,
+            column_kind kind,
+            row_tombstone tomb,
+            gc_clock::time_point query_time,
+            can_gc_fn&,
+            gc_clock::time_point gc_before);
 
     row difference(const schema&, column_kind, const row& other) const;