lsm: fix snapshot::get() skipping tombstones in SST files

A run (buildkite/82930, h/t Willem) of DebugRowsTest
test_read_and_write_rows caught a metastore validation failure where
partition metadata appeared to still exist after all rows had been
deleted. The partition_validator found stale metadata (via a get())
while correctly seeing that extents and terms were deleted (via iterator
scans).

In this particular test, a metastore flush happened to happen in between
deleting our rows and the post-test validation running, which unearthed
a bug in the LSM.

The SST point lookup path (snapshot::get -> impl::get -> version::get
-> sst::reader::internal_get) constructs an internal seek key with
the snapshot's seqno and value_type::value (=0). However, our LSM
encodes tombstone=1 and value=0. Internal keys sort by (seqno, type)
descending, so a tombstone (type=1) sorts before a value (type=0) at the
same seqno. The seek for the first entry >= the lookup key therefore
skips the tombstone and lands on the stale value from a prior seqno.

Iterator-based reads are unaffected because the db_iterator resolves
tombstones during iteration by deduplicating entries per user key.
The memtable's get() is also unaffected because it strips the type
before doing a lower_bound lookup.

This commit fixes snapshot::get() and database::get() to use
value_type::tombstone for the seek key, to use the highest type value,
so the seek key is the earliest possible internal key for a given
(user_key, seqno) pair. Also removes the now-stale dassert in
memtable::get() that rejected non-value lookup key types.

(cherry picked from commit 76009fc)

Author: andrwng

src/v/lsm/db/memtable.cc

