Resolve distributor bucket space in distributor component methods.

3 files changed, 31 insertions, 33 deletions

diff --git a/storage/src/vespa/storage/distributor/distributorcomponent.cpp b/storage/src/vespa/storage/distributor/distributorcomponent.cpp
index baa9f9916b2..f8a0a5504ec 100644
--- a/storage/src/vespa/storage/distributor/distributorcomponent.cpp
+++ b/storage/src/vespa/storage/distributor/distributorcomponent.cpp
@@ -3,10 +3,13 @@
 #include <vespa/storage/common/bucketoperationlogger.h>
 #include <vespa/storageapi/messageapi/storagereply.h>
 #include <vespa/vdslib/distribution/distribution.h>
+#include <vespa/storage/distributor/distributor_bucket_space_repo.h>
 
 #include <vespa/log/log.h>
 LOG_SETUP(".distributorstoragelink");
 
+using document::BucketSpace;
+
 namespace storage {
 
 namespace distributor {
@@ -45,7 +48,8 @@ DistributorComponent::getClusterState() const
 std::vector<uint16_t>
 DistributorComponent::getIdealNodes(const document::Bucket &bucket) const
 {
-    return getDistribution().getIdealStorageNodes(
+    auto &bucketSpace(_bucketSpaceRepo.get(bucket.getBucketSpace()));
+    return bucketSpace.getDistribution().getIdealStorageNodes(
             getClusterState(),
             bucket.getBucketId(),
             _distributor.getStorageNodeUpStates());
@@ -82,8 +86,9 @@ BucketOwnership
 DistributorComponent::checkOwnershipInPendingAndCurrentState(
         const document::Bucket &bucket) const
 {
+    auto &bucketSpace(_bucketSpaceRepo.get(bucket.getBucketSpace()));
     return checkOwnershipInPendingAndGivenState(
-            getDistribution(), getClusterState(), bucket);
+            bucketSpace.getDistribution(), getClusterState(), bucket);
 }
 
 bool
@@ -112,14 +117,15 @@ DistributorComponent::ownsBucketInState(
         const lib::ClusterState& clusterState,
         const document::Bucket &bucket) const
 {
-    return ownsBucketInState(getDistribution(), clusterState, bucket);
+    auto &bucketSpace(_bucketSpaceRepo.get(bucket.getBucketSpace()));
+    return ownsBucketInState(bucketSpace.getDistribution(), clusterState, bucket);
 }
 
 bool
-DistributorComponent::ownsBucketInCurrentState(
-        const document::Bucket &bucket) const
+DistributorComponent::ownsBucketInCurrentState(const document::Bucket &bucket) const
 {
-    return ownsBucketInState(getDistribution(), getClusterState(), bucket);
+    auto &bucketSpace(_bucketSpaceRepo.get(bucket.getBucketSpace()));
+    return ownsBucketInState(bucketSpace.getDistribution(), getClusterState(), bucket);
 }
 
 api::StorageMessageAddress
@@ -165,7 +171,8 @@ void
 DistributorComponent::removeNodesFromDB(const document::Bucket &bucket,
                                           const std::vector<uint16_t>& nodes)
 {
-    BucketDatabase::Entry dbentry = getBucketDatabase().get(bucket.getBucketId());
+    auto &bucketSpace(_bucketSpaceRepo.get(bucket.getBucketSpace()));
+    BucketDatabase::Entry dbentry = bucketSpace.getBucketDatabase().get(bucket.getBucketId());
 
     if (dbentry.valid()) {
         for (uint32_t i = 0; i < nodes.size(); ++i) {
@@ -179,14 +186,14 @@ DistributorComponent::removeNodesFromDB(const document::Bucket &bucket,
         }
 
         if (dbentry->getNodeCount() != 0) {
-            getBucketDatabase().update(dbentry);
+            bucketSpace.getBucketDatabase().update(dbentry);
         } else {
             LOG(debug,
                 "After update, bucket %s now has no copies. "
                 "Removing from database.",
                 bucket.toString().c_str());
 
-            getBucketDatabase().remove(bucket.getBucketId());
+            bucketSpace.getBucketDatabase().remove(bucket.getBucketId());
         }
     }
 }
@@ -222,8 +229,9 @@ DistributorComponent::updateBucketDatabase(
         const std::vector<BucketCopy>& changedNodes,
         uint32_t updateFlags)
 {
+    auto &bucketSpace(_bucketSpaceRepo.get(bucket.getBucketSpace()));
     assert(!(bucket.getBucketId() == document::BucketId()));
-    BucketDatabase::Entry dbentry = getBucketDatabase().get(bucket.getBucketId());
+    BucketDatabase::Entry dbentry = bucketSpace.getBucketDatabase().get(bucket.getBucketId());
 
     BucketOwnership ownership(checkOwnershipInPendingAndCurrentState(bucket));
     if (!ownership.isOwned()) {
@@ -277,7 +285,7 @@ DistributorComponent::updateBucketDatabase(
     if (updateFlags & DatabaseUpdate::RESET_TRUSTED) {
         dbentry->resetTrusted();
     }
-    getBucketDatabase().update(dbentry);
+    bucketSpace.getBucketDatabase().update(dbentry);
 }
 
 void
@@ -332,19 +340,12 @@ DistributorComponent::getSibling(const document::BucketId& bid) const {
 };
 
 BucketDatabase::Entry
-DistributorComponent::createAppropriateBucket(const document::BucketId& bid)
-{
-    return getBucketDatabase().createAppropriateBucket(
-            _distributor.getConfig().getMinimalBucketSplit(),
-            bid);
-}
-
-document::BucketId
-DistributorComponent::getAppropriateBucket(const document::BucketId& bid)
+DistributorComponent::createAppropriateBucket(const document::Bucket &bucket)
 {
-    return getBucketDatabase().getAppropriateBucket(
+    auto &bucketSpace(_bucketSpaceRepo.get(bucket.getBucketSpace()));
+    return bucketSpace.getBucketDatabase().createAppropriateBucket(
             _distributor.getConfig().getMinimalBucketSplit(),
-            bid);
+            bucket.getBucketId());
 }
 
 bool
diff --git a/storage/src/vespa/storage/distributor/distributorcomponent.h b/storage/src/vespa/storage/distributor/distributorcomponent.h
index b89e7d0a5f2..307ddc20299 100644
--- a/storage/src/vespa/storage/distributor/distributorcomponent.h
+++ b/storage/src/vespa/storage/distributor/distributorcomponent.h
@@ -173,13 +173,10 @@ public:
     document::BucketId getSibling(const document::BucketId& bid) const;
 
     /**
-     * Gets a bucket that is split correctly according to other buckets that
-     * are in the bucket database. For instance, if you have a sibling bucket of
-     * the bucket, a similarly split bucket should be created.
+     * Create a bucket that is split correctly according to other buckets that
+     * are in the bucket database.
      */
-    document::BucketId getAppropriateBucket(const document::BucketId& bid);
-
-    BucketDatabase::Entry createAppropriateBucket(const document::BucketId& bid);
+    BucketDatabase::Entry createAppropriateBucket(const document::Bucket &bucket);
 
     /**
      * Returns true if the node is currently initializing.
diff --git a/storage/src/vespa/storage/distributor/operations/external/multioperationoperation.cpp b/storage/src/vespa/storage/distributor/operations/external/multioperationoperation.cpp
index 19c693e2a7f..404cb286ecf 100644
--- a/storage/src/vespa/storage/distributor/operations/external/multioperationoperation.cpp
+++ b/storage/src/vespa/storage/distributor/operations/external/multioperationoperation.cpp
@@ -143,18 +143,18 @@ MultiOperationOperation::onStart(DistributorMessageSender& sender)
     {
         if (operationIt->valid()) {
             document::DocumentId docId = operationIt->getDocumentId();
-            document::BucketId bucketId(
-                    _manager.getBucketIdFactory().getBucketId(docId));
+            document::Bucket bucket(_msg->getBucket().getBucketSpace(),
+                                    _manager.getBucketIdFactory().getBucketId(docId));
 
-            LOG(debug, "Operation with documentid %s mapped to bucketid %s", docId.toString().c_str(), bucketId.toString().c_str());
+            LOG(debug, "Operation with documentid %s mapped to bucket %s", docId.toString().c_str(), bucket.toString().c_str());
 
             // OK, we have a bucket ID, must now know which buckets this belongs
             // to
             std::vector<BucketDatabase::Entry> entries;
-            _manager.getBucketDatabase().getParents(bucketId, entries);
+            _manager.getBucketDatabase().getParents(bucket.getBucketId(), entries);
 
             if (entries.empty()) {
-                entries.push_back(_manager.createAppropriateBucket(bucketId));
+                entries.push_back(_manager.createAppropriateBucket(bucket));
             }
 
             for (uint32_t i = 0; i < entries.size(); ++i) {