From 6b7473be538663bf5544a4e59b09f46a026ee911 Mon Sep 17 00:00:00 2001
From: Tomasz Grabiec <tgrabiec@scylladb.com>
Date: Mon, 21 Nov 2022 18:23:18 +0100
Subject: [PATCH] mvcc: partition_snapshot_row_cursor: Handle non-evictable
 snapshots

This is a prerequisite for using the cursor in memtable readers.

Non-evictable snapshots are those which live in memtables. Unlike
evictable snapshots, they don't have a dummy entry at position after
all clustering rows. In evictable snapshots, lookup always finds an
entry, not so with non-evictable snapshots. The cursor was not
prepared for this case, this patch handles it.
---
 partition_snapshot_row_cursor.hh | 43 +++++++++++++++++++++++---------
 1 file changed, 31 insertions(+), 12 deletions(-)

diff --git a/partition_snapshot_row_cursor.hh b/partition_snapshot_row_cursor.hh
index c712cde569..7ea9eb5780 100644
--- a/partition_snapshot_row_cursor.hh
+++ b/partition_snapshot_row_cursor.hh
@@ -250,6 +250,19 @@ class partition_snapshot_row_cursor final {
                 if (pos) [[likely]] {
                     _heap.emplace_back(position_in_version{pos, std::move(rows), version_no, unique_owner, cont});
                 }
+            } else {
+                if (_reversed) [[unlikely]] {
+                    if (!rows.empty()) {
+                        pos = std::prev(rows.end());
+                    } else {
+                        _background_continuity = true;
+                    }
+                } else {
+                    _background_continuity = true; // Default continuity past the last entry
+                }
+                if (pos) [[likely]] {
+                    _heap.emplace_back(position_in_version{pos, std::move(rows), version_no, unique_owner, is_continuous::yes});
+                }
             }
             ++version_no;
             first = false;
@@ -380,22 +393,28 @@ public:
             _latest_it = it;
             auto heap_i = boost::find_if(_heap, [](auto&& v) { return v.version_no == 0; });
 
-            is_continuous cont = it->continuous();
-            if (_reversed) [[unlikely]] {
-                if (!match) {
-                    // lower_bound() in reverse order points to predecessor of it unless the keys are equal.
-                    if (it == rows.begin()) {
-                        _background_continuity |= bool(it->continuous());
-                        it = {};
+            is_continuous cont;
+            if (it) {
+                if (_reversed) [[unlikely]] {
+                    if (!match) {
+                        // lower_bound() in reverse order points to predecessor of it unless the keys are equal.
+                        if (it == rows.begin()) {
+                            _background_continuity |= bool(it->continuous());
+                            it = {};
+                        } else {
+                            cont = it->continuous();
+                            --it;
+                        }
                     } else {
-                        cont = it->continuous();
-                        --it;
+                        // We can put anything in the match case since this continuity will not be used
+                        // when advancing the cursor.
+                        cont = is_continuous::no;
                     }
                 } else {
-                    // We can put anything in the match case since this continuity will not be used
-                    // when advancing the cursor.
-                    cont = is_continuous::no;
+                    cont = it->continuous();
                 }
+            } else {
+                _background_continuity = true; // Default continuity past the last entry.
             }
 
             if (!it) {