Merge pull request #29576 from andrwng/lsm-refresh

lsm: add refresh to read-only database

Author: andrwng

src/v/lsm/db/impl.cc

@@ -155,7 +155,7 @@ ss::future<lookup_result> impl::get(internal::key_view key) {
     auto current = _versions->current();
     version::get_stats stats{};
     auto result = co_await current->get(key, &stats);
-    if (current->update_stats(stats)) {
+    if (!_opts->readonly && current->update_stats(stats)) {
         maybe_schedule_compaction();
     }
     co_return result;
@@ -177,7 +177,7 @@ impl::create_iterator(iterator_options opts) {
           [this](internal::key_view key) {
               return _versions->current()->record_read_sample(key).then(
                 [this](bool compaction_needed) {
-                    if (compaction_needed) {
+                    if (!_opts->readonly && compaction_needed) {
                         maybe_schedule_compaction();
                     }
                 });
@@ -237,6 +237,14 @@ ss::future<> impl::flush() {
     return impl::flush(ssx::instant::infinite_future());
 }
 
+ss::future<bool> impl::refresh() {
+    if (!_opts->readonly) {
+        throw invalid_argument_exception(
+          "refresh() can only be called on a read-only database");
+    }
+    co_return co_await _versions->refresh();
+}
+
 ss::future<> impl::close() {
     vlog(log.trace, "close_start");
     _as.request_abort_ex(abort_requested_exception("database closing"));
@@ -274,7 +282,7 @@ ss::future<> impl::recover() {
 }
 
 void impl::maybe_schedule_compaction() {
-    if (_as.abort_requested()) {
+    if (_as.abort_requested() || _opts->readonly) {
         return;
     }
     if (!_flush_task && _imm) {

src/v/lsm/db/impl.h

@@ -88,6 +88,10 @@ class impl {
     // Flush with no deadline.
     ss::future<> flush();
 
+    // Reload the manifest from disk, adding a new version to the version set.
+    // Only valid if in read-only mode.
+    ss::future<bool> refresh();
+
     // Close the database, no more operations should happen to the database at
     // this point.
     //

src/v/lsm/db/tests/BUILD

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

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

@@ -10,6 +10,7 @@
  */
 
 #include "gtest/gtest.h"
+#include "lsm/core/exceptions.h"
 #include "lsm/core/internal/keys.h"
 #include "lsm/core/internal/options.h"
 #include "lsm/db/impl.h"
@@ -126,7 +127,17 @@ class ImplTest : public testing::Test {
     void SetUp() override {
         // Make a smaller sized database so we get some actual leveling
         // happening.
-        _options = ss::make_lw_shared<lsm::internal::options>({
+        _options = make_options();
+        _underlying_data_persistence = lsm::io::make_memory_data_persistence();
+        _meta_persistence = lsm::io::make_memory_metadata_persistence(
+          &_meta_persistence_controller);
+        _tracking_data = std::make_unique<tracking_data_persistence>(
+          _underlying_data_persistence.get());
+        open();
+    }
+
+    ss::lw_shared_ptr<lsm::internal::options> make_options() {
+        return ss::make_lw_shared<lsm::internal::options>({
           .levels = lsm::internal::options::make_levels(
             {.max_total_bytes = 1_MiB, .max_file_size = 256_KiB},
             /*multiplier=*/2,
@@ -136,22 +147,22 @@ class ImplTest : public testing::Test {
           .write_buffer_size = 256_KiB,
           .level_one_compaction_trigger = 2,
         });
-        _underlying_data_persistence = lsm::io::make_memory_data_persistence();
-        _meta_persistence = lsm::io::make_memory_metadata_persistence(
-          &_meta_persistence_controller);
-        _tracking_data = std::make_unique<tracking_data_persistence>(
-          _underlying_data_persistence.get());
-        open();
     }
 
     void TearDown() override {
         _db->close().get();
+        for (auto& db : _other_dbs) {
+            db->close().get();
+        }
         _underlying_data_persistence->close().get();
         _meta_persistence->close().get();
         _shadow.clear();
     }
 
-    void write_at_least(size_t size) {
+    void write_at_least(size_t size, lsm::db::impl* db = nullptr) {
+        if (!db) {
+            db = _db.get();
+        }
         auto batch = ss::make_lw_shared<lsm::db::memtable>();
         decltype(_shadow) shadow_batch;
         auto seqno = _db->max_applied_seqno().value_or(0_seqno);
@@ -168,15 +179,15 @@ class ImplTest : public testing::Test {
               shadow_entry(value.share(), key.seqno()));
             batch->put(key, value.share());
         }
-        _db->apply(std::move(batch)).get();
+        db->apply(std::move(batch)).get();
         // Only apply writes if db write was a success
         for (auto& [k, v] : shadow_batch) {
             _shadow.insert_or_assign(k, std::move(v));
         }
     }
 
-    testing::AssertionResult matches_shadow() {
-        return matches_shadow(_shadow, nullptr);
+    testing::AssertionResult matches_shadow(lsm::db::impl* db = nullptr) {
+        return matches_shadow(_shadow, nullptr, db);
     }
 
     shadow_map clone_shadow_map() {
@@ -187,9 +198,14 @@ class ImplTest : public testing::Test {
         return s;
     }
 
-    testing::AssertionResult
-    matches_shadow(const shadow_map& shadow, lsm::db::snapshot* snapshot) {
-        auto iter = _db->create_iterator({.snapshot = snapshot}).get();
+    testing::AssertionResult matches_shadow(
+      const shadow_map& shadow,
+      lsm::db::snapshot* snapshot,
+      lsm::db::impl* db = nullptr) {
+        if (!db) {
+            db = _db.get();
+        }
+        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()) {
@@ -232,16 +248,24 @@ class ImplTest : public testing::Test {
         _db->close().get();
         _tracking_data->reset_tracking();
     }
-    void open() {
-        _db = lsm::db::impl::open(
-                _options,
-                {
-                  .data = std::make_unique<proxy_data_persistence>(
-                    _tracking_data.get()),
-                  .metadata = std::make_unique<proxy_metadata_persistence>(
-                    _meta_persistence.get()),
-                })
-                .get();
+    void open() { _db = do_open(_options); }
+
+    lsm::db::impl* open(ss::lw_shared_ptr<lsm::internal::options> opts) {
+        _other_dbs.emplace_back(do_open(std::move(opts)));
+        return _other_dbs.back().get();
+    }
+
+    std::unique_ptr<lsm::db::impl>
+    do_open(ss::lw_shared_ptr<lsm::internal::options> opts) {
+        return lsm::db::impl::open(
+                 opts,
+                 {
+                   .data = std::make_unique<proxy_data_persistence>(
+                     _tracking_data.get()),
+                   .metadata = std::make_unique<proxy_metadata_persistence>(
+                     _meta_persistence.get()),
+                 })
+          .get();
     }
 
     auto max_applied_seqno() { return _db->max_applied_seqno(); }
@@ -268,6 +292,7 @@ class ImplTest : public testing::Test {
     std::unique_ptr<lsm::io::metadata_persistence> _meta_persistence;
     std::unique_ptr<tracking_data_persistence> _tracking_data;
     std::unique_ptr<lsm::db::impl> _db;
+    std::vector<std::unique_ptr<lsm::db::impl>> _other_dbs;
 };
 
 TEST_F(ImplTest, MemtableIsFlushed) {
@@ -527,4 +552,125 @@ TEST_F(ImplTest, GetFindsKeysWithDifferentSeqno) {
     EXPECT_FALSE(result.is_missing());
 }
 
+TEST_F(ImplTest, RefreshOnWritableDbThrows) {
+    EXPECT_THROW(_db->refresh().get(), lsm::invalid_argument_exception);
+}
+
+TEST_F(ImplTest, RefreshEmpty) {
+    auto read_opts = make_options();
+    read_opts->readonly = true;
+
+    // Sanity check that the read-only database sees nothing.
+    auto* read_db = open(read_opts);
+    EXPECT_TRUE(matches_shadow(read_db));
+
+    // Refresh shouldn't have issues with the persistence being empty.
+    read_db->refresh().get();
+    EXPECT_TRUE(matches_shadow(read_db));
+    EXPECT_FALSE(read_db->max_applied_seqno().has_value());
+}
+
+TEST_F(ImplTest, RefreshNoOp) {
+    write_at_least(512_KiB);
+    _db->flush().get();
+
+    auto read_opts = make_options();
+    read_opts->readonly = true;
+    auto* read_db = open(read_opts);
+    EXPECT_TRUE(matches_shadow(read_db));
+
+    auto max_persisted = _db->max_persisted_seqno();
+    write_at_least(512_KiB);
+
+    // Without flushing, refreshing should be a no-op.
+    EXPECT_FALSE(read_db->refresh().get());
+    EXPECT_LT(read_db->max_applied_seqno(), _db->max_applied_seqno());
+    EXPECT_EQ(read_db->max_persisted_seqno(), max_persisted);
+}
+
+TEST_F(ImplTest, RefreshSeesChanges) {
+    write_at_least(512_KiB);
+    _db->flush().get();
+    write_at_least(512_KiB);
+    _db->flush().get();
+
+    auto read_opts = make_options();
+    read_opts->readonly = true;
+    auto* read_db = open(read_opts);
+    EXPECT_TRUE(matches_shadow(read_db));
+    EXPECT_EQ(read_db->max_applied_seqno(), _db->max_applied_seqno());
+    EXPECT_EQ(read_db->max_persisted_seqno(), _db->max_persisted_seqno());
+
+    // Write new data.
+    write_at_least(512_KiB);
+    EXPECT_FALSE(matches_shadow(read_db));
+
+    // Even when flushing the data shouldn't be visible.
+    _db->flush().get();
+    EXPECT_FALSE(matches_shadow(read_db));
+    EXPECT_NE(read_db->max_applied_seqno(), _db->max_applied_seqno());
+    EXPECT_NE(read_db->max_persisted_seqno(), _db->max_persisted_seqno());
+
+    // But once we refresh it should match.
+    EXPECT_TRUE(read_db->refresh().get());
+    EXPECT_TRUE(matches_shadow(read_db));
+    EXPECT_EQ(read_db->max_applied_seqno(), _db->max_applied_seqno());
+    EXPECT_EQ(read_db->max_persisted_seqno(), _db->max_persisted_seqno());
+}
+
+TEST_F(ImplTest, RefreshWithSnapshot) {
+    write_at_least(512_KiB);
+    _db->flush().get();
+
+    auto read_opts = make_options();
+    read_opts->readonly = true;
+    auto* read_db = open(read_opts);
+    EXPECT_TRUE(matches_shadow(read_db));
+
+    auto snap = read_db->create_snapshot();
+    auto shadow = clone_shadow_map();
+    EXPECT_TRUE(matches_shadow(shadow, snap->get(), read_db));
+
+    write_at_least(512_KiB);
+    _db->flush().get();
+
+    // We should only match one we refresh.
+    EXPECT_FALSE(matches_shadow(read_db));
+    EXPECT_TRUE(read_db->refresh().get());
+    EXPECT_TRUE(matches_shadow(read_db));
+
+    // The snapshot should still match.
+    EXPECT_TRUE(matches_shadow(shadow, snap->get(), read_db));
+}
+
+TEST_F(ImplTest, RefreshFailsForLowerSeqno) {
+    write_at_least(512_KiB);
+    _db->flush().get();
+
+    // Nefarious case: open another database so it gets a low seqno and we can
+    // flush a seqno lower than the refreshed database below.
+    auto* another_db = open(make_options());
+
+    // Flush a couple more times to bump the seqno.
+    write_at_least(512_KiB);
+    _db->flush().get();
+    write_at_least(512_KiB);
+    _db->flush().get();
+
+    // Open a read-only database.
+    auto read_opts = make_options();
+    read_opts->readonly = true;
+    auto* read_db = open(read_opts);
+
+    // Write some data to the other database and flush, lowering the sequence
+    // number.
+    write_at_least(512_KiB, another_db);
+    another_db->flush().get();
+
+    // Refreshing shouldn't work.
+    auto before_refresh = read_db->max_applied_seqno();
+    EXPECT_ANY_THROW(read_db->refresh().get());
+    EXPECT_EQ(read_db->max_applied_seqno(), before_refresh);
+}
+
 } // namespace

src/v/lsm/db/version_set.cc

@@ -564,7 +564,42 @@ ss::future<> version_set::recover() {
     _last_seqno = m->last_seqno;
 }
 
+ss::future<bool> version_set::refresh() {
+    auto m = co_await read_manifest();
+    if (!m) {
+        co_return false;
+    }
+
+    if (m->next_file_id < _next_file_id) {
+        throw corruption_exception(
+          "manifest next_file_id {} is less than current {}",
+          m->next_file_id(),
+          _next_file_id());
+    }
+    if (_last_seqno.has_value() && m->last_seqno < _last_seqno.value()) {
+        throw corruption_exception(
+          "manifest last_seqno {} is less than current {}",
+          m->last_seqno(),
+          _last_seqno.value()());
+    }
+
+    if (m->next_file_id == _next_file_id && m->last_seqno == _last_seqno) {
+        co_return false;
+    }
+
+    finalize(m->version.get());
+    set_current(std::move(m->version));
+    _next_file_id = m->next_file_id;
+    _last_seqno = m->last_seqno;
+    co_return true;
+}
+
 void version_set::finalize(version* v) {
+    if (_options->readonly) {
+        // No need to compute any compaction states, as we won't run compaction
+        // in read-only mode.
+        return;
+    }
     // Precompute the best level for the next compaction
     internal::level best_level = 0_level;
     double best_score = static_cast<double>(v->_files[0_level].size())

src/v/lsm/db/version_set.h

@@ -162,6 +162,10 @@ class version_set : public file_id_allocator {
     // layer.
     ss::future<> recover();
 
+    // Reload the manifest from disk. Returns true if the manifest was updated,
+    // false if no change. Throws if the manifest would regress state.
+    ss::future<bool> refresh();
+
     // The latest seqno applied to the LSM tree.
     std::optional<internal::sequence_number> last_seqno() const {
         return _last_seqno;

src/v/lsm/lsm.cc

@@ -217,6 +217,8 @@ snapshot database::create_snapshot() {
 
 write_batch database::create_write_batch() { return write_batch{_impl.get()}; }
 
+ss::future<bool> database::refresh() { return _impl->refresh(); }
+
 write_batch::write_batch(db::impl* db)
   : _batch(ss::make_lw_shared<db::memtable>())
   , _db(db) {}