Merge pull request #4222 from vespa-engine/vekterli/use-content-bucket-space-for-bucket-mover-runs

Use content bucket space for bucket mover runs

11 files changed, 99 insertions, 128 deletions

diff --git a/storage/src/vespa/storage/bucketmover/bucketmover.cpp b/storage/src/vespa/storage/bucketmover/bucketmover.cpp
index b9073ca2cdc..bc9a8b0c428 100644
--- a/storage/src/vespa/storage/bucketmover/bucketmover.cpp
+++ b/storage/src/vespa/storage/bucketmover/bucketmover.cpp
@@ -4,6 +4,7 @@
 #include "htmltable.h"
 #include <vespa/storage/config/config-stor-server.h>
 #include <vespa/storage/common/bucketmessages.h>
+#include <vespa/storage/common/content_bucket_space_repo.h>
 #include <vespa/storage/common/nodestateupdater.h>
 #include <vespa/storage/storageutil/log.h>
 #include <vespa/config/common/exceptions.h>
@@ -27,7 +28,7 @@ BucketMover::BucketMover(const config::ConfigUri & configUri,
       _cycleCount(0),
       _nextRun(0),
       _configFetcher(configUri.getContext()),
-      _diskDistribution(_component.getDistribution()->getDiskDistribution()),
+      _diskDistribution(currentDiskDistribution()),
       _maxSleepTime(60 * 60)
 {
     if (!configUri.empty()) {
@@ -40,7 +41,7 @@ BucketMover::BucketMover(const config::ConfigUri & configUri,
 
 BucketMover::~BucketMover()
 {
-    if (_thread.get() != 0) {
+    if (_thread) {
         LOG(error, "BucketMover deleted without calling close() first");
        onClose();
     }
@@ -61,10 +62,10 @@ BucketMover::onClose()
     // Avoid getting config during shutdown
     _configFetcher.close();
     // Close thread to ensure we don't send anything more down after
-    if (_thread.get()) {
+    if (_thread) {
         _thread->interruptAndJoin(&_wait);
         LOG(debug, "Bucket mover worker thread closed.");
-        _thread.reset(0);
+        _thread.reset();
     }
 }
 
@@ -111,12 +112,14 @@ BucketMover::startNewRun()
     // If not in a run but time to start another one, do so
     LOG(debug, "Starting new move cycle at time %s.",
         _component.getClock().getTimeInSeconds().toString().c_str());
-    _currentRun.reset(new bucketmover::Run(
-            _component.getBucketDatabase(BucketSpace::placeHolder()),
-            _component.getDistribution(),
+    // TODO consider if we should invoke bucket moving across all bucket spaces. Not likely to ever be needed.
+    // If so, we have to spawn off an individual Run per space, as it encompasses
+    // both a (disk) distribution and a bucket database.
+    _currentRun = std::make_unique<bucketmover::Run>(
+            _component.getBucketSpaceRepo().get(document::BucketSpace::placeHolder()),
             *_component.getStateUpdater().getReportedNodeState(),
             _component.getIndex(),
-            _component.getClock()));
+            _component.getClock());
 }
 
 void
@@ -124,8 +127,7 @@ BucketMover::queueNewMoves()
 {
     // If we have too few pending, send some new moves, if there are more
     // moves to perform.
-    while (_pendingMoves.size() < uint32_t(_config->maxPending))
-    {
+    while (_pendingMoves.size() < uint32_t(_config->maxPending)) {
         Move nextMove = _currentRun->getNextMove();
 
         // If no more moves to do, stop attempting to send more.
@@ -133,11 +135,8 @@ BucketMover::queueNewMoves()
             break;
         }
         _pendingMoves.push_back(nextMove);
-        document::Bucket bucket(BucketSpace::placeHolder(), nextMove.getBucketId());
-        std::shared_ptr<BucketDiskMoveCommand> cmd(
-                new BucketDiskMoveCommand(bucket,
-                                          nextMove.getSourceDisk(),
-                                          nextMove.getTargetDisk()));
+        auto cmd = std::make_shared<BucketDiskMoveCommand>(
+                nextMove.getBucket(), nextMove.getSourceDisk(), nextMove.getTargetDisk());
         cmd->setPriority(nextMove.getPriority());
         _newMoves.push_back(cmd);
     }
@@ -171,11 +170,9 @@ BucketMover::finishCurrentRun()
 void
 BucketMover::sendNewMoves()
 {
-    for (std::list<BucketDiskMoveCommand::SP>::iterator it
-             = _newMoves.begin(); it != _newMoves.end(); ++it)
-    {
-        LOG(debug, "Moving bucket: %s", (**it).toString().c_str());
-        sendDown(*it);
+    for (auto& move : _newMoves) {
+        LOG(debug, "Moving bucket: %s", move->toString().c_str());
+        sendDown(move);
 
         // Be able to sleep a bit between moves for debugging to see
         // what is happening. (Cannot use wait() here as reply of
@@ -196,7 +193,7 @@ BucketMover::tick()
 
         framework::SecondTime currentTime(_component.getClock().getTimeInSeconds());
 
-        if (_currentRun.get() == 0) {
+        if (!_currentRun) {
             if (currentTime >= _nextRun) {
                 startNewRun();
             } else {
@@ -287,25 +284,24 @@ bool
 BucketMover::onInternalReply(
         const std::shared_ptr<api::InternalReply>& internalReply)
 {
-        // We only care about move disk bucket replies
-    std::shared_ptr<BucketDiskMoveReply> reply(
-            std::dynamic_pointer_cast<BucketDiskMoveReply>(internalReply));
-    if (!reply.get()) return false;
+    // We only care about move disk bucket replies
+    auto reply = std::dynamic_pointer_cast<BucketDiskMoveReply>(internalReply);
+    if (!reply) {
+        return false;
+    }
 
-        // Warn if we see move replies outside of a run. Should not be possible.
+    // Warn if we see move replies outside of a run. Should not be possible.
     vespalib::MonitorGuard monitor(_wait);
-    if (_currentRun.get() == 0) {
+    if (!_currentRun) {
         LOG(warning, "Got a bucket disk move reply while no run is active. "
                      "This should not happen, as runs should stay active until "
                      "all requests are answered.");
         return true;
     }
-        // Match move against pending ones
+    // Match move against pending ones
     Move move;
-    for (std::list<Move>::iterator it = _pendingMoves.begin();
-         it != _pendingMoves.end(); ++it)
-    {
-        if (it->getBucketId() == reply->getBucketId()
+    for (auto it = _pendingMoves.begin(); it != _pendingMoves.end(); ++it) {
+        if (it->getBucket() == reply->getBucket()
             && it->getSourceDisk() == reply->getSrcDisk()
             && it->getTargetDisk() == reply->getDstDisk())
         {
@@ -338,18 +334,20 @@ BucketMover::onInternalReply(
     return true;
 }
 
+// TODO if we start supporting disk moves for other spaces than the default space
+// we also have to check all disk distributions here.
 void
 BucketMover::storageDistributionChanged()
 {
-    lib::Distribution::SP distribution = _component.getDistribution();
-
-        // Verify that the actual disk distribution changed, if not ignore
-    lib::Distribution::DiskDistribution newDistr(distribution->getDiskDistribution());
+    // Verify that the actual disk distribution changed, if not ignore
+    lib::Distribution::DiskDistribution newDistr(currentDiskDistribution());
 
-    if (_diskDistribution == newDistr) return;
+    if (_diskDistribution == newDistr) {
+        return;
+    }
 
     vespalib::MonitorGuard monitor(_wait);
-    if (_currentRun.get() != 0) {
+    if (_currentRun) {
         LOG(info, "Aborting bucket mover run as disk distribution changed "
                   "from %s to %s.",
             lib::Distribution::getDiskDistributionName(_diskDistribution).c_str(),
@@ -365,9 +363,14 @@ BucketMover::storageDistributionChanged()
     _nextRun = framework::SecondTime(0);
 }
 
+lib::Distribution::DiskDistribution BucketMover::currentDiskDistribution() const {
+    auto distribution = _component.getBucketSpaceRepo().get(document::BucketSpace::placeHolder()).getDistribution();
+    return distribution->getDiskDistribution();
+}
+
 bool BucketMover::isWorkingOnCycle() const {
     vespalib::MonitorGuard monitor(_wait);
-    return (_currentRun.get() != 0);
+    return (_currentRun.get() != nullptr);
 }
 
 uint32_t BucketMover::getCycleCount() const {
@@ -382,7 +385,7 @@ BucketMover::print(std::ostream& out, bool verbose,
     (void) verbose; (void) indent;
     vespalib::MonitorGuard monitor(_wait);
     out << "BucketMover() {";
-    if (_currentRun.get() != 0) {
+    if (_currentRun) {
         out << "\n" << indent << "  ";
         _currentRun->print(out, verbose, indent + "  ");
     } else {
@@ -390,11 +393,9 @@ BucketMover::print(std::ostream& out, bool verbose,
     }
     if (verbose && !_history.empty()) {
         out << "\n" << indent << "  History:";
-        for (std::list<RunStatistics>::const_iterator it = _history.begin();
-             it != _history.end(); ++it)
-        {
+        for (auto& entry : _history) {
             out << "\n" << indent << "    ";
-            it->print(out, true, indent + "    ");
+            entry.print(out, true, indent + "    ");
         }
     }
     out << "\n" << indent << "}";
@@ -412,17 +413,14 @@ BucketMover::reportHtmlStatus(std::ostream& out,
         printCurrentStatus(out, *_history.begin());
     }
     out << "<h2>Current move cycle</h2>\n";
-    if (_currentRun.get() != 0) {
+    if (_currentRun) {
         printRunHtml(out, *_currentRun);
         if (_currentRun->getPendingMoves().empty()) {
             out << "<blockquote>No pending moves.</blockquote>\n";
         } else {
             out << "<blockquote>Pending bucket moves:<ul>\n";
-            for (std::list<Move>::const_iterator it
-                    = _currentRun->getPendingMoves().begin();
-                 it != _currentRun->getPendingMoves().end(); ++it)
-            {
-                out << "<li>" << *it << "</li>\n";
+            for (auto& entry : _currentRun->getPendingMoves()) {
+                out << "<li>" << entry << "</li>\n";
             }
             out << "</ul></blockquote>\n";
         }
@@ -432,7 +430,7 @@ BucketMover::reportHtmlStatus(std::ostream& out,
         framework::SecondTime currentTime(
                 _component.getClock().getTimeInSeconds());
         if (_nextRun <= currentTime) {
-            if (_thread.get() != 0) {
+            if (_thread) {
                 out << "Next run to start immediately.";
                     // Wake up thread, so user sees it starts immediately :)
                 monitor.signal();
@@ -454,10 +452,8 @@ BucketMover::reportHtmlStatus(std::ostream& out,
     }
     if (!_history.empty()) {
         out << "<h2>Statistics from previous bucket mover cycles</h2>\n";
-        for (std::list<RunStatistics>::const_iterator it = _history.begin();
-             it != _history.end(); ++it)
-        {
-            printRunStatisticsHtml(out, *it);
+        for (auto& entry : _history) {
+            printRunStatisticsHtml(out, entry);
         }
     }
 }
diff --git a/storage/src/vespa/storage/bucketmover/bucketmover.h b/storage/src/vespa/storage/bucketmover/bucketmover.h
index 0b9cfc04455..42cbf693fcb 100644
--- a/storage/src/vespa/storage/bucketmover/bucketmover.h
+++ b/storage/src/vespa/storage/bucketmover/bucketmover.h
@@ -83,6 +83,7 @@ private:
     void run(framework::ThreadHandle&) override;
     bool onInternalReply(const std::shared_ptr<api::InternalReply>&) override;
     void storageDistributionChanged() override;
+    lib::Distribution::DiskDistribution currentDiskDistribution() const;
 
     framework::SecondTime calculateWaitTimeOfNextRun() const;
 
diff --git a/storage/src/vespa/storage/bucketmover/move.cpp b/storage/src/vespa/storage/bucketmover/move.cpp
index 7d6f96c46ba..b91329fad36 100644
--- a/storage/src/vespa/storage/bucketmover/move.cpp
+++ b/storage/src/vespa/storage/bucketmover/move.cpp
@@ -9,13 +9,13 @@ namespace bucketmover {
 Move::Move()
     : _sourceDisk(0),
       _targetDisk(0),
-      _bucket(0),
+      _bucket(document::BucketSpace::placeHolder(), document::BucketId(0)),
       _totalDocSize(0),
       _priority(255)
 {
 }
 
-Move::Move(uint16_t source, uint16_t target, const document::BucketId& bucket,
+Move::Move(uint16_t source, uint16_t target, const document::Bucket& bucket,
            uint32_t totalDocSize)
     : _sourceDisk(source),
       _targetDisk(target),
diff --git a/storage/src/vespa/storage/bucketmover/move.h b/storage/src/vespa/storage/bucketmover/move.h
index e2401e5abed..92d05e4e0ae 100644
--- a/storage/src/vespa/storage/bucketmover/move.h
+++ b/storage/src/vespa/storage/bucketmover/move.h
@@ -8,7 +8,7 @@
 
 #pragma once
 
-#include <vespa/document/bucket/bucketid.h>
+#include <vespa/document/bucket/bucket.h>
 #include <vespa/vespalib/util/printable.h>
 
 namespace storage {
@@ -17,22 +17,22 @@ namespace bucketmover {
 class Move : public vespalib::Printable {
     uint16_t _sourceDisk;
     uint16_t _targetDisk;
-    document::BucketId _bucket;
+    document::Bucket _bucket;
     uint32_t _totalDocSize;
     uint8_t _priority;
 
 public:
     Move();
-    Move(uint16_t source, uint16_t target, const document::BucketId& bucket,
+    Move(uint16_t source, uint16_t target, const document::Bucket& bucket,
          uint32_t totalDocSize);
 
     /** False if invalid move. (Empty constructor) Indicates end of run. */
-    bool isDefined() const { return (_bucket.getRawId() != 0); }
+    bool isDefined() const { return (_bucket.getBucketId().getRawId() != 0); }
 
     // Only valid to call if move is defined
     uint16_t getSourceDisk() const { return _sourceDisk; }
     uint16_t getTargetDisk() const { return _targetDisk; }
-    const document::BucketId& getBucketId() const { return _bucket; }
+    const document::Bucket& getBucket() const { return _bucket; }
     uint8_t getPriority() const { return _priority; }
     uint32_t getTotalDocSize() const { return _totalDocSize; }
 
diff --git a/storage/src/vespa/storage/bucketmover/run.cpp b/storage/src/vespa/storage/bucketmover/run.cpp
index 6a0ef2079ce..22bcfa55f15 100644
--- a/storage/src/vespa/storage/bucketmover/run.cpp
+++ b/storage/src/vespa/storage/bucketmover/run.cpp
@@ -12,24 +12,24 @@ LOG_SETUP(".bucketmover.run");
 namespace storage {
 namespace bucketmover {
 
-Run::Run(StorBucketDatabase& db,
-         lib::Distribution::SP distribution,
+Run::Run(ContentBucketSpace& bucketSpace,
          const lib::NodeState& nodeState,
          uint16_t nodeIndex,
          framework::Clock& clock)
-    : _bucketDatabase(db),
-      _distribution(distribution),
+    : _bucketSpace(bucketSpace),
+      _distribution(bucketSpace.getDistribution()),
       _nodeState(nodeState),
       _nodeIndex(nodeIndex),
       _entries(),
       _iterationDone(false),
-      _statistics(distribution->getDiskDistribution(), clock, nodeState),
+      _statistics(_distribution->getDiskDistribution(), clock, nodeState),
       _aborted(false)
 {
 }
 
 namespace {
     struct BucketIterator {
+        document::BucketSpace _iteratedBucketSpace;
         const lib::Distribution& _distribution;
         const lib::NodeState& _nodeState;
         RunStatistics& _statistics;
@@ -39,10 +39,12 @@ namespace {
         uint32_t _bucketsVisited;
         document::BucketId _firstBucket;
 
-        BucketIterator(const lib::Distribution& d, const lib::NodeState& ns,
+        BucketIterator(document::BucketSpace iteratedBucketSpace,
+                       const lib::Distribution& d, const lib::NodeState& ns,
                        uint16_t nodeIndex, RunStatistics& stats,
                        std::list<Move>& entries)
-            : _distribution(d),
+            : _iteratedBucketSpace(iteratedBucketSpace),
+              _distribution(d),
               _nodeState(ns),
               _statistics(stats),
               _entries(entries),
@@ -57,12 +59,12 @@ namespace {
         operator()(document::BucketId::Type revId,
                    StorBucketDatabase::Entry& entry)
         {
-            document::BucketId bucket(document::BucketId::keyToBucketId(revId));
-            if (bucket == _firstBucket) {
+            document::BucketId bucketId(document::BucketId::keyToBucketId(revId));
+            if (bucketId == _firstBucket) {
                 return StorBucketDatabase::CONTINUE;
             }
             uint16_t idealDisk = _distribution.getIdealDisk(
-                    _nodeState, _nodeIndex, bucket,
+                    _nodeState, _nodeIndex, bucketId,
                     lib::Distribution::IDEAL_DISK_EVEN_IF_DOWN);
             RunStatistics::DiskData& diskData(
                     _statistics._diskData[entry.disk]);
@@ -72,10 +74,11 @@ namespace {
                 diskData._bucketSize += entry.getBucketInfo().getTotalDocumentSize();
                 ++diskData._bucketsFoundOnCorrectDisk;
             } else {
+                document::Bucket bucket(_iteratedBucketSpace, bucketId);
                 _entries.push_back(Move(
                             entry.disk, idealDisk, bucket, entry.getBucketInfo().getTotalDocumentSize()));
             }
-            _statistics._lastBucketVisited = bucket;
+            _statistics._lastBucketVisited = bucketId;
             if (++_bucketsVisited >= _maxBucketsToIterateAtOnce) {
                 return StorBucketDatabase::ABORT;
             }
@@ -104,18 +107,16 @@ Run::getNextMove()
 
             if (!_statistics._diskData[e.getTargetDisk()]._diskDisabled) {
                 _pending.push_back(e);
-                _statistics._lastBucketProcessed = e.getBucketId();
-                _statistics._lastBucketProcessedTime
-                    = _statistics._clock->getTimeInSeconds();
+                _statistics._lastBucketProcessed = e.getBucket(); // Only used for printing
+                _statistics._lastBucketProcessedTime = _statistics._clock->getTimeInSeconds();
                 return e;
             }
         }
 
         // Cache more entries
-        BucketIterator it(*_distribution, _nodeState, _nodeIndex, _statistics,
-                          _entries);
-        _bucketDatabase.all(it, "bucketmover::Run",
-                            _statistics._lastBucketVisited.toKey());
+        BucketIterator it(_bucketSpace.bucketSpace(), *_distribution,
+                          _nodeState, _nodeIndex, _statistics, _entries);
+        _bucketSpace.bucketDatabase().all(it, "bucketmover::Run", _statistics._lastBucketVisited.toKey());
         if (it._bucketsVisited == 0) {
             _iterationDone = true;
             if (_pending.empty()) {
@@ -128,31 +129,6 @@ Run::getNextMove()
 }
 
 void
-Run::depleteMoves()
-{
-    while (true) {
-            // Cache more entries
-        BucketIterator bi(*_distribution, _nodeState, _nodeIndex, _statistics,
-                          _entries);
-        _bucketDatabase.all(bi, "bucketmover::depleteMoves",
-                            _statistics._lastBucketVisited.toKey());
-        if (bi._bucketsVisited == 0) {
-            break;
-        }
-        for (std::list<Move>::const_iterator it = _entries.begin();
-             it != _entries.end(); ++it)
-        {
-            ++_statistics._diskData[it->getSourceDisk()][it->getTargetDisk()]
-                ._bucketsLeftOnWrongDisk;
-            uint32_t size = it->getTotalDocSize();
-            _statistics._diskData[it->getSourceDisk()]._bucketSize += size;
-        }
-        _entries.clear();
-    }
-    finalize();
-}
-
-void
 Run::finalize()
 {
     _statistics._endTime = _statistics._clock->getTimeInSeconds();
@@ -162,10 +138,8 @@ void
 Run::removePending(Move& move)
 {
     bool foundPending = false;
-    for (std::list<Move>::iterator it = _pending.begin(); it != _pending.end();
-         ++it)
-    {
-        if (it->getBucketId() == move.getBucketId()) {
+    for (auto it = _pending.begin(); it != _pending.end(); ++it) {
+        if (it->getBucket() == move.getBucket()) {
             _pending.erase(it);
             foundPending = true;
             break;
@@ -173,7 +147,7 @@ Run::removePending(Move& move)
     }
     if (!foundPending) {
         LOG(warning, "Got answer for %s that was not in the pending list.",
-            move.getBucketId().toString().c_str());
+            move.getBucket().toString().c_str());
         return;
     }
     if (_iterationDone && _pending.empty()) {
diff --git a/storage/src/vespa/storage/bucketmover/run.h b/storage/src/vespa/storage/bucketmover/run.h
index 11f2cf0763c..eb7a6df2d17 100644
--- a/storage/src/vespa/storage/bucketmover/run.h
+++ b/storage/src/vespa/storage/bucketmover/run.h
@@ -18,6 +18,7 @@
 
 #include "move.h"
 #include "runstatistics.h"
+#include <vespa/storage/common/content_bucket_space.h>
 #include <vespa/vdslib/distribution/distribution.h>
 #include <vespa/vdslib/state/nodestate.h>
 #include <list>
@@ -33,8 +34,8 @@ class Clock;
 namespace bucketmover {
 
 class Run : public document::Printable {
-    StorBucketDatabase& _bucketDatabase;
-    lib::Distribution::SP _distribution;
+    ContentBucketSpace& _bucketSpace;
+    std::shared_ptr<const lib::Distribution> _distribution;
     lib::NodeState _nodeState;
     uint16_t _nodeIndex;
     uint32_t _maxEntriesToKeep;
@@ -48,8 +49,7 @@ class Run : public document::Printable {
 public:
     Run(const Run &) = delete;
     Run & operator = (const Run &) = delete;
-    Run(StorBucketDatabase&,
-        lib::Distribution::SP,
+    Run(ContentBucketSpace& bucketSpace,
         const lib::NodeState&,
         uint16_t nodeIndex,
         framework::Clock&);
@@ -78,12 +78,6 @@ public:
      */
     Move getNextMove();
 
-    /**
-     * Run through the database not doing any moves. Useful to do a run only
-     * to gather statistics of current state.
-     */
-    void depleteMoves();
-
     void moveOk(Move& move);
     void moveFailedBucketNotFound(Move& move);
     void moveFailed(Move& move);
diff --git a/storage/src/vespa/storage/bucketmover/runstatistics.cpp b/storage/src/vespa/storage/bucketmover/runstatistics.cpp
index 8f7fe67fcf3..314f04a0d66 100644
--- a/storage/src/vespa/storage/bucketmover/runstatistics.cpp
+++ b/storage/src/vespa/storage/bucketmover/runstatistics.cpp
@@ -39,7 +39,7 @@ RunStatistics::RunStatistics(DiskDistribution d, framework::Clock& clock,
                              const lib::NodeState& ns)
     : _clock(&clock),
       _distribution(d),
-      _lastBucketProcessed(0),
+      _lastBucketProcessed(),
       _lastBucketVisited(0),
       _diskData(ns.getDiskCount(), DiskData(ns.getDiskCount())),
       _startTime(_clock->getTimeInSeconds()),
@@ -149,13 +149,14 @@ RunStatistics::getWronglyPlacedRatio() const
     return static_cast<double>(wrong) / total;
 }
 
+// FIXME does not cover multiple spaces (but only used for printing)
 double
 RunStatistics::getProgress() const
 {
     if (_endTime.isSet()) return 1.0;
     double result = 0;
     double weight = 0.5;
-    uint64_t key = _lastBucketProcessed.toKey();
+    uint64_t key = _lastBucketProcessed.getBucketId().toKey();
     for (uint16_t i=0; i<64; ++i) {
         uint64_t flag = uint64_t(1) << (63 - i);
         if ((key & flag) == flag) {
diff --git a/storage/src/vespa/storage/bucketmover/runstatistics.h b/storage/src/vespa/storage/bucketmover/runstatistics.h
index da51be9ef7d..908f345b307 100644
--- a/storage/src/vespa/storage/bucketmover/runstatistics.h
+++ b/storage/src/vespa/storage/bucketmover/runstatistics.h
@@ -38,7 +38,7 @@
 
 #include <vespa/vdslib/state/nodestate.h>
 #include <vespa/vdslib/distribution/distribution.h>
-#include <vespa/document/bucket/bucketid.h>
+#include <vespa/document/bucket/bucket.h>
 #include <vespa/vespalib/util/printable.h>
 #include <vespa/storageframework/generic/clock/time.h>
 
@@ -76,7 +76,7 @@ struct RunStatistics : public document::Printable {
 
     framework::Clock* _clock;
     DiskDistribution _distribution;
-    document::BucketId _lastBucketProcessed;
+    document::Bucket _lastBucketProcessed;
     document::BucketId _lastBucketVisited; // Invalid bucket for starting point
     std::vector<DiskData> _diskData;
     framework::SecondTime _startTime;
diff --git a/storage/src/vespa/storage/common/content_bucket_space.cpp b/storage/src/vespa/storage/common/content_bucket_space.cpp
index 0f84a2d0e38..4344bccc785 100644
--- a/storage/src/vespa/storage/common/content_bucket_space.cpp
+++ b/storage/src/vespa/storage/common/content_bucket_space.cpp
@@ -4,8 +4,9 @@
 
 namespace storage {
 
-ContentBucketSpace::ContentBucketSpace()
-    : _bucketDatabase(),
+ContentBucketSpace::ContentBucketSpace(document::BucketSpace bucketSpace)
+    : _bucketSpace(bucketSpace),
+      _bucketDatabase(),
       _lock(),
       _distribution()
 {
diff --git a/storage/src/vespa/storage/common/content_bucket_space.h b/storage/src/vespa/storage/common/content_bucket_space.h
index 65524121ca6..3b3dddade4f 100644
--- a/storage/src/vespa/storage/common/content_bucket_space.h
+++ b/storage/src/vespa/storage/common/content_bucket_space.h
@@ -1,6 +1,7 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 #pragma once
 
+#include <vespa/document/bucket/bucketspace.h>
 #include <vespa/storage/bucketdb/storbucketdb.h>
 #include <mutex>
 
@@ -13,13 +14,16 @@ namespace lib { class Distribution; }
  */
 class ContentBucketSpace {
 private:
+    document::BucketSpace _bucketSpace;
     StorBucketDatabase _bucketDatabase;
     mutable std::mutex _lock;
     std::shared_ptr<const lib::Distribution> _distribution;
 
 public:
     using UP = std::unique_ptr<ContentBucketSpace>;
-    ContentBucketSpace();
+    ContentBucketSpace(document::BucketSpace bucketSpace);
+
+    document::BucketSpace bucketSpace() const noexcept { return _bucketSpace; }
     StorBucketDatabase &bucketDatabase() { return _bucketDatabase; }
     void setDistribution(std::shared_ptr<const lib::Distribution> distribution);
     std::shared_ptr<const lib::Distribution> getDistribution() const;
diff --git a/storage/src/vespa/storage/common/content_bucket_space_repo.cpp b/storage/src/vespa/storage/common/content_bucket_space_repo.cpp
index 3761b6fc20d..1846c132c0a 100644
--- a/storage/src/vespa/storage/common/content_bucket_space_repo.cpp
+++ b/storage/src/vespa/storage/common/content_bucket_space_repo.cpp
@@ -9,7 +9,7 @@ namespace storage {
 ContentBucketSpaceRepo::ContentBucketSpaceRepo()
     : _map()
 {
-    _map.emplace(BucketSpace::placeHolder(), std::make_unique<ContentBucketSpace>());
+    _map.emplace(BucketSpace::placeHolder(), std::make_unique<ContentBucketSpace>(BucketSpace::placeHolder()));
 }
 
 ContentBucketSpace &