diff options
Diffstat (limited to 'src/mongo/s/chunk_manager.cpp')
| -rw-r--r-- | src/mongo/s/chunk_manager.cpp | 695 |
1 files changed, 195 insertions, 500 deletions
diff --git a/src/mongo/s/chunk_manager.cpp b/src/mongo/s/chunk_manager.cpp index 1d3ed85e925..888ba644b52 100644 --- a/src/mongo/s/chunk_manager.cpp +++ b/src/mongo/s/chunk_manager.cpp @@ -33,8 +33,6 @@ #include "mongo/s/chunk_manager.h" -#include <algorithm> - #include "mongo/bson/simple_bsonobj_comparator.h" #include "mongo/db/matcher/extensions_callback_noop.h" #include "mongo/db/query/collation/collation_index_key.h" @@ -65,42 +63,18 @@ void checkAllElementsAreOfType(BSONType type, const BSONObj& o) { allElementsAreOfType(type, o)); } -bool overlaps(const ChunkInfo& a, const ChunkInfo& b) { - // Microbenchmarks results showed that comparing keystrings - // is more performant than comparing BSONObj - const auto aMinKeyStr = ShardKeyPattern::toKeyString(a.getMin()); - const auto& aMaxKeyStr = a.getMaxKeyString(); - const auto bMinKeyStr = ShardKeyPattern::toKeyString(b.getMin()); - const auto& bMaxKeyStr = b.getMaxKeyString(); - - return aMinKeyStr < bMaxKeyStr && aMaxKeyStr > bMinKeyStr; -} - -void checkChunksAreContiguous(const ChunkInfo& left, const ChunkInfo& right) { - const auto& leftKeyString = left.getMaxKeyString(); - const auto rightKeyString = ShardKeyPattern::toKeyString(right.getMin()); - if (leftKeyString == rightKeyString) { - return; - } - - if (SimpleBSONObjComparator::kInstance.evaluate(left.getMax() < right.getMin())) { - uasserted(ErrorCodes::ConflictingOperationInProgress, - str::stream() << "Gap exists in the routing table between chunks " - << left.getRange().toString() << " and " - << right.getRange().toString()); +void appendChunkTo(std::vector<std::shared_ptr<ChunkInfo>>& chunks, + const std::shared_ptr<ChunkInfo>& chunk) { + if (!chunks.empty() && chunk->getRange().overlaps(chunks.back()->getRange())) { + if (chunks.back()->getLastmod().isOlderThan(chunk->getLastmod())) { + chunks.pop_back(); + chunks.push_back(chunk); + } } else { - uasserted(ErrorCodes::ConflictingOperationInProgress, - str::stream() << "Overlap exists in the routing table between chunks " - << left.getRange().toString() << " and " - << right.getRange().toString()); + chunks.push_back(chunk); } - - MONGO_UNREACHABLE; } -using ChunkVector = ChunkMap::ChunkVector; -using ChunkVectorMap = ChunkMap::ChunkVectorMap; - // This function processes the passed in chunks by removing the older versions of any overlapping // chunks. The resulting chunks must be ordered by the maximum bound and not have any // overlapping chunks. In order to process the original set of chunks correctly which may have @@ -125,18 +99,10 @@ std::vector<std::shared_ptr<ChunkInfo>> flatten(const std::vector<ChunkType>& ch std::vector<std::shared_ptr<ChunkInfo>> flattened; flattened.reserve(changedChunkInfos.size()); - flattened.emplace_back(std::move(changedChunkInfos[0])); + flattened.push_back(changedChunkInfos[0]); for (size_t i = 1; i < changedChunkInfos.size(); ++i) { - auto& chunk = changedChunkInfos[i]; - if (overlaps(*chunk, *flattened.back())) { - if (flattened.back()->getLastmod().isOlderThan(chunk->getLastmod())) { - flattened.pop_back(); - flattened.emplace_back(std::move(chunk)); - } - } else { - flattened.emplace_back(std::move(chunk)); - } + appendChunkTo(flattened, changedChunkInfos[i]); } std::reverse(flattened.begin(), flattened.end()); @@ -144,489 +110,205 @@ std::vector<std::shared_ptr<ChunkInfo>> flatten(const std::vector<ChunkType>& ch return flattened; } -} // namespace - -size_t ChunkMap::size() const { - size_t totalChunks{0}; - for (const auto& mapIt : _chunkVectorMap) { - totalChunks += mapIt.second->size(); - } - return totalChunks; -} - -std::shared_ptr<ChunkInfo> ChunkMap::findIntersectingChunk(const BSONObj& shardKey) const { - const auto shardKeyString = ShardKeyPattern::toKeyString(shardKey); - - const auto it = _chunkVectorMap.upper_bound(shardKeyString); - if (it == _chunkVectorMap.end()) { - // upper_bound() will miss the last chunkVector if shardKey is actually the MaxKey, - // thus we need to check explicitly if shardKey is contained in the last chunk. - if (const auto& lastChunk = std::prev(_chunkVectorMap.end())->second->back(); - lastChunk->containsKey(shardKey)) { - return lastChunk; - } else { - return {}; - } - } - - const auto& chunkVector = *(it->second); - const auto chunkIt = _findIntersectingChunkIterator( - shardKeyString, chunkVector.begin(), chunkVector.end(), true /*isMaxInclusive*/); - - if (chunkIt == chunkVector.end()) { - return {}; - } +void validateChunkIsNotOlderThan(const std::shared_ptr<ChunkInfo>& chunk, + const ChunkVersion& version) { + uassert(ErrorCodes::ConflictingOperationInProgress, + str::stream() << "Changed chunk " << chunk->toString() + << " has timestamp different from that of the collection " + << version.getTimestamp(), + version.getTimestamp() == chunk->getLastmod().getTimestamp()); - return *chunkIt; + uassert(626840, + str::stream() + << "Changed chunk " << chunk->toString() + << " doesn't have version that's greater or equal than that of the collection " + << version.toString(), + version.isOlderOrEqualThan(chunk->getLastmod())); } -ChunkMap ChunkMap::createMerged(std::vector<std::shared_ptr<ChunkInfo>> changedChunks) const { - auto updatedChunkMap = _makeUpdated(std::move(changedChunks)); - tassert(6752900, - "Chunk map found to be empty after refresh", - updatedChunkMap._chunkVectorMap.size() && - updatedChunkMap._chunkVectorMap.begin()->second->size()); - return updatedChunkMap; -} - -void ChunkMap::_commitUpdatedChunkVector(std::shared_ptr<ChunkVector>&& chunkVectorPtr, - bool checkMaxKeyConsistency) { - - invariant(!chunkVectorPtr->empty()); - - const auto& vectorMaxKeyString = chunkVectorPtr->back()->getMaxKeyString(); - const auto nextMapIt = _chunkVectorMap.lower_bound(vectorMaxKeyString); +} // namespace - // Check lower bound is consistent - if (nextMapIt == _chunkVectorMap.begin()) { - checkAllElementsAreOfType(MinKey, chunkVectorPtr->front()->getMin()); - } else { - checkChunksAreContiguous(*(std::prev(nextMapIt)->second->back()), - *(chunkVectorPtr->front())); - } +ShardVersionMap ChunkMap::constructShardVersionMap() const { + ShardVersionMap shardVersions; + ChunkVector::const_iterator current = _chunkMap.cbegin(); + + boost::optional<BSONObj> firstMin = boost::none; + boost::optional<BSONObj> lastMax = boost::none; + + while (current != _chunkMap.cend()) { + const auto& firstChunkInRange = *current; + const auto& currentRangeShardId = firstChunkInRange->getShardIdAt(boost::none); + + // Tracks the max shard version for the shard on which the current range will reside + auto shardVersionIt = shardVersions.find(currentRangeShardId); + if (shardVersionIt == shardVersions.end()) { + shardVersionIt = shardVersions + .emplace(std::piecewise_construct, + std::forward_as_tuple(currentRangeShardId), + std::forward_as_tuple(_collectionVersion.epoch(), + _collectionVersion.getTimestamp())) + .first; + } - if (checkMaxKeyConsistency) { - // Check upper bound is consistent - if (nextMapIt == _chunkVectorMap.end()) { - checkAllElementsAreOfType(MaxKey, chunkVectorPtr->back()->getMax()); - } else { - checkChunksAreContiguous(*(chunkVectorPtr->back()), *(nextMapIt->second->front())); + auto& maxShardVersion = shardVersionIt->second.shardVersion; + + current = + std::find_if(current, + _chunkMap.cend(), + [¤tRangeShardId, &maxShardVersion](const auto& currentChunk) { + if (currentChunk->getShardIdAt(boost::none) != currentRangeShardId) + return true; + + if (maxShardVersion.isOlderThan(currentChunk->getLastmod())) + maxShardVersion = currentChunk->getLastmod(); + + return false; + }); + + const auto rangeLast = *std::prev(current); + + const auto& rangeMin = firstChunkInRange->getMin(); + const auto& rangeMax = rangeLast->getMax(); + + // Check the continuity of the chunks map + if (lastMax && !SimpleBSONObjComparator::kInstance.evaluate(*lastMax == rangeMin)) { + if (SimpleBSONObjComparator::kInstance.evaluate(*lastMax < rangeMin)) + uasserted(ErrorCodes::ConflictingOperationInProgress, + str::stream() << "Gap exists in the routing table between chunks " + << findIntersectingChunk(*lastMax)->getRange().toString() + << " and " << rangeLast->getRange().toString()); + else + uasserted(ErrorCodes::ConflictingOperationInProgress, + str::stream() << "Overlap exists in the routing table between chunks " + << findIntersectingChunk(*lastMax)->getRange().toString() + << " and " << rangeLast->getRange().toString()); } - } - auto minVectorSize = _maxChunkVectorSize / 2; - if (chunkVectorPtr->size() < minVectorSize) { - _mergeAndCommitUpdatedChunkVector(nextMapIt, std::move(chunkVectorPtr)); - } else { - _splitAndCommitUpdatedChunkVector(nextMapIt, std::move(chunkVectorPtr)); - } -} + if (!firstMin) + firstMin = rangeMin; -void ChunkMap::_mergeAndCommitUpdatedChunkVector(ChunkVectorMap::const_iterator pos, - std::shared_ptr<ChunkVector>&& smallVectorPtr) { - if (pos == _chunkVectorMap.begin()) { - // Vector will be placed at the head of the map, - // thus there is not previous vector we could merge with - smallVectorPtr->shrink_to_fit(); - _chunkVectorMap.emplace_hint( - pos, smallVectorPtr->back()->getMaxKeyString(), std::move(smallVectorPtr)); + lastMax = rangeMax; - return; + // If a shard has chunks it must have a shard version, otherwise we have an invalid chunk + // somewhere, which should have been caught at chunk load time + invariant(maxShardVersion.isSet()); } - auto prevVectorPtr = _chunkVectorMap.extract(std::prev(pos)).mapped(); - auto mergeVectorPtr = std::make_shared<ChunkVector>(); - mergeVectorPtr->reserve(prevVectorPtr->size() + smallVectorPtr->size()); - - // Fill initial part of merged vector with a copy of old vector (prevVectorPtr) - // Note that the old vector is potentially shared with previous ChunkMap instances, - // thus we copy rather than moving elements to maintain its integrity. - mergeVectorPtr->insert(mergeVectorPtr->end(), prevVectorPtr->begin(), prevVectorPtr->end()); - - // Fill the rest of merged vector with the small updated vector - mergeVectorPtr->insert(mergeVectorPtr->end(), - std::make_move_iterator(smallVectorPtr->begin()), - std::make_move_iterator(smallVectorPtr->end())); + if (!_chunkMap.empty()) { + invariant(!shardVersions.empty()); + invariant(firstMin.is_initialized()); + invariant(lastMax.is_initialized()); - _chunkVectorMap.emplace_hint( - pos, mergeVectorPtr->back()->getMaxKeyString(), std::move(mergeVectorPtr)); -} - -/* - * Split the given chunk vector into pieces not bigger than _maxChunkVectorSize - * and add them to the chunkVector map. - * - * When chunks can't be divided equally among all generated pieces, - * this algorithm guarantee that the size difference between all pieces will be minimal and that - * smaller pieces will be placed at the end. - */ -void ChunkMap::_splitAndCommitUpdatedChunkVector(ChunkVectorMap::const_iterator pos, - std::shared_ptr<ChunkVector>&& chunkVectorPtr) { - auto& chunkVector = *chunkVectorPtr; - const long long totalSize = chunkVector.size(); - const long long numPieces = (totalSize + _maxChunkVectorSize - 1) / _maxChunkVectorSize; - const long long largePieceSize = (totalSize + numPieces - 1) / numPieces; - const long long numLargePieces = totalSize % numPieces; - const long long smallPieceSize = totalSize / numPieces; - - auto lastPos = pos; - auto chunkIt = chunkVector.end(); - for (int pieceCount = 1; pieceCount < numPieces; pieceCount++) { - auto tmpVectorPtr = std::make_shared<ChunkVector>(); - auto targetPieceSize = - (numPieces - pieceCount) < numLargePieces ? largePieceSize : smallPieceSize; - tmpVectorPtr->insert(tmpVectorPtr->end(), - std::make_move_iterator(chunkIt - targetPieceSize), - std::make_move_iterator(chunkIt)); - chunkIt -= targetPieceSize; - lastPos = _chunkVectorMap.emplace_hint( - lastPos, tmpVectorPtr->back()->getMaxKeyString(), std::move(tmpVectorPtr)); + checkAllElementsAreOfType(MinKey, firstMin.get()); + checkAllElementsAreOfType(MaxKey, lastMax.get()); } - invariant(std::distance(chunkVector.begin(), chunkIt) == largePieceSize); - chunkVector.resize(largePieceSize); - chunkVector.shrink_to_fit(); - _chunkVectorMap.emplace_hint( - lastPos, chunkVector.back()->getMaxKeyString(), std::move(chunkVectorPtr)); + return shardVersions; } -void ChunkMap::_updateShardVersionFromDiscardedChunk(const ChunkInfo& chunk) { - auto shardVersionIt = _shardVersions.find(chunk.getShardId()); - if (shardVersionIt != _shardVersions.end() && - shardVersionIt->second.shardVersion == chunk.getLastmod()) { - _shardVersions.erase(shardVersionIt); +void ChunkMap::appendChunk(const std::shared_ptr<ChunkInfo>& chunk) { + appendChunkTo(_chunkMap, chunk); + const auto chunkVersion = chunk->getLastmod(); + if (_collectionVersion.isOlderThan(chunkVersion)) { + _collectionVersion = ChunkVersion(chunkVersion.majorVersion(), + chunkVersion.minorVersion(), + chunkVersion.epoch(), + _collTimestamp); } } -void ChunkMap::_updateShardVersionFromUpdateChunk(const ChunkInfo& chunk, - const ShardVersionMap& oldShardVersions) { - const auto& newVersion = chunk.getLastmod(); - const auto newValidAfter = [&] { - auto thisChunkValidAfter = chunk.getHistory().empty() - ? Timestamp{0, 0} - : chunk.getHistory().front().getValidAfter(); - - auto oldShardVersionIt = oldShardVersions.find(chunk.getShardId()); - auto oldShardValidAfter = oldShardVersionIt == oldShardVersions.end() - ? Timestamp{0, 0} - : oldShardVersionIt->second.validAfter; - - return std::max(thisChunkValidAfter, oldShardValidAfter); - }(); - - // Version for this chunk shard got updated - bool versionUpdated{false}; - - auto [shardVersionIt, created] = - _shardVersions.try_emplace(chunk.getShardId(), newVersion, newValidAfter); - - if (created) { - // We just created a new entry in the _shardVersions map with latest version and latest - // valid after. - versionUpdated = true; - } else { - // _shardVersions map already contained an entry for this chunk shard - - // Update version for this shard - if (shardVersionIt->second.shardVersion.isOlderThan(newVersion)) { - shardVersionIt->second.shardVersion = newVersion; - versionUpdated = true; - } +std::shared_ptr<ChunkInfo> ChunkMap::findIntersectingChunk(const BSONObj& shardKey) const { + const auto it = _findIntersectingChunk(shardKey); - // Update validAfter for this shard - if (newValidAfter > shardVersionIt->second.validAfter) { - shardVersionIt->second.validAfter = newValidAfter; - } - } + if (it != _chunkMap.end()) + return *it; - // Update version for the entire collection - if (versionUpdated && _collectionVersion.isOlderThan(newVersion)) { - _collectionVersion = ChunkVersion{newVersion.majorVersion(), - newVersion.minorVersion(), - newVersion.epoch(), - _collectionVersion.getTimestamp()}; - } + return std::shared_ptr<ChunkInfo>(); } -ChunkMap ChunkMap::_makeUpdated(ChunkVector&& updateChunks) const { - ChunkMap newMap{*this}; +ChunkMap ChunkMap::createMerged( + const std::vector<std::shared_ptr<ChunkInfo>>& changedChunks) const { + size_t chunkMapIndex = 0; + size_t changedChunkIndex = 0; - if (updateChunks.empty()) { - // No updates, just clone the original map - return newMap; - } + ChunkMap updatedChunkMap( + getVersion().epoch(), getVersion().getTimestamp(), _chunkMap.size() + changedChunks.size()); - std::shared_ptr<ChunkVector> oldVectorPtr; - ChunkVector::const_iterator oldChunkIt; - ChunkVector::iterator updateChunkIt; - ChunkVector::const_iterator updateChunkWrittenBytesIt; - std::shared_ptr<ChunkVector> newVectorPtr; - bool lastCommittedIsNew; - - const auto processOldChunk = [&](const std::shared_ptr<ChunkInfo>& nextChunkPtr, - bool discard = false) { - if (discard) { - // Discard chunk from oldVector - - while (updateChunkWrittenBytesIt != updateChunks.end() && - overlaps(*nextChunkPtr, **updateChunkWrittenBytesIt)) { - // Copy writtenBytes to all new overlapping chunks - (*updateChunkWrittenBytesIt++) - ->getWritesTracker() - ->addBytesWritten(nextChunkPtr->getWritesTracker()->getBytesWritten()); - } - - newMap._updateShardVersionFromDiscardedChunk(*nextChunkPtr); - // Since we are discarding the old chunk rather than committing, - // we do not update `lastCommitedIsNew` flag. - } else { - if (!newVectorPtr->empty() && lastCommittedIsNew) { - checkChunksAreContiguous(*newVectorPtr->back(), *nextChunkPtr); - } - lastCommittedIsNew = false; - newVectorPtr->emplace_back(nextChunkPtr); - } - }; - - const auto processUpdateChunk = [&](std::shared_ptr<ChunkInfo>&& nextChunkPtr) { - newMap._updateShardVersionFromUpdateChunk(*nextChunkPtr, _shardVersions); - uassert(ErrorCodes::ConflictingOperationInProgress, - str::stream() << "Changed chunk " << nextChunkPtr->toString() - << " has timestamp different from that of the collection " - << _collectionVersion.getTimestamp(), - nextChunkPtr->getLastmod().getTimestamp() == _collectionVersion.getTimestamp()); - - uassert(626840, - str::stream() - << "Changed chunk " << nextChunkPtr->toString() - << " doesn't have version that's greater or equal than that of the collection " - << _collectionVersion.toString(), - _collectionVersion.isOlderOrEqualThan(nextChunkPtr->getLastmod())); - - if (!newVectorPtr->empty()) { - checkChunksAreContiguous(*newVectorPtr->back(), *nextChunkPtr); - } - lastCommittedIsNew = true; - newVectorPtr->emplace_back(std::move(nextChunkPtr)); - }; - - const auto processOneChunk = [&] { - dassert(oldChunkIt != oldVectorPtr->end() || updateChunkIt != updateChunks.end()); - if (updateChunkIt == updateChunks.end()) { - // no more updates - processOldChunk(*(oldChunkIt++)); - return; - } - if (oldChunkIt == oldVectorPtr->end()) { - // No more old chunks - processUpdateChunk(std::move(*(updateChunkIt++))); - return; + while (chunkMapIndex < _chunkMap.size() || changedChunkIndex < changedChunks.size()) { + if (chunkMapIndex >= _chunkMap.size()) { + validateChunkIsNotOlderThan(changedChunks[changedChunkIndex], getVersion()); + updatedChunkMap.appendChunk(changedChunks[changedChunkIndex++]); + continue; } - const auto& oldChunk = **oldChunkIt; - auto& updateChunk = **updateChunkIt; - - // We have both update and old chunk to peak from - // If they overlaps we discard the old chunk otherwise we process the one with smaller key - if (overlaps(updateChunk, oldChunk)) { - processOldChunk(*(oldChunkIt++), true /* discard */); - return; - } else { - // Ranges do not overlap so we yield the chunk with smaller max key - if (updateChunk.getMaxKeyString() < oldChunk.getMaxKeyString()) { - processUpdateChunk(std::move(*(updateChunkIt++))); - return; - } else { - processOldChunk(*(oldChunkIt++)); - return; - } + if (changedChunkIndex >= changedChunks.size()) { + updatedChunkMap.appendChunk(_chunkMap[chunkMapIndex++]); + continue; } - }; - - updateChunkIt = updateChunks.begin(); - updateChunkWrittenBytesIt = updateChunkIt; - // Skip first vectors that were not affected by this update since we don't need to modify them - auto mapIt = newMap._chunkVectorMap.upper_bound( - ShardKeyPattern::toKeyString((*updateChunkIt)->getRange().getMin())); - oldVectorPtr = - mapIt != newMap._chunkVectorMap.end() ? mapIt->second : std::make_shared<ChunkVector>(); - oldChunkIt = oldVectorPtr->begin(); - // Prepare newVector used as destination of merge sort algorithm - newVectorPtr = std::make_shared<ChunkVector>(); - newVectorPtr->reserve(mapIt != newMap._chunkVectorMap.end() - ? oldVectorPtr->size() - : std::min(_maxChunkVectorSize, updateChunks.size())); - lastCommittedIsNew = false; - - // Iterate until we drained all updates and old vectors - while (updateChunkIt != updateChunks.end() || mapIt != newMap._chunkVectorMap.end()) { - processOneChunk(); - - // Keep processing chunks until we reach the end of the current old vector - if (oldChunkIt == oldVectorPtr->end()) { - if (mapIt == newMap._chunkVectorMap.end()) { - // Only updates left - if (newVectorPtr->size() >= _maxChunkVectorSize) { - auto checkMaxKeyConsistency = updateChunkIt == updateChunks.end(); - newMap._commitUpdatedChunkVector(std::move(newVectorPtr), - checkMaxKeyConsistency); - newVectorPtr = std::make_shared<ChunkVector>(); - // Allocate space only for the remaining updates - newVectorPtr->reserve(newVectorPtr->size() + - std::min(_maxChunkVectorSize, - static_cast<size_t>(std::distance( - updateChunkIt, updateChunks.end())))); - } - } else { - // drained all chunks from old vector in use, - // remove old vector from the new map since we are going to replace it. - auto followingMapIt = newMap._chunkVectorMap.erase(mapIt); - - // Advance the map iterator to the next old vector to update - mapIt = [&] { - if (followingMapIt == newMap._chunkVectorMap.end()) { - // No more old vector to process - return newMap._chunkVectorMap.end(); - } - if (updateChunkIt == updateChunks.end()) { - // No more updates skip all remaining vectors - return newMap._chunkVectorMap.end(); - } + auto overlap = _chunkMap[chunkMapIndex]->getRange().overlaps( + changedChunks[changedChunkIndex]->getRange()); - if (newVectorPtr->size() < _maxChunkVectorSize / 2) { - // New vector is too small, keep accumulating next oldVector - return followingMapIt; - } + if (overlap) { + auto& changedChunk = changedChunks[changedChunkIndex++]; + auto& chunkInfo = _chunkMap[chunkMapIndex]; - // next update doesn't overlap with current old vector so we need to jump - // forward to the first overlapping old vector. - // This is an optimization to skip vectors that are not affected by any updates. - auto nextOvelappingMapIt = newMap._chunkVectorMap.upper_bound( - ShardKeyPattern::toKeyString((*updateChunkIt)->getRange().getMin())); - invariant(nextOvelappingMapIt != newMap._chunkVectorMap.end()); - return nextOvelappingMapIt; - }(); - - // Commit chunks accumulated in new vector if - // - We are skipping next old vector, thus next old chunk is not adjacent to last - // committed chunk - // - We already reached maxChunkSize and next update is not adjacent to last - // committed chunk - if (mapIt != followingMapIt || - (newVectorPtr->size() >= _maxChunkVectorSize && - (updateChunkIt == updateChunks.end() || - ShardKeyPattern::toKeyString((*updateChunkIt)->getRange().getMin()) != - newVectorPtr->back()->getMaxKeyString()))) { - newMap._commitUpdatedChunkVector(std::move(newVectorPtr), true); - newVectorPtr = std::make_shared<ChunkVector>(); - } + auto bytesInReplacedChunk = chunkInfo->getWritesTracker()->getBytesWritten(); + changedChunk->getWritesTracker()->addBytesWritten(bytesInReplacedChunk); - if (mapIt != newMap._chunkVectorMap.end()) { - // Update references to oldVector - oldVectorPtr = mapIt->second; - oldChunkIt = oldVectorPtr->begin(); - // Reserve space for next chunks, - // we cannot know before traversing the next old vector how many chunks will be - // added to the new vector, thus this reservation is just best effort. - newVectorPtr->reserve(newVectorPtr->size() + oldVectorPtr->size()); - } else { - // Only updates left, allocate space only for the remaining updates - newVectorPtr->reserve(newVectorPtr->size() + - std::min(_maxChunkVectorSize, - static_cast<size_t>(std::distance( - updateChunkIt, updateChunks.end())))); - } - } + validateChunkIsNotOlderThan(changedChunk, getVersion()); + updatedChunkMap.appendChunk(changedChunk); + } else { + updatedChunkMap.appendChunk(_chunkMap[chunkMapIndex++]); } } - if (!newVectorPtr->empty()) { - newMap._commitUpdatedChunkVector(std::move(newVectorPtr), true); - } - - return newMap; + return updatedChunkMap; } BSONObj ChunkMap::toBSON() const { BSONObjBuilder builder; getVersion().serializeToBSON("startingVersion"_sd, &builder); - builder.append("chunkCount", static_cast<int64_t>(size())); + builder.append("chunkCount", static_cast<int64_t>(_chunkMap.size())); { BSONArrayBuilder arrayBuilder(builder.subarrayStart("chunks"_sd)); - for (const auto& mapIt : _chunkVectorMap) { - for (const auto& chunkInfoPtr : *mapIt.second) { - arrayBuilder.append(chunkInfoPtr->toString()); - } + for (const auto& chunk : _chunkMap) { + arrayBuilder.append(chunk->toString()); } } return builder.obj(); } -std::string ChunkMap::toString() const { - StringBuilder sb; - - sb << "Bucket size: " << _maxChunkVectorSize << "\n"; - sb << "Num buckets: " << _chunkVectorMap.size() << "\n"; - sb << "Num chunks: " << size() << "\n"; - sb << "Chunks:\n"; - size_t vectorCount{0}; - for (const auto& mapIt : _chunkVectorMap) { - sb << "\t vector[" << vectorCount++ << "] key: " << mongo::base64::encode(mapIt.first) - << ", size: " << mapIt.second->size() << "\n"; - for (const auto& chunkInfoPtr : *mapIt.second) { - sb << "\t" << chunkInfoPtr->toString() << '\n'; - } - } - - sb << "Shard versions:\n"; - for (const auto& entry : _shardVersions) { - sb << "\t" << entry.first << ": " << entry.second.shardVersion.toString() << '\n'; - } - - sb << "Collection version:" << _collectionVersion.toString() << '\n'; - - return sb.str(); -} - -ChunkVector::const_iterator ChunkMap::_findIntersectingChunkIterator( - const std::string& shardKeyString, - ChunkVector::const_iterator first, - ChunkVector::const_iterator last, - bool isMaxInclusive) const { +ChunkMap::ChunkVector::const_iterator ChunkMap::_findIntersectingChunk(const BSONObj& shardKey, + bool isMaxInclusive) const { + auto shardKeyString = ShardKeyPattern::toKeyString(shardKey); if (!isMaxInclusive) { - return std::lower_bound(first, - last, - shardKeyString, - [&](const auto& chunkInfo, const std::string& shardKeyString) { + return std::lower_bound(_chunkMap.begin(), + _chunkMap.end(), + shardKey, + [&shardKeyString](const auto& chunkInfo, const BSONObj& shardKey) { return chunkInfo->getMaxKeyString() < shardKeyString; }); } else { - return std::upper_bound(first, - last, - shardKeyString, - [&](const std::string& shardKeyString, const auto& chunkInfo) { + return std::upper_bound(_chunkMap.begin(), + _chunkMap.end(), + shardKey, + [&shardKeyString](const BSONObj& shardKey, const auto& chunkInfo) { return shardKeyString < chunkInfo->getMaxKeyString(); }); } } - -std::pair<ChunkVectorMap::const_iterator, ChunkVectorMap::const_iterator> -ChunkMap::_overlappingVectorSlotBounds(const std::string& minShardKeyStr, - const std::string& maxShardKeyStr, - bool isMaxInclusive) const { - - const auto itMin = _chunkVectorMap.upper_bound(minShardKeyStr); +std::pair<ChunkMap::ChunkVector::const_iterator, ChunkMap::ChunkVector::const_iterator> +ChunkMap::_overlappingBounds(const BSONObj& min, const BSONObj& max, bool isMaxInclusive) const { + const auto itMin = _findIntersectingChunk(min); const auto itMax = [&]() { - auto it = isMaxInclusive ? _chunkVectorMap.upper_bound(maxShardKeyStr) - : _chunkVectorMap.lower_bound(maxShardKeyStr); - - return it == _chunkVectorMap.end() ? it : ++it; + auto it = _findIntersectingChunk(max, isMaxInclusive); + return it == _chunkMap.end() ? it : ++it; }(); return {itMin, itMax}; @@ -656,7 +338,7 @@ RoutingTableHistory::RoutingTableHistory( _maxChunkSizeBytes(maxChunkSizeBytes), _allowMigrations(allowMigrations), _chunkMap(std::move(chunkMap)), - _shardVersions(_chunkMap.getShardVersionsMap()) {} + _shardVersions(_chunkMap.constructShardVersionMap()) {} void RoutingTableHistory::setShardStale(const ShardId& shardId) { if (gEnableFinerGrainedCatalogCacheRefresh) { @@ -706,7 +388,7 @@ Chunk ChunkManager::findIntersectingChunk(const BSONObj& shardKey, uassert(ErrorCodes::ShardKeyNotFound, str::stream() << "Cannot target single shard using key " << shardKey << " for namespace " << _rt->optRt->nss(), - chunkInfo); + chunkInfo && chunkInfo->containsKey(shardKey)); return Chunk(*chunkInfo, _clusterTime); } @@ -719,14 +401,15 @@ bool ChunkManager::keyBelongsToShard(const BSONObj& shardKey, const ShardId& sha if (!chunkInfo) return false; + invariant(chunkInfo->containsKey(shardKey)); + return chunkInfo->getShardIdAt(_clusterTime) == shardId; } void ChunkManager::getShardIdsForQuery(boost::intrusive_ptr<ExpressionContext> expCtx, const BSONObj& query, const BSONObj& collation, - std::set<ShardId>* shardIds, - bool bypassIsFieldHashedCheck) const { + std::set<ShardId>* shardIds) const { auto findCommand = std::make_unique<FindCommandRequest>(_rt->optRt->nss()); findCommand->setFilter(query.getOwned()); @@ -752,7 +435,7 @@ void ChunkManager::getShardIdsForQuery(boost::intrusive_ptr<ExpressionContext> e auto shardKeyToFind = _rt->optRt->getShardKeyPattern().extractShardKeyFromQuery(*cq); if (!shardKeyToFind.isEmpty()) { try { - auto chunk = findIntersectingChunk(shardKeyToFind, collation, bypassIsFieldHashedCheck); + auto chunk = findIntersectingChunk(shardKeyToFind, collation); shardIds->insert(chunk.getShardId()); return; } catch (const DBException&) { @@ -792,7 +475,7 @@ void ChunkManager::getShardIdsForQuery(boost::intrusive_ptr<ExpressionContext> e // For now, we satisfy that assumption by adding a shard with no matches rather than returning // an empty set of shards. if (shardIds->empty()) { - _rt->optRt->forEachChunk([&](const auto& chunkInfo) { + _rt->optRt->forEachChunk([&](const std::shared_ptr<ChunkInfo>& chunkInfo) { shardIds->insert(chunkInfo->getShardIdAt(_clusterTime)); return false; }); @@ -812,7 +495,7 @@ void ChunkManager::getShardIdsForRange(const BSONObj& min, return; } - _rt->optRt->forEachOverlappingChunk(min, max, true, [&](const auto& chunkInfo) { + _rt->optRt->forEachOverlappingChunk(min, max, true, [&](auto& chunkInfo) { shardIds->insert(chunkInfo->getShardIdAt(_clusterTime)); // No need to iterate through the rest of the ranges, because we already know we need to use @@ -832,7 +515,7 @@ bool ChunkManager::rangeOverlapsShard(const ChunkRange& range, const ShardId& sh bool overlapFound = false; _rt->optRt->forEachOverlappingChunk( - range.getMin(), range.getMax(), false, [&](const auto& chunkInfo) { + range.getMin(), range.getMax(), false, [&](auto& chunkInfo) { if (chunkInfo->getShardIdAt(_clusterTime) == shardId) { overlapFound = true; return false; @@ -849,7 +532,7 @@ boost::optional<Chunk> ChunkManager::getNextChunkOnShard(const BSONObj& shardKey boost::optional<Chunk> chunk; _rt->optRt->forEachChunk( - [&](const auto& chunkInfo) { + [&](auto& chunkInfo) { if (chunkInfo->getShardIdAt(_clusterTime) == shardId) { chunk.emplace(*chunkInfo, _clusterTime); return false; @@ -922,7 +605,7 @@ IndexBounds ChunkManager::getIndexBoundsForQuery(const BSONObj& key, // Use query framework to generate index bounds QueryPlannerParams plannerParams; // Must use "shard key" index - plannerParams.options = QueryPlannerParams::STRICT_NO_TABLE_SCAN; + plannerParams.options = QueryPlannerParams::NO_TABLE_SCAN; IndexEntry indexEntry(key, indexType, IndexDescriptor::kLatestIndexVersion, @@ -1051,15 +734,21 @@ std::string ChunkManager::toString() const { return _rt->optRt ? _rt->optRt->toString() : "UNSHARDED"; } -ShardVersionTargetingInfo RoutingTableHistory::_getVersion(const ShardId& shardName, - bool throwOnStaleShard) const { +bool RoutingTableHistory::compatibleWith(const RoutingTableHistory& other, + const ShardId& shardName) const { + // Return true if the shard version is the same in the two chunk managers + // TODO: This doesn't need to be so strong, just major vs + return other.getVersion(shardName) == getVersion(shardName); +} + +ChunkVersion RoutingTableHistory::_getVersion(const ShardId& shardName, + bool throwOnStaleShard) const { auto it = _shardVersions.find(shardName); if (it == _shardVersions.end()) { // Shards without explicitly tracked shard versions (meaning they have no chunks) always // have a version of (0, 0, epoch, timestamp) const auto collVersion = _chunkMap.getVersion(); - return ShardVersionTargetingInfo( - ChunkVersion(0, 0, collVersion.epoch(), collVersion.getTimestamp()), Timestamp(0, 0)); + return ChunkVersion(0, 0, collVersion.epoch(), collVersion.getTimestamp()); } if (throwOnStaleShard && gEnableFinerGrainedCatalogCacheRefresh) { @@ -1068,21 +757,30 @@ ShardVersionTargetingInfo RoutingTableHistory::_getVersion(const ShardId& shardN !it->second.isStale.load()); } - const auto& shardVersionTargetingInfo = it->second; - return ShardVersionTargetingInfo(shardVersionTargetingInfo.shardVersion, - shardVersionTargetingInfo.validAfter); + return it->second.shardVersion; +} + +ChunkVersion RoutingTableHistory::getVersion(const ShardId& shardName) const { + return _getVersion(shardName, true); +} + +ChunkVersion RoutingTableHistory::getVersionForLogging(const ShardId& shardName) const { + return _getVersion(shardName, false); } std::string RoutingTableHistory::toString() const { StringBuilder sb; sb << "RoutingTableHistory: " << _nss.ns() << " key: " << _shardKeyPattern.toString() << '\n'; - sb << _chunkMap.toString(); + sb << "Chunks:\n"; + _chunkMap.forEach([&sb](const auto& chunk) { + sb << "\t" << chunk->toString() << '\n'; + return true; + }); sb << "Shard versions:\n"; for (const auto& entry : _shardVersions) { - sb << "\t" << entry.first << ": " << entry.second.shardVersion.toString() << " @ " - << entry.second.validAfter.toString() << '\n'; + sb << "\t" << entry.first << ": " << entry.second.shardVersion.toString() << '\n'; } return sb.str(); @@ -1103,19 +801,16 @@ RoutingTableHistory RoutingTableHistory::makeNew( const std::vector<ChunkType>& chunks) { auto changedChunkInfos = flatten(chunks); - - return RoutingTableHistory( - std::move(nss), - std::move(uuid), - std::move(shardKeyPattern), - std::move(defaultCollator), - std::move(unique), - std::move(timeseriesFields), - std::move(reshardingFields), - boost::none, - allowMigrations, - ChunkMap{epoch, timestamp, static_cast<size_t>(gRoutingTableCacheChunkBucketSize)} - .createMerged(std::move(changedChunkInfos))); + return RoutingTableHistory(std::move(nss), + std::move(uuid), + std::move(shardKeyPattern), + std::move(defaultCollator), + std::move(unique), + std::move(timeseriesFields), + std::move(reshardingFields), + maxChunkSizeBytes, + allowMigrations, + ChunkMap{epoch, timestamp}.createMerged(changedChunkInfos)); } // Note that any new parameters added to RoutingTableHistory::makeUpdated() must also be added to @@ -1129,7 +824,7 @@ RoutingTableHistory RoutingTableHistory::makeUpdated( const std::vector<ChunkType>& changedChunks) const { auto changedChunkInfos = flatten(changedChunks); - auto chunkMap = _chunkMap.createMerged(std::move(changedChunkInfos)); + auto chunkMap = _chunkMap.createMerged(changedChunkInfos); // Only update the same collection. invariant(getVersion().isSameCollection(chunkMap.getVersion())); |