@@ -186,10 +186,6 @@ void memtable::merge(ss::lw_shared_ptr<memtable> other) {
 }
 
 lookup_result memtable::get(internal::key_view key) {
-    dassert(
-      key.type() == internal::value_type::value,
-      "when getting from the memtable, keys must be of value type",
-      key.decode());
     auto it = _table.lower_bound(key.without_type());
     if (it != _table.end() && it->first.user_key() == key.user_key()) {
         if (it->first.type() == internal::value_type::tombstone) {

src/v/lsm/db/tests/BUILD

@@ -168,6 +168,7 @@ redpanda_cc_gtest(
     deps = [
         "//src/v/base",
         "//src/v/bytes:iobuf",
+        "//src/v/lsm",
         "//src/v/lsm/core:exceptions",
         "//src/v/lsm/core/internal:keys",
         "//src/v/lsm/core/internal:options",

src/v/lsm/db/tests/impl_test.cc

@@ -12,6 +12,7 @@
 #include "lsm/core/internal/options.h"
 #include "lsm/db/impl.h"
 #include "lsm/io/memory_persistence.h"
+#include "lsm/lsm.h"
 #include "random/generators.h"
 #include "ssx/clock.h"
 #include "test_utils/async.h"
@@ -154,6 +155,7 @@ class ImplTest : public testing::Test {
         _underlying_data_persistence->close().get();
         _meta_persistence->close().get();
         _shadow.clear();
+        _deleted_keys.clear();
     }
 
     void write_at_least(size_t size, lsm::db::impl* db = nullptr) {
@@ -179,6 +181,7 @@ class ImplTest : public testing::Test {
         db->apply(std::move(batch)).get();
         // Only apply writes if db write was a success
         for (auto& [k, v] : shadow_batch) {
+            _deleted_keys.erase(k);
             _shadow.insert_or_assign(k, std::move(v));
         }
     }
@@ -210,6 +213,7 @@ class ImplTest : public testing::Test {
         db->apply(std::move(batch)).get();
         for (auto& k : keys_to_delete) {
             _shadow.erase(k);
+            _deleted_keys.insert(k);
         }
     }
 
@@ -232,12 +236,14 @@ class ImplTest : public testing::Test {
         if (!db) {
             db = _db.get();
         }
+        std::vector<std::string> errors;
+
+        // Validate via iterator (range scan path).
         auto iter = db->create_iterator({.snapshot = snapshot}).get();
         auto it = shadow.begin();
-        std::vector<std::string> errors;
         for (iter->seek_to_first().get(); iter->valid(); iter->next().get()) {
             if (it == shadow.end()) {
-                errors.emplace_back("extra elements");
+                errors.emplace_back("iter: extra elements");
                 break;
             }
             bool key_eq = it->first == iter->key().user_key();
@@ -258,8 +264,44 @@ class ImplTest : public testing::Test {
             ++it;
         }
         if (it != shadow.end()) {
-            errors.emplace_back("missing elements");
+            errors.emplace_back("iter: missing elements");
+        }
+
+        // Validate via point lookups (get() path). Construct the lookup
+        // key the same way snapshot::get() / database::get() does.
+        auto get_seqno = snapshot ? snapshot->seqno()
+                                  : lsm::internal::sequence_number::max();
+        for (const auto& [user_key, entry] : shadow) {
+            auto lookup_key = lsm::internal::key::encode({
+              .key = lsm::user_key_view(user_key),
+              .seqno = get_seqno,
+              .type = lsm::internal::value_type::tombstone,
+            });
+            auto result = db->get(lookup_key).get();
+            if (result.is_missing()) {
+                errors.push_back(fmt::format("get: key {} missing", user_key));
+            } else if (result.is_tombstone()) {
+                errors.push_back(
+                  fmt::format("get: key {} unexpectedly tombstoned", user_key));
+            }
         }
+        for (const auto& user_key : _deleted_keys) {
+            if (shadow.contains(user_key)) {
+                continue;
+            }
+            auto lookup_key = lsm::internal::key::encode({
+              .key = lsm::user_key_view(user_key),
+              .seqno = get_seqno,
+              .type = lsm::internal::value_type::tombstone,
+            });
+            auto result = db->get(lookup_key).get();
+            if (!result.is_missing() && !result.is_tombstone()) {
+                errors.push_back(
+                  fmt::format(
+                    "get: deleted key {} returned a value", user_key));
+            }
+        }
+
         if (errors.empty()) {
             return testing::AssertionSuccess();
         }
@@ -313,6 +355,7 @@ class ImplTest : public testing::Test {
 
 protected:
     shadow_map _shadow;
+    std::set<ss::sstring> _deleted_keys;
     ss::lw_shared_ptr<lsm::internal::options> _options;
     std::unique_ptr<lsm::io::data_persistence> _underlying_data_persistence;
     lsm::io::memory_persistence_controller _meta_persistence_controller;
@@ -385,6 +428,12 @@ TEST_F(ImplTest, RandomizedWithDeletes) {
         EXPECT_TRUE(matches_shadow());
         delete_random_keys();
         EXPECT_TRUE(matches_shadow());
+
+        // Also test after flushes, with and without a snapshot.
+        _db->flush().get();
+        EXPECT_TRUE(matches_shadow());
+        auto snap = _db->create_snapshot();
+        EXPECT_TRUE(matches_shadow(_shadow, snap->get()));
     }
 }
 
@@ -593,6 +642,53 @@ TEST_F(ImplTest, GetFindsKeysWithDifferentSeqno) {
     EXPECT_FALSE(result.is_missing());
 }
 
+// Regression test: snapshot::get() must detect tombstones in SST files.
+TEST_F(ImplTest, SnapshotGetDetectsTombstoneInSST) {
+    using lsm::sequence_number;
+    auto data_persistence = lsm::io::make_memory_data_persistence();
+    lsm::io::memory_persistence_controller meta_ctl;
+    auto meta_persistence = lsm::io::make_memory_metadata_persistence(
+      &meta_ctl);
+    auto db = lsm::database::open(
+                lsm::options{},
+                {
+                  .data = std::move(data_persistence),
+                  .metadata = std::move(meta_persistence),
+                })
+                .get();
+
+    // Write a value and then tombstone it.
+    {
+        auto wb = db.create_write_batch();
+        wb.put("foo", iobuf::from("bar"), sequence_number(1));
+        db.apply(std::move(wb)).get();
+    }
+    {
+        auto wb = db.create_write_batch();
+        wb.remove("foo", sequence_number(2));
+        db.apply(std::move(wb)).get();
+    }
+
+    // Flush so both the value and tombstone move to an SST file.
+    db.flush(ssx::instant::infinite_future()).get();
+    ASSERT_EQ(db.max_persisted_seqno(), sequence_number(2));
+
+    auto snap = db.create_snapshot();
+
+    // Iterator correctly sees no live keys (extents/terms path).
+    {
+        auto iter = snap.create_iterator().get();
+        iter.seek_to_first().get();
+        EXPECT_FALSE(iter.valid()) << "iterator should see no live keys";
+    }
+
+    // snapshot::get() must also return nullopt for the deleted key.
+    auto result = snap.get("foo").get();
+    EXPECT_FALSE(result.has_value())
+      << "snapshot::get() should not find a deleted key";
+    db.close().get();
+}
+
 TEST_F(ImplTest, RefreshOnWritableDbThrows) {
     EXPECT_THROW(_db->refresh().get(), lsm::invalid_argument_exception);
 }

src/v/lsm/lsm.cc

@@ -194,7 +194,7 @@ ss::future<std::optional<iobuf>> database::get(std::string_view target) {
     auto key = internal::key::encode({
       .key = lsm::user_key_view(target),
       .seqno = internal::sequence_number::max(),
-      .type = internal::value_type::value,
+      .type = internal::value_type::tombstone,
     });
     auto result = co_await _impl->get(key);
     co_return std::move(result).take_value();
@@ -277,7 +277,7 @@ ss::future<std::optional<iobuf>> snapshot::get(std::string_view target) {
     auto key = internal::key::encode({
       .key = lsm::user_key_view(target),
       .seqno = _snap->seqno(),
-      .type = internal::value_type::value,
+      .type = internal::value_type::tombstone,
     });
     auto result = co_await _db->get(key);
     co_return std::move(result).take_value();