Add striped implementation of B-tree content node bucket database

Abstracts away multiple underlying B-tree DBs that each hold a subset
of the super bucket space. Offers ordered iteration via a priority-queue
based view over the sub DBs.

Not yet ready for prime time, as the striping inherently requires an
absolute lower bound on the bucket bits used in the system, which is
currently not enforced.

17 files changed, 517 insertions, 76 deletions

diff --git a/storage/src/tests/bucketdb/lockablemaptest.cpp b/storage/src/tests/bucketdb/lockablemaptest.cpp
index f5e68dd3de1..c84293ca9f9 100644
--- a/storage/src/tests/bucketdb/lockablemaptest.cpp
+++ b/storage/src/tests/bucketdb/lockablemaptest.cpp
@@ -2,6 +2,7 @@
 
 #include <vespa/vespalib/util/document_runnable.h>
 #include <vespa/storage/bucketdb/btree_lockable_map.hpp>
+#include <vespa/storage/bucketdb/striped_btree_lockable_map.hpp>
 #include <vespa/vespalib/gtest/gtest.h>
 #include <gmock/gmock.h>
 
@@ -9,6 +10,8 @@
 LOG_SETUP(".lockable_map_test");
 
 // FIXME these old tests may have the least obvious semantics and worst naming in the entire storage module
+// FIXME the non-bucket ID based tests only "accidentally" work with the striped DB implementation
+// since they just all happen to look like zero-buckets with count-bits above the minimum threshold.
 
 using namespace ::testing;
 using document::BucketId;
@@ -55,8 +58,7 @@ struct LockableMapTest : ::testing::Test {
     using Map = MapT;
 };
 
-// TODO add striped B-tree DB to this type set once ready
-using MapTypes = ::testing::Types<bucketdb::BTreeLockableMap<A>>;
+using MapTypes = ::testing::Types<bucketdb::BTreeLockableMap<A>, bucketdb::StripedBTreeLockableMap<A>>;
 VESPA_GTEST_TYPED_TEST_SUITE(LockableMapTest, MapTypes);
 
 // Disable warnings emitted by gtest generated files when using typed tests
@@ -102,46 +104,6 @@ TYPED_TEST(LockableMapTest, simple_usage) {
     EXPECT_TRUE(map.empty());
 }
 
-TYPED_TEST(LockableMapTest, comparison) {
-    TypeParam map1;
-    TypeParam map2;
-    bool preExisted;
-
-    // Check empty state is correct
-    EXPECT_EQ(map1, map2);
-    EXPECT_FALSE(map1 < map2);
-    EXPECT_FALSE(map1 != map2);
-
-    // Check that different lengths are ok
-    map1.insert(4, A(1, 2, 3), "foo", preExisted);
-    EXPECT_FALSE(map1 == map2);
-    EXPECT_FALSE(map1 < map2);
-    EXPECT_LT(map2, map1);
-    EXPECT_NE(map1, map2);
-
-    // Check that equal elements are ok
-    map2.insert(4, A(1, 2, 3), "foo", preExisted);
-    EXPECT_EQ(map1, map2);
-    EXPECT_FALSE(map1 < map2);
-    EXPECT_FALSE(map1 != map2);
-
-    // Check that non-equal values are ok
-    map1.insert(6, A(1, 2, 6), "foo", preExisted);
-    map2.insert(6, A(1, 2, 3), "foo", preExisted);
-    EXPECT_FALSE(map1 == map2);
-    EXPECT_FALSE(map1 < map2);
-    EXPECT_LT(map2, map1);
-    EXPECT_NE(map1, map2);
-
-    // Check that non-equal keys are ok
-    map1.erase(6, "foo");
-    map1.insert(7, A(1, 2, 3), "foo", preExisted);
-    EXPECT_FALSE(map1 == map2);
-    EXPECT_FALSE(map1 < map2);
-    EXPECT_LT(map2, map1);
-    EXPECT_NE(map1, map2);
-}
-
 namespace {
 
 template <typename Map>
@@ -721,16 +683,16 @@ TYPED_TEST(LockableMapTest, find_all_inconsistently_split_6) {
 TYPED_TEST(LockableMapTest, find_all_inconsistent_below_16_bits) {
     TypeParam map;
 
-    document::BucketId id1(1, 0x1); // contains id2-id3
-    document::BucketId id2(3, 0x1);
-    document::BucketId id3(4, 0xD);
+    document::BucketId id1(8,  0b0000'0000'0001); // contains id2-id3
+    document::BucketId id2(10, 0b0011'0000'0001);
+    document::BucketId id3(11, 0b0101'0000'0001);
 
     bool preExisted;
     map.insert(id1.stripUnused().toKey(), A(1,2,3), "foo", preExisted);
     map.insert(id2.stripUnused().toKey(), A(2,3,4), "foo", preExisted);
     map.insert(id3.stripUnused().toKey(), A(3,4,5), "foo", preExisted);
 
-    document::BucketId id(3, 0x5);
+    document::BucketId id(10, 0b0001'0000'0001);
 
     auto results = map.getAll(id, "foo");
 
diff --git a/storage/src/vespa/storage/bucketdb/CMakeLists.txt b/storage/src/vespa/storage/bucketdb/CMakeLists.txt
index 23974cf2522..20510d895e7 100644
--- a/storage/src/vespa/storage/bucketdb/CMakeLists.txt
+++ b/storage/src/vespa/storage/bucketdb/CMakeLists.txt
@@ -10,6 +10,7 @@ vespa_add_library(storage_bucketdb OBJECT
     bucketmanagermetrics.cpp
     generic_btree_bucket_database.cpp
     storbucketdb.cpp
+    striped_btree_lockable_map.cpp
     DEPENDS
 )
 vespa_generate_config(storage_bucketdb stor-bucketdb.def)
diff --git a/storage/src/vespa/storage/bucketdb/abstract_bucket_map.h b/storage/src/vespa/storage/bucketdb/abstract_bucket_map.h
index eedef2ec0ff..bf46f7c3992 100644
--- a/storage/src/vespa/storage/bucketdb/abstract_bucket_map.h
+++ b/storage/src/vespa/storage/bucketdb/abstract_bucket_map.h
@@ -177,12 +177,8 @@ public:
         do_for_each_mutable_unordered(std::move(func), clientId);
     }
 
-    void for_each(std::function<Decision(uint64_t, const ValueT&)> func,
-                  const char* clientId,
-                  const key_type& first = 0,
-                  const key_type& last = UINT64_MAX)
-    {
-        do_for_each(std::move(func), clientId, first, last);
+    void for_each(std::function<Decision(uint64_t, const ValueT&)> func, const char* clientId) {
+        do_for_each(std::move(func), clientId);
     }
 
     std::unique_ptr<bucketdb::ReadGuard<ValueT>> acquire_read_guard() const {
@@ -206,9 +202,7 @@ private:
     virtual void do_for_each_mutable_unordered(std::function<Decision(uint64_t, ValueT&)> func,
                                                const char* clientId) = 0;
     virtual void do_for_each(std::function<Decision(uint64_t, const ValueT&)> func,
-                             const char* clientId,
-                             const key_type& first,
-                             const key_type& last) = 0;
+                             const char* clientId) = 0;
     virtual std::unique_ptr<bucketdb::ReadGuard<ValueT>> do_acquire_read_guard() const = 0;
 };
 
diff --git a/storage/src/vespa/storage/bucketdb/btree_bucket_database.cpp b/storage/src/vespa/storage/bucketdb/btree_bucket_database.cpp
index 602cdbdc5d5..d547830d765 100644
--- a/storage/src/vespa/storage/bucketdb/btree_bucket_database.cpp
+++ b/storage/src/vespa/storage/bucketdb/btree_bucket_database.cpp
@@ -195,7 +195,9 @@ vespalib::MemoryUsage BTreeBucketDatabase::memory_usage() const noexcept {
     return _impl->memory_usage();
 }
 
-class BTreeBucketDatabase::ReadGuardImpl final : public bucketdb::ReadGuard<Entry> {
+class BTreeBucketDatabase::ReadGuardImpl final
+    : public bucketdb::ReadGuard<Entry, ConstEntryRef>
+{
     ImplType::ReadSnapshot _snapshot;
 public:
     explicit ReadGuardImpl(const BTreeBucketDatabase& db);
@@ -204,6 +206,7 @@ public:
     std::vector<Entry> find_parents_and_self(const document::BucketId& bucket) const override;
     std::vector<Entry> find_parents_self_and_children(const document::BucketId& bucket) const override;
     void for_each(std::function<void(uint64_t, const Entry&)> func) const override;
+    std::unique_ptr<bucketdb::ConstIterator<ConstEntryRef>> create_iterator() const override;
     [[nodiscard]] uint64_t generation() const noexcept override;
 };
 
@@ -235,11 +238,17 @@ void BTreeBucketDatabase::ReadGuardImpl::for_each(std::function<void(uint64_t, c
     _snapshot.for_each<ByValue>(std::move(func));
 }
 
+std::unique_ptr<bucketdb::ConstIterator<ConstEntryRef>>
+BTreeBucketDatabase::ReadGuardImpl::create_iterator() const {
+    return _snapshot.create_iterator(); // TODO test
+}
+
 uint64_t BTreeBucketDatabase::ReadGuardImpl::generation() const noexcept {
     return _snapshot.generation();
 }
 
-std::unique_ptr<bucketdb::ReadGuard<Entry>> BTreeBucketDatabase::acquire_read_guard() const {
+std::unique_ptr<bucketdb::ReadGuard<Entry, ConstEntryRef>>
+BTreeBucketDatabase::acquire_read_guard() const {
     return std::make_unique<ReadGuardImpl>(*this);
 }
 
diff --git a/storage/src/vespa/storage/bucketdb/btree_bucket_database.h b/storage/src/vespa/storage/bucketdb/btree_bucket_database.h
index 122c6eeb0fb..2821e360792 100644
--- a/storage/src/vespa/storage/bucketdb/btree_bucket_database.h
+++ b/storage/src/vespa/storage/bucketdb/btree_bucket_database.h
@@ -58,7 +58,7 @@ private:
     class ReadGuardImpl;
     friend class ReadGuardImpl;
 public:
-    std::unique_ptr<bucketdb::ReadGuard<Entry>> acquire_read_guard() const override;
+    std::unique_ptr<bucketdb::ReadGuard<Entry, ConstEntryRef>> acquire_read_guard() const override;
 
     vespalib::MemoryUsage memory_usage() const noexcept override;
 };
diff --git a/storage/src/vespa/storage/bucketdb/btree_lockable_map.h b/storage/src/vespa/storage/bucketdb/btree_lockable_map.h
index 6e42a721732..5667c75c5d7 100644
--- a/storage/src/vespa/storage/bucketdb/btree_lockable_map.h
+++ b/storage/src/vespa/storage/bucketdb/btree_lockable_map.h
@@ -73,6 +73,8 @@ public:
     void showLockClients(vespalib::asciistream & out) const override;
 
 private:
+    template <typename T1> friend class StripedBTreeLockableMap;
+
     struct hasher {
         size_t operator () (const LockId & lid) const { return lid.hash(); }
     };
@@ -121,9 +123,7 @@ private:
                                        const char* clientId) override;
 
     void do_for_each(std::function<Decision(uint64_t, const mapped_type&)> func,
-                     const char* clientId,
-                     const key_type& first,
-                     const key_type& last) override;
+                     const char* clientId) override;
 
     void do_for_each_chunked(std::function<Decision(uint64_t, const mapped_type&)> func,
                              const char* client_id,
diff --git a/storage/src/vespa/storage/bucketdb/btree_lockable_map.hpp b/storage/src/vespa/storage/bucketdb/btree_lockable_map.hpp
index d291cfd371a..c8597915e44 100644
--- a/storage/src/vespa/storage/bucketdb/btree_lockable_map.hpp
+++ b/storage/src/vespa/storage/bucketdb/btree_lockable_map.hpp
@@ -1,4 +1,6 @@
 // Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#pragma once
+
 #include "btree_lockable_map.h"
 #include "generic_btree_bucket_database.hpp"
 #include <vespa/vespalib/btree/btreebuilder.h>
@@ -293,15 +295,13 @@ void BTreeLockableMap<T>::do_for_each_mutable_unordered(std::function<Decision(u
 
 template <typename T>
 void BTreeLockableMap<T>::do_for_each(std::function<Decision(uint64_t, const mapped_type&)> func,
-                                      const char* clientId,
-                                      const key_type& first,
-                                      const key_type& last)
+                                      const char* clientId)
 {
-    key_type key = first;
+    key_type key = 0;
     mapped_type val;
     std::unique_lock guard(_lock);
     while (true) {
-        if (findNextKey(key, val, clientId, guard) || key > last) {
+        if (findNextKey(key, val, clientId, guard)) {
             return;
         }
         Decision d(func(key, val));
@@ -363,6 +363,7 @@ public:
     std::vector<T> find_parents_and_self(const document::BucketId& bucket) const override;
     std::vector<T> find_parents_self_and_children(const document::BucketId& bucket) const override;
     void for_each(std::function<void(uint64_t, const T&)> func) const override;
+    std::unique_ptr<ConstIterator<const T&>> create_iterator() const override;
     [[nodiscard]] uint64_t generation() const noexcept override;
 };
 
@@ -404,6 +405,12 @@ void BTreeLockableMap<T>::ReadGuardImpl::for_each(std::function<void(uint64_t, c
 }
 
 template <typename T>
+std::unique_ptr<ConstIterator<const T&>>
+BTreeLockableMap<T>::ReadGuardImpl::create_iterator() const {
+    return _snapshot.template create_iterator(); // TODO test
+}
+
+template <typename T>
 uint64_t BTreeLockableMap<T>::ReadGuardImpl::generation() const noexcept {
     return _snapshot.generation();
 }
diff --git a/storage/src/vespa/storage/bucketdb/bucketdatabase.h b/storage/src/vespa/storage/bucketdb/bucketdatabase.h
index 1ee165a739c..cd94a698358 100644
--- a/storage/src/vespa/storage/bucketdb/bucketdatabase.h
+++ b/storage/src/vespa/storage/bucketdb/bucketdatabase.h
@@ -141,10 +141,10 @@ public:
 
     virtual uint32_t childCount(const document::BucketId&) const = 0;
 
-    using ReadGuard = bucketdb::ReadGuard<Entry>;
+    using ReadGuard = bucketdb::ReadGuard<Entry, ConstEntryRef>;
 
     virtual std::unique_ptr<ReadGuard> acquire_read_guard() const {
-        return std::unique_ptr<bucketdb::ReadGuard<Entry>>();
+        return std::unique_ptr<bucketdb::ReadGuard<Entry, ConstEntryRef>>();
     }
 
     [[nodiscard]] virtual vespalib::MemoryUsage memory_usage() const noexcept = 0;
diff --git a/storage/src/vespa/storage/bucketdb/const_iterator.h b/storage/src/vespa/storage/bucketdb/const_iterator.h
new file mode 100644
index 00000000000..fd42ba971e9
--- /dev/null
+++ b/storage/src/vespa/storage/bucketdb/const_iterator.h
@@ -0,0 +1,18 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#pragma once
+
+#include <cstdint>
+
+namespace storage::bucketdb {
+
+template <typename ConstRefT>
+class ConstIterator {
+public:
+    virtual ~ConstIterator() = default;
+    virtual void next() noexcept = 0;
+    [[nodiscard]] virtual bool valid() const noexcept = 0;
+    [[nodiscard]] virtual uint64_t key() const noexcept = 0;
+    [[nodiscard]] virtual ConstRefT value() const = 0;
+};
+
+}
\ No newline at end of file
diff --git a/storage/src/vespa/storage/bucketdb/generic_btree_bucket_database.h b/storage/src/vespa/storage/bucketdb/generic_btree_bucket_database.h
index 9534b583cd5..ea6d47f26bb 100644
--- a/storage/src/vespa/storage/bucketdb/generic_btree_bucket_database.h
+++ b/storage/src/vespa/storage/bucketdb/generic_btree_bucket_database.h
@@ -1,6 +1,7 @@
 // Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 #pragma once
 
+#include "const_iterator.h"
 #include "db_merger.h"
 #include <vespa/document/bucket/bucketid.h>
 #include <vespa/vespalib/btree/btree.h>
@@ -120,6 +121,8 @@ public:
         const GenericBTreeBucketDatabase*  _db;
         vespalib::GenerationHandler::Guard _guard;
         typename BTree::FrozenView         _frozen_view;
+
+        class ConstIteratorImpl;
     public:
         explicit ReadSnapshot(const GenericBTreeBucketDatabase& db);
         ~ReadSnapshot();
@@ -133,6 +136,7 @@ public:
         void find_parents_self_and_children(const document::BucketId& bucket, Func func) const;
         template <typename IterValueExtractor, typename Func>
         void for_each(Func func) const;
+        std::unique_ptr<ConstIterator<ConstValueRef>> create_iterator() const;
         [[nodiscard]] uint64_t generation() const noexcept;
     };
 private:
diff --git a/storage/src/vespa/storage/bucketdb/generic_btree_bucket_database.hpp b/storage/src/vespa/storage/bucketdb/generic_btree_bucket_database.hpp
index e53fad84db2..839efae4c1b 100644
--- a/storage/src/vespa/storage/bucketdb/generic_btree_bucket_database.hpp
+++ b/storage/src/vespa/storage/bucketdb/generic_btree_bucket_database.hpp
@@ -565,6 +565,41 @@ void GenericBTreeBucketDatabase<DataStoreTraitsT>::ReadSnapshot::for_each(Func f
 }
 
 template <typename DataStoreTraitsT>
+class GenericBTreeBucketDatabase<DataStoreTraitsT>::ReadSnapshot::ConstIteratorImpl
+    : public ConstIterator<typename DataStoreTraitsT::ConstValueRef>
+{
+    const typename GenericBTreeBucketDatabase<DataStoreTraitsT>::ReadSnapshot& _snapshot;
+    typename BTree::ConstIterator _iter;
+public:
+    using ConstValueRef = typename DataStoreTraitsT::ConstValueRef;
+
+    explicit ConstIteratorImpl(const GenericBTreeBucketDatabase<DataStoreTraitsT>::ReadSnapshot& snapshot)
+        : _snapshot(snapshot),
+          _iter(_snapshot._frozen_view.begin())
+    {}
+    ~ConstIteratorImpl() override = default;
+
+    void next() noexcept override {
+        ++_iter;
+    }
+    bool valid() const noexcept override {
+        return _iter.valid();
+    }
+    uint64_t key() const noexcept override {
+        return _iter.getKey();
+    }
+    ConstValueRef value() const override {
+        return ByConstRef::apply(*_snapshot._db, _iter);
+    }
+};
+
+template <typename DataStoreTraitsT>
+std::unique_ptr<ConstIterator<typename DataStoreTraitsT::ConstValueRef>>
+GenericBTreeBucketDatabase<DataStoreTraitsT>::ReadSnapshot::create_iterator() const {
+    return std::make_unique<ConstIteratorImpl>(*this);
+}
+
+template <typename DataStoreTraitsT>
 uint64_t GenericBTreeBucketDatabase<DataStoreTraitsT>::ReadSnapshot::generation() const noexcept {
     return _guard.getGeneration();
 }
diff --git a/storage/src/vespa/storage/bucketdb/read_guard.h b/storage/src/vespa/storage/bucketdb/read_guard.h
index bacb0a38e2f..68a0c8578a8 100644
--- a/storage/src/vespa/storage/bucketdb/read_guard.h
+++ b/storage/src/vespa/storage/bucketdb/read_guard.h
@@ -1,12 +1,15 @@
 // Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 #pragma once
 
+#include "const_iterator.h"
 #include <vespa/document/bucket/bucketid.h>
 #include <functional>
+#include <memory>
 #include <vector>
 
 namespace storage::bucketdb {
 
+
 /*
  * Read guard for accessing the bucket tree of an underlying bucket database
  * in a thread-safe, read-only manner.
@@ -26,7 +29,7 @@ namespace storage::bucketdb {
  * memory to be retained by the backing DB until released.
  */
 
-template <typename ValueT>
+template <typename ValueT, typename ConstRefT = const ValueT&>
 class ReadGuard {
 public:
     ReadGuard() = default;
@@ -40,6 +43,7 @@ public:
     virtual std::vector<ValueT> find_parents_and_self(const document::BucketId& bucket) const = 0;
     virtual std::vector<ValueT> find_parents_self_and_children(const document::BucketId& bucket) const = 0;
     virtual void for_each(std::function<void(uint64_t, const ValueT&)> func) const = 0;
+    virtual std::unique_ptr<ConstIterator<ConstRefT>> create_iterator() const = 0;
     // If the underlying guard represents a snapshot, returns its monotonically
     // increasing generation. Otherwise returns 0.
     [[nodiscard]] virtual uint64_t generation() const noexcept = 0;
diff --git a/storage/src/vespa/storage/bucketdb/storbucketdb.cpp b/storage/src/vespa/storage/bucketdb/storbucketdb.cpp
index b6a5a2320b1..c92436c26ab 100644
--- a/storage/src/vespa/storage/bucketdb/storbucketdb.cpp
+++ b/storage/src/vespa/storage/bucketdb/storbucketdb.cpp
@@ -123,11 +123,9 @@ void StorBucketDatabase::for_each_mutable_unordered(
 
 void StorBucketDatabase::for_each(
         std::function<Decision(uint64_t, const bucketdb::StorageBucketInfo&)> func,
-        const char* clientId,
-        const key_type& first,
-        const key_type& last)
+        const char* clientId)
 {
-    _impl->for_each(std::move(func), clientId, first, last);
+    _impl->for_each(std::move(func), clientId);
 }
 
 std::unique_ptr<bucketdb::ReadGuard<StorBucketDatabase::Entry>>
diff --git a/storage/src/vespa/storage/bucketdb/storbucketdb.h b/storage/src/vespa/storage/bucketdb/storbucketdb.h
index 61d8522ae9d..e7ccac30c37 100644
--- a/storage/src/vespa/storage/bucketdb/storbucketdb.h
+++ b/storage/src/vespa/storage/bucketdb/storbucketdb.h
@@ -64,9 +64,7 @@ public:
                                     const char* clientId);
 
     void for_each(std::function<Decision(uint64_t, const bucketdb::StorageBucketInfo&)> func,
-                  const char* clientId,
-                  const key_type& first = key_type(),
-                  const key_type& last = key_type() - 1);
+                  const char* clientId);
 
     [[nodiscard]] std::unique_ptr<bucketdb::ReadGuard<Entry>> acquire_read_guard() const;
 
diff --git a/storage/src/vespa/storage/bucketdb/striped_btree_lockable_map.cpp b/storage/src/vespa/storage/bucketdb/striped_btree_lockable_map.cpp
new file mode 100644
index 00000000000..90f7ec5ae7b
--- /dev/null
+++ b/storage/src/vespa/storage/bucketdb/striped_btree_lockable_map.cpp
@@ -0,0 +1,8 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#include "striped_btree_lockable_map.hpp"
+
+namespace storage::bucketdb {
+
+template class StripedBTreeLockableMap<StorageBucketInfo>; // Forced instantiation.
+
+}
diff --git a/storage/src/vespa/storage/bucketdb/striped_btree_lockable_map.h b/storage/src/vespa/storage/bucketdb/striped_btree_lockable_map.h
new file mode 100644
index 00000000000..690db6e08d0
--- /dev/null
+++ b/storage/src/vespa/storage/bucketdb/striped_btree_lockable_map.h
@@ -0,0 +1,88 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#pragma once
+
+#include "btree_lockable_map.h"
+#include <memory>
+#include <vector>
+
+namespace storage::bucketdb {
+
+// Bucket database implementation that stripes all superbuckets across
+// a set of disjoint sub-DBs. All locking is handled by the individual
+// sub-DBs, meaning that accessing one does not cause contention for
+// readers/writers accessing another.
+// Ordered iteration is transparently provided by the const for_each method
+// and by read guards.
+template <typename T>
+class StripedBTreeLockableMap final : public AbstractBucketMap<T> {
+public:
+    using ParentType   = AbstractBucketMap<T>;
+    using WrappedEntry = typename ParentType::WrappedEntry;
+    using key_type     = typename ParentType::key_type;
+    using mapped_type  = typename ParentType::mapped_type;
+    using LockId       = typename ParentType::LockId;
+    using EntryMap     = typename ParentType::EntryMap;
+    using Decision     = typename ParentType::Decision;
+    using BucketId     = document::BucketId;
+
+    constexpr static uint16_t MaxStripeBits = 8;
+private:
+    using StripedDBType = BTreeLockableMap<T>;
+    uint16_t _n_stripe_bits;
+    size_t _n_stripes;
+    std::vector<std::unique_ptr<StripedDBType>> _stripes;
+public:
+    explicit StripedBTreeLockableMap(uint16_t n_stripe_bits = 4);
+    ~StripedBTreeLockableMap() override;
+
+    size_t size() const noexcept override;
+    size_t getMemoryUsage() const noexcept override;
+    vespalib::MemoryUsage detailed_memory_usage() const noexcept override;
+    bool empty() const noexcept override;
+
+    WrappedEntry get(const key_type& key, const char* clientId, bool createIfNonExisting) override;
+    WrappedEntry get(const key_type& key, const char* clientId) {
+        return get(key, clientId, false);
+    }
+    bool erase(const key_type& key, const char* clientId, bool has_lock) override;
+    void insert(const key_type& key, const mapped_type& value,
+                const char* client_id, bool has_lock, bool& pre_existed) override;
+
+    bool erase(const key_type& key, const char* client_id) {
+        return erase(key, client_id, false);
+    }
+    void insert(const key_type& key, const mapped_type& value,
+                const char* client_id, bool& pre_existed) {
+        return insert(key, value, client_id, false, pre_existed);
+    }
+    void clear();
+    void print(std::ostream& out, bool verbose, const std::string& indent) const override;
+    EntryMap getContained(const BucketId& bucketId, const char* clientId) override;
+    EntryMap getAll(const BucketId& bucketId, const char* clientId) override;
+    bool isConsistent(const WrappedEntry& entry) override;
+    void showLockClients(vespalib::asciistream & out) const override;
+
+private:
+    class ReadGuardImpl;
+
+    void unlock(const key_type& key) override;
+
+    void do_for_each_mutable_unordered(std::function<Decision(uint64_t, mapped_type&)> func,
+                                       const char* client_id) override;
+
+    void do_for_each(std::function<Decision(uint64_t, const mapped_type&)> func,
+                     const char* client_id) override;
+
+    void do_for_each_chunked(std::function<Decision(uint64_t, const mapped_type&)> func,
+                             const char* client_id,
+                             vespalib::duration yield_time,
+                             uint32_t chunk_size) override;
+
+    std::unique_ptr<ReadGuard<T>> do_acquire_read_guard() const override;
+
+    [[nodiscard]] size_t stripe_of(key_type) const noexcept;
+    [[nodiscard]] StripedDBType& db_for(key_type) noexcept;
+    [[nodiscard]] const StripedDBType& db_for(key_type) const noexcept;
+};
+
+}
diff --git a/storage/src/vespa/storage/bucketdb/striped_btree_lockable_map.hpp b/storage/src/vespa/storage/bucketdb/striped_btree_lockable_map.hpp
new file mode 100644
index 00000000000..ba7d435846c
--- /dev/null
+++ b/storage/src/vespa/storage/bucketdb/striped_btree_lockable_map.hpp
@@ -0,0 +1,315 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#pragma once
+
+#include "striped_btree_lockable_map.h"
+#include "btree_lockable_map.hpp"
+#include <algorithm>
+#include <cassert>
+#include <queue>
+
+namespace storage::bucketdb {
+
+namespace {
+
+constexpr uint8_t used_bits_of(uint64_t key) noexcept {
+    return static_cast<uint8_t>(key & 0b11'1111ULL);
+}
+
+}
+
+// TODO rename to sharded_btree_lockable_map instead?
+
+template <typename T>
+StripedBTreeLockableMap<T>::StripedBTreeLockableMap(uint16_t n_stripe_bits)
+    : _n_stripe_bits(n_stripe_bits),
+      _n_stripes(1ULL << _n_stripe_bits),
+      _stripes()
+{
+    assert(_n_stripe_bits > 0);
+    assert(_n_stripe_bits <= MaxStripeBits);
+    _stripes.reserve(_n_stripes);
+    for (size_t i = 0; i < _n_stripes; ++i) {
+        // TODO reduce initial sub-DB data store memory usage based on number of stripes
+        _stripes.emplace_back(std::make_unique<BTreeLockableMap<T>>());
+    }
+}
+
+template <typename T>
+StripedBTreeLockableMap<T>::~StripedBTreeLockableMap() = default;
+
+template <typename T>
+size_t StripedBTreeLockableMap<T>::stripe_of(key_type key) const noexcept {
+    assert(used_bits_of(key) >= _n_stripe_bits);
+    // Since bucket keys have count-bits at the LSB positions, we want to look at the MSBs instead.
+    return (key >> (64 - _n_stripe_bits));
+}
+
+template <typename T>
+typename StripedBTreeLockableMap<T>::StripedDBType&
+StripedBTreeLockableMap<T>::db_for(key_type key) noexcept {
+    return *_stripes[stripe_of(key)];
+}
+
+template <typename T>
+const typename StripedBTreeLockableMap<T>::StripedDBType&
+StripedBTreeLockableMap<T>::db_for(key_type key) const noexcept {
+    return *_stripes[stripe_of(key)];
+}
+
+template <typename T>
+size_t StripedBTreeLockableMap<T>::size() const noexcept {
+    size_t sz = 0;
+    for (auto& s : _stripes) {
+        sz += s->size();
+    }
+    return sz;
+}
+
+template <typename T>
+size_t StripedBTreeLockableMap<T>::getMemoryUsage() const noexcept {
+    size_t mem_usage = 0;
+    for (auto& s : _stripes) {
+        mem_usage += s->getMemoryUsage();
+    }
+    return mem_usage;
+}
+
+template <typename T>
+vespalib::MemoryUsage StripedBTreeLockableMap<T>::detailed_memory_usage() const noexcept {
+    vespalib::MemoryUsage mem_usage;
+    for (auto& s : _stripes) {
+        mem_usage.merge(s->detailed_memory_usage());
+    }
+    return mem_usage;
+}
+
+template <typename T>
+bool StripedBTreeLockableMap<T>::empty() const noexcept {
+    return std::all_of(_stripes.begin(), _stripes.end(), [](auto& s){ return s->empty(); });
+}
+
+template <typename T>
+typename StripedBTreeLockableMap<T>::WrappedEntry
+StripedBTreeLockableMap<T>::get(const key_type& key, const char* clientId, bool createIfNonExisting) {
+    return db_for(key).get(key, clientId, createIfNonExisting);
+}
+
+template <typename T>
+bool StripedBTreeLockableMap<T>::erase(const key_type& key, const char* client_id, bool has_lock) {
+    return db_for(key).erase(key, client_id, has_lock);
+}
+
+template <typename T>
+void StripedBTreeLockableMap<T>::insert(const key_type& key, const mapped_type& value,
+                                        const char* clientId, bool has_lock, bool& pre_existed)
+{
+    db_for(key).insert(key, value, clientId, has_lock, pre_existed);
+}
+
+template <typename T>
+void StripedBTreeLockableMap<T>::clear() {
+    for (auto& s : _stripes) {
+        s->clear();
+    }
+}
+
+template <typename T>
+void StripedBTreeLockableMap<T>::print(std::ostream& out, bool verbose,
+                                       const std::string& indent) const
+{
+    // TODO more wrapped printing?
+    for (auto& s : _stripes) {
+        s->print(out, verbose, indent);
+    }
+}
+
+template <typename T>
+typename StripedBTreeLockableMap<T>::EntryMap
+StripedBTreeLockableMap<T>::getContained(const BucketId& bucket, const char* clientId) {
+    return db_for(bucket.toKey()).getContained(bucket, clientId);
+}
+
+template <typename T>
+typename StripedBTreeLockableMap<T>::EntryMap
+StripedBTreeLockableMap<T>::getAll(const BucketId& bucket, const char* clientId) {
+    return db_for(bucket.toKey()).getAll(bucket, clientId);
+}
+
+template <typename T>
+bool StripedBTreeLockableMap<T>::isConsistent(const StripedBTreeLockableMap::WrappedEntry& entry) {
+    return db_for(entry.getKey()).isConsistent(entry);
+}
+
+template <typename T>
+void StripedBTreeLockableMap<T>::showLockClients(vespalib::asciistream& out) const {
+    for (auto& s : _stripes) {
+        s->showLockClients(out);
+    }
+}
+
+template <typename T>
+void StripedBTreeLockableMap<T>::do_for_each_mutable_unordered(std::function<Decision(uint64_t, mapped_type&)> func,
+                                                               const char* client_id)
+{
+    // This is by definition unordered in terms of bucket keys
+    for (auto& stripe : _stripes) {
+        // TODO pass functor by ref instead
+        stripe->for_each_mutable_unordered(func, client_id);
+    }
+}
+
+template <typename T>
+void StripedBTreeLockableMap<T>::unlock(const key_type& key) {
+    db_for(key).unlock(key);
+}
+
+template <typename T>
+void StripedBTreeLockableMap<T>::do_for_each(std::function<Decision(uint64_t, const mapped_type&)> func,
+                                             [[maybe_unused]] const char* client_id)
+{
+    auto guard = do_acquire_read_guard();
+    for (auto iter = guard->create_iterator(); iter->valid(); iter->next()) {
+        if (func(iter->key(), iter->value()) != Decision::CONTINUE) {
+            break;
+        }
+    }
+}
+
+template <typename T>
+void StripedBTreeLockableMap<T>::do_for_each_chunked(std::function<Decision(uint64_t, const mapped_type&)> func,
+                                                     const char* client_id,
+                                                     [[maybe_unused]] vespalib::duration yield_time,
+                                                     [[maybe_unused]] uint32_t chunk_size)
+{
+    do_for_each(std::move(func), client_id);
+}
+
+template <typename T>
+class StripedBTreeLockableMap<T>::ReadGuardImpl final
+    : public bucketdb::ReadGuard<T, const T&>
+{
+    const StripedBTreeLockableMap<T>& _db;
+    // There is a 1-1 relationship between DB stripes and guards.
+    // This is essential to be able to choose the correct guard.
+    std::vector<std::unique_ptr<bucketdb::ReadGuard<T, const T&>>> _stripe_guards;
+public:
+    explicit ReadGuardImpl(const StripedBTreeLockableMap<T>& db);
+    ~ReadGuardImpl() override;
+
+    std::vector<T> find_parents_and_self(const document::BucketId& bucket) const override;
+    std::vector<T> find_parents_self_and_children(const document::BucketId& bucket) const override;
+    void for_each(std::function<void(uint64_t, const T&)> func) const override;
+    std::unique_ptr<ConstIterator<const T&>> create_iterator() const override;
+    [[nodiscard]] uint64_t generation() const noexcept override { return 0; /*TODO*/ }
+};
+
+template <typename T>
+StripedBTreeLockableMap<T>::ReadGuardImpl::ReadGuardImpl(const StripedBTreeLockableMap<T>& db)
+    : _db(db),
+      _stripe_guards()
+{
+    _stripe_guards.reserve(_db._stripes.size());
+    for (auto& s : _db._stripes) {
+        _stripe_guards.emplace_back(s->acquire_read_guard());
+    }
+}
+
+template <typename T>
+StripedBTreeLockableMap<T>::ReadGuardImpl::~ReadGuardImpl() = default;
+
+template <typename T>
+std::vector<T>
+StripedBTreeLockableMap<T>::ReadGuardImpl::find_parents_and_self(const document::BucketId& bucket) const {
+    return _stripe_guards[_db.stripe_of(bucket.toKey())]->find_parents_and_self(bucket);
+}
+
+template <typename T>
+std::vector<T>
+StripedBTreeLockableMap<T>::ReadGuardImpl::find_parents_self_and_children(const document::BucketId& bucket) const {
+    return _stripe_guards[_db.stripe_of(bucket.toKey())]->find_parents_self_and_children(bucket);
+}
+
+template <typename T>
+void StripedBTreeLockableMap<T>::ReadGuardImpl::for_each(std::function<void(uint64_t, const T&)> func) const {
+    for (auto iter = create_iterator(); iter->valid(); iter->next()) {
+        func(iter->key(), iter->value());
+    }
+}
+
+template <typename T>
+std::unique_ptr<ReadGuard<T>> StripedBTreeLockableMap<T>::do_acquire_read_guard() const {
+    return std::make_unique<ReadGuardImpl>(*this);
+}
+
+namespace {
+
+template <typename T> using Iter = ConstIterator<const T&>;
+template <typename T> using KeyAndIterPtr = std::pair<uint64_t, Iter<T>*>;
+
+template <typename T>
+struct CompareFirstGreater {
+    bool operator()(const KeyAndIterPtr<T>& lhs, const KeyAndIterPtr<T>& rhs) const noexcept {
+        return (lhs.first > rhs.first);
+    }
+};
+
+template <typename T>
+class ConstIteratorImpl final : public ConstIterator<const T&> {
+    using PriorityQueueType = std::priority_queue<KeyAndIterPtr<T>,
+                                                  std::vector<KeyAndIterPtr<T>>,
+                                                  CompareFirstGreater<T>>;
+    // This is pretty heavy weight, but this iterator is only used for full DB sweeps
+    // used by background maintenance operations, not by any realtime traffic.
+    std::vector<std::unique_ptr<Iter<T>>> _iters;
+    PriorityQueueType                     _iter_queue;
+public:
+    // Precondition: all iterators must be initially valid
+    explicit ConstIteratorImpl(std::vector<std::unique_ptr<Iter<T>>> iters)
+        : _iters(std::move(iters))
+    {
+        for (auto& iter : _iters) {
+            _iter_queue.push(std::make_pair(iter->key(), iter.get()));
+        }
+    }
+
+    void next() noexcept override {
+        assert(!_iter_queue.empty());
+        auto* top_iter = _iter_queue.top().second;
+        top_iter->next();
+        _iter_queue.pop();
+        if (top_iter->valid()) {
+            _iter_queue.push(std::make_pair(top_iter->key(), top_iter));
+        }
+    }
+
+    bool valid() const noexcept override {
+        return !_iter_queue.empty();
+    }
+
+    uint64_t key() const noexcept override {
+        return _iter_queue.top().first;
+    }
+
+    const T& value() const override {
+        return _iter_queue.top().second->value();
+    }
+};
+
+}
+
+template <typename T>
+std::unique_ptr<ConstIterator<const T&>>
+StripedBTreeLockableMap<T>::ReadGuardImpl::create_iterator() const {
+    std::vector<std::unique_ptr<Iter<T>>> iters;
+    iters.reserve(_db._n_stripes);
+    for (auto& g : _stripe_guards) {
+        auto iter = g->create_iterator();
+        if (!iter->valid()) {
+            continue;
+        }
+        iters.emplace_back(std::move(iter));
+    }
+    return std::make_unique<ConstIteratorImpl<T>>(std::move(iters));
+}
+
+}