KVStore::remove Implementation Part 2

src/turtle_kv/core/merge_compactor.test.cpp

@@ -666,4 +666,4 @@ TEST_F(ResultSetConcatTest, ConcatDeath)
                "the second ResultSet!");
 }
 
-}  // namespace
+}  // namespace

src/turtle_kv/kv_store_scanner.cpp

@@ -469,6 +469,9 @@ Status KVStoreScanner::set_next_item()
 
   for (;;) {
     if (this->heap_.empty()) {
+      if (this->next_item_ && this->next_item_->value.is_delete()) {
+        this->next_item_ = None;
+      }
       return OkStatus();
     }
 

src/turtle_kv/tree/active_pivots_set.hpp

@@ -111,6 +111,11 @@ class ActivePivotsSetBase
     this->bit_set_ = insert_bit(this->bit_set_, i, v);
   }
 
+  BATT_ALWAYS_INLINE void remove(i32 i)
+  {
+    this->bit_set_ = remove_bit(this->bit_set_, i);
+  }
+
   BATT_ALWAYS_INLINE bool is_empty() const
   {
     return this->bit_set_ == Bitset{};
@@ -153,11 +158,11 @@ class ActivePivotsSet128 : public ActivePivotsSetBase<std::array<u64, 2>>
   BATT_ALWAYS_INLINE auto printable() const
   {
     return [this](std::ostream& out) {
-      out << std::bitset<64>{this->bit_set_[1]} << "," << std::bitset<64>{this->bit_set_[1]};
+      out << std::bitset<64>{this->bit_set_[1]} << "," << std::bitset<64>{this->bit_set_[0]};
     };
   }
 
-  /** \brief Removes the specified number (`count`) pivots from the bit set.
+  /** \brief Removes the specified number (`count`) pivots from the beginning of the bit set.
    */
   BATT_ALWAYS_INLINE void pop_front_pivots(i32 count)
   {
@@ -173,6 +178,33 @@ class ActivePivotsSet128 : public ActivePivotsSetBase<std::array<u64, 2>>
       this->bit_set_[1] = 0;
     }
   }
+
+  /** \brief Removes the specified number (`count`) pivots from the end of the bit set.
+   */
+  BATT_ALWAYS_INLINE void pop_back_pivots(i32 count)
+  {
+    if (count < 1) {
+      return;
+    }
+
+    if (count < 64) {
+      this->bit_set_[1] = (this->bit_set_[1] << count) | (this->bit_set_[0] >> (64 - count));
+      this->bit_set_[0] <<= count;
+    } else {
+      this->bit_set_[1] = this->bit_set_[0] << (count - 64);
+      this->bit_set_[0] = 0;
+    }
+  }
+
+  /** \brief Performs a bit wise OR operation with another bit set.
+   */
+  BATT_ALWAYS_INLINE Self& operator|=(const Self& other)
+  {
+    this->bit_set_[0] |= other.bit_set_[0];
+    this->bit_set_[1] |= other.bit_set_[1];
+
+    return *this;
+  }
 };
 
 static_assert(ActivePivotsSet<ActivePivotsSet128>);

src/turtle_kv/tree/algo/segmented_levels.hpp

@@ -187,15 +187,21 @@ struct SegmentedLevelAlgorithms {
 
       const PackedLeafPage& leaf = PackedLeafPage::view_of(pinned_page.get_page_buffer());
 
-      auto flushed_last = leaf.lower_bound(max_key);
-      if (flushed_last != leaf.items_end() && get_key(*flushed_last) == max_key) {
-        ++flushed_last;
-      }
-
       CInterval<KeyView> flush_key_crange{pivot_lower_bound_key, max_key};
       segment.drop_key_range(flush_key_crange, leaf.items_slice());
 
-      if (flushed_last == leaf.items_end() || get_key(*flushed_last) >= pivot_upper_bound_key) {
+      auto pivot_first = leaf.lower_bound(pivot_lower_bound_key);
+      usize pivot_first_i = std::distance(leaf.items_begin(), pivot_first);
+
+      auto pivot_last = leaf.lower_bound(pivot_upper_bound_key);
+      usize pivot_last_i = std::distance(leaf.items_begin(), pivot_last);
+
+      Interval<u32> pivot_live_range =
+          segment.get_live_item_range(this->level_,
+                                      Interval<u32>{BATT_CHECKED_CAST(u32, pivot_first_i),
+                                                    BATT_CHECKED_CAST(u32, pivot_last_i)});
+
+      if (pivot_live_range.empty()) {
         segment.set_pivot_active(pivot_i, false);
       }
 
@@ -283,6 +289,18 @@ struct SegmentedLevelAlgorithms {
     return OkStatus();
   }
 
+  /** \brief Merges the two given pivots, effectively erasing `right_pivot`.
+   */
+  void merge_pivots(i32 left_pivot, i32 right_pivot)
+  {
+    const usize segment_count = this->level_.segment_count();
+
+    for (usize segment_i = 0; segment_i < segment_count; ++segment_i) {
+      SegmentT& segment = this->level_.get_segment(segment_i);
+      in_segment(segment).merge_pivots(left_pivot, right_pivot, this->level_);
+    }
+  }
+
   /** \brief Invokes `fn` for each SegmentT& selected by `pivot_selector`.
    *
    * `pivot_selector` can be:

src/turtle_kv/tree/algo/segments.hpp

@@ -141,6 +141,18 @@ struct SegmentAlgorithms {
     return true;
   }
 
+  /** \brief Merges the two given pivots, effectively erasing `right_pivot`.
+   */
+  template <typename LevelT>
+  [[nodiscard]] void merge_pivots(i32 left_pivot, i32 right_pivot, const LevelT& level)
+  {
+    bool left_is_active = this->segment_.is_pivot_active(left_pivot);
+    bool right_is_active = this->segment_.is_pivot_active(right_pivot);
+    this->segment_.set_pivot_active(left_pivot, left_is_active | right_is_active);
+
+    this->segment_.remove_pivot(right_pivot);
+  }
+
   /** \brief Invokes the speficied `fn` for each active pivot in the specified range, passing a
    * reference to the segment and the pivot index (i32).
    */

src/turtle_kv/tree/in_memory_leaf.cpp

@@ -6,8 +6,53 @@
 #include <turtle_kv/tree/packed_leaf_page.hpp>
 #include <turtle_kv/tree/the_key.hpp>
 
+#include <batteries/algo/parallel_transform.hpp>
+
 namespace turtle_kv {
 
+//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
+//
+/*static*/ std::unique_ptr<InMemoryLeaf> InMemoryLeaf::unpack(
+    llfs::PinnedPage&& pinned_leaf_page,
+    const TreeOptions& tree_options,
+    const PackedLeafPage& packed_leaf,
+    batt::WorkerPool& worker_pool) noexcept
+{
+  std::unique_ptr<InMemoryLeaf> new_leaf =
+      std::make_unique<InMemoryLeaf>(batt::make_copy(pinned_leaf_page), tree_options);
+
+  Slice<const PackedKeyValue> packed_items = packed_leaf.items_slice();
+  std::vector<EditView> buffer;
+  buffer.reserve(packed_items.size());
+
+  {
+    batt::ScopedWorkContext context{worker_pool};
+
+    const ParallelAlgoDefaults& algo_defaults = parallel_algo_defaults();
+    const batt::TaskCount max_tasks{worker_pool.size() + 1};
+
+    batt::parallel_transform(
+        context,
+        packed_items.begin(),
+        packed_items.end(),
+        buffer.data(),
+        [](const PackedKeyValue& pkv) -> EditView {
+          return to_edit_view(pkv);
+        },
+        /*min_task_size = */ algo_defaults.copy_edits.min_task_size,
+        /*max_tasks = */ max_tasks);
+  }
+
+  MergeCompactor::ResultSet</*decay_to_items=*/true> result_set;
+  const ItemView* first_edit = (const ItemView*)buffer.data();
+  result_set.append(std::move(buffer), as_slice(first_edit, packed_items.size()));
+  new_leaf->result_set = std::move(result_set);
+
+  new_leaf->set_edit_size_totals(compute_running_total(worker_pool, *(new_leaf->result_set)));
+
+  return {std::move(new_leaf)};
+}
+
 //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
 //
 SubtreeViability InMemoryLeaf::get_viability()
@@ -150,6 +195,40 @@ auto InMemoryLeaf::make_split_plan() const -> StatusOr<SplitPlan>
   return plan;
 }
 
+//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
+//
+Status InMemoryLeaf::try_merge(BatchUpdateContext& context,
+                               std::unique_ptr<InMemoryLeaf> sibling) noexcept
+{
+  BATT_CHECK(this->result_set);
+  BATT_CHECK(sibling->result_set);
+
+  if (sibling->result_set->empty()) {
+    BATT_CHECK(batt::is_case<Viable>(this->get_viability()))
+        << "Sibling leaf is not viable, so this leaf must be viable!";
+    return OkStatus();
+  }
+
+  if (this->result_set->empty()) {
+    BATT_CHECK(batt::is_case<Viable>(sibling->get_viability()))
+        << "This leaf is not viable, so sibling leaf must be viable!";
+    this->pinned_leaf_page_ = std::move(sibling->pinned_leaf_page_);
+    this->result_set = std::move(sibling->result_set);
+    this->shared_edit_size_totals_ = sibling->shared_edit_size_totals_;
+    this->edit_size_totals = std::move(sibling->edit_size_totals);
+    return OkStatus();
+  }
+
+  BATT_CHECK_LT(this->get_max_key(), sibling->get_min_key());
+
+  this->result_set = MergeCompactor::ResultSet<true>::concat(std::move(*this->result_set),
+                                                             std::move(*(sibling->result_set)));
+
+  this->set_edit_size_totals(context.compute_running_total(*this->result_set));
+
+  return OkStatus();
+}
+
 //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
 //
 Status InMemoryLeaf::apply_batch_update(BatchUpdate& update) noexcept

src/turtle_kv/tree/in_memory_leaf.hpp

@@ -40,6 +40,13 @@ struct InMemoryLeaf {
 
   //+++++++++++-+-+--+----- --- -- -  -  -   -
 
+  static std::unique_ptr<InMemoryLeaf> unpack(llfs::PinnedPage&& pinned_leaf_page,
+                                              const TreeOptions& tree_options,
+                                              const PackedLeafPage& packed_leaf,
+                                              batt::WorkerPool& worker_pool) noexcept;
+
+  //+++++++++++-+-+--+----- --- -- -  -  -   -
+
   explicit InMemoryLeaf(llfs::PinnedPage&& pinned_leaf_page,
                         const TreeOptions& tree_options_arg) noexcept
       : pinned_leaf_page_{std::move(pinned_leaf_page)}
@@ -98,6 +105,8 @@ struct InMemoryLeaf {
 
   StatusOr<SplitPlan> make_split_plan() const;
 
+  Status try_merge(BatchUpdateContext& context, std::unique_ptr<InMemoryLeaf> sibling) noexcept;
+
   Status apply_batch_update(BatchUpdate& update) noexcept;
 
   Status start_serialize(TreeSerializeContext& context);