From d14ec86e7d5d72b79b167ccc4228d44192d97efc Mon Sep 17 00:00:00 2001
From: Piotr Jastrzebski <piotr@scylladb.com>
Date: Mon, 12 Apr 2021 20:57:15 +0200
Subject: [PATCH] read_context: add _partition_exists

This new state stores the information whether current partition
represented by _key is present in underlying.

Refs #8435.

Signed-off-by: Piotr Jastrzebski <piotr@scylladb.com>
(cherry picked from commit ceab5f026d34dc309c56197fc8248c2ce77af43b)
---
 read_context.hh | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

diff --git a/read_context.hh b/read_context.hh
index e16f0f5a55..6261537ab2 100644
--- a/read_context.hh
+++ b/read_context.hh
@@ -141,6 +141,7 @@ class read_context final : public enable_lw_shared_from_this<read_context> {
     mutation_source_opt _underlying_snapshot;
     dht::partition_range _sm_range;
     std::optional<dht::decorated_key> _key;
+    bool _partition_exists;
     row_cache::phase_type _phase;
 public:
     read_context(row_cache& cache,
@@ -189,15 +190,23 @@ public:
     autoupdating_underlying_reader& underlying() { return _underlying; }
     row_cache::phase_type phase() const { return _phase; }
     const dht::decorated_key& key() const { return *_key; }
+    bool partition_exists() const { return _partition_exists; }
     void on_underlying_created() { ++_underlying_created; }
     bool digest_requested() const { return _slice.options.contains<query::partition_slice::option::with_digest>(); }
 public:
     future<> ensure_underlying(db::timeout_clock::time_point timeout) {
         if (_underlying_snapshot) {
             return create_underlying(timeout).then([this, timeout] {
-                return _underlying.underlying()(timeout).discard_result();
+                return _underlying.underlying()(timeout).then([this] (mutation_fragment_opt&& mfopt) {
+                    _partition_exists = bool(mfopt);
+                });
             });
         }
+        // We know that partition exists because all the callers of
+        // enter_partition(const dht::decorated_key&, row_cache::phase_type)
+        // check that and there's no other way of setting _underlying_snapshot
+        // to empty. Except for calling create_underlying.
+        _partition_exists = true;
         return make_ready_future<>();
     }
 public:
@@ -207,6 +216,8 @@ public:
         _underlying_snapshot = snapshot;
         _key = dk;
     }
+    // Precondition: each caller needs to make sure that partition with |dk| key
+    //               exists in underlying before calling this function.
     void enter_partition(const dht::decorated_key& dk, row_cache::phase_type phase) {
         _phase = phase;
         _underlying_snapshot = {};