Merge pull request #28856 from rockwotj/lsm_followup_memtable

lsm: support read-your-own-uncommitted-writes

Author: rockwotj

src/v/lsm/BUILD

@@ -9,11 +9,11 @@ redpanda_cc_library(
         "lsm.h",
     ],
     implementation_deps = [
-        "//src/v/lsm/core/internal:batch",
         "//src/v/lsm/core/internal:files",
         "//src/v/lsm/core/internal:iterator",
         "//src/v/lsm/core/internal:keys",
         "//src/v/lsm/db:impl",
+        "//src/v/lsm/db:memtable",
     ],
     visibility = ["//visibility:public"],
     deps = [

src/v/lsm/core/internal/BUILD

@@ -94,15 +94,3 @@ redpanda_cc_library(
         "@seastar",
     ],
 )
-
-redpanda_cc_library(
-    name = "batch",
-    hdrs = ["batch.h"],
-    deps = [
-        ":keys",
-        "//src/v/base",
-        "//src/v/bytes:iobuf",
-        "@abseil-cpp//absl/container:btree",
-        "@seastar",
-    ],
-)

src/v/lsm/core/internal/batch.h

@@ -70,7 +70,6 @@ redpanda_cc_library(
         "//src/v/base",
         "//src/v/bytes:iobuf",
         "//src/v/lsm/core:exceptions",
-        "//src/v/lsm/core/internal:batch",
         "//src/v/lsm/core/internal:files",
         "//src/v/lsm/core/internal:iterator",
         "//src/v/lsm/core/internal:keys",
@@ -94,7 +93,6 @@ redpanda_cc_library(
         "//src/v/base",
         "//src/v/bytes:iobuf",
         "//src/v/lsm/core:lookup_result",
-        "//src/v/lsm/core/internal:batch",
         "//src/v/lsm/core/internal:iterator",
         "//src/v/lsm/core/internal:keys",
         "@abseil-cpp//absl/container:btree",

src/v/lsm/db/BUILD

@@ -103,12 +103,12 @@ ss::future<std::unique_ptr<impl>> impl::open(
     co_return db;
 }
 
-ss::future<> impl::apply(internal::write_batch batch) {
-    if (batch.empty()) {
+ss::future<> impl::apply(ss::lw_shared_ptr<memtable> batch) {
+    if (batch->empty()) {
         co_return;
     }
     co_await make_room_for_write();
-    _mem->apply(std::move(batch));
+    _mem->merge(std::move(batch));
 }
 
 ss::future<> impl::make_room_for_write() {
@@ -186,13 +186,22 @@ ss::future<lookup_result> impl::get(internal::key_view key) {
     co_return result;
 }
 
-ss::future<std::unique_ptr<internal::iterator>> impl::create_iterator() {
-    auto iter = co_await create_internal_iterator();
+ss::future<std::unique_ptr<internal::iterator>>
+impl::create_iterator(ss::optimized_optional<ss::lw_shared_ptr<memtable>> wb) {
+    auto iter = co_await create_internal_iterator(wb->get());
+    internal::sequence_number iter_seqno = max_applied_seqno();
+    if (wb) {
+        auto wb_seqno = (*wb)->last_seqno().value_or(iter_seqno);
+        vassert(
+          wb_seqno > iter_seqno,
+          "write memtable seqno must be greater than the current "
+          "max_applied_seqno: {} > {}",
+          wb_seqno,
+          iter_seqno);
+        iter_seqno = wb_seqno;
+    }
     co_return create_db_iterator(
-      std::move(iter),
-      max_applied_seqno(),
-      _opts,
-      [this](internal::key_view key) {
+      std::move(iter), iter_seqno, _opts, [this](internal::key_view key) {
           return _versions->current()->record_read_sample(key).then(
             [this](bool compaction_needed) {
                 if (compaction_needed) {
@@ -203,8 +212,11 @@ ss::future<std::unique_ptr<internal::iterator>> impl::create_iterator() {
 }
 
 ss::future<std::unique_ptr<internal::iterator>>
-impl::create_internal_iterator() {
+impl::create_internal_iterator(ss::optimized_optional<memtable*> wb) {
     chunked_vector<std::unique_ptr<internal::iterator>> list;
+    if (wb) {
+        list.push_back((*wb)->create_iterator());
+    }
     list.push_back(_mem->create_iterator());
     if (_imm) {
         list.push_back((*_imm)->create_iterator());

src/v/lsm/db/impl.cc

@@ -12,8 +12,6 @@
 #pragma once
 
 #include "base/seastarx.h"
-#include "bytes/iobuf.h"
-#include "lsm/core/internal/batch.h"
 #include "lsm/core/internal/iterator.h"
 #include "lsm/core/internal/keys.h"
 #include "lsm/core/internal/options.h"
@@ -54,15 +52,17 @@ class impl {
       open(ss::lw_shared_ptr<internal::options>, io::persistence);
 
     // Apply a batch of writes to the database atomically.
-    ss::future<> apply(internal::write_batch);
+    ss::future<> apply(ss::lw_shared_ptr<memtable>);
 
     // Get a key from the database
     ss::future<lookup_result> get(internal::key_view);
 
-    // Create an iterator over the database. Note that this iterator
-    // results in ALL entries from the database, a deduplicating iterator
-    // needs to be added on top to give a traditional iterator view.
-    ss::future<std::unique_ptr<internal::iterator>> create_iterator();
+    // Create an interator over the database.
+    //
+    // If a non-null memtable is passed in, then a frozen state of the memtable
+    // is applied ontop of the existing database.
+    ss::future<std::unique_ptr<internal::iterator>>
+      create_iterator(ss::optimized_optional<ss::lw_shared_ptr<memtable>>);
 
     // Flush any pending state in memtables to disk.
     ss::future<> flush();
@@ -74,7 +74,14 @@ class impl {
     ss::future<> close();
 
 private:
-    ss::future<std::unique_ptr<internal::iterator>> create_internal_iterator();
+    // Create an iterator over the database. Note that this iterator
+    // results in ALL entries from the database, a deduplicating iterator
+    // needs to be added on top to give a traditional iterator view.
+    //
+    // If a non-null memtable is passed in, then a frozen state of the memtable
+    // is applied ontop of the existing database.
+    ss::future<std::unique_ptr<internal::iterator>>
+      create_internal_iterator(ss::optimized_optional<memtable*>);
 
     ss::future<> recover();
 

src/v/lsm/db/impl.h

@@ -23,6 +23,8 @@
 
 namespace lsm::db {
 
+using internal::operator""_seqno;
+
 class memtable::iterator : public internal::iterator {
 public:
     // The dummy iterator is only a place holder for the linked list in the
@@ -140,19 +142,50 @@ class memtable::iterator : public internal::iterator {
         _it;
 };
 
-void memtable::apply(internal::write_batch batch) {
-    if (_last_seqno) {
-        dassert(
-          _last_seqno < batch.last_seqno(),
-          "expected new batch seqno to be greater than what is applied: {} < "
-          "{}",
-          _last_seqno.value(),
-          batch.last_seqno());
-    }
+void memtable::put(internal::key key, iobuf value) {
+    internal::key_view::parts parts = internal::key_view{key}.decode();
+    vassert(
+      parts.type == internal::value_type::value,
+      "when add a put to a memtable, keys much be of value type",
+      key);
+    vassert(
+      parts.seqno >= _last_seqno,
+      "seqno should only go up: {} >= {}",
+      parts.seqno);
+    invalidate_iterators();
+    _memory_usage += key.memory_usage() + value.memory_usage();
+    _last_seqno = parts.seqno;
+    _table.emplace(std::move(key), std::move(value));
+}
+
+void memtable::remove(internal::key key) {
+    internal::key_view::parts parts = internal::key_view{key}.decode();
+    vassert(
+      parts.type == internal::value_type::tombstone,
+      "when add a tombstone to a memtable, keys much be of tombstone type",
+      key);
+    vassert(
+      parts.seqno >= _last_seqno,
+      "seqno should only go up: {} >= {}",
+      parts.seqno);
+    invalidate_iterators();
+    iobuf value;
+    _memory_usage += key.memory_usage() + value.memory_usage();
+    _last_seqno = parts.seqno;
+    _table.emplace(std::move(key), std::move(value));
+}
+
+void memtable::merge(ss::lw_shared_ptr<memtable> other) {
+    vassert(
+      _last_seqno < other->last_seqno(),
+      "expected new batch seqno to be greater than what is applied: {} < "
+      "{}",
+      _last_seqno.value_or(0_seqno),
+      other->last_seqno().value_or(0_seqno));
     invalidate_iterators();
-    _memory_usage += batch.memory_usage();
-    _last_seqno = batch.last_seqno();
-    _table.merge(std::move(batch.entries()));
+    _memory_usage += other->approximate_memory_usage();
+    _last_seqno = other->last_seqno();
+    _table.merge(std::move(other->_table));
 }
 
 lookup_result memtable::get(internal::key_view key) {

src/v/lsm/db/memtable.cc

@@ -14,7 +14,6 @@
 #include "absl/container/btree_map.h"
 #include "base/seastarx.h"
 #include "bytes/iobuf.h"
-#include "lsm/core/internal/batch.h"
 #include "lsm/core/internal/iterator.h"
 #include "lsm/core/internal/keys.h"
 #include "lsm/core/lookup_result.h"
@@ -43,11 +42,24 @@ class memtable : public ss::enable_lw_shared_from_this<memtable> {
     memtable& operator=(memtable&&) = delete;
     ~memtable();
 
-    // Add batch of key-value pairs to the memtable.
+    // Add a key-value pair to the database. Overwrites any previous value.
     //
-    // REQUIRES: the sequence numbers in the write batch are >= to the last
-    // applied sequence number in this memtable.
-    void apply(internal::write_batch);
+    // REQUIRES: key.value_type is value
+    void put(internal::key key, iobuf value);
+
+    // Remove the value for a given key.
+    //
+    // REQUIRES: key.value_type is tombstone
+    void remove(internal::key key);
+
+    // Merge the supplied memtable into this memtable.
+    // The resulting memtable should not be used anymore after passed into
+    // this function as the data is *taken* from the memtable and applied to
+    // this memtable.
+    //
+    // REQUIRES: the sequence numbers in the supplied memtable are >= to the
+    // last applied sequence number in this memtable.
+    void merge(ss::lw_shared_ptr<memtable>);
 
     // Get the value for a given key.
     //
@@ -56,7 +68,8 @@ class memtable : public ss::enable_lw_shared_from_this<memtable> {
 
     // Create an iterator for this memtable.
     //
-    // This iterator is safe to use in face of concurrent updates
+    // This iterator is safe to use in face of concurrent updates, but users can
+    // see newer values that are added since the iterator is created.
     std::unique_ptr<internal::iterator> create_iterator();
 
     // The approximate amount of memory used for this memtable.

src/v/lsm/db/memtable.h

@@ -10,11 +10,11 @@ redpanda_cc_binary(
         "//src/v/base",
         "//src/v/bytes:iobuf",
         "//src/v/lsm/core:compression",
-        "//src/v/lsm/core/internal:batch",
         "//src/v/lsm/core/internal:iterator",
         "//src/v/lsm/core/internal:keys",
         "//src/v/lsm/core/internal:options",
         "//src/v/lsm/db:impl",
+        "//src/v/lsm/db:memtable",
         "//src/v/lsm/io:disk_persistence",
         "//src/v/lsm/io:memory_persistence",
         "//src/v/random:generators",
@@ -35,7 +35,6 @@ redpanda_cc_gtest(
         "//src/v/base",
         "//src/v/bytes:iobuf",
         "//src/v/lsm/core:lookup_result",
-        "//src/v/lsm/core/internal:batch",
         "//src/v/lsm/core/internal:keys",
         "//src/v/lsm/core/internal/tests:iterator_test_harness",
         "//src/v/lsm/db:memtable",
@@ -131,7 +130,6 @@ redpanda_cc_gtest(
     deps = [
         "//src/v/base",
         "//src/v/bytes:iobuf",
-        "//src/v/lsm/core/internal:batch",
         "//src/v/lsm/core/internal:iterator",
         "//src/v/lsm/core/internal:keys",
         "//src/v/lsm/core/internal:options",
@@ -152,10 +150,10 @@ redpanda_cc_gtest(
     deps = [
         "//src/v/base",
         "//src/v/bytes:iobuf",
-        "//src/v/lsm/core/internal:batch",
         "//src/v/lsm/core/internal:keys",
         "//src/v/lsm/core/internal:options",
         "//src/v/lsm/db:impl",
+        "//src/v/lsm/db:memtable",
         "//src/v/lsm/io:memory_persistence",
         "//src/v/lsm/sst:builder",
         "//src/v/random:generators",