diff options
Diffstat (limited to 'src/mongo/s')
101 files changed, 3586 insertions, 1182 deletions
diff --git a/src/mongo/s/SConscript b/src/mongo/s/SConscript index 336e7a2ff98..c144eed4ca8 100644 --- a/src/mongo/s/SConscript +++ b/src/mongo/s/SConscript @@ -49,13 +49,13 @@ env.Library( 'cluster_commands_helpers.cpp', 'collection_uuid_mismatch.cpp', 'multi_statement_transaction_requests_sender.cpp', + 'router_role.cpp', 'router_transactions_metrics.cpp', 'router_transactions_stats.idl', - 'router.cpp', 'session_catalog_router.cpp', 'stale_shard_version_helpers.cpp', - 'transaction_router_resource_yielder.cpp', 'transaction_router.cpp', + 'transaction_router_resource_yielder.cpp', ], LIBDEPS=[ '$BUILD_DIR/mongo/db/commands/txn_cmd_request', @@ -187,6 +187,7 @@ env.Library( 'refine_collection_shard_key_coordinator_feature_flags.idl', 'request_types/abort_reshard_collection.idl', 'request_types/add_shard_request_type.cpp', + 'request_types/get_stats_for_balancing.idl', 'request_types/add_shard_to_zone_request_type.cpp', 'request_types/auto_split_vector.idl', 'request_types/balance_chunk_request_type.cpp', @@ -617,6 +618,7 @@ env.CppUnitTest( 'catalog/type_mongos_test.cpp', 'catalog/type_shard_test.cpp', 'catalog/type_tags_test.cpp', + 'chunks_test_util.cpp', 'chunk_manager_index_bounds_test.cpp', 'chunk_manager_query_test.cpp', 'chunk_manager_targeter_test.cpp', diff --git a/src/mongo/s/async_requests_sender.cpp b/src/mongo/s/async_requests_sender.cpp index 1fceefc428e..a5100472f69 100644 --- a/src/mongo/s/async_requests_sender.cpp +++ b/src/mongo/s/async_requests_sender.cpp @@ -37,6 +37,7 @@ #include <memory> #include "mongo/client/remote_command_targeter.h" +#include "mongo/db/curop.h" #include "mongo/executor/remote_command_request.h" #include "mongo/logv2/log.h" #include "mongo/rpc/get_status_from_command_result.h" @@ -87,6 +88,8 @@ AsyncRequestsSender::AsyncRequestsSender(OperationContext* opCtx, // Kick off requests immediately. _remotes.emplace_back(this, request.shardId, request.cmdObj).executeRequest(); } + + CurOp::get(_opCtx)->ensureRecordRemoteOpWait(); } AsyncRequestsSender::Response AsyncRequestsSender::next() noexcept { @@ -117,9 +120,22 @@ AsyncRequestsSender::Response AsyncRequestsSender::next() noexcept { _resourceYielder->yield(_opCtx); } + auto curOp = CurOp::get(_opCtx); + // Calculating the total wait time for remote operations relies on the CurOp's timing + // measurement facility and we can't use such facility when the current operation is marked + // as done. Some commands such as 'analyzeShardKey' command may send remote operations using + // AsyncRequestsSender even after marking the current operation done and so we need to check + // whether the current operation is still in progress. + auto curOpInProgress = !curOp->isDone(); + if (curOpInProgress) { + curOp->startRemoteOpWaitTimer(); + } // Only wait for the next result without popping it, so an error unyielding doesn't // discard an already popped response. auto waitStatus = _responseQueue.waitForNonEmptyNoThrow(_opCtx); + if (curOpInProgress) { + curOp->stopRemoteOpWaitTimer(); + } auto unyieldStatus = _resourceYielder ? _resourceYielder->unyieldNoThrow(_opCtx) : Status::OK(); @@ -170,9 +186,10 @@ AsyncRequestsSender::RemoteData::RemoteData(AsyncRequestsSender* ars, BSONObj cmdObj) : _ars(ars), _shardId(std::move(shardId)), _cmdObj(std::move(cmdObj)) {} -std::shared_ptr<Shard> AsyncRequestsSender::RemoteData::getShard() { - // TODO: Pass down an OperationContext* to use here. - return Grid::get(getGlobalServiceContext())->shardRegistry()->getShardNoReload(_shardId); +SemiFuture<std::shared_ptr<Shard>> AsyncRequestsSender::RemoteData::getShard() noexcept { + return Grid::get(getGlobalServiceContext()) + ->shardRegistry() + ->getShard(*_ars->_subBaton, _shardId); } void AsyncRequestsSender::RemoteData::executeRequest() { @@ -192,7 +209,12 @@ void AsyncRequestsSender::RemoteData::executeRequest() { auto AsyncRequestsSender::RemoteData::scheduleRequest() -> SemiFuture<RemoteCommandOnAnyCallbackArgs> { - return resolveShardIdToHostAndPorts(_ars->_readPreference) + return getShard() + .thenRunOn(*_ars->_subBaton) + .then([this](auto&& shard) { + return shard->getTargeter()->findHosts(_ars->_readPreference, + CancellationToken::uncancelable()); + }) .thenRunOn(*_ars->_subBaton) .then([this](auto&& hostAndPorts) { _shardHostAndPort.emplace(hostAndPorts.front()); @@ -202,17 +224,6 @@ auto AsyncRequestsSender::RemoteData::scheduleRequest() .semi(); } -SemiFuture<std::vector<HostAndPort>> AsyncRequestsSender::RemoteData::resolveShardIdToHostAndPorts( - const ReadPreferenceSetting& readPref) { - const auto shard = getShard(); - if (!shard) { - return Status(ErrorCodes::ShardNotFound, - str::stream() << "Could not find shard " << _shardId); - } - - return shard->getTargeter()->findHosts(readPref, CancellationToken::uncancelable()); -} - auto AsyncRequestsSender::RemoteData::scheduleRemoteCommand(std::vector<HostAndPort>&& hostAndPorts) -> SemiFuture<RemoteCommandOnAnyCallbackArgs> { hangBeforeSchedulingRemoteCommand.executeIf( @@ -278,43 +289,47 @@ auto AsyncRequestsSender::RemoteData::handleResponse(RemoteCommandOnAnyCallbackA } // There was an error with either the response or the command. - auto shard = getShard(); - if (!shard) { - uasserted(ErrorCodes::ShardNotFound, str::stream() << "Could not find shard " << _shardId); - } else { - std::vector<HostAndPort> failedTargets; - - if (rcr.response.target) { - failedTargets = {*rcr.response.target}; - } else { - failedTargets = rcr.request.target; - } + return getShard() + .thenRunOn(*_ars->_subBaton) + .then([this, status = std::move(status), rcr = std::move(rcr)]( + std::shared_ptr<mongo::Shard>&& shard) { + std::vector<HostAndPort> failedTargets; - shard->updateReplSetMonitor(failedTargets.front(), status); - bool isStartingTransaction = _cmdObj.getField("startTransaction").booleanSafe(); - if (!_ars->_stopRetrying && shard->isRetriableError(status.code(), _ars->_retryPolicy) && - _retryCount < kMaxNumFailedHostRetryAttempts && !isStartingTransaction) { - - LOGV2_DEBUG(4615637, - 1, - "Command to remote {shardId} for hosts {hosts} failed with retryable error " - "{error} and will be retried", - "Command to remote shard failed with retryable error and will be retried", - "shardId"_attr = _shardId, - "hosts"_attr = failedTargets, - "error"_attr = redact(status)); - ++_retryCount; - _shardHostAndPort.reset(); - // retry through recursion - return scheduleRequest(); - } - } + if (rcr.response.target) { + failedTargets = {*rcr.response.target}; + } else { + failedTargets = rcr.request.target; + } - // Status' in the response.status field that aren't retried get converted to top level errors - uassertStatusOK(rcr.response.status); + shard->updateReplSetMonitor(failedTargets.front(), status); + bool isStartingTransaction = _cmdObj.getField("startTransaction").booleanSafe(); + if (!_ars->_stopRetrying && + shard->isRetriableError(status.code(), _ars->_retryPolicy) && + _retryCount < kMaxNumFailedHostRetryAttempts && !isStartingTransaction) { + + LOGV2_DEBUG( + 4615637, + 1, + "Command to remote {shardId} for hosts {hosts} failed with retryable error " + "{error} and will be retried", + "Command to remote shard failed with retryable error and will be retried", + "shardId"_attr = _shardId, + "hosts"_attr = failedTargets, + "error"_attr = redact(status)); + ++_retryCount; + _shardHostAndPort.reset(); + // retry through recursion + return scheduleRequest(); + } + + // Status' in the response.status field that aren't retried get converted to top level + // errors + uassertStatusOK(rcr.response.status); - // We're not okay (on the remote), but still not going to retry - return std::move(rcr); + // We're not okay (on the remote), but still not going to retry + return Future<RemoteCommandOnAnyCallbackArgs>::makeReady(std::move(rcr)).semi(); + }) + .semi(); }; } // namespace mongo diff --git a/src/mongo/s/async_requests_sender.h b/src/mongo/s/async_requests_sender.h index 432227356cc..27d22bc5511 100644 --- a/src/mongo/s/async_requests_sender.h +++ b/src/mongo/s/async_requests_sender.h @@ -179,9 +179,15 @@ private: RemoteData(AsyncRequestsSender* ars, ShardId shardId, BSONObj cmdObj); /** - * Returns the Shard object associated with this remote. + * Returns a SemiFuture containing a shard object associated with this remote. + * + * This will return a SemiFuture with a ShardNotFound error status in case the shard is not + * found. + * + * Additionally this call can trigger a refresh of the ShardRegistry so it could possibly + * return other network error status related to the refresh. */ - std::shared_ptr<Shard> getShard(); + SemiFuture<std::shared_ptr<Shard>> getShard() noexcept; /** * Returns true if we've already queued a response from the remote. diff --git a/src/mongo/s/catalog/sharding_catalog_client.h b/src/mongo/s/catalog/sharding_catalog_client.h index 05ea0bf1658..e98e24516d8 100644 --- a/src/mongo/s/catalog/sharding_catalog_client.h +++ b/src/mongo/s/catalog/sharding_catalog_client.h @@ -131,11 +131,14 @@ public: /** * Retrieves all collections under a specified database (or in the system). If the dbName * parameter is empty, returns all collections. + * + * @param sort Fields to use for sorting the results. If empty, no sorting is performed. */ virtual std::vector<CollectionType> getCollections( OperationContext* opCtx, StringData db, - repl::ReadConcernLevel readConcernLevel = repl::ReadConcernLevel::kMajorityReadConcern) = 0; + repl::ReadConcernLevel readConcernLevel = repl::ReadConcernLevel::kMajorityReadConcern, + const BSONObj& sort = BSONObj()) = 0; /** * Returns the set of collections for the specified database, which have been marked as sharded. @@ -281,15 +284,22 @@ public: repl::ReadConcernLevel readConcern) = 0; /** - * Returns keys for the given purpose and with an expiresAt value greater than newerThanThis. + * Returns internal keys for the given purpose and have an expiresAt value greater than + * newerThanThis. */ - virtual StatusWith<std::vector<KeysCollectionDocument>> getNewKeys( + virtual StatusWith<std::vector<KeysCollectionDocument>> getNewInternalKeys( OperationContext* opCtx, StringData purpose, const LogicalTime& newerThanThis, repl::ReadConcernLevel readConcernLevel) = 0; /** + * Returns all external (i.e. validation-only) keys for the given purpose. + */ + virtual StatusWith<std::vector<ExternalKeysCollectionDocument>> getAllExternalKeys( + OperationContext* opCtx, StringData purpose, repl::ReadConcernLevel readConcernLevel) = 0; + + /** * Directly inserts a document in the specified namespace on the config server. The document * must have an _id index. Must only be used for insertions in the 'config' database. * diff --git a/src/mongo/s/catalog/sharding_catalog_client_impl.cpp b/src/mongo/s/catalog/sharding_catalog_client_impl.cpp index bf98a08c7aa..793f05229e2 100644 --- a/src/mongo/s/catalog/sharding_catalog_client_impl.cpp +++ b/src/mongo/s/catalog/sharding_catalog_client_impl.cpp @@ -302,6 +302,45 @@ AggregateCommandRequest makeCollectionAndChunksAggregation(OperationContext* opC return AggregateCommandRequest(CollectionType::ConfigNS, std::move(serializedPipeline)); } +/** + * Returns keys for the given purpose and have an expiresAt value greater than newerThanThis on the + * given shard. + */ +template <typename KeyDocumentType> +StatusWith<std::vector<KeyDocumentType>> _getNewKeys(OperationContext* opCtx, + std::shared_ptr<Shard> shard, + const NamespaceString& nss, + StringData purpose, + const LogicalTime& newerThanThis, + repl::ReadConcernLevel readConcernLevel) { + BSONObjBuilder queryBuilder; + queryBuilder.append("purpose", purpose); + queryBuilder.append("expiresAt", BSON("$gt" << newerThanThis.asTimestamp())); + + auto findStatus = shard->exhaustiveFindOnConfig(opCtx, + kConfigReadSelector, + readConcernLevel, + nss, + queryBuilder.obj(), + BSON("expiresAt" << 1), + boost::none); + if (!findStatus.isOK()) { + return findStatus.getStatus(); + } + const auto& objs = findStatus.getValue().docs; + + std::vector<KeyDocumentType> keyDocs; + keyDocs.reserve(objs.size()); + for (auto&& obj : objs) { + try { + keyDocs.push_back(KeyDocumentType::parse(IDLParserErrorContext("keyDoc"), obj)); + } catch (...) { + return exceptionToStatus(); + } + } + return keyDocs; +} + } // namespace ShardingCatalogClientImpl::ShardingCatalogClientImpl() = default; @@ -457,7 +496,10 @@ CollectionType ShardingCatalogClientImpl::getCollection(OperationContext* opCtx, } std::vector<CollectionType> ShardingCatalogClientImpl::getCollections( - OperationContext* opCtx, StringData dbName, repl::ReadConcernLevel readConcernLevel) { + OperationContext* opCtx, + StringData dbName, + repl::ReadConcernLevel readConcernLevel, + const BSONObj& sort) { BSONObjBuilder b; if (!dbName.empty()) b.appendRegex(CollectionType::kNssFieldName, @@ -469,7 +511,7 @@ std::vector<CollectionType> ShardingCatalogClientImpl::getCollections( readConcernLevel, CollectionType::ConfigNS, b.obj(), - BSONObj(), + sort, boost::none)) .value; std::vector<CollectionType> collections; @@ -482,7 +524,7 @@ std::vector<CollectionType> ShardingCatalogClientImpl::getCollections( std::vector<NamespaceString> ShardingCatalogClientImpl::getAllShardedCollectionsForDb( OperationContext* opCtx, StringData dbName, repl::ReadConcernLevel readConcern) { - auto collectionsOnConfig = getCollections(opCtx, dbName, readConcern); + auto collectionsOnConfig = getCollections(opCtx, dbName, readConcern, BSONObj()); std::vector<NamespaceString> collectionsToReturn; collectionsToReturn.reserve(collectionsOnConfig.size()); @@ -1213,41 +1255,32 @@ StatusWith<repl::OpTimeWith<vector<BSONObj>>> ShardingCatalogClientImpl::_exhaus response.getValue().opTime); } -StatusWith<std::vector<KeysCollectionDocument>> ShardingCatalogClientImpl::getNewKeys( +StatusWith<std::vector<KeysCollectionDocument>> ShardingCatalogClientImpl::getNewInternalKeys( OperationContext* opCtx, StringData purpose, const LogicalTime& newerThanThis, repl::ReadConcernLevel readConcernLevel) { - auto config = Grid::get(opCtx)->shardRegistry()->getConfigShard(); - - BSONObjBuilder queryBuilder; - queryBuilder.append("purpose", purpose); - queryBuilder.append("expiresAt", BSON("$gt" << newerThanThis.asTimestamp())); - - auto findStatus = config->exhaustiveFindOnConfig(opCtx, - kConfigReadSelector, - readConcernLevel, - NamespaceString::kKeysCollectionNamespace, - queryBuilder.obj(), - BSON("expiresAt" << 1), - boost::none); - - if (!findStatus.isOK()) { - return findStatus.getStatus(); - } - - const auto& keyDocs = findStatus.getValue().docs; - std::vector<KeysCollectionDocument> keys; - keys.reserve(keyDocs.size()); - for (auto&& keyDoc : keyDocs) { - try { - keys.push_back(KeysCollectionDocument::parse(IDLParserErrorContext("keyDoc"), keyDoc)); - } catch (...) { - return exceptionToStatus(); - } - } + auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard(); + return _getNewKeys<KeysCollectionDocument>(opCtx, + configShard, + NamespaceString::kKeysCollectionNamespace, + purpose, + newerThanThis, + readConcernLevel); +} - return keys; +StatusWith<std::vector<ExternalKeysCollectionDocument>> +ShardingCatalogClientImpl::getAllExternalKeys(OperationContext* opCtx, + StringData purpose, + repl::ReadConcernLevel readConcernLevel) { + auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard(); + return _getNewKeys<ExternalKeysCollectionDocument>( + opCtx, + configShard, + NamespaceString::kExternalKeysCollectionNamespace, + purpose, + LogicalTime(), + readConcernLevel); } } // namespace mongo diff --git a/src/mongo/s/catalog/sharding_catalog_client_impl.h b/src/mongo/s/catalog/sharding_catalog_client_impl.h index 99c86ef03a1..51bd97acc32 100644 --- a/src/mongo/s/catalog/sharding_catalog_client_impl.h +++ b/src/mongo/s/catalog/sharding_catalog_client_impl.h @@ -82,7 +82,8 @@ public: std::vector<CollectionType> getCollections(OperationContext* opCtx, StringData db, - repl::ReadConcernLevel readConcernLevel) override; + repl::ReadConcernLevel readConcernLevel, + const BSONObj& sort) override; std::vector<NamespaceString> getAllShardedCollectionsForDb( OperationContext* opCtx, StringData dbName, repl::ReadConcernLevel readConcern) override; @@ -169,12 +170,17 @@ public: const WriteConcernOptions& writeConcern, boost::optional<BSONObj> hint = boost::none) override; - StatusWith<std::vector<KeysCollectionDocument>> getNewKeys( + StatusWith<std::vector<KeysCollectionDocument>> getNewInternalKeys( OperationContext* opCtx, StringData purpose, const LogicalTime& newerThanThis, repl::ReadConcernLevel readConcernLevel) override; + StatusWith<std::vector<ExternalKeysCollectionDocument>> getAllExternalKeys( + OperationContext* opCtx, + StringData purpose, + repl::ReadConcernLevel readConcernLevel) override; + private: /** * Updates a single document (if useMultiUpdate is false) or multiple documents (if diff --git a/src/mongo/s/catalog/sharding_catalog_client_mock.cpp b/src/mongo/s/catalog/sharding_catalog_client_mock.cpp index 3ad6ab3f54a..36bd79ec6b0 100644 --- a/src/mongo/s/catalog/sharding_catalog_client_mock.cpp +++ b/src/mongo/s/catalog/sharding_catalog_client_mock.cpp @@ -69,7 +69,10 @@ CollectionType ShardingCatalogClientMock::getCollection(OperationContext* opCtx, } std::vector<CollectionType> ShardingCatalogClientMock::getCollections( - OperationContext* opCtx, StringData dbName, repl::ReadConcernLevel readConcernLevel) { + OperationContext* opCtx, + StringData dbName, + repl::ReadConcernLevel readConcernLevel, + const BSONObj& sort) { uasserted(ErrorCodes::InternalError, "Method not implemented"); } @@ -191,7 +194,7 @@ Status ShardingCatalogClientMock::createDatabase(OperationContext* opCtx, return {ErrorCodes::InternalError, "Method not implemented"}; } -StatusWith<std::vector<KeysCollectionDocument>> ShardingCatalogClientMock::getNewKeys( +StatusWith<std::vector<KeysCollectionDocument>> ShardingCatalogClientMock::getNewInternalKeys( OperationContext* opCtx, StringData purpose, const LogicalTime& newerThanThis, @@ -199,6 +202,13 @@ StatusWith<std::vector<KeysCollectionDocument>> ShardingCatalogClientMock::getNe return {ErrorCodes::InternalError, "Method not implemented"}; } +StatusWith<std::vector<ExternalKeysCollectionDocument>> +ShardingCatalogClientMock::getAllExternalKeys(OperationContext* opCtx, + StringData purpose, + repl::ReadConcernLevel readConcernLevel) { + return {ErrorCodes::InternalError, "Method not implemented"}; +} + StatusWith<repl::OpTimeWith<std::vector<BSONObj>>> ShardingCatalogClientMock::_exhaustiveFindOnConfig(OperationContext* opCtx, const ReadPreferenceSetting& readPref, diff --git a/src/mongo/s/catalog/sharding_catalog_client_mock.h b/src/mongo/s/catalog/sharding_catalog_client_mock.h index fdab949c024..0ce80f42b77 100644 --- a/src/mongo/s/catalog/sharding_catalog_client_mock.h +++ b/src/mongo/s/catalog/sharding_catalog_client_mock.h @@ -58,7 +58,8 @@ public: std::vector<CollectionType> getCollections(OperationContext* opCtx, StringData db, - repl::ReadConcernLevel readConcernLevel) override; + repl::ReadConcernLevel readConcernLevel, + const BSONObj& sort) override; std::vector<NamespaceString> getAllShardedCollectionsForDb( OperationContext* opCtx, StringData dbName, repl::ReadConcernLevel readConcern) override; @@ -148,12 +149,17 @@ public: Status createDatabase(OperationContext* opCtx, StringData dbName, ShardId primaryShard); - StatusWith<std::vector<KeysCollectionDocument>> getNewKeys( + StatusWith<std::vector<KeysCollectionDocument>> getNewInternalKeys( OperationContext* opCtx, StringData purpose, const LogicalTime& newerThanThis, repl::ReadConcernLevel readConcernLevel) override; + StatusWith<std::vector<ExternalKeysCollectionDocument>> getAllExternalKeys( + OperationContext* opCtx, + StringData purpose, + repl::ReadConcernLevel readConcernLevel) override; + private: StatusWith<repl::OpTimeWith<std::vector<BSONObj>>> _exhaustiveFindOnConfig( OperationContext* opCtx, diff --git a/src/mongo/s/catalog/sharding_catalog_client_test.cpp b/src/mongo/s/catalog/sharding_catalog_client_test.cpp index 1c9e8bca3b9..e0bbacae764 100644 --- a/src/mongo/s/catalog/sharding_catalog_client_test.cpp +++ b/src/mongo/s/catalog/sharding_catalog_client_test.cpp @@ -1351,10 +1351,10 @@ TEST_F(ShardingCatalogClientTest, GetNewKeys) { repl::ReadConcernLevel readConcernLevel(repl::ReadConcernLevel::kMajorityReadConcern); auto future = launchAsync([this, purpose, currentTime, readConcernLevel] { - auto status = - catalogClient()->getNewKeys(operationContext(), purpose, currentTime, readConcernLevel); - ASSERT_OK(status.getStatus()); - return status.getValue(); + auto swKeys = catalogClient()->getNewInternalKeys( + operationContext(), purpose, currentTime, readConcernLevel); + ASSERT_OK(swKeys.getStatus()); + return swKeys.getValue(); }); LogicalTime dummyTime(Timestamp(9876, 5432)); @@ -1416,10 +1416,10 @@ TEST_F(ShardingCatalogClientTest, GetNewKeysWithEmptyCollection) { repl::ReadConcernLevel readConcernLevel(repl::ReadConcernLevel::kMajorityReadConcern); auto future = launchAsync([this, purpose, currentTime, readConcernLevel] { - auto status = - catalogClient()->getNewKeys(operationContext(), purpose, currentTime, readConcernLevel); - ASSERT_OK(status.getStatus()); - return status.getValue(); + auto swKeys = catalogClient()->getNewInternalKeys( + operationContext(), purpose, currentTime, readConcernLevel); + ASSERT_OK(swKeys.getStatus()); + return swKeys.getValue(); }); onFindCommand([this](const RemoteCommandRequest& request) { diff --git a/src/mongo/s/catalog/type_chunk.cpp b/src/mongo/s/catalog/type_chunk.cpp index 61c964af93d..485f448529c 100644 --- a/src/mongo/s/catalog/type_chunk.cpp +++ b/src/mongo/s/catalog/type_chunk.cpp @@ -88,6 +88,15 @@ Status extractObject(const BSONObj& obj, const std::string& fieldName, BSONEleme return Status::OK(); } +bool allElementsAreMaxKey(const BSONObj& obj) { + for (auto&& elem : obj) { + if (elem.type() != MaxKey) { + return false; + } + } + return true; +} + } // namespace ChunkRange::ChunkRange(BSONObj minKey, BSONObj maxKey) @@ -124,7 +133,8 @@ StatusWith<ChunkRange> ChunkRange::fromBSON(const BSONObj& obj) { } bool ChunkRange::containsKey(const BSONObj& key) const { - return _minKey.woCompare(key) <= 0 && key.woCompare(_maxKey) < 0; + return (_minKey.woCompare(key) <= 0 && key.woCompare(_maxKey) < 0) || + MONGO_unlikely(allElementsAreMaxKey(key) && key.binaryEqual(_maxKey)); } void ChunkRange::append(BSONObjBuilder* builder) const { diff --git a/src/mongo/s/catalog/type_chunk_test.cpp b/src/mongo/s/catalog/type_chunk_test.cpp index 52c172a529a..3de5d37ce88 100644 --- a/src/mongo/s/catalog/type_chunk_test.cpp +++ b/src/mongo/s/catalog/type_chunk_test.cpp @@ -352,6 +352,69 @@ TEST(ChunkRange, Union) { target.unionWith(ChunkRange(BSON("x" << 9), BSON("x" << 15)))); } +TEST(ChunkRange, ContainsKey) { + auto target = ChunkRange(BSON("x" << 5), BSON("x" << 10)); + ASSERT_FALSE(target.containsKey(BSON("x" << MINKEY))); + ASSERT_FALSE(target.containsKey(BSON("x" << 2))); + ASSERT_TRUE(target.containsKey(BSON("x" << 5))); + ASSERT_TRUE(target.containsKey(BSON("x" << 7))); + ASSERT_FALSE(target.containsKey(BSON("x" << 10))); + ASSERT_FALSE(target.containsKey(BSON("x" << 15))); + ASSERT_FALSE(target.containsKey(BSON("x" << MAXKEY))); + + target = ChunkRange(BSON("x" << MINKEY), BSON("x" << 5)); + ASSERT_TRUE(target.containsKey(BSON("x" << MINKEY))); + ASSERT_TRUE(target.containsKey(BSON("x" << 2))); + ASSERT_FALSE(target.containsKey(BSON("x" << 5))); + ASSERT_FALSE(target.containsKey(BSON("x" << 10))); + ASSERT_FALSE(target.containsKey(BSON("x" << MAXKEY))); + + target = ChunkRange(BSON("x" << 5), BSON("x" << MAXKEY)); + ASSERT_FALSE(target.containsKey(BSON("x" << MINKEY))); + ASSERT_FALSE(target.containsKey(BSON("x" << 2))); + ASSERT_TRUE(target.containsKey(BSON("x" << 5))); + ASSERT_TRUE(target.containsKey(BSON("x" << 10))); + ASSERT_TRUE(target.containsKey(BSON("x" << MAXKEY))); +} + +TEST(ChunkRange, ContainsKeyCompound) { + auto target = ChunkRange(BSON("x" << 5 << "y" << 100), BSON("x" << 10 << "y" << 120)); + ASSERT_FALSE(target.containsKey(BSON("x" << MINKEY))); + ASSERT_FALSE(target.containsKey(BSON("x" << MINKEY << "y" << 110))); + ASSERT_FALSE(target.containsKey(BSON("x" << 2))); + ASSERT_FALSE(target.containsKey(BSON("x" << 2 << "y" << 110))); + ASSERT_FALSE(target.containsKey(BSON("x" << 5))); + ASSERT_FALSE(target.containsKey(BSON("x" << 5 << "y" << 0))); + ASSERT_TRUE(target.containsKey(BSON("x" << 5 << "y" << 120))); + ASSERT_TRUE(target.containsKey(BSON("x" << 5 << "y" << MAXKEY))); + ASSERT_TRUE(target.containsKey(BSON("x" << 5 << "y" << 100))); + ASSERT_TRUE(target.containsKey(BSON("x" << 5 << "y" << 110))); + ASSERT_TRUE(target.containsKey(BSON("x" << 7 << "y" << MINKEY))); + ASSERT_TRUE(target.containsKey(BSON("x" << 7 << "y" << 100))); + ASSERT_TRUE(target.containsKey(BSON("x" << 7 << "y" << 110))); + ASSERT_TRUE(target.containsKey(BSON("x" << 7 << "y" << 120))); + ASSERT_TRUE(target.containsKey(BSON("x" << 7 << "y" << MAXKEY))); + ASSERT_TRUE(target.containsKey(BSON("x" << 10))); + ASSERT_TRUE(target.containsKey(BSON("x" << 10 << "y" << 0))); + ASSERT_TRUE(target.containsKey(BSON("x" << 10 << "y" << 100))); + ASSERT_TRUE(target.containsKey(BSON("x" << 10 << "y" << 110))); + ASSERT_FALSE(target.containsKey(BSON("x" << 10 << "y" << 120))); + ASSERT_FALSE(target.containsKey(BSON("x" << 10 << "y" << MAXKEY))); + ASSERT_FALSE(target.containsKey(BSON("x" << 15))); + ASSERT_FALSE(target.containsKey(BSON("x" << MAXKEY))); + ASSERT_FALSE(target.containsKey(BSON("x" << MAXKEY << "y" << 0))); + ASSERT_FALSE(target.containsKey(BSON("x" << MAXKEY << "y" << 100))); + ASSERT_FALSE(target.containsKey(BSON("x" << MAXKEY << "y" << 110))); + ASSERT_FALSE(target.containsKey(BSON("x" << MAXKEY << "y" << 120))); + ASSERT_FALSE(target.containsKey(BSON("x" << MAXKEY << "y" << MAXKEY))); + + target = ChunkRange(BSON("x" << 5 << "y" << 5), BSON("x" << MAXKEY << "y" << MAXKEY)); + ASSERT_FALSE(target.containsKey(BSON("x" << MINKEY << "y" << MAXKEY))); + ASSERT_FALSE(target.containsKey(BSON("x" << 2 << "y" << MAXKEY))); + ASSERT_TRUE(target.containsKey(BSON("x" << 5 << "y" << MAXKEY))); + ASSERT_TRUE(target.containsKey(BSON("x" << MAXKEY << "y" << MAXKEY))); +} + TEST(ChunkRange, MinGreaterThanMaxShouldError) { auto parseStatus = ChunkRange::fromBSON(BSON("min" << BSON("x" << 10) << "max" << BSON("x" << 0))); diff --git a/src/mongo/s/catalog/type_collection.cpp b/src/mongo/s/catalog/type_collection.cpp index 5cc64eca8b5..320382f9e98 100644 --- a/src/mongo/s/catalog/type_collection.cpp +++ b/src/mongo/s/catalog/type_collection.cpp @@ -54,7 +54,7 @@ CollectionType::CollectionType(NamespaceString nss, std::move(creationTime), std::move(uuid), std::move(keyPattern)) { - invariant(creationTime != Timestamp(0, 0)); + invariant(getTimestamp() != Timestamp(0, 0)); setEpoch(std::move(epoch)); } diff --git a/src/mongo/s/catalog_cache.cpp b/src/mongo/s/catalog_cache.cpp index 60314d67ef3..fff547935bc 100644 --- a/src/mongo/s/catalog_cache.cpp +++ b/src/mongo/s/catalog_cache.cpp @@ -37,6 +37,8 @@ #include "mongo/s/catalog_cache.h" +#include <fmt/format.h> + #include "mongo/bson/bsonobjbuilder.h" #include "mongo/db/query/collation/collator_factory_interface.h" #include "mongo/db/repl/optime_with.h" @@ -57,6 +59,7 @@ namespace mongo { namespace { MONGO_FAIL_POINT_DEFINE(blockCollectionCacheLookup); +MONGO_FAIL_POINT_DEFINE(blockDatabaseCacheLookup); // How many times to try refreshing the routing info if the set of chunks loaded from the config // server is found to be inconsistent. @@ -68,9 +71,6 @@ const int kCollectionCacheSize = 10000; const OperationContext::Decoration<bool> operationShouldBlockBehindCatalogCacheRefresh = OperationContext::declareDecoration<bool>(); -const OperationContext::Decoration<bool> operationBlockedBehindCatalogCacheRefresh = - OperationContext::declareDecoration<bool>(); - std::shared_ptr<RoutingTableHistory> createUpdatedRoutingTableHistory( OperationContext* opCtx, const NamespaceString& nss, @@ -81,28 +81,31 @@ std::shared_ptr<RoutingTableHistory> createUpdatedRoutingTableHistory( if (isIncremental && collectionAndChunks.changedChunks.size() == 1 && collectionAndChunks.changedChunks[0].getVersion() == existingHistory->optRt->getVersion()) { - invariant(collectionAndChunks.allowMigrations == existingHistory->optRt->allowMigrations(), - str::stream() << "allowMigrations field of " << nss - << " collection changed without changing the collection version " - << existingHistory->optRt->getVersion().toString() - << ". Old value: " << existingHistory->optRt->allowMigrations() - << ", new value: " << collectionAndChunks.allowMigrations); + tassert(7032310, + fmt::format("allowMigrations field of collection '{}' changed without changing the " + "collection version {}. Old value: {}, new value: {}", + nss.toString(), + existingHistory->optRt->getVersion().toString(), + existingHistory->optRt->allowMigrations(), + collectionAndChunks.allowMigrations), + collectionAndChunks.allowMigrations == existingHistory->optRt->allowMigrations()); const auto& oldReshardingFields = existingHistory->optRt->getReshardingFields(); const auto& newReshardingFields = collectionAndChunks.reshardingFields; - invariant( - [&] { - if (oldReshardingFields && newReshardingFields) - return oldReshardingFields->toBSON().woCompare(newReshardingFields->toBSON()) == - 0; - else - return !oldReshardingFields && !newReshardingFields; - }(), - str::stream() << "reshardingFields field of " << nss - << " collection changed without changing the collection version " - << existingHistory->optRt->getVersion().toString() - << ". Old value: " << oldReshardingFields->toBSON() - << ", new value: " << newReshardingFields->toBSON()); + tassert(7032311, + fmt::format("reshardingFields field of collection '{}' changed without changing " + "the collection version {}. Old value: {}, new value: {}", + nss.toString(), + existingHistory->optRt->getVersion().toString(), + oldReshardingFields->toBSON().toString(), + newReshardingFields->toBSON().toString()), + [&] { + if (oldReshardingFields && newReshardingFields) + return oldReshardingFields->toBSON().woCompare( + newReshardingFields->toBSON()) == 0; + else + return !oldReshardingFields && !newReshardingFields; + }()); return existingHistory->optRt; } @@ -114,7 +117,11 @@ std::shared_ptr<RoutingTableHistory> createUpdatedRoutingTableHistory( return 0; } if (collectionAndChunks.maxChunkSizeBytes) { - invariant(collectionAndChunks.maxChunkSizeBytes.get() > 0); + tassert(7032312, + fmt::format("Invalid maxChunkSizeBytes value {} for collection '{}'", + nss.toString(), + collectionAndChunks.maxChunkSizeBytes.get()), + collectionAndChunks.maxChunkSizeBytes.get() > 0); return uint64_t(*collectionAndChunks.maxChunkSizeBytes); } return boost::none; @@ -242,26 +249,51 @@ CatalogCache::CatalogCache(ServiceContext* const service, CatalogCacheLoader& ca } CatalogCache::~CatalogCache() { - // The executor is used by the Database and Collection caches, - // so it must be joined, before these caches are destroyed, - // per the contract of ReadThroughCache. + // The executor is used by all the caches that correspond to the router role, so it must be + // joined before these caches are destroyed, per the contract of ReadThroughCache. + shutDownAndJoin(); +} + +void CatalogCache::shutDownAndJoin() { _executor->shutdown(); _executor->join(); } StatusWith<CachedDatabaseInfo> CatalogCache::getDatabase(OperationContext* opCtx, - StringData dbName, - bool allowLocks) { - if (!allowLocks) { - invariant( - !opCtx->lockState() || !opCtx->lockState()->isLocked(), + StringData dbName) { + return _getDatabase(opCtx, dbName); +} + +StatusWith<CachedDatabaseInfo> CatalogCache::_getDatabase(OperationContext* opCtx, + StringData dbName, + bool allowLocks) { + tassert(7032313, "Do not hold a lock while refreshing the catalog cache. Doing so would potentially " "hold the lock during a network call, and can lead to a deadlock as described in " - "SERVER-37398."); - } + "SERVER-37398.", + allowLocks || !opCtx->lockState() || !opCtx->lockState()->isLocked()); try { - auto dbEntry = _databaseCache.acquire(opCtx, dbName, CacheCausalConsistency::kLatestKnown); + auto dbEntryFuture = + _databaseCache.acquireAsync(dbName, CacheCausalConsistency::kLatestKnown); + + if (allowLocks) { + // When allowLocks is true we may be holding a lock, so we don't want to block the + // current thread: if the future is ready let's use it, otherwise return an error. + if (dbEntryFuture.isReady()) { + return dbEntryFuture.get(opCtx); + } else { + // This error only contains the database name and must be handled by any callers of + // _getDatabase with the potential for allowLocks to be true. The caller should + // convert this to ErrorCodes::ShardCannotRefreshDueToLocksHeld with the full + // namespace. + return Status{ShardCannotRefreshDueToLocksHeldInfo(NamespaceString(dbName)), + "Database info refresh did not complete"}; + } + } + + // From this point we can guarantee that allowLocks is false. + auto dbEntry = dbEntryFuture.get(opCtx); uassert(ErrorCodes::NamespaceNotFound, str::stream() << "database " << dbName << " not found", dbEntry); @@ -277,18 +309,28 @@ StatusWith<ChunkManager> CatalogCache::_getCollectionRoutingInfoAt( const NamespaceString& nss, boost::optional<Timestamp> atClusterTime, bool allowLocks) { - if (!allowLocks) { - invariant(!opCtx->lockState() || !opCtx->lockState()->isLocked(), - "Do not hold a lock while refreshing the catalog cache. Doing so would " - "potentially hold " - "the lock during a network call, and can lead to a deadlock as described in " - "SERVER-37398."); - } + tassert(7032314, + "Do not hold a lock while refreshing the catalog cache. Doing so would potentially " + "hold the lock during a network call, and can lead to a deadlock as described in " + "SERVER-37398.", + allowLocks || !opCtx->lockState() || !opCtx->lockState()->isLocked()); try { - const auto swDbInfo = getDatabase(opCtx, nss.db(), allowLocks); + const auto swDbInfo = _getDatabase(opCtx, nss.db(), allowLocks); if (!swDbInfo.isOK()) { - if (swDbInfo == ErrorCodes::NamespaceNotFound) { + if (swDbInfo == ErrorCodes::ShardCannotRefreshDueToLocksHeld) { + // Since collection refreshes always imply database refreshes, it is ok to transform + // this error into a collection error rather than a database error. + auto dbRefreshInfo = + swDbInfo.getStatus().extraInfo<ShardCannotRefreshDueToLocksHeldInfo>(); + LOGV2_DEBUG(7850500, + 2, + "Adding collection name to ShardCannotRefreshDueToLocksHeld error", + "dbName"_attr = dbRefreshInfo->getNss().db(), + "nss"_attr = nss); + return Status{ShardCannotRefreshDueToLocksHeldInfo(nss), + "Routing info refresh did not complete"}; + } else if (swDbInfo == ErrorCodes::NamespaceNotFound) { LOGV2_FOR_CATALOG_REFRESH( 4947103, 2, @@ -326,9 +368,6 @@ StatusWith<ChunkManager> CatalogCache::_getCollectionRoutingInfoAt( } // From this point we can guarantee that allowLocks is false - - operationBlockedBehindCatalogCacheRefresh(opCtx) = true; - size_t acquireTries = 0; Timer t; @@ -525,37 +564,6 @@ void CatalogCache::report(BSONObjBuilder* builder) const { _collectionCache.reportStats(&cacheStatsBuilder); } -void CatalogCache::checkAndRecordOperationBlockedByRefresh(OperationContext* opCtx, - mongo::LogicalOp opType) { - if (!isMongos() || !operationBlockedBehindCatalogCacheRefresh(opCtx)) { - return; - } - - auto& opsBlockedByRefresh = _stats.operationsBlockedByRefresh; - - opsBlockedByRefresh.countAllOperations.fetchAndAddRelaxed(1); - - switch (opType) { - case LogicalOp::opInsert: - opsBlockedByRefresh.countInserts.fetchAndAddRelaxed(1); - break; - case LogicalOp::opQuery: - opsBlockedByRefresh.countQueries.fetchAndAddRelaxed(1); - break; - case LogicalOp::opUpdate: - opsBlockedByRefresh.countUpdates.fetchAndAddRelaxed(1); - break; - case LogicalOp::opDelete: - opsBlockedByRefresh.countDeletes.fetchAndAddRelaxed(1); - break; - case LogicalOp::opCommand: - opsBlockedByRefresh.countCommands.fetchAndAddRelaxed(1); - break; - default: - MONGO_UNREACHABLE; - } -} - void CatalogCache::invalidateDatabaseEntry_LINEARIZABLE(const StringData& dbName) { _databaseCache.invalidateKey(dbName); } @@ -568,26 +576,6 @@ void CatalogCache::Stats::report(BSONObjBuilder* builder) const { builder->append("countStaleConfigErrors", countStaleConfigErrors.load()); builder->append("totalRefreshWaitTimeMicros", totalRefreshWaitTimeMicros.load()); - - if (isMongos()) { - BSONObjBuilder operationsBlockedByRefreshBuilder( - builder->subobjStart("operationsBlockedByRefresh")); - - operationsBlockedByRefreshBuilder.append( - "countAllOperations", operationsBlockedByRefresh.countAllOperations.load()); - operationsBlockedByRefreshBuilder.append("countInserts", - operationsBlockedByRefresh.countInserts.load()); - operationsBlockedByRefreshBuilder.append("countQueries", - operationsBlockedByRefresh.countQueries.load()); - operationsBlockedByRefreshBuilder.append("countUpdates", - operationsBlockedByRefresh.countUpdates.load()); - operationsBlockedByRefreshBuilder.append("countDeletes", - operationsBlockedByRefresh.countDeletes.load()); - operationsBlockedByRefreshBuilder.append("countCommands", - operationsBlockedByRefresh.countCommands.load()); - - operationsBlockedByRefreshBuilder.done(); - } } CatalogCache::DatabaseCache::DatabaseCache(ServiceContext* service, @@ -610,6 +598,10 @@ CatalogCache::DatabaseCache::LookupResult CatalogCache::DatabaseCache::_lookupDa const std::string& dbName, const DatabaseTypeValueHandle& previousDbType, const ComparableDatabaseVersion& previousDbVersion) { + if (MONGO_unlikely(blockDatabaseCacheLookup.shouldFail())) { + LOGV2(8023400, "Hanging before refreshing cached database entry"); + blockDatabaseCacheLookup.pauseWhileSet(); + } // TODO (SERVER-34164): Track and increment stats for database refreshes LOGV2_FOR_CATALOG_REFRESH(24102, 2, "Refreshing cached database entry", "db"_attr = dbName); @@ -701,7 +693,7 @@ CatalogCache::CollectionCache::LookupResult CatalogCache::CollectionCache::_look OperationContext* opCtx, const NamespaceString& nss, const RoutingTableHistoryValueHandle& existingHistory, - const ComparableChunkVersion& previousVersion) { + const ComparableChunkVersion& timeInStore) { const bool isIncremental(existingHistory && existingHistory->optRt); _updateRefreshesStats(isIncremental, true); blockCollectionCacheLookup.pauseWhileSet(opCtx); @@ -720,7 +712,7 @@ CatalogCache::CollectionCache::LookupResult CatalogCache::CollectionCache::_look "Refreshing cached collection", "namespace"_attr = nss, "lookupSinceVersion"_attr = lookupVersion, - "timeInStore"_attr = previousVersion); + "timeInStore"_attr = timeInStore); auto collectionAndChunks = _catalogCacheLoader.getChunksSince(nss, lookupVersion).get(); @@ -742,14 +734,18 @@ CatalogCache::CollectionCache::LookupResult CatalogCache::CollectionCache::_look const ChunkVersion newVersion = newRoutingHistory->getVersion(); newComparableVersion.setChunkVersion(newVersion); - LOGV2_FOR_CATALOG_REFRESH(4619901, - isIncremental || newComparableVersion != previousVersion ? 0 : 1, - "Refreshed cached collection", - "namespace"_attr = nss, - "lookupSinceVersion"_attr = lookupVersion, - "newVersion"_attr = newComparableVersion, - "timeInStore"_attr = previousVersion, - "duration"_attr = Milliseconds(t.millis())); + // The log below is logged at debug(0) (equivalent to info level) only if the new placement + // version is different than the one we already had (if any). + LOGV2_FOR_CATALOG_REFRESH( + 4619901, + (!isIncremental || newVersion != existingHistory->optRt->getVersion()) ? 0 : 1, + "Refreshed cached collection", + "namespace"_attr = nss, + "lookupSinceVersion"_attr = lookupVersion, + "newVersion"_attr = newComparableVersion, + "timeInStore"_attr = timeInStore, + "duration"_attr = Milliseconds(t.millis())); + _updateRefreshesStats(isIncremental, false); return LookupResult(OptionalRoutingTableHistory(std::move(newRoutingHistory)), diff --git a/src/mongo/s/catalog_cache.h b/src/mongo/s/catalog_cache.h index 48c781d27ad..1b40ffb5a53 100644 --- a/src/mongo/s/catalog_cache.h +++ b/src/mongo/s/catalog_cache.h @@ -140,13 +140,16 @@ public: virtual ~CatalogCache(); /** + * Shuts down and joins the executor used by all the caches to run their blocking work. + */ + void shutDownAndJoin(); + + /** * Blocking method that ensures the specified database is in the cache, loading it if necessary, * and returns it. If the database was not in cache, all the sharded collections will be in the * 'needsRefresh' state. */ - StatusWith<CachedDatabaseInfo> getDatabase(OperationContext* opCtx, - StringData dbName, - bool allowLocks = false); + StatusWith<CachedDatabaseInfo> getDatabase(OperationContext* opCtx, StringData dbName); /** * Blocking method to get the routing information for a specific collection at a given cluster @@ -254,13 +257,6 @@ public: void report(BSONObjBuilder* builder) const; /** - * Checks if the current operation was ever marked as needing refresh. If the curent operation - * was marked as needing refresh, updates the relevant counters inside the Stats struct. - */ - void checkAndRecordOperationBlockedByRefresh(OperationContext* opCtx, mongo::LogicalOp opType); - - - /** * Non-blocking method that marks the current database entry for the dbName as needing * refresh. Will cause all further targetting attempts to block on a catalog cache refresh, * even if they do not require causal consistency. @@ -336,6 +332,13 @@ private: void _updateRefreshesStats(bool isIncremental, bool add); }; + // Callers of this internal function that are passing allowLocks must handle allowLocks failures + // by checking for ErrorCodes::ShardCannotRefreshDueToLocksHeld and addint the full namespace to + // the exception. + StatusWith<CachedDatabaseInfo> _getDatabase(OperationContext* opCtx, + StringData dbName, + bool allowLocks = false); + StatusWith<ChunkManager> _getCollectionRoutingInfoAt(OperationContext* opCtx, const NamespaceString& nss, boost::optional<Timestamp> atClusterTime, @@ -363,18 +366,6 @@ private: // combined AtomicWord<long long> totalRefreshWaitTimeMicros{0}; - // Cumulative, always-increasing counter of how many operations have been blocked by a - // catalog cache refresh. Broken down by operation type to match the operations tracked - // by the OpCounters class. - struct OperationsBlockedByRefresh { - AtomicWord<long long> countAllOperations{0}; - AtomicWord<long long> countInserts{0}; - AtomicWord<long long> countQueries{0}; - AtomicWord<long long> countUpdates{0}; - AtomicWord<long long> countDeletes{0}; - AtomicWord<long long> countCommands{0}; - } operationsBlockedByRefresh; - /** * Reports the accumulated statistics for serverStatus. */ diff --git a/src/mongo/s/catalog_cache_loader.h b/src/mongo/s/catalog_cache_loader.h index 18a26324fbc..af2bdb4045b 100644 --- a/src/mongo/s/catalog_cache_loader.h +++ b/src/mongo/s/catalog_cache_loader.h @@ -126,6 +126,11 @@ public: virtual void onStepUp() = 0; /** + * Interrupts ongoing refreshes on rollback. + */ + virtual void onReplicationRollback() = 0; + + /** * Transitions into shut down and cleans up state. Once this transitions to shut down, should * not be able to transition back to normal. Should be safe to be called more than once. */ diff --git a/src/mongo/s/catalog_cache_loader_mock.cpp b/src/mongo/s/catalog_cache_loader_mock.cpp index 09b3b60e3f4..ffca2e3786e 100644 --- a/src/mongo/s/catalog_cache_loader_mock.cpp +++ b/src/mongo/s/catalog_cache_loader_mock.cpp @@ -61,6 +61,10 @@ void CatalogCacheLoaderMock::onStepUp() { MONGO_UNREACHABLE; } +void CatalogCacheLoaderMock::onReplicationRollback() { + MONGO_UNREACHABLE; +} + void CatalogCacheLoaderMock::shutDown() {} void CatalogCacheLoaderMock::notifyOfCollectionVersionUpdate(const NamespaceString& nss) { diff --git a/src/mongo/s/catalog_cache_loader_mock.h b/src/mongo/s/catalog_cache_loader_mock.h index 13538c4f69c..dd015b0258d 100644 --- a/src/mongo/s/catalog_cache_loader_mock.h +++ b/src/mongo/s/catalog_cache_loader_mock.h @@ -52,6 +52,7 @@ public: void initializeReplicaSetRole(bool isPrimary) override; void onStepDown() override; void onStepUp() override; + void onReplicationRollback() override; void shutDown() override; void notifyOfCollectionVersionUpdate(const NamespaceString& nss) override; void waitForCollectionFlush(OperationContext* opCtx, const NamespaceString& nss) override; diff --git a/src/mongo/s/catalog_cache_test.cpp b/src/mongo/s/catalog_cache_test.cpp index be608cbb68c..b894d6cdee6 100644 --- a/src/mongo/s/catalog_cache_test.cpp +++ b/src/mongo/s/catalog_cache_test.cpp @@ -36,6 +36,7 @@ #include "mongo/s/catalog/type_database_gen.h" #include "mongo/s/catalog_cache.h" #include "mongo/s/catalog_cache_loader_mock.h" +#include "mongo/s/shard_cannot_refresh_due_to_locks_held_exception.h" #include "mongo/s/sharding_router_test_fixture.h" #include "mongo/s/stale_exception.h" #include "mongo/s/type_collection_common_types_gen.h" @@ -299,6 +300,69 @@ TEST_F(CatalogCacheTest, OnStaleShardVersionWithGraterVersion) { ASSERT(status == ErrorCodes::InternalError); } +TEST_F(CatalogCacheTest, GetCollectionRoutingInfoAllowLocksReturnsImmediately) { + const auto dbVersion = DatabaseVersion(UUID::gen(), Timestamp(1, 1)); + const auto cachedCollVersion = ChunkVersion(1, 0, OID::gen(), Timestamp(1, 1)); + + loadDatabases({DatabaseType(kNss.db().toString(), kShards[0], dbVersion)}); + loadCollection(cachedCollVersion); + + const auto swCm = + _catalogCache->getCollectionRoutingInfo(operationContext(), kNss, true /* allowLocks */); + ASSERT_OK(swCm.getStatus()); + + ASSERT(swCm.getValue().getVersion() == cachedCollVersion); +} + +TEST_F(CatalogCacheTest, GetCollectionRoutingInfoAllowLocksNeedsToFetchNewCollInfo) { + const auto dbVersion = DatabaseVersion(UUID::gen(), Timestamp(1, 1)); + const auto cachedCollVersion = ChunkVersion(1, 0, OID::gen(), Timestamp(1, 1)); + const auto wantedCollVersion = + ChunkVersion(2, 0, cachedCollVersion.epoch(), cachedCollVersion.getTimestamp()); + + loadDatabases({DatabaseType(kNss.db().toString(), kShards[0], dbVersion)}); + loadCollection(cachedCollVersion); + _catalogCache->invalidateShardOrEntireCollectionEntryForShardedCollection( + kNss, wantedCollVersion, kShards[0]); + + { + FailPointEnableBlock failPoint("blockCollectionCacheLookup"); + + const auto status = + _catalogCache->getCollectionRoutingInfo(operationContext(), kNss, true /* allowLocks */) + .getStatus(); + + ASSERT(status == ErrorCodes::ShardCannotRefreshDueToLocksHeld); + auto refreshInfo = status.extraInfo<ShardCannotRefreshDueToLocksHeldInfo>(); + ASSERT(refreshInfo); + } + // Cancel ongoing refresh + _catalogCache->invalidateCollectionEntry_LINEARIZABLE(kNss); +} + +TEST_F(CatalogCacheTest, GetCollectionRoutingInfoAllowLocksNeedsToFetchNewDBInfo) { + const auto dbVersion = DatabaseVersion(UUID::gen(), Timestamp(1, 1)); + const auto cachedCollVersion = ChunkVersion(1, 0, OID::gen(), Timestamp(1, 1)); + + loadDatabases({DatabaseType(kNss.db().toString(), kShards[0], dbVersion)}); + loadCollection(cachedCollVersion); + _catalogCache->invalidateDatabaseEntry_LINEARIZABLE(kNss.db()); + + { + FailPointEnableBlock failPoint("blockDatabaseCacheLookup"); + + const auto status = + _catalogCache->getCollectionRoutingInfo(operationContext(), kNss, true /* allowLocks */) + .getStatus(); + + ASSERT(status == ErrorCodes::ShardCannotRefreshDueToLocksHeld); + auto refreshInfo = status.extraInfo<ShardCannotRefreshDueToLocksHeldInfo>(); + ASSERT(refreshInfo); + } + // Cancel ongoing refresh + _catalogCache->invalidateDatabaseEntry_LINEARIZABLE(kNss.db()); +} + TEST_F(CatalogCacheTest, TimeseriesFieldsAreProperlyPropagatedOnCC) { const auto dbVersion = DatabaseVersion(UUID::gen(), Timestamp(1, 1)); const auto epoch = OID::gen(); diff --git a/src/mongo/s/chunk.cpp b/src/mongo/s/chunk.cpp index e59c9143ecf..0566e6c4314 100644 --- a/src/mongo/s/chunk.cpp +++ b/src/mongo/s/chunk.cpp @@ -107,12 +107,15 @@ void ChunkInfo::throwIfMovedSince(const Timestamp& ts) const { } bool ChunkInfo::containsKey(const BSONObj& shardKey) const { - return getMin().woCompare(shardKey) <= 0 && shardKey.woCompare(getMax()) < 0; + return _range.containsKey(shardKey); } std::string ChunkInfo::toString() const { + auto writtenBytesStr = + _writesTracker ? std::to_string(_writesTracker->getBytesWritten()) : "null"; return str::stream() << ChunkType::shard() << ": " << _shardId << ", " << ChunkType::lastmod() - << ": " << _lastmod.toString() << ", " << _range.toString(); + << ": " << _lastmod.toString() << ", range: " << _range.toString() + << ", writtenBytes: " << writtenBytesStr; } void ChunkInfo::markAsJumbo() { diff --git a/src/mongo/s/chunk_manager.cpp b/src/mongo/s/chunk_manager.cpp index 888ba644b52..bdc69c7214d 100644 --- a/src/mongo/s/chunk_manager.cpp +++ b/src/mongo/s/chunk_manager.cpp @@ -33,6 +33,8 @@ #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" @@ -63,18 +65,42 @@ void checkAllElementsAreOfType(BSONType type, const BSONObj& o) { allElementsAreOfType(type, o)); } -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); - } +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()); } else { - chunks.push_back(chunk); + uasserted(ErrorCodes::ConflictingOperationInProgress, + str::stream() << "Overlap exists in the routing table between chunks " + << left.getRange().toString() << " and " + << right.getRange().toString()); } + + 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 @@ -99,10 +125,18 @@ std::vector<std::shared_ptr<ChunkInfo>> flatten(const std::vector<ChunkType>& ch std::vector<std::shared_ptr<ChunkInfo>> flattened; flattened.reserve(changedChunkInfos.size()); - flattened.push_back(changedChunkInfos[0]); + flattened.emplace_back(std::move(changedChunkInfos[0])); for (size_t i = 1; i < changedChunkInfos.size(); ++i) { - appendChunkTo(flattened, changedChunkInfos[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)); + } } std::reverse(flattened.begin(), flattened.end()); @@ -110,205 +144,452 @@ std::vector<std::shared_ptr<ChunkInfo>> flatten(const std::vector<ChunkType>& ch return flattened; } -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()); +} // namespace - 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())); +size_t ChunkMap::size() const { + size_t totalChunks{0}; + for (const auto& mapIt : _chunkVectorMap) { + totalChunks += mapIt.second->size(); + } + return totalChunks; } -} // namespace - -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; +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 {}; } + } - 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()); - } + const auto& chunkVector = *(it->second); + const auto chunkIt = _findIntersectingChunkIterator( + shardKeyString, chunkVector.begin(), chunkVector.end(), true /*isMaxInclusive*/); - if (!firstMin) - firstMin = rangeMin; + if (chunkIt == chunkVector.end()) { + return {}; + } - lastMax = rangeMax; + return *chunkIt; +} - // 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()); - } +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; +} - if (!_chunkMap.empty()) { - invariant(!shardVersions.empty()); - invariant(firstMin.is_initialized()); - invariant(lastMax.is_initialized()); +void ChunkMap::_commitUpdatedChunkVector(std::shared_ptr<ChunkVector>&& chunkVectorPtr, + bool checkMaxKeyConsistency) { - checkAllElementsAreOfType(MinKey, firstMin.get()); - checkAllElementsAreOfType(MaxKey, lastMax.get()); + invariant(!chunkVectorPtr->empty()); + + const auto& vectorMaxKeyString = chunkVectorPtr->back()->getMaxKeyString(); + const auto nextMapIt = _chunkVectorMap.lower_bound(vectorMaxKeyString); + + // Check lower bound is consistent + if (nextMapIt == _chunkVectorMap.begin()) { + checkAllElementsAreOfType(MinKey, chunkVectorPtr->front()->getMin()); + } else { + checkChunksAreContiguous(*(std::prev(nextMapIt)->second->back()), + *(chunkVectorPtr->front())); + } + + if (checkMaxKeyConsistency) { + // Check upper bound is consistent + if (nextMapIt == _chunkVectorMap.end()) { + checkAllElementsAreOfType(MaxKey, chunkVectorPtr->back()->getMax()); + } else { + checkChunksAreContiguous(*(chunkVectorPtr->back()), *(nextMapIt->second->front())); + } } - return shardVersions; + auto minVectorSize = _maxChunkVectorSize / 2; + if (chunkVectorPtr->size() < minVectorSize) { + _mergeAndCommitUpdatedChunkVector(nextMapIt, std::move(chunkVectorPtr)); + } else { + _splitAndCommitUpdatedChunkVector(nextMapIt, std::move(chunkVectorPtr)); + } } -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::_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)); + + return; } + + 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 oldVector + mergeVectorPtr->insert(mergeVectorPtr->end(), + std::make_move_iterator(prevVectorPtr->begin()), + std::make_move_iterator(prevVectorPtr->end())); + mergeVectorPtr->insert(mergeVectorPtr->end(), + std::make_move_iterator(smallVectorPtr->begin()), + std::make_move_iterator(smallVectorPtr->end())); + _chunkVectorMap.emplace_hint( + pos, mergeVectorPtr->back()->getMaxKeyString(), std::move(mergeVectorPtr)); } -std::shared_ptr<ChunkInfo> ChunkMap::findIntersectingChunk(const BSONObj& shardKey) const { - const auto it = _findIntersectingChunk(shardKey); +/* + * 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)); + } - if (it != _chunkMap.end()) - return *it; + 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 std::shared_ptr<ChunkInfo>(); +void ChunkMap::_updateShardVersionFromDiscardedChunk(const ChunkInfo& chunk) { + auto shardVersionIt = _shardVersions.find(chunk.getShardId()); + if (shardVersionIt != _shardVersions.end() && + shardVersionIt->second.shardVersion == chunk.getLastmod()) { + _shardVersions.erase(shardVersionIt); + } } -ChunkMap ChunkMap::createMerged( - const std::vector<std::shared_ptr<ChunkInfo>>& changedChunks) const { - size_t chunkMapIndex = 0; - size_t changedChunkIndex = 0; +void ChunkMap::_updateShardVersionFromUpdateChunk(const ChunkInfo& chunk) { + auto [shardVersionIt, created] = _shardVersions.try_emplace( + chunk.getShardId(), _collectionVersion.epoch(), _collectionVersion.getTimestamp()); + + if (created || shardVersionIt->second.shardVersion.isOlderThan(chunk.getLastmod())) { + const auto& newVersion = chunk.getLastmod(); + // Update shard version with the most recent one from new chunk + shardVersionIt->second.shardVersion = newVersion; + if (_collectionVersion.isOlderThan(newVersion)) { + _collectionVersion = ChunkVersion{newVersion.majorVersion(), + newVersion.minorVersion(), + newVersion.epoch(), + _collectionVersion.getTimestamp()}; + } + } +} - ChunkMap updatedChunkMap( - getVersion().epoch(), getVersion().getTimestamp(), _chunkMap.size() + changedChunks.size()); +ChunkMap ChunkMap::_makeUpdated(ChunkVector&& updateChunks) const { + ChunkMap newMap{*this}; - while (chunkMapIndex < _chunkMap.size() || changedChunkIndex < changedChunks.size()) { - if (chunkMapIndex >= _chunkMap.size()) { - validateChunkIsNotOlderThan(changedChunks[changedChunkIndex], getVersion()); - updatedChunkMap.appendChunk(changedChunks[changedChunkIndex++]); - continue; + if (updateChunks.empty()) { + // No updates, just clone the original map + return newMap; + } + + 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); + 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; } - if (changedChunkIndex >= changedChunks.size()) { - updatedChunkMap.appendChunk(_chunkMap[chunkMapIndex++]); - 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; + } } + }; + + 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(); + } - auto overlap = _chunkMap[chunkMapIndex]->getRange().overlaps( - changedChunks[changedChunkIndex]->getRange()); + if (updateChunkIt == updateChunks.end()) { + // No more updates skip all remaining vectors + return newMap._chunkVectorMap.end(); + } - if (overlap) { - auto& changedChunk = changedChunks[changedChunkIndex++]; - auto& chunkInfo = _chunkMap[chunkMapIndex]; + if (newVectorPtr->size() < _maxChunkVectorSize / 2) { + // New vector is too small, keep accumulating next oldVector + return followingMapIt; + } - auto bytesInReplacedChunk = chunkInfo->getWritesTracker()->getBytesWritten(); - changedChunk->getWritesTracker()->addBytesWritten(bytesInReplacedChunk); + // 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>(); + } - validateChunkIsNotOlderThan(changedChunk, getVersion()); - updatedChunkMap.appendChunk(changedChunk); - } else { - updatedChunkMap.appendChunk(_chunkMap[chunkMapIndex++]); + 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())))); + } + } } } - return updatedChunkMap; + if (!newVectorPtr->empty()) { + newMap._commitUpdatedChunkVector(std::move(newVectorPtr), true); + } + + return newMap; } BSONObj ChunkMap::toBSON() const { BSONObjBuilder builder; getVersion().serializeToBSON("startingVersion"_sd, &builder); - builder.append("chunkCount", static_cast<int64_t>(_chunkMap.size())); + builder.append("chunkCount", static_cast<int64_t>(size())); { BSONArrayBuilder arrayBuilder(builder.subarrayStart("chunks"_sd)); - for (const auto& chunk : _chunkMap) { - arrayBuilder.append(chunk->toString()); + for (const auto& mapIt : _chunkVectorMap) { + for (const auto& chunkInfoPtr : *mapIt.second) { + arrayBuilder.append(chunkInfoPtr->toString()); + } } } return builder.obj(); } -ChunkMap::ChunkVector::const_iterator ChunkMap::_findIntersectingChunk(const BSONObj& shardKey, - bool isMaxInclusive) const { - auto shardKeyString = ShardKeyPattern::toKeyString(shardKey); +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 { if (!isMaxInclusive) { - return std::lower_bound(_chunkMap.begin(), - _chunkMap.end(), - shardKey, - [&shardKeyString](const auto& chunkInfo, const BSONObj& shardKey) { + return std::lower_bound(first, + last, + shardKeyString, + [&](const auto& chunkInfo, const std::string& shardKeyString) { return chunkInfo->getMaxKeyString() < shardKeyString; }); } else { - return std::upper_bound(_chunkMap.begin(), - _chunkMap.end(), - shardKey, - [&shardKeyString](const BSONObj& shardKey, const auto& chunkInfo) { + return std::upper_bound(first, + last, + shardKeyString, + [&](const std::string& shardKeyString, const auto& chunkInfo) { return shardKeyString < chunkInfo->getMaxKeyString(); }); } } -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); + +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); const auto itMax = [&]() { - auto it = _findIntersectingChunk(max, isMaxInclusive); - return it == _chunkMap.end() ? it : ++it; + auto it = isMaxInclusive ? _chunkVectorMap.upper_bound(maxShardKeyStr) + : _chunkVectorMap.lower_bound(maxShardKeyStr); + + return it == _chunkVectorMap.end() ? it : ++it; }(); return {itMin, itMax}; @@ -338,7 +619,7 @@ RoutingTableHistory::RoutingTableHistory( _maxChunkSizeBytes(maxChunkSizeBytes), _allowMigrations(allowMigrations), _chunkMap(std::move(chunkMap)), - _shardVersions(_chunkMap.constructShardVersionMap()) {} + _shardVersions(_chunkMap.getShardVersionsMap()) {} void RoutingTableHistory::setShardStale(const ShardId& shardId) { if (gEnableFinerGrainedCatalogCacheRefresh) { @@ -388,7 +669,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->containsKey(shardKey)); + chunkInfo); return Chunk(*chunkInfo, _clusterTime); } @@ -401,15 +682,14 @@ 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) const { + std::set<ShardId>* shardIds, + bool bypassIsFieldHashedCheck) const { auto findCommand = std::make_unique<FindCommandRequest>(_rt->optRt->nss()); findCommand->setFilter(query.getOwned()); @@ -435,7 +715,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); + auto chunk = findIntersectingChunk(shardKeyToFind, collation, bypassIsFieldHashedCheck); shardIds->insert(chunk.getShardId()); return; } catch (const DBException&) { @@ -475,7 +755,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 std::shared_ptr<ChunkInfo>& chunkInfo) { + _rt->optRt->forEachChunk([&](const auto& chunkInfo) { shardIds->insert(chunkInfo->getShardIdAt(_clusterTime)); return false; }); @@ -495,7 +775,7 @@ void ChunkManager::getShardIdsForRange(const BSONObj& min, return; } - _rt->optRt->forEachOverlappingChunk(min, max, true, [&](auto& chunkInfo) { + _rt->optRt->forEachOverlappingChunk(min, max, true, [&](const 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 @@ -515,7 +795,7 @@ bool ChunkManager::rangeOverlapsShard(const ChunkRange& range, const ShardId& sh bool overlapFound = false; _rt->optRt->forEachOverlappingChunk( - range.getMin(), range.getMax(), false, [&](auto& chunkInfo) { + range.getMin(), range.getMax(), false, [&](const auto& chunkInfo) { if (chunkInfo->getShardIdAt(_clusterTime) == shardId) { overlapFound = true; return false; @@ -532,7 +812,7 @@ boost::optional<Chunk> ChunkManager::getNextChunkOnShard(const BSONObj& shardKey boost::optional<Chunk> chunk; _rt->optRt->forEachChunk( - [&](auto& chunkInfo) { + [&](const auto& chunkInfo) { if (chunkInfo->getShardIdAt(_clusterTime) == shardId) { chunk.emplace(*chunkInfo, _clusterTime); return false; @@ -772,11 +1052,7 @@ std::string RoutingTableHistory::toString() const { StringBuilder sb; sb << "RoutingTableHistory: " << _nss.ns() << " key: " << _shardKeyPattern.toString() << '\n'; - sb << "Chunks:\n"; - _chunkMap.forEach([&sb](const auto& chunk) { - sb << "\t" << chunk->toString() << '\n'; - return true; - }); + sb << _chunkMap.toString(); sb << "Shard versions:\n"; for (const auto& entry : _shardVersions) { @@ -801,16 +1077,19 @@ 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), - maxChunkSizeBytes, - allowMigrations, - ChunkMap{epoch, timestamp}.createMerged(changedChunkInfos)); + + 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))); } // Note that any new parameters added to RoutingTableHistory::makeUpdated() must also be added to @@ -824,7 +1103,7 @@ RoutingTableHistory RoutingTableHistory::makeUpdated( const std::vector<ChunkType>& changedChunks) const { auto changedChunkInfos = flatten(changedChunks); - auto chunkMap = _chunkMap.createMerged(changedChunkInfos); + auto chunkMap = _chunkMap.createMerged(std::move(changedChunkInfos)); // Only update the same collection. invariant(getVersion().isSameCollection(chunkMap.getVersion())); diff --git a/src/mongo/s/chunk_manager.h b/src/mongo/s/chunk_manager.h index 00c75957d37..2bd28a2386b 100644 --- a/src/mongo/s/chunk_manager.h +++ b/src/mongo/s/chunk_manager.h @@ -41,7 +41,6 @@ #include "mongo/s/resharding/type_collection_fields_gen.h" #include "mongo/s/shard_key_pattern.h" #include "mongo/s/type_collection_common_types_gen.h" -#include "mongo/stdx/unordered_map.h" #include "mongo/util/concurrency/ticketholder.h" #include "mongo/util/read_through_cache.h" @@ -58,6 +57,9 @@ struct ShardVersionTargetingInfo { // Max chunk version for the shard ChunkVersion shardVersion; + ShardVersionTargetingInfo(const ShardVersionTargetingInfo& info) + : isStale(info.isStale.load()), shardVersion(info.shardVersion) {} + ShardVersionTargetingInfo(const OID& epoch, const Timestamp& timestamp); }; @@ -71,73 +73,173 @@ using ShardVersionMap = stdx::unordered_map<ShardId, ShardVersionTargetingInfo, * underlying implementation. */ class ChunkMap { - // Vector of chunks ordered by max key. +public: + // Vector of chunks ordered by max key in ascending order. using ChunkVector = std::vector<std::shared_ptr<ChunkInfo>>; + using ChunkVectorMap = std::map<std::string, std::shared_ptr<ChunkVector>>; -public: - explicit ChunkMap(OID epoch, const Timestamp& timestamp, size_t initialCapacity = 0) - : _collectionVersion(0, 0, epoch, timestamp), _collTimestamp(timestamp) { - _chunkMap.reserve(initialCapacity); - } + explicit ChunkMap(OID epoch, const Timestamp& timestamp, size_t chunkVectorSize) + : _collectionVersion(0, 0, epoch, timestamp), + _collTimestamp(timestamp), + _maxChunkVectorSize(chunkVectorSize) {} - size_t size() const { - return _chunkMap.size(); - } + size_t size() const; + // Max version across all chunks ChunkVersion getVersion() const { return _collectionVersion; } + size_t getMaxChunkVectorSize() const { + return _maxChunkVectorSize; + } + + const ShardVersionMap& getShardVersionsMap() const { + return _shardVersions; + } + + const ChunkVectorMap& getChunkVectorMap() const { + return _chunkVectorMap; + } + + + /* + * Invoke the given handler for each std::shared_ptr<ChunkInfo> contained in this chunk map + * until either all matching chunks have been processed or @handler returns false. + * + * Chunks are yielded in ascending order of shardkey (e.g. minKey to maxKey); + * + * When shardKey is provided the function will start yileding from the chunk that contains the + * given shard key. + */ template <typename Callable> void forEach(Callable&& handler, const BSONObj& shardKey = BSONObj()) const { - auto it = shardKey.isEmpty() ? _chunkMap.begin() : _findIntersectingChunk(shardKey); + if (shardKey.isEmpty()) { + for (const auto& mapIt : _chunkVectorMap) { + for (const auto& chunkInfoPtr : *(mapIt.second)) { + if (!handler(chunkInfoPtr)) + return; + } + } + + return; + } - for (; it != _chunkMap.end(); ++it) { - if (!handler(*it)) - break; + auto shardKeyString = ShardKeyPattern::toKeyString(shardKey); + + const auto mapItBegin = _chunkVectorMap.upper_bound(shardKeyString); + for (auto mapIt = mapItBegin; mapIt != _chunkVectorMap.end(); mapIt++) { + const auto& chunkVector = *(mapIt->second); + auto it = mapIt == mapItBegin ? _findIntersectingChunkIterator(shardKeyString, + chunkVector.begin(), + chunkVector.end(), + true /*isMaxInclusive*/) + : chunkVector.begin(); + for (; it != chunkVector.end(); ++it) { + if (!handler(*it)) + return; + } } } + + /* + * Invoke the given @handler for each std::shared_ptr<ChunkInfo> that overlaps with range [@min, + * @max] until either all matching chunks have been processed or @handler returns false. + * + * Chunks are yielded in ascending order of shardkey (e.g. minKey to maxKey); + * + * When @isMaxInclusive is true also the chunk whose minKey is equal to @max will be yielded. + */ template <typename Callable> void forEachOverlappingChunk(const BSONObj& min, const BSONObj& max, bool isMaxInclusive, Callable&& handler) const { - const auto bounds = _overlappingBounds(min, max, isMaxInclusive); - - for (auto it = bounds.first; it != bounds.second; ++it) { - if (!handler(*it)) - break; + const auto minShardKeyStr = ShardKeyPattern::toKeyString(min); + const auto maxShardKeyStr = ShardKeyPattern::toKeyString(max); + const auto bounds = + _overlappingVectorSlotBounds(minShardKeyStr, maxShardKeyStr, isMaxInclusive); + for (auto mapIt = bounds.first; mapIt != bounds.second; ++mapIt) { + + const auto& chunkVector = *(mapIt->second); + + const auto chunkItBegin = [&] { + if (mapIt == bounds.first) { + // On first vector we need to start from chunk that contain the given minKey + return _findIntersectingChunkIterator(minShardKeyStr, + chunkVector.begin(), + chunkVector.end(), + true /* isMaxInclusive */); + } + return chunkVector.begin(); + }(); + + const auto chunkItEnd = [&] { + if (mapIt == std::prev(bounds.second)) { + // On last vector we need to skip all chunks that are greater than the give + // maxKey + auto it = _findIntersectingChunkIterator( + maxShardKeyStr, chunkItBegin, chunkVector.end(), isMaxInclusive); + return it == chunkVector.end() ? it : ++it; + } + return chunkVector.end(); + }(); + + for (auto chunkIt = chunkItBegin; chunkIt != chunkItEnd; ++chunkIt) { + if (!handler(*chunkIt)) + return; + } } } - ShardVersionMap constructShardVersionMap() const; std::shared_ptr<ChunkInfo> findIntersectingChunk(const BSONObj& shardKey) const; - void appendChunk(const std::shared_ptr<ChunkInfo>& chunk); - - ChunkMap createMerged(const std::vector<std::shared_ptr<ChunkInfo>>& changedChunks) const; + ChunkMap createMerged(ChunkVector changedChunks) const; BSONObj toBSON() const; -private: - ChunkVector::const_iterator _findIntersectingChunk(const BSONObj& shardKey, - bool isMaxInclusive = true) const; - std::pair<ChunkVector::const_iterator, ChunkVector::const_iterator> _overlappingBounds( - const BSONObj& min, const BSONObj& max, bool isMaxInclusive) const; + std::string toString() const; - ChunkVector _chunkMap; +private: + ChunkVector::const_iterator _findIntersectingChunkIterator(const std::string& shardKeyString, + ChunkVector::const_iterator first, + ChunkVector::const_iterator last, + bool isMaxInclusive) const; + + std::pair<ChunkVectorMap::const_iterator, ChunkVectorMap::const_iterator> + _overlappingVectorSlotBounds(const std::string& minShardKeyStr, + const std::string& maxShardKeyStr, + bool isMaxInclusive) const; + ChunkMap _makeUpdated(ChunkVector&& changedChunks) const; + + void _updateShardVersionFromDiscardedChunk(const ChunkInfo& chunk); + void _updateShardVersionFromUpdateChunk(const ChunkInfo& chunk); + void _commitUpdatedChunkVector(std::shared_ptr<ChunkVector>&& chunkVectorPtr, + bool checkMaxKeyConsistency); + void _mergeAndCommitUpdatedChunkVector(ChunkVectorMap::const_iterator pos, + std::shared_ptr<ChunkVector>&& chunkVectorPtr); + void _splitAndCommitUpdatedChunkVector(ChunkVectorMap::const_iterator pos, + std::shared_ptr<ChunkVector>&& chunkVectorPtr); + + ChunkVectorMap _chunkVectorMap; // Max version across all chunks ChunkVersion _collectionVersion; + // The representation of shard versions and staleness indicators for this namespace. If a + // shard does not exist, it will not have an entry in the map. + // Note: this declaration must not be moved before _chunkMap since it is initialized by using + // the _chunkVectorMap instance. + ShardVersionMap _shardVersions; + // Represents the timestamp present in config.collections for this ChunkMap. - // - // Note that due to the way Phase 1 of the FCV upgrade writes timestamps to chunks - // (non-atomically), it is possible that chunks exist with timestamps, but the corresponding - // config.collections entry doesn't. In this case, the chunks timestamp should be ignored when - // computing the collection version and we should use _collTimestamp instead. Timestamp _collTimestamp; + + // Maximum size of chunk vectors stored in the chunk vector map. + // Bigger vectors will imply slower incremental refreshes (more chunks to copy) but + // faster map copy (less chunk vector pointers to copy). + size_t _maxChunkVectorSize; }; /** @@ -228,6 +330,7 @@ public: */ void setAllShardsRefreshed(); + // Max version across all chunks ChunkVersion getVersion() const { return _chunkMap.getVersion(); } @@ -625,12 +728,15 @@ public: /** * Finds the shard IDs for a given filter and collation. If collation is empty, we use the - * collection default collation for targeting. + * collection default collation for targeting. If 'bypassIsFieldHashedCheck' is true, it skips + * checking if the shard key was hashed and assumes that any non-collatable shard key was not + * hashed from a collatable type. */ void getShardIdsForQuery(boost::intrusive_ptr<ExpressionContext> expCtx, const BSONObj& query, const BSONObj& collation, - std::set<ShardId>* shardIds) const; + std::set<ShardId>* shardIds, + bool bypassIsFieldHashedCheck = false) const; /** * Returns all shard ids which contain chunks overlapping the range [min, max]. Please note the diff --git a/src/mongo/s/chunk_manager_refresh_bm.cpp b/src/mongo/s/chunk_manager_refresh_bm.cpp index 3c7f3adb6b3..c3577944fab 100644 --- a/src/mongo/s/chunk_manager_refresh_bm.cpp +++ b/src/mongo/s/chunk_manager_refresh_bm.cpp @@ -50,17 +50,17 @@ RoutingTableHistoryValueHandle makeStandaloneRoutingTableHistory(RoutingTableHis ComparableChunkVersion::makeComparableChunkVersion(version)); } +ShardId getShardId(int i) { + return {std::string(str::stream() << "shard_" << i)}; +} + ChunkRange getRangeForChunk(int i, int nChunks) { invariant(i >= 0); invariant(nChunks > 0); invariant(i < nChunks); - if (i == 0) { - return {BSON("_id" << MINKEY), BSON("_id" << 0)}; - } - if (i + 1 == nChunks) { - return {BSON("_id" << (i - 1) * 100), BSON("_id" << MAXKEY)}; - } - return {BSON("_id" << (i - 1) * 100), BSON("_id" << i * 100)}; + auto min = (i == 0) ? BSON("_id" << MINKEY) : BSON("_id" << (i - 1) * 100); + auto max = (i == nChunks - 1) ? BSON("_id" << MAXKEY) : BSON("_id" << i * 100); + return {std::move(min), std::move(max)}; } template <typename ShardSelectorFn> @@ -93,21 +93,21 @@ CollectionMetadata makeChunkManagerWithShardSelector(int nShards, boost::none /* chunkSizeBytes */, true, chunks); - return CollectionMetadata(ChunkManager(ShardId("Shard0"), + return CollectionMetadata(ChunkManager(getShardId(0), DatabaseVersion(UUID::gen(), Timestamp(1, 0)), makeStandaloneRoutingTableHistory(std::move(rt)), boost::none), - ShardId("shard0")); + getShardId(0)); } ShardId pessimalShardSelector(int i, int nShards, int nChunks) { - return ShardId(str::stream() << "shard" << (i % nShards)); + return getShardId(i % nShards); } ShardId optimalShardSelector(int i, int nShards, int nChunks) { invariant(nShards <= nChunks); const auto shardNum = (int64_t(i) * nShards / nChunks) % nShards; - return ShardId(str::stream() << "shard" << shardNum); + return getShardId(shardNum); } MONGO_COMPILER_NOINLINE auto makeChunkManagerWithPessimalBalancedDistribution(int nShards, @@ -124,35 +124,133 @@ MONGO_COMPILER_NOINLINE auto runIncrementalUpdate(const CollectionMetadata& cm, const std::vector<ChunkType>& newChunks) { auto rt = cm.getChunkManager()->getRoutingTableHistory_ForTest().makeUpdated( boost::none /* timeseriesFields */, boost::none, boost::none, true, newChunks); - return CollectionMetadata(ChunkManager(ShardId("shard0"), + return CollectionMetadata(ChunkManager(getShardId(0), DatabaseVersion(UUID::gen(), Timestamp(1, 0)), makeStandaloneRoutingTableHistory(std::move(rt)), boost::none), - ShardId("shard0")); + getShardId(0)); } -void BM_IncrementalRefreshOfPessimalBalancedDistribution(benchmark::State& state) { +/* + * Simulate a refresh of the ChunkManager where a number of chunks is migrated from one shard to + * another. + * + * The chunks modified in the routing table are equally spaced. + */ +void BM_IncrementalSpacedRefreshMoveChunks(benchmark::State& state) { const int nShards = state.range(0); const int nChunks = state.range(1); + const int nUpdates = state.range(2); auto metadata = makeChunkManagerWithPessimalBalancedDistribution(nShards, nChunks); - auto postMoveVersion = metadata.getChunkManager()->getVersion(); - const UUID uuid = metadata.getUUID(); + auto lastVersion = metadata.getCollVersion(); + std::vector<ChunkType> newChunks; - postMoveVersion.incMajor(); - newChunks.emplace_back(uuid, getRangeForChunk(1, nChunks), postMoveVersion, ShardId("shard0")); - postMoveVersion.incMajor(); - newChunks.emplace_back(uuid, getRangeForChunk(3, nChunks), postMoveVersion, ShardId("shard1")); + newChunks.reserve(nUpdates); + const auto updateSpacing = nChunks / nUpdates; + for (int i = 0; i < nUpdates; i++) { + const auto idx = i * updateSpacing; + lastVersion.incMajor(); + newChunks.emplace_back(metadata.getUUID(), + getRangeForChunk(idx, nChunks), + lastVersion, + pessimalShardSelector(idx, nShards, nChunks)); + } - for (auto keepRunning : state) { + std::mt19937 g; + g.seed(456); + std::shuffle(newChunks.begin(), newChunks.end(), g); + + for (auto _ : state) { benchmark::DoNotOptimize(runIncrementalUpdate(metadata, newChunks)); } } -BENCHMARK(BM_IncrementalRefreshOfPessimalBalancedDistribution) - ->Args({2, 50000}) - ->Args({2, 250000}) - ->Args({2, 500000}); +BENCHMARK(BM_IncrementalSpacedRefreshMoveChunks) + ->Args({4, 1, 1}) + ->Args({4, 10, 1}) + ->Args({4, 100, 1}) + ->Args({4, 1000, 1}) + ->Args({4, 10000, 1}) + ->Args({4, 100000, 1}) + ->Args({4, 10000, 10}) + ->Args({4, 10000, 100}) + ->Args({4, 10000, 1000}) + ->Args({4, 10000, 10000}); + +/* + * Simulate a refresh of the ChunkManager where a number of chunks is merged together. + */ +void BM_IncrementalSpacedRefreshMergeChunks(benchmark::State& state) { + const int nShards = state.range(0); + const int nChunks = state.range(1); + const int nUpdates = state.range(2); + auto metadata = makeChunkManagerWithOptimalBalancedDistribution(nShards, nChunks); + + auto lastVersion = metadata.getCollVersion(); + + std::vector<ChunkType> newChunks; + newChunks.reserve(nUpdates); + invariant(nUpdates <= nShards); + const auto shardSpacing = nShards / (nUpdates + 1); + std::set<ShardId> shardsToMerge; + for (int i = 0; i < nUpdates; i++) { + invariant(i * shardSpacing <= nShards); + shardsToMerge.emplace(getShardId(i * shardSpacing)); + } + + ShardId shardId; + std::vector<ChunkRange> rangesToMerge; + + const auto flushRanges = [&] { + if (rangesToMerge.empty()) { + return; + } + + lastVersion.incMajor(); + newChunks.emplace_back( + metadata.getUUID(), + ChunkRange(rangesToMerge.front().getMin(), rangesToMerge.back().getMax()), + lastVersion, + shardId); + rangesToMerge.clear(); + }; + + for (int i = 0; i < nChunks; i++) { + auto nextShardId = pessimalShardSelector(i, nShards, nChunks); + if (nextShardId != shardId) { + flushRanges(); + shardId = nextShardId; + } + if (shardsToMerge.count(shardId) == 1) { + rangesToMerge.emplace_back(getRangeForChunk(i, nChunks)); + } + } + flushRanges(); + + std::mt19937 g; + g.seed(456); + std::shuffle(newChunks.begin(), newChunks.end(), g); + + for (auto _ : state) { + benchmark::DoNotOptimize(runIncrementalUpdate(metadata, newChunks)); + } +} + +/* + * Simulate chunks merge on a routing table of 10000 chunks partitioned among 4 shards. + * + * [ 0, 2500) -> shard0 + * [2500, 5000) -> shard1 + * [5000, 7500) -> shard2 + * [7500, 10000) -> shard3 + */ + +BENCHMARK(BM_IncrementalSpacedRefreshMergeChunks) + ->Args({4, 10000, 1}) // merge all chunks on shard2 + ->Args({4, 10000, 2}) // merge all chunks on shard2 and shard3 + ->Args({4, 10000, 3}) // merge all chunks on shard1, shard2 and shard3 + ->Args({4, 10000, 4}); // merge all chunks on shard1, shard2, shard3 and shard4 template <typename ShardSelectorFn> auto BM_FullBuildOfChunkManager(benchmark::State& state, ShardSelectorFn selectShard) { @@ -187,11 +285,11 @@ auto BM_FullBuildOfChunkManager(benchmark::State& state, ShardSelectorFn selectS true, chunks); benchmark::DoNotOptimize( - CollectionMetadata(ChunkManager(ShardId("shard0"), + CollectionMetadata(ChunkManager(getShardId(0), DatabaseVersion(UUID::gen(), Timestamp(1, 0)), makeStandaloneRoutingTableHistory(std::move(rt)), boost::none), - ShardId("shard0"))); + getShardId(0))); } } @@ -413,11 +511,14 @@ MONGO_INITIALIZER(RegisterBenchmarks)(InitializerContext* context) { }; for (auto bmCase : bmCases) { - bmCase->Args({2, 50000}) - ->Args({10, 50000}) - ->Args({100, 50000}) - ->Args({1000, 50000}) - ->Args({2, 2}); + bmCase->Args({2, 2}) + ->Args({1, 10000}) + ->Args({10, 10000}) + ->Args({100, 10000}) + ->Args({1000, 10000}) + ->Args({10, 10}) + ->Args({10, 100}) + ->Args({10, 1000}); } } diff --git a/src/mongo/s/chunk_manager_targeter.cpp b/src/mongo/s/chunk_manager_targeter.cpp index bdb386420c6..7858dd0b752 100644 --- a/src/mongo/s/chunk_manager_targeter.cpp +++ b/src/mongo/s/chunk_manager_targeter.cpp @@ -43,6 +43,7 @@ #include "mongo/db/query/canonical_query.h" #include "mongo/db/query/collation/collation_index_key.h" #include "mongo/db/query/collation/collator_factory_interface.h" +#include "mongo/db/stats/counters.h" #include "mongo/db/timeseries/timeseries_constants.h" #include "mongo/db/timeseries/timeseries_options.h" #include "mongo/db/timeseries/timeseries_update_delete_util.h" @@ -386,6 +387,10 @@ std::vector<ShardEndpoint> ChunkManagerTargeter::targetUpdate(OperationContext* const auto& updateOp = itemRef.getUpdate(); + if (updateOp.getMulti()) { + updateManyCount.increment(1); + } + // If the collection is not sharded, forward the update to the primary shard. if (!_cm.isSharded()) { // TODO (SERVER-51070): Remove the boost::none when the config server can support @@ -502,8 +507,11 @@ std::vector<ShardEndpoint> ChunkManagerTargeter::targetDelete(OperationContext* itemRef.getLet(), itemRef.getLegacyRuntimeConstants()); + if (deleteOp.getMulti()) { + deleteManyCount.increment(1); + } + BSONObj deleteQuery = deleteOp.getQ(); - BSONObj shardKey; if (_cm.isSharded()) { if (_isRequestOnTimeseriesViewNamespace) { uassert(ErrorCodes::NotImplemented, @@ -529,20 +537,13 @@ std::vector<ShardEndpoint> ChunkManagerTargeter::targetDelete(OperationContext* deleteQuery = BSONObj(); } } - - // Sharded collections have the following further requirements for targeting: - // - // Limit-1 deletes must be targeted exactly by shard key *or* exact _id - shardKey = - uassertStatusOK(_cm.getShardKeyPattern().extractShardKeyFromQuery(expCtx, deleteQuery)); } - // Target the shard key or delete query - if (!shardKey.isEmpty()) { - auto swEndpoint = _targetShardKey(shardKey, collation); - if (swEndpoint.isOK()) { - return std::vector{std::move(swEndpoint.getValue())}; - } + // We first try to target based on the delete's query. It is always valid to forward any delete + // to a single shard, so return immediately if we are able to target a single shard. + auto endpoints = uassertStatusOK(_targetQuery(expCtx, deleteQuery, collation)); + if (endpoints.size() == 1) { + return endpoints; } // We failed to target a single shard. @@ -571,7 +572,7 @@ std::vector<ShardEndpoint> ChunkManagerTargeter::targetDelete(OperationContext* << ", shard key pattern: " << _cm.getShardKeyPattern().toString(), !_cm.isSharded() || deleteOp.getMulti() || isExactIdQuery(opCtx, *cq, _cm)); - return uassertStatusOK(_targetQuery(expCtx, deleteQuery, collation)); + return endpoints; } StatusWith<std::vector<ShardEndpoint>> ChunkManagerTargeter::_targetQuery( diff --git a/src/mongo/s/chunk_manager_targeter_test.cpp b/src/mongo/s/chunk_manager_targeter_test.cpp index 14070303a0f..f7e3642579e 100644 --- a/src/mongo/s/chunk_manager_targeter_test.cpp +++ b/src/mongo/s/chunk_manager_targeter_test.cpp @@ -303,22 +303,29 @@ TEST_F(ChunkManagerTargeterTest, TargetDeleteWithRangePrefixHashedShardKey) { << "hashed"), splitPoints); - // Cannot delete without full shardkey in the query. - auto requestPartialKey = buildDelete(kNss, fromjson("{'a.b': {$gt : 2}}")); + // Can delete with partial shard key in the query if the query only targets one shard. + auto requestPartialKey = buildDelete(kNss, fromjson("{'a.b': {$gt : 101}}")); + auto res = cmTargeter.targetDelete(operationContext(), BatchItemRef(&requestPartialKey, 0)); + ASSERT_EQUALS(res.size(), 1); + ASSERT_EQUALS(res[0].shardName, "4"); + + // Cannot delete with partial shard key in the query if the query targets multiple shards. + auto requestPartialKey2 = buildDelete(kNss, fromjson("{'a.b': {$gt: 0}}")); ASSERT_THROWS_CODE( - cmTargeter.targetDelete(operationContext(), BatchItemRef(&requestPartialKey, 0)), + cmTargeter.targetDelete(operationContext(), BatchItemRef(&requestPartialKey2, 0)), DBException, ErrorCodes::ShardKeyNotFound); - auto requestPartialKey2 = buildDelete(kNss, fromjson("{'a.b': -101}")); + // Cannot delete without at least a partial shard key. + auto requestNoShardKey = buildDelete(kNss, fromjson("{'k': 0}")); ASSERT_THROWS_CODE( - cmTargeter.targetDelete(operationContext(), BatchItemRef(&requestPartialKey2, 0)), + cmTargeter.targetDelete(operationContext(), BatchItemRef(&requestNoShardKey, 0)), DBException, ErrorCodes::ShardKeyNotFound); // Delete targeted correctly with full shard key in query. auto requestFullKey = buildDelete(kNss, fromjson("{'a.b': -101, 'c.d': 5}")); - auto res = cmTargeter.targetDelete(operationContext(), BatchItemRef(&requestFullKey, 0)); + res = cmTargeter.targetDelete(operationContext(), BatchItemRef(&requestFullKey, 0)); ASSERT_EQUALS(res.size(), 1); ASSERT_EQUALS(res[0].shardName, "1"); diff --git a/src/mongo/s/chunk_map_test.cpp b/src/mongo/s/chunk_map_test.cpp index 6514fc00745..96980c009f7 100644 --- a/src/mongo/s/chunk_map_test.cpp +++ b/src/mongo/s/chunk_map_test.cpp @@ -27,18 +27,66 @@ * it in the license file. */ +#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kTest + #include "mongo/platform/basic.h" +#include "mongo/logv2/log.h" +#include "mongo/platform/random.h" #include "mongo/s/chunk_manager.h" +#include "mongo/s/chunk_writes_tracker.h" +#include "mongo/s/chunks_test_util.h" #include "mongo/unittest/unittest.h" namespace mongo { +using chunks_test_util::assertEqualChunkInfo; +using chunks_test_util::calculateCollVersion; +using chunks_test_util::calculateIntermediateShardKey; +using chunks_test_util::genChunkVector; +using chunks_test_util::genRandomSplitPoints; +using chunks_test_util::performRandomChunkOperations; + namespace { -const NamespaceString kNss("TestDB", "TestColl"); +PseudoRandom _random{SecureRandom().nextInt64()}; + const ShardId kThisShard("testShard"); +ShardVersionMap getShardVersionMap(const ChunkMap& chunkMap) { + return chunkMap.getShardVersionsMap(); +} + +std::map<ShardId, ChunkVersion> calculateShardVersions( + const std::vector<std::shared_ptr<ChunkInfo>>& chunkVector) { + std::map<ShardId, ChunkVersion> svMap; + for (const auto& chunk : chunkVector) { + auto mapIt = svMap.find(chunk->getShardId()); + if (mapIt == svMap.end()) { + svMap.emplace(chunk->getShardId(), chunk->getLastmod()); + continue; + } + if (mapIt->second.isOlderThan(chunk->getLastmod())) { + mapIt->second = chunk->getLastmod(); + } + } + return svMap; +} + +std::vector<std::shared_ptr<ChunkInfo>> toChunkInfoPtrVector( + const std::vector<ChunkType>& chunkTypes, bool initializeWriteTrackerRandom = true) { + std::vector<std::shared_ptr<ChunkInfo>> chunkPtrs; + chunkPtrs.reserve(chunkTypes.size()); + for (const auto& chunkType : chunkTypes) { + auto chunkInfoPtr = std::make_shared<ChunkInfo>(chunkType); + if (initializeWriteTrackerRandom) { + chunkInfoPtr->getWritesTracker()->addBytesWritten(_random.nextInt64(30)); + } + chunkPtrs.push_back(std::move(chunkInfoPtr)); + } + return chunkPtrs; +} + class ChunkMapTest : public unittest::Test { public: const KeyPattern& getShardKeyPattern() const { @@ -49,16 +97,36 @@ public: return _uuid; } + const OID& collEpoch() const { + return _epoch; + } + + const Timestamp& collTimestamp() const { + return _collTimestamp; + } + + ChunkMap makeChunkMap(const std::vector<std::shared_ptr<ChunkInfo>>& chunks) const { + const auto chunkBucketSize = + static_cast<size_t>(_random.nextInt64(chunks.size() * 1.2) + 1); + LOGV2(7162701, "Creating new chunk map", "chunkBucketSize"_attr = chunkBucketSize); + return ChunkMap{collEpoch(), collTimestamp(), chunkBucketSize}.createMerged(chunks); + } + + std::vector<ChunkType> genRandomChunkVector(size_t maxNumChunks = 30, + size_t minNumChunks = 1) const { + return chunks_test_util::genRandomChunkVector( + _uuid, _epoch, _collTimestamp, maxNumChunks, minNumChunks); + } + private: - KeyPattern _shardKeyPattern{BSON("a" << 1)}; + KeyPattern _shardKeyPattern{chunks_test_util::kShardKeyPattern}; const UUID _uuid = UUID::gen(); + const OID _epoch{OID::gen()}; + const Timestamp _collTimestamp{1, 1}; }; -} // namespace - TEST_F(ChunkMapTest, TestAddChunk) { - const OID epoch = OID::gen(); - ChunkVersion version{1, 0, epoch, Timestamp(1, 1)}; + ChunkVersion version{1, 0, collEpoch(), collTimestamp()}; auto chunk = std::make_shared<ChunkInfo>( ChunkType{uuid(), @@ -66,18 +134,258 @@ TEST_F(ChunkMapTest, TestAddChunk) { version, kThisShard}); - ChunkMap chunkMap{epoch, Timestamp(1, 1)}; - auto newChunkMap = chunkMap.createMerged({chunk}); + auto newChunkMap = makeChunkMap({chunk}); ASSERT_EQ(newChunkMap.size(), 1); } +TEST_F(ChunkMapTest, ConstructChunkMapRandom) { + auto chunkVector = toChunkInfoPtrVector(genRandomChunkVector()); + + const auto expectedShardVersions = calculateShardVersions(chunkVector); + const auto expectedCollVersion = calculateCollVersion(expectedShardVersions); + + const auto chunkMap = makeChunkMap(chunkVector); + + // Check that it contains all the chunks + ASSERT_EQ(chunkVector.size(), chunkMap.size()); + // Check collection version + ASSERT_EQ(expectedCollVersion, chunkMap.getVersion()); + + size_t i = 0; + chunkMap.forEach([&](const auto& chunkPtr) { + const auto& expectedChunkPtr = chunkVector[i++]; + assertEqualChunkInfo(*expectedChunkPtr, *chunkPtr); + return true; + }); + + // Validate all shard versions + const auto shardVersions = getShardVersionMap(chunkMap); + ASSERT_EQ(expectedShardVersions.size(), shardVersions.size()); + for (const auto& mapIt : shardVersions) { + ASSERT_EQ(expectedShardVersions.at(mapIt.first), mapIt.second.shardVersion); + } + + // Check that vectors are balanced in size + auto maxVectorSize = static_cast<size_t>(std::lround(chunkMap.getMaxChunkVectorSize() * 1.5)); + auto minVectorSize = std::min( + chunkMap.size(), static_cast<size_t>(std::lround(chunkMap.getMaxChunkVectorSize() / 2))); + + for (const auto& [maxKeyString, chunkVectorPtr] : chunkMap.getChunkVectorMap()) { + ASSERT_GTE(chunkVectorPtr->size(), minVectorSize); + ASSERT_LTE(chunkVectorPtr->size(), maxVectorSize); + } +} + +TEST_F(ChunkMapTest, ConstructChunkMapRandomAllChunksSameVersion) { + auto chunkVector = genRandomChunkVector(); + auto commonVersion = chunkVector.front().getVersion(); + + // Set same version on all chunks + for (auto& chunk : chunkVector) { + chunk.setVersion(commonVersion); + } + + auto chunkInfoVector = toChunkInfoPtrVector(chunkVector); + const auto expectedShardVersions = calculateShardVersions(chunkInfoVector); + const auto expectedCollVersion = calculateCollVersion(expectedShardVersions); + + ASSERT_EQ(commonVersion, expectedCollVersion); + + const auto chunkMap = makeChunkMap(chunkInfoVector); + + // Check that it contains all the chunks + ASSERT_EQ(chunkInfoVector.size(), chunkMap.size()); + // Check collection version + ASSERT_EQ(expectedCollVersion, chunkMap.getVersion()); + + size_t i = 0; + chunkMap.forEach([&](const auto& chunkPtr) { + const auto& expectedChunkPtr = chunkInfoVector[i++]; + assertEqualChunkInfo(*expectedChunkPtr, *chunkPtr); + return true; + }); + + // Validate all shard versions + const auto shardVersions = getShardVersionMap(chunkMap); + ASSERT_EQ(expectedShardVersions.size(), shardVersions.size()); + for (const auto& mapIt : shardVersions) { + ASSERT_EQ(expectedShardVersions.at(mapIt.first), mapIt.second.shardVersion); + } +} + +/* + * Check that constucting a ChunkMap with chunks that have mismatching timestamp fails. + */ +TEST_F(ChunkMapTest, ConstructChunkMapMismatchingTimestamp) { + auto chunkVector = toChunkInfoPtrVector(genRandomChunkVector()); + + // Set a different epoch in one of the chunks + const Timestamp wrongTimestamp{Date_t::now()}; + ASSERT_NE(wrongTimestamp, collTimestamp()); + const auto wrongChunkIdx = _random.nextInt32(chunkVector.size()); + const auto oldChunk = chunkVector.at(wrongChunkIdx); + const auto oldVersion = oldChunk->getLastmod(); + const ChunkVersion wrongVersion{ + oldVersion.majorVersion(), oldVersion.minorVersion(), collEpoch(), wrongTimestamp}; + chunkVector[wrongChunkIdx] = std::make_shared<ChunkInfo>( + ChunkType{uuid(), oldChunk->getRange(), wrongVersion, oldChunk->getShardId()}); + + ASSERT_THROWS_CODE( + makeChunkMap(chunkVector), AssertionException, ErrorCodes::ConflictingOperationInProgress); +} + +TEST_F(ChunkMapTest, UpdateMapNotLeaveSmallVectors) { + const ChunkVersion initialVersion{1, 0, collEpoch(), collTimestamp()}; + auto chunkVector = toChunkInfoPtrVector( + genChunkVector(uuid(), genRandomSplitPoints(8), initialVersion, 1 /*numShards*/)); + + const auto chunkBucketSize = 4; + LOGV2(7162703, "Constructing new chunk map", "chunkBucketSize"_attr = chunkBucketSize); + const auto initialChunkMap = + ChunkMap(collEpoch(), collTimestamp(), chunkBucketSize).createMerged(chunkVector); + + // Check that it contains all the chunks + ASSERT_EQ(chunkVector.size(), initialChunkMap.size()); + + auto mergedVersion = initialChunkMap.getVersion(); + mergedVersion.incMinor(); + + auto mergedChunk = std::make_shared<ChunkInfo>(ChunkType{ + uuid(), + ChunkRange{chunkVector[4]->getRange().getMin(), chunkVector.back()->getRange().getMax()}, + mergedVersion, + kThisShard}); + const auto chunkMap = initialChunkMap.createMerged({mergedChunk}); + + // Check that vectors are balanced in size + auto maxVectorSize = std::lround(chunkMap.getMaxChunkVectorSize() * 1.5); + auto minVectorSize = std::min( + chunkMap.size(), static_cast<size_t>(std::lround(chunkMap.getMaxChunkVectorSize() / 2))); + + for (const auto& [maxKeyString, chunkVectorPtr] : chunkMap.getChunkVectorMap()) { + ASSERT_GTE(chunkVectorPtr->size(), minVectorSize); + ASSERT_LTE(chunkVectorPtr->size(), maxVectorSize); + } +} + +/* + * Check that updating a ChunkMap with chunks that have mismatching timestamp fails. + */ +TEST_F(ChunkMapTest, UpdateChunkMapMismatchingTimestamp) { + auto chunkVector = toChunkInfoPtrVector(genRandomChunkVector()); + + auto chunkMap = makeChunkMap(chunkVector); + auto collVersion = chunkMap.getVersion(); + + // Set a different epoch in one of the chunks + const Timestamp wrongTimestamp{Date_t::now()}; + const auto wrongChunkIdx = _random.nextInt32(chunkVector.size()); + const auto oldChunk = chunkVector.at(wrongChunkIdx); + const ChunkVersion wrongVersion{ + collVersion.majorVersion(), collVersion.minorVersion(), collEpoch(), wrongTimestamp}; + auto updateChunk = std::make_shared<ChunkInfo>( + ChunkType{uuid(), oldChunk->getRange(), wrongVersion, oldChunk->getShardId()}); + + ASSERT_THROWS_CODE(chunkMap.createMerged({updateChunk}), + AssertionException, + ErrorCodes::ConflictingOperationInProgress); +} + +/* + * Check that updating a ChunkMap with chunks that have lower version fails. + */ +TEST_F(ChunkMapTest, UpdateChunkMapLowerVersion) { + auto chunkVector = toChunkInfoPtrVector(genRandomChunkVector()); + + auto chunkMap = makeChunkMap(chunkVector); + + const auto wrongChunkIdx = _random.nextInt32(chunkVector.size()); + const auto oldChunk = chunkVector.at(wrongChunkIdx); + const ChunkVersion wrongVersion{0, 1, collEpoch(), collTimestamp()}; + auto updateChunk = std::make_shared<ChunkInfo>( + ChunkType{uuid(), oldChunk->getRange(), wrongVersion, oldChunk->getShardId()}); + + ASSERT_THROWS_CODE(chunkMap.createMerged({updateChunk}), AssertionException, 626840); +} +/* + * Test update of ChunkMap with random chunk manipulation (splits/merges/moves); + */ +TEST_F(ChunkMapTest, UpdateChunkMapRandom) { + auto initialChunks = genRandomChunkVector(); + auto initialChunksInfo = toChunkInfoPtrVector(initialChunks); + + const auto initialChunkMap = makeChunkMap(initialChunksInfo); + + const auto initialShardVersions = calculateShardVersions(initialChunksInfo); + const auto initialCollVersion = calculateCollVersion(initialShardVersions); + + auto chunks = initialChunks; + + const auto maxNumChunkOps = 2 * initialChunks.size(); + const auto numChunkOps = _random.nextInt32(maxNumChunkOps); + performRandomChunkOperations(&chunks, numChunkOps); + + auto chunksInfo = toChunkInfoPtrVector(initialChunks, false /* initializeWriteTrackerRandom */); + + std::vector<std::shared_ptr<ChunkInfo>> updatedChunksInfo; + for (auto& chunkPtr : chunksInfo) { + // First overlapping chunk in the initial vector + const auto& overlapInitChunk = + **std::lower_bound(initialChunksInfo.begin(), + initialChunksInfo.end(), + ShardKeyPattern::toKeyString(chunkPtr->getRange().getMin()), + [](const auto& chunkInfo, const std::string& shardKeyString) { + return chunkInfo->getMaxKeyString() <= shardKeyString; + }); + // The new chunks inherits the written bytes from the first overlapping old chunk + chunkPtr->getWritesTracker()->addBytesWritten( + overlapInitChunk.getWritesTracker()->getBytesWritten()); + + if (!chunkPtr->getLastmod().isOlderOrEqualThan(initialCollVersion)) { + updatedChunksInfo.push_back(std::make_shared<ChunkInfo>(ChunkType{ + uuid(), chunkPtr->getRange(), chunkPtr->getLastmod(), chunkPtr->getShardId()})); + } + } + + const auto expectedShardVersions = calculateShardVersions(chunksInfo); + const auto expectedCollVersion = calculateCollVersion(expectedShardVersions); + auto chunkMap = initialChunkMap.createMerged(updatedChunksInfo); + + // Check that it contains all the chunks + ASSERT_EQ(chunksInfo.size(), chunkMap.size()); + // Check collection version + ASSERT_EQ(expectedCollVersion, chunkMap.getVersion()); + + size_t i = 0; + chunkMap.forEach([&](const auto& chunkPtr) { + const auto& expectedChunkPtr = chunksInfo[i++]; + assertEqualChunkInfo(*expectedChunkPtr, *chunkPtr); + return true; + }); + + // Validate all shard versions + const auto shardVersions = getShardVersionMap(chunkMap); + ASSERT_EQ(expectedShardVersions.size(), shardVersions.size()); + for (const auto& mapIt : shardVersions) { + ASSERT_EQ(expectedShardVersions.at(mapIt.first), mapIt.second.shardVersion); + } + + // Check that vectors are balanced in size + auto maxVectorSize = static_cast<size_t>(std::lround(chunkMap.getMaxChunkVectorSize() * 1.5)); + auto minVectorSize = std::min( + chunkMap.size(), static_cast<size_t>(std::lround(chunkMap.getMaxChunkVectorSize() / 2))); + + for (const auto& [maxKeyString, chunkVectorPtr] : chunkMap.getChunkVectorMap()) { + ASSERT_GTE(chunkVectorPtr->size(), minVectorSize); + ASSERT_LTE(chunkVectorPtr->size(), maxVectorSize); + } +} + TEST_F(ChunkMapTest, TestEnumerateAllChunks) { - const OID epoch = OID::gen(); - ChunkMap chunkMap{epoch, Timestamp(1, 1)}; - ChunkVersion version{1, 0, epoch, Timestamp(1, 1)}; + ChunkVersion version{1, 0, collEpoch(), collTimestamp()}; - auto newChunkMap = chunkMap.createMerged( + auto newChunkMap = makeChunkMap( {std::make_shared<ChunkInfo>( ChunkType{uuid(), ChunkRange{getShardKeyPattern().globalMin(), BSON("a" << 0)}, @@ -107,12 +415,11 @@ TEST_F(ChunkMapTest, TestEnumerateAllChunks) { ASSERT_EQ(count, newChunkMap.size()); } + TEST_F(ChunkMapTest, TestIntersectingChunk) { - const OID epoch = OID::gen(); - ChunkMap chunkMap{epoch, Timestamp(1, 1)}; - ChunkVersion version{1, 0, epoch, Timestamp(1, 1)}; + ChunkVersion version{1, 0, collEpoch(), collTimestamp()}; - auto newChunkMap = chunkMap.createMerged( + auto newChunkMap = makeChunkMap( {std::make_shared<ChunkInfo>( ChunkType{uuid(), ChunkRange{getShardKeyPattern().globalMin(), BSON("a" << 0)}, @@ -135,14 +442,33 @@ TEST_F(ChunkMapTest, TestIntersectingChunk) { SimpleBSONObjComparator::kInstance.evaluate(intersectingChunk->getMin() == BSON("a" << 0))); ASSERT(SimpleBSONObjComparator::kInstance.evaluate(intersectingChunk->getMax() == BSON("a" << 100))); + + // findIntersectingChunks returns last chunk if invoked with MaxKey + intersectingChunk = + newChunkMap.findIntersectingChunk(BSON("a" << getShardKeyPattern().globalMax())); + ASSERT(SimpleBSONObjComparator::kInstance.evaluate(intersectingChunk->getMin() == + BSON("a" << 100))); + ASSERT(SimpleBSONObjComparator::kInstance.evaluate(intersectingChunk->getMax() == + getShardKeyPattern().globalMax())); +} + +TEST_F(ChunkMapTest, TestIntersectingChunkRandom) { + auto chunks = toChunkInfoPtrVector(genRandomChunkVector()); + + const auto chunkMap = makeChunkMap(chunks); + + auto targetChunkIt = chunks.begin() + _random.nextInt64(chunks.size()); + auto intermediateKey = calculateIntermediateShardKey( + (*targetChunkIt)->getMin(), (*targetChunkIt)->getMax(), 0.2 /* minKeyProb */); + + auto intersectingChunkPtr = chunkMap.findIntersectingChunk(intermediateKey); + assertEqualChunkInfo(**(targetChunkIt), *intersectingChunkPtr); } TEST_F(ChunkMapTest, TestEnumerateOverlappingChunks) { - const OID epoch = OID::gen(); - ChunkMap chunkMap{epoch, Timestamp(1, 1)}; - ChunkVersion version{1, 0, epoch, Timestamp(1, 1)}; + ChunkVersion version{1, 0, collEpoch(), collTimestamp()}; - auto newChunkMap = chunkMap.createMerged( + auto newChunkMap = makeChunkMap( {std::make_shared<ChunkInfo>( ChunkType{uuid(), ChunkRange{getShardKeyPattern().globalMin(), BSON("a" << 0)}, @@ -160,14 +486,101 @@ TEST_F(ChunkMapTest, TestEnumerateOverlappingChunks) { auto min = BSON("a" << -50); auto max = BSON("a" << 150); - int count = 0; newChunkMap.forEachOverlappingChunk(min, max, true, [&](const auto& chunk) { count++; return true; }); + ASSERT_EQ(count, 3); + min = BSON("a" << -50); + max = BSON("a" << getShardKeyPattern().globalMax()); + count = 0; + newChunkMap.forEachOverlappingChunk(min, max, false, [&](const auto& chunk) { + count++; + return true; + }); ASSERT_EQ(count, 3); + + min = BSON("a" << 50); + max = BSON("a" << 100); + count = 0; + newChunkMap.forEachOverlappingChunk(min, max, true, [&](const auto& chunk) { + count++; + return true; + }); + ASSERT_EQ(count, 2); + + min = BSON("a" << 50); + max = BSON("a" << 100); + count = 0; + newChunkMap.forEachOverlappingChunk(min, max, false, [&](const auto& chunk) { + count++; + return true; + }); + ASSERT_EQ(count, 1); +} + +TEST_F(ChunkMapTest, ForEachNoShardKey) { + auto chunks = toChunkInfoPtrVector(genRandomChunkVector()); + + const auto chunkMap = makeChunkMap(chunks); + + auto lastChunkIdx = std::max(_random.nextInt64(chunks.size()), static_cast<int64_t>(1)); + + int i = 0; + chunkMap.forEach([&](const auto& chunkInfo) { + assertEqualChunkInfo(*chunks[i], *chunkInfo); + return ++i < lastChunkIdx; + }); + + ASSERT_EQ(i, lastChunkIdx); +} + +TEST_F(ChunkMapTest, ForEachWithShardKey) { + auto chunks = toChunkInfoPtrVector(genRandomChunkVector()); + + const auto chunkMap = makeChunkMap(chunks); + + auto firstChunkIdx = static_cast<size_t>(_random.nextInt64(chunks.size())); + const auto& firstChunk = chunks[firstChunkIdx]; + auto skey = calculateIntermediateShardKey( + firstChunk->getMin(), firstChunk->getMax(), 0.2 /* minKeyProb */); + + size_t i = firstChunkIdx; + auto lastChunkIdx = firstChunkIdx + + std::max(_random.nextInt64(chunks.size() - firstChunkIdx), static_cast<int64_t>(1)); + chunkMap.forEach( + [&](const auto& chunkInfo) { + assertEqualChunkInfo(*chunks[i], *chunkInfo); + return ++i < lastChunkIdx; + }, + skey); + + ASSERT_EQ(i, lastChunkIdx); } +TEST_F(ChunkMapTest, TestEnumerateOverlappingChunksRandom) { + auto chunks = toChunkInfoPtrVector(genRandomChunkVector()); + + const auto chunkMap = makeChunkMap(chunks); + + auto firstChunkIt = chunks.begin() + _random.nextInt64(chunks.size()); + auto lastChunkIt = firstChunkIt + _random.nextInt64(std::distance(firstChunkIt, chunks.end())); + + auto minBound = calculateIntermediateShardKey( + (*firstChunkIt)->getMin(), (*firstChunkIt)->getMax(), 0.2 /* minKeyProb */); + auto maxBound = calculateIntermediateShardKey( + (*lastChunkIt)->getMin(), (*lastChunkIt)->getMax(), 0.2 /* minKeyProb */); + + auto it = firstChunkIt; + chunkMap.forEachOverlappingChunk(minBound, maxBound, true, [&](const auto& chunkInfoPtr) { + assertEqualChunkInfo(**(it++), *chunkInfoPtr); + return true; + }); + ASSERT_EQ(0, std::distance(it, std::next(lastChunkIt))); +} + +} // namespace + } // namespace mongo diff --git a/src/mongo/s/chunks_test_util.cpp b/src/mongo/s/chunks_test_util.cpp new file mode 100644 index 00000000000..ac3025ecffd --- /dev/null +++ b/src/mongo/s/chunks_test_util.cpp @@ -0,0 +1,305 @@ +/** + * Copyright (C) 2023-present MongoDB, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * <http://www.mongodb.com/licensing/server-side-public-license>. + * + * As a special exception, the copyright holders give permission to link the + * code of portions of this program with the OpenSSL library under certain + * conditions as described in each individual source file and distribute + * linked combinations including the program with the OpenSSL library. You + * must comply with the Server Side Public License in all respects for + * all of the code used other than as permitted herein. If you modify file(s) + * with this exception, you may extend this exception to your version of the + * file(s), but you are not obligated to do so. If you do not wish to do so, + * delete this exception statement from your version. If you delete this + * exception statement from all source files in the program, then also delete + * it in the license file. + */ + +#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kTest + +#include "mongo/s/chunks_test_util.h" +#include "mongo/db/namespace_string.h" +#include "mongo/logv2/log.h" +#include "mongo/platform/random.h" +#include "mongo/s/shard_key_pattern.h" +#include "mongo/unittest/unittest.h" + +namespace mongo::chunks_test_util { +namespace { + +PseudoRandom _random{SecureRandom().nextInt64()}; + +} // namespace + +void assertEqualChunkInfo(const ChunkInfo& x, const ChunkInfo& y) { + ASSERT_BSONOBJ_EQ(x.getMin(), y.getMin()); + ASSERT_BSONOBJ_EQ(x.getMax(), y.getMax()); + ASSERT_EQ(x.getMaxKeyString(), y.getMaxKeyString()); + ASSERT_EQ(x.getShardId(), y.getShardId()); + ASSERT_EQ(x.getLastmod(), y.getLastmod()); + ASSERT_EQ(x.isJumbo(), y.isJumbo()); + ASSERT_EQ(x.getWritesTracker()->getBytesWritten(), y.getWritesTracker()->getBytesWritten()); +} + +ShardId getShardId(int shardIdx) { + return {std::string(str::stream() << "shard_" << shardIdx)}; +} + +std::vector<BSONObj> genRandomSplitPoints(size_t numChunks) { + std::vector<BSONObj> splitPoints; + splitPoints.reserve(numChunks + 1); + splitPoints.emplace_back(kShardKeyPattern.globalMin()); + int nextSplit{-1000}; + for (size_t i = 0; i < numChunks - 1; ++i) { + nextSplit += i * 10 * (_random.nextInt32(10) + 1); + splitPoints.emplace_back(BSON(kSKey << nextSplit)); + } + splitPoints.emplace_back(kShardKeyPattern.globalMax()); + return splitPoints; +} + +std::vector<ChunkVersion> genRandomVersions(size_t num, const ChunkVersion& initialVersion) { + std::vector<ChunkVersion> versions; + versions.reserve(num); + auto major = initialVersion.majorVersion(); + auto minor = initialVersion.minorVersion(); + + for (size_t i = 0; i < num; ++i) { + if (_random.nextInt32(2)) { + ++major; + minor = 0; + } else { + ++minor; + } + versions.emplace_back(major, minor, initialVersion.epoch(), initialVersion.getTimestamp()); + } + std::shuffle(versions.begin(), versions.end(), _random.urbg()); + return versions; +} + +std::vector<ChunkType> genChunkVector(const UUID& uuid, + const std::vector<BSONObj>& splitPoints, + const ChunkVersion& initialVersion, + size_t numShards) { + + return genChunkVector( + uuid, splitPoints, genRandomVersions(splitPoints.size() - 1, initialVersion), numShards); +} + +std::vector<ChunkType> genChunkVector(const UUID& uuid, + const std::vector<BSONObj>& splitPoints, + const std::vector<ChunkVersion>& versions, + size_t numShards) { + + invariant(SimpleBSONObjComparator::kInstance.evaluate(splitPoints.front() == + kShardKeyPattern.globalMin())); + invariant(SimpleBSONObjComparator::kInstance.evaluate(splitPoints.back() == + kShardKeyPattern.globalMax())); + const auto numChunks = splitPoints.size() - 1; + invariant(numChunks == versions.size()); + + std::vector<ChunkType> chunks; + chunks.reserve(numChunks); + auto minKey = splitPoints.front(); + for (size_t i = 0; i < numChunks; ++i) { + auto maxKey = splitPoints.at(i + 1); + const auto shard = getShardId(_random.nextInt64(numShards)); + const auto version = versions.at(i); + chunks.emplace_back(uuid, ChunkRange{minKey, maxKey}, version, shard); + minKey = std::move(maxKey); + } + return chunks; +} + +ChunkVersion calculateCollVersion(const std::map<ShardId, ChunkVersion>& shardVersions) { + return std::max_element(shardVersions.begin(), + shardVersions.end(), + [](const std::pair<ShardId, ChunkVersion>& p1, + const std::pair<ShardId, ChunkVersion>& p2) { + return p1.second.isOlderThan(p2.second); + }) + ->second; +} + +std::map<ShardId, ChunkVersion> calculateShardVersions(const std::vector<ChunkType>& chunkVector) { + std::map<ShardId, ChunkVersion> svMap; + for (const auto& chunk : chunkVector) { + auto mapIt = svMap.find(chunk.getShard()); + if (mapIt == svMap.end()) { + svMap.emplace(chunk.getShard(), chunk.getVersion()); + continue; + } + if (mapIt->second.isOlderThan(chunk.getVersion())) { + mapIt->second = chunk.getVersion(); + } + } + return svMap; +} + +std::vector<ChunkType> genRandomChunkVector(const UUID& uuid, + const OID& epoch, + const Timestamp& timestamp, + size_t maxNumChunks, + size_t minNumChunks) { + invariant(minNumChunks <= maxNumChunks); + const auto numChunks = minNumChunks + _random.nextInt32((maxNumChunks - minNumChunks) + 1); + const auto numShards = _random.nextInt32(numChunks) + 1; + const ChunkVersion initialVersion{1, 0, epoch, timestamp}; + + LOGV2(7162700, + "Generating random chunk vector", + "numChunks"_attr = numChunks, + "numShards"_attr = numShards); + + return genChunkVector(uuid, genRandomSplitPoints(numChunks), initialVersion, numShards); +} + +BSONObj calculateIntermediateShardKey(const BSONObj& leftKey, + const BSONObj& rightKey, + double minKeyProb, + double maxKeyProb) { + invariant(0 <= minKeyProb && minKeyProb <= 1, "minKeyProb out of range [0, 1]"); + invariant(0 <= maxKeyProb && maxKeyProb <= 1, "maxKeyProb out of range [0, 1]"); + + if (_random.nextInt32(100) < minKeyProb * 100) { + return leftKey; + } + + if (_random.nextInt32(100) < maxKeyProb * 100) { + return rightKey; + } + + const auto isMinKey = leftKey.woCompare(kShardKeyPattern.globalMin()) == 0; + const auto isMaxKey = rightKey.woCompare(kShardKeyPattern.globalMax()) == 0; + + int splitPoint; + if (isMinKey && isMaxKey) { + // [min, max] -> split at 0 + splitPoint = 0; + } else if (!isMinKey && !isMaxKey) { + // [x, y] -> split in the middle + auto min = leftKey.firstElement().numberInt(); + auto max = rightKey.firstElement().numberInt(); + invariant(min + 1 < max, + str::stream() << "Can't split range [" << min << ", " << max << "]"); + splitPoint = min + ((max - min) / 2); + } else if (isMaxKey) { + // [x, maxKey] -> split at x*2; + auto prevBound = leftKey.firstElement().numberInt(); + auto increment = prevBound ? prevBound : _random.nextInt32(100) + 1; + splitPoint = prevBound + std::abs(increment); + } else if (isMinKey) { + // [minKey, x] -> split at x*2; + auto prevBound = rightKey.firstElement().numberInt(); + auto increment = prevBound ? prevBound : _random.nextInt32(100) + 1; + splitPoint = prevBound - std::abs(increment); + } else { + MONGO_UNREACHABLE; + } + + return BSON(kSKey << splitPoint); +} + +void performRandomChunkOperations(std::vector<ChunkType>* chunksPtr, size_t numOperations) { + auto& chunks = *chunksPtr; + auto collVersion = calculateCollVersion(calculateShardVersions(chunks)); + + auto moveChunk = [&] { + auto& chunkToMigrate = chunks[_random.nextInt32(chunks.size())]; + collVersion.incMajor(); + + auto controlChunkIt = std::find_if(chunks.begin(), chunks.end(), [&](const auto& chunk) { + return chunk.getShard() == chunkToMigrate.getShard() && + !chunk.getRange().overlaps(chunkToMigrate.getRange()); + }); + if (controlChunkIt != chunks.end()) { + controlChunkIt->setVersion(collVersion); + collVersion.incMinor(); + } + auto newShard = getShardId(_random.nextInt64(chunks.size())); + chunkToMigrate.setShard(newShard); + chunkToMigrate.setVersion(collVersion); + }; + + auto splitChunk = [&] { + auto chunkToSplitIt = chunks.begin() + _random.nextInt32(chunks.size()); + while (chunkToSplitIt != chunks.begin() && chunkToSplitIt != std::prev(chunks.end()) && + (chunkToSplitIt->getMax().firstElement().numberInt() - + chunkToSplitIt->getMin().firstElement().numberInt()) < 2) { + // If the chunk is unsplittable select another one + chunkToSplitIt = chunks.begin() + _random.nextInt32(chunks.size()); + } + + const auto& chunkToSplit = *chunkToSplitIt; + + auto splitKey = calculateIntermediateShardKey(chunkToSplit.getMin(), chunkToSplit.getMax()); + + collVersion.incMinor(); + const ChunkRange leftRange{chunkToSplit.getMin(), splitKey}; + ChunkType leftChunk{ + chunkToSplit.getCollectionUUID(), leftRange, collVersion, chunkToSplit.getShard()}; + + collVersion.incMinor(); + const ChunkRange rightRange{splitKey, chunkToSplit.getMax()}; + ChunkType rightChunk{ + chunkToSplit.getCollectionUUID(), rightRange, collVersion, chunkToSplit.getShard()}; + + auto it = chunks.erase(chunkToSplitIt); + it = chunks.insert(it, std::move(rightChunk)); + it = chunks.insert(it, std::move(leftChunk)); + }; + + auto mergeChunks = [&] { + const auto firstChunkIt = chunks.begin() + _random.nextInt32(chunks.size()); + const auto& shardId = firstChunkIt->getShard(); + auto lastChunkIt = std::find_if(firstChunkIt, chunks.end(), [&](const auto& chunk) { + return chunk.getShard() != shardId; + }); + const auto numContiguosChunks = std::distance(firstChunkIt, lastChunkIt); + if (numContiguosChunks < 2) { + // nothing to merge + return; + } + const auto numChunkToMerge = _random.nextInt32(numContiguosChunks - 1) + 2; + lastChunkIt = firstChunkIt + numChunkToMerge; + const auto& firstChunk = *firstChunkIt; + collVersion.incMinor(); + const ChunkRange mergedRange{firstChunk.getMin(), std::prev(lastChunkIt)->getMax()}; + ChunkType mergedChunk{ + firstChunk.getCollectionUUID(), mergedRange, collVersion, firstChunk.getShard()}; + + auto it = chunks.erase(firstChunkIt, lastChunkIt); + it = chunks.insert(it, mergedChunk); + }; + + for (size_t i = 0; i < numOperations; i++) { + switch (_random.nextInt32(3)) { + case 0: + moveChunk(); + break; + case 1: + splitChunk(); + break; + case 2: + mergeChunks(); + break; + default: + MONGO_UNREACHABLE; + break; + } + } +} + +} // namespace mongo::chunks_test_util diff --git a/src/mongo/s/chunks_test_util.h b/src/mongo/s/chunks_test_util.h new file mode 100644 index 00000000000..90499904da8 --- /dev/null +++ b/src/mongo/s/chunks_test_util.h @@ -0,0 +1,113 @@ +/** + * Copyright (C) 2023-present MongoDB, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * <http://www.mongodb.com/licensing/server-side-public-license>. + * + * As a special exception, the copyright holders give permission to link the + * code of portions of this program with the OpenSSL library under certain + * conditions as described in each individual source file and distribute + * linked combinations including the program with the OpenSSL library. You + * must comply with the Server Side Public License in all respects for + * all of the code used other than as permitted herein. If you modify file(s) + * with this exception, you may extend this exception to your version of the + * file(s), but you are not obligated to do so. If you do not wish to do so, + * delete this exception statement from your version. If you delete this + * exception statement from all source files in the program, then also delete + * it in the license file. + */ + +#include "mongo/platform/basic.h" + +#include "mongo/s/chunk.h" +#include "mongo/s/chunk_writes_tracker.h" + +namespace mongo::chunks_test_util { + +static const std::string kSKey{"a"}; +static const KeyPattern kShardKeyPattern{BSON(kSKey << 1)}; + +/* + * Assert that all the fields contained in the provided ChunkInfo are equal. + * + * This is needed since ChunkInfo class does not provide an equal operator. + */ +void assertEqualChunkInfo(const ChunkInfo& x, const ChunkInfo& y); + +ShardId getShardId(int shardIdx); + +/** + * Return a vector of randomly generated split points. + * covering the entire shard key space including the boundaries [minKey, maxKey) + * + * e.g. {"a": <int>} + */ +std::vector<BSONObj> genRandomSplitPoints(size_t numChunks); + +/* + * Generate a shuffled list of random chunk versions. + * + * The generated versions are all strictly greater than the provided initialVersion. + */ +std::vector<ChunkVersion> genRandomVersions(size_t num, const ChunkVersion& initialVersion); + +/* + * Generate a vector of chunks whose boundaries are defined by the provided split points and random + * chunk versions. + */ +std::vector<ChunkType> genChunkVector(const UUID& uuid, + const std::vector<BSONObj>& splitPoints, + const ChunkVersion& initialVersion, + size_t numShards); + +/* + * Generate a vector of chunks. + */ +std::vector<ChunkType> genChunkVector(const UUID& uuid, + const std::vector<BSONObj>& splitPoints, + const std::vector<ChunkVersion>& versions, + size_t numShards); + +/* + * Return a randomly generated vector of chunks. + */ +std::vector<ChunkType> genRandomChunkVector(const UUID& uuid, + const OID& epoch, + const Timestamp& timestamp, + size_t maxNumChunks, + size_t minNumChunks = 1); + +std::map<ShardId, ChunkVersion> calculateShardVersions(const std::vector<ChunkType>& chunkVector); + +ChunkVersion calculateCollVersion(const std::map<ShardId, ChunkVersion>& shardVersions); + +/* + * Return a shardkey that is in between the given range [leftKey, rightKey] + */ +BSONObj calculateIntermediateShardKey(const BSONObj& leftKey, + const BSONObj& rightKey, + double minKeyProb = 0.0, + double maxKeyProb = 0.0); + +/* + * Perform a series of random operations on the given list of chunks. + * + * The operations performed resemble all possible operations that could happen to a routing table in + * a production cluster (move, merge, split, etc...) + * + * @chunks: list of chunks ordered by minKey + * @numOperations: number of operations to perform + */ +void performRandomChunkOperations(std::vector<ChunkType>* chunks, size_t numOperations); + +} // namespace mongo::chunks_test_util diff --git a/src/mongo/s/client/shard_registry.cpp b/src/mongo/s/client/shard_registry.cpp index d722a9116ec..0b1691af82b 100644 --- a/src/mongo/s/client/shard_registry.cpp +++ b/src/mongo/s/client/shard_registry.cpp @@ -130,12 +130,8 @@ ShardRegistry::Cache::LookupResult ShardRegistry::_lookup(OperationContext* opCt // Check if we need to refresh from the configsvrs. If so, then do that and get the results, // otherwise (this is a lookup only to incorporate updated connection strings from the RSM), // then get the equivalent values from the previously cached data. - auto [returnData, - returnTopologyTime, - returnForceReloadIncrement, - removedShards, - fetchedFromConfigServers] = [&]() - -> std::tuple<ShardRegistryData, Timestamp, Increment, ShardRegistryData::ShardMap, bool> { + auto [returnData, returnTopologyTime, returnForceReloadIncrement, removedShards] = + [&]() -> std::tuple<ShardRegistryData, Timestamp, Increment, ShardRegistryData::ShardMap> { if (timeInStore.topologyTime > cachedData.getTime().topologyTime || timeInStore.forceReloadIncrement > cachedData.getTime().forceReloadIncrement) { auto [reloadedData, maxTopologyTime] = @@ -144,14 +140,12 @@ ShardRegistry::Cache::LookupResult ShardRegistry::_lookup(OperationContext* opCt auto [mergedData, removedShards] = ShardRegistryData::mergeExisting(*cachedData, reloadedData); - return { - mergedData, maxTopologyTime, timeInStore.forceReloadIncrement, removedShards, true}; + return {mergedData, maxTopologyTime, timeInStore.forceReloadIncrement, removedShards}; } else { return {*cachedData, cachedData.getTime().topologyTime, cachedData.getTime().forceReloadIncrement, - {}, - false}; + {}}; } }(); @@ -186,11 +180,6 @@ ShardRegistry::Cache::LookupResult ShardRegistry::_lookup(OperationContext* opCt } } - // The registry is "up" once there has been a successful lookup from the config servers. - if (fetchedFromConfigServers) { - _isUp.store(true); - } - Time returnTime{returnTopologyTime, rsmIncrementForConnStrings, returnForceReloadIncrement}; LOGV2_DEBUG(4620251, 2, @@ -218,9 +207,9 @@ void ShardRegistry::startupPeriodicReloader(OperationContext* opCtx) { AsyncTry([this] { LOGV2_DEBUG(22726, 1, "Reloading shardRegistry"); - return _reloadInternal(); + return _reloadAsyncNoRetry(); }) - .until([](auto sw) { + .until([](auto&& sw) { if (!sw.isOK()) { LOGV2(22727, "Error running periodic reload of shard registry", @@ -232,7 +221,7 @@ void ShardRegistry::startupPeriodicReloader(OperationContext* opCtx) { }) .withDelayBetweenIterations(kRefreshPeriod) // This call is optional. .on(_executor, CancellationToken::uncancelable()) - .getAsync([](auto sw) { + .getAsync([](auto&& sw) { LOGV2_DEBUG(22725, 1, "Exiting periodic shard registry reloader", @@ -295,6 +284,49 @@ StatusWith<std::shared_ptr<Shard>> ShardRegistry::getShard(OperationContext* opC return {ErrorCodes::ShardNotFound, str::stream() << "Shard " << shardId << " not found"}; } +SemiFuture<std::shared_ptr<Shard>> ShardRegistry::getShard(ExecutorPtr executor, + const ShardId& shardId) noexcept { + + // Fetch the shard registry data associated to the latest known topology time + return _getDataAsync() + .thenRunOn(executor) + .then([this, executor, shardId](auto&& cachedData) { + // First check if this is a non config shard lookup + if (auto shard = cachedData->findShard(shardId)) { + return SemiFuture<std::shared_ptr<Shard>>::makeReady(std::move(shard)); + } + + // then check if this is a config shard (this call is blocking in any case) + { + stdx::lock_guard<Latch> lk(_mutex); + if (auto shard = _configShardData.findShard(shardId)) { + return SemiFuture<std::shared_ptr<Shard>>::makeReady(std::move(shard)); + } + } + + // If the shard was not found, force reload the shard regitry data and try again. + // + // This is to cover the following scenario: + // 1. Primary of the replicaset fetch the list of shards and store it on disk + // 2. Primary crash before the latest VectorClock topology time is majority written to + // disk + // 3. A new primary with a stale ShardRegistry is elected and read the set of shards + // from disk and calls ShardRegistry::getShard + + return _reloadAsync() + .thenRunOn(executor) + .then([this, executor, shardId](auto&& cachedData) -> std::shared_ptr<Shard> { + auto shard = cachedData->findShard(shardId); + uassert(ErrorCodes::ShardNotFound, + str::stream() << "Shard " << shardId << " not found", + shard); + return shard; + }) + .semi(); + }) + .semi(); +} + std::vector<ShardId> ShardRegistry::getAllShardIds(OperationContext* opCtx) { auto shardIds = _getData(opCtx)->getAllShardIds(); if (shardIds.empty()) { @@ -380,8 +412,18 @@ std::unique_ptr<Shard> ShardRegistry::createConnection(const ConnectionString& c return _shardFactory->createUniqueShard(ShardId("<unnamed>"), connStr); } -bool ShardRegistry::isUp() const { - return _isUp.load(); +bool ShardRegistry::isUp() { + if (_isUp.load()) + return true; + + // Before the first lookup is completed, the latest cached value is either empty or it is + // associated to the default constructed time + const auto latestCached = _cache->peekLatestCached(_kSingleton); + if (latestCached && latestCached.getTime() != Time()) { + _isUp.store(true); + return true; + } + return false; } void ShardRegistry::toBSON(BSONObjBuilder* result) const { @@ -400,23 +442,26 @@ void ShardRegistry::toBSON(BSONObjBuilder* result) const { } void ShardRegistry::reload(OperationContext* opCtx) { + _reloadAsync().get(opCtx); +} + +SharedSemiFuture<ShardRegistry::Cache::ValueHandle> ShardRegistry::_reloadAsync() { if (MONGO_unlikely(TestingProctor::instance().isEnabled())) { // Some unit tests don't support running the reload's AsyncTry on the fixed executor. - _reloadInternal().get(opCtx); + return _reloadAsyncNoRetry(); } else { - AsyncTry([=]() mutable { return _reloadInternal(); }) + return AsyncTry([=]() mutable { return _reloadAsyncNoRetry(); }) .until([](auto sw) mutable { return sw.getStatus() != ErrorCodes::ReadConcernMajorityNotAvailableYet; }) .withBackoffBetweenIterations(kExponentialBackoff) - .on(Grid::get(opCtx)->getExecutorPool()->getFixedExecutor(), + .on(Grid::get(getGlobalServiceContext())->getExecutorPool()->getFixedExecutor(), CancellationToken::uncancelable()) - .semi() - .get(opCtx); + .share(); } } -SharedSemiFuture<ShardRegistry::Cache::ValueHandle> ShardRegistry::_reloadInternal() { +SharedSemiFuture<ShardRegistry::Cache::ValueHandle> ShardRegistry::_reloadAsyncNoRetry() { // Make the next acquire do a lookup. auto value = _forceReloadIncrement.addAndFetch(1); LOGV2_DEBUG(4620253, 2, "Forcing ShardRegistry reload", "newForceReloadIncrement"_attr = value); diff --git a/src/mongo/s/client/shard_registry.h b/src/mongo/s/client/shard_registry.h index 2c329f11b4c..aceb45db9d2 100644 --- a/src/mongo/s/client/shard_registry.h +++ b/src/mongo/s/client/shard_registry.h @@ -239,6 +239,9 @@ public: */ StatusWith<std::shared_ptr<Shard>> getShard(OperationContext* opCtx, const ShardId& shardId); + SemiFuture<std::shared_ptr<Shard>> getShard(ExecutorPtr executor, + const ShardId& shardId) noexcept; + /** * Returns a vector containing all known shard IDs. * The order of the elements is not guaranteed. @@ -272,7 +275,7 @@ public: * The ShardRegistry is "up" once a successful lookup from the config servers has been * completed. */ - bool isUp() const; + bool isUp(); void toBSON(BSONObjBuilder* result) const; @@ -438,7 +441,8 @@ private: void _initializeCacheIfNecessary() const; - SharedSemiFuture<Cache::ValueHandle> _reloadInternal(); + SharedSemiFuture<Cache::ValueHandle> _reloadAsync(); + SharedSemiFuture<Cache::ValueHandle> _reloadAsyncNoRetry(); /** * Factory to create shards. Never changed after startup so safe to access outside of _mutex. diff --git a/src/mongo/s/client/sharding_connection_hook.cpp b/src/mongo/s/client/sharding_connection_hook.cpp index 0a7374a240e..bc72c6ea064 100644 --- a/src/mongo/s/client/sharding_connection_hook.cpp +++ b/src/mongo/s/client/sharding_connection_hook.cpp @@ -73,14 +73,14 @@ void ShardingConnectionHook::onCreate(DBClientBase* conn) { if (conn->type() == ConnectionString::ConnectionType::kStandalone) { - BSONObj isMasterResponse; - if (!conn->runCommand("admin", BSON("ismaster" << 1), isMasterResponse)) { - uassertStatusOK(getStatusFromCommandResult(isMasterResponse)); + BSONObj helloResponse; + if (!conn->runCommand("admin", BSON("hello" << 1), helloResponse)) { + uassertStatusOK(getStatusFromCommandResult(helloResponse)); } long long configServerModeNumber; Status status = - bsonExtractIntegerField(isMasterResponse, "configsvr", &configServerModeNumber); + bsonExtractIntegerField(helloResponse, "configsvr", &configServerModeNumber); if (status == ErrorCodes::NoSuchKey) { // This isn't a config server we're talking to. diff --git a/src/mongo/s/client/sharding_network_connection_hook.cpp b/src/mongo/s/client/sharding_network_connection_hook.cpp index a6229316eaa..60368d2eb0f 100644 --- a/src/mongo/s/client/sharding_network_connection_hook.cpp +++ b/src/mongo/s/client/sharding_network_connection_hook.cpp @@ -48,12 +48,12 @@ namespace mongo { Status ShardingNetworkConnectionHook::validateHost( const HostAndPort& remoteHost, const BSONObj&, - const executor::RemoteCommandResponse& isMasterReply) { - return validateHostImpl(remoteHost, isMasterReply); + const executor::RemoteCommandResponse& helloReply) { + return validateHostImpl(remoteHost, helloReply); } Status ShardingNetworkConnectionHook::validateHostImpl( - const HostAndPort& remoteHost, const executor::RemoteCommandResponse& isMasterReply) { + const HostAndPort& remoteHost, const executor::RemoteCommandResponse& helloReply) { auto shard = Grid::get(getGlobalServiceContext())->shardRegistry()->getShardForHostNoReload(remoteHost); if (!shard) { @@ -62,11 +62,11 @@ Status ShardingNetworkConnectionHook::validateHostImpl( } long long configServerModeNumber; - auto status = bsonExtractIntegerField(isMasterReply.data, "configsvr", &configServerModeNumber); + auto status = bsonExtractIntegerField(helloReply.data, "configsvr", &configServerModeNumber); switch (status.code()) { case ErrorCodes::OK: { - // The ismaster response indicates remoteHost is a config server. + // The hello response indicates remoteHost is a config server. if (!shard->isConfig()) { return {ErrorCodes::InvalidOptions, str::stream() << "Surprised to discover that " << remoteHost.toString() @@ -75,7 +75,7 @@ Status ShardingNetworkConnectionHook::validateHostImpl( return Status::OK(); } case ErrorCodes::NoSuchKey: { - // The ismaster response indicates that remoteHost is not a config server, or that + // The hello response indicates that remoteHost is not a config server, or that // the config server is running a version prior to the 3.1 development series. if (!shard->isConfig()) { return Status::OK(); @@ -86,7 +86,7 @@ Status ShardingNetworkConnectionHook::validateHostImpl( << " does not believe it is a config server"}; } default: - // The ismaster response was malformed. + // The hello response was malformed. return status; } } diff --git a/src/mongo/s/client/sharding_network_connection_hook.h b/src/mongo/s/client/sharding_network_connection_hook.h index c77428e56d9..d3715dcec1b 100644 --- a/src/mongo/s/client/sharding_network_connection_hook.h +++ b/src/mongo/s/client/sharding_network_connection_hook.h @@ -45,18 +45,18 @@ public: /** * Checks that the given host is valid to be used in this sharded cluster, based on its - * isMaster response. + * "hello" response. */ Status validateHost(const HostAndPort& remoteHost, const BSONObj& request, - const executor::RemoteCommandResponse& isMasterReply) override; + const executor::RemoteCommandResponse& helloReply) override; /** * Implementation of validateHost can be called without a ShardingNetworkConnectionHook * instance. */ static Status validateHostImpl(const HostAndPort& remoteHost, - const executor::RemoteCommandResponse& isMasterReply); + const executor::RemoteCommandResponse& helloReply); /** * Makes a SetShardVersion request for initializing sharding information on the new connection. diff --git a/src/mongo/s/cluster_commands_helpers.cpp b/src/mongo/s/cluster_commands_helpers.cpp index 5c0469ab245..805be79a7b9 100644 --- a/src/mongo/s/cluster_commands_helpers.cpp +++ b/src/mongo/s/cluster_commands_helpers.cpp @@ -131,13 +131,14 @@ namespace { * caller. */ std::vector<AsyncRequestsSender::Request> buildVersionedRequestsForTargetedShards( - OperationContext* opCtx, + boost::intrusive_ptr<ExpressionContext> expCtx, const NamespaceString& nss, const ChunkManager& cm, const std::set<ShardId>& shardsToSkip, const BSONObj& cmdObj, const BSONObj& query, const BSONObj& collation) { + auto opCtx = expCtx->opCtx; auto cmdToSend = cmdObj; @@ -170,7 +171,6 @@ std::vector<AsyncRequestsSender::Request> buildVersionedRequestsForTargetedShard CollatorFactoryInterface::get(opCtx->getServiceContext())->makeFromBSON(collation)); } - auto expCtx = make_intrusive<ExpressionContext>(opCtx, std::move(collator), nss); cm.getShardIdsForQuery(expCtx, query, collation, &shardIds); for (const ShardId& shardId : shardIds) { @@ -401,11 +401,28 @@ std::vector<AsyncRequestsSender::Response> scatterGatherVersionedTargetByRouting const ReadPreferenceSetting& readPref, Shard::RetryPolicy retryPolicy, const BSONObj& query, + const BSONObj& collation, + const boost::optional<BSONObj>& letParameters, + const boost::optional<LegacyRuntimeConstants>& runtimeConstants) { + auto expCtx = makeExpressionContextWithDefaultsForTargeter( + opCtx, nss, collation, boost::none /*explainVerbosity*/, letParameters, runtimeConstants); + return scatterGatherVersionedTargetByRoutingTable( + expCtx, dbName, nss, cm, cmdObj, readPref, retryPolicy, query, collation); +} + +[[nodiscard]] std::vector<AsyncRequestsSender::Response> scatterGatherVersionedTargetByRoutingTable( + boost::intrusive_ptr<ExpressionContext> expCtx, + StringData dbName, + const NamespaceString& nss, + const ChunkManager& cm, + const BSONObj& cmdObj, + const ReadPreferenceSetting& readPref, + Shard::RetryPolicy retryPolicy, + const BSONObj& query, const BSONObj& collation) { const auto requests = buildVersionedRequestsForTargetedShards( - opCtx, nss, cm, {} /* shardsToSkip */, cmdObj, query, collation); - - return gatherResponses(opCtx, dbName, readPref, retryPolicy, requests); + expCtx, nss, cm, {} /* shardsToSkip */, cmdObj, query, collation); + return gatherResponses(expCtx->opCtx, dbName, readPref, retryPolicy, requests); } std::vector<AsyncRequestsSender::Response> @@ -419,9 +436,13 @@ scatterGatherVersionedTargetByRoutingTableNoThrowOnStaleShardVersionErrors( const ReadPreferenceSetting& readPref, Shard::RetryPolicy retryPolicy, const BSONObj& query, - const BSONObj& collation) { + const BSONObj& collation, + const boost::optional<BSONObj>& letParameters, + const boost::optional<LegacyRuntimeConstants>& runtimeConstants) { + auto expCtx = makeExpressionContextWithDefaultsForTargeter( + opCtx, nss, collation, boost::none /*explainVerbosity*/, letParameters, runtimeConstants); const auto requests = buildVersionedRequestsForTargetedShards( - opCtx, nss, cm, shardsToSkip, cmdObj, query, collation); + expCtx, nss, cm, shardsToSkip, cmdObj, query, collation); return gatherResponsesNoThrowOnStaleShardVersionErrors( opCtx, dbName, readPref, retryPolicy, requests); @@ -456,6 +477,12 @@ AsyncRequestsSender::Response executeCommandAgainstShardWithMinKeyChunk( const BSONObj& cmdObj, const ReadPreferenceSetting& readPref, Shard::RetryPolicy retryPolicy) { + auto expCtx = makeExpressionContextWithDefaultsForTargeter(opCtx, + nss, + BSONObj() /*collation*/, + boost::none /*explainVerbosity*/, + boost::none /*letParameters*/, + boost::none /*runtimeConstants*/); const auto query = cm.isSharded() ? cm.getShardKeyPattern().getKeyPattern().globalMin() : BSONObj(); @@ -466,7 +493,7 @@ AsyncRequestsSender::Response executeCommandAgainstShardWithMinKeyChunk( readPref, retryPolicy, buildVersionedRequestsForTargetedShards( - opCtx, nss, cm, {} /* shardsToSkip */, cmdObj, query, BSONObj() /* collation */)); + expCtx, nss, cm, {} /* shardsToSkip */, cmdObj, query, BSONObj() /* collation */)); return std::move(responses.front()); } @@ -649,10 +676,14 @@ std::vector<std::pair<ShardId, BSONObj>> getVersionedRequestsForTargetedShards( const ChunkManager& cm, const BSONObj& cmdObj, const BSONObj& query, - const BSONObj& collation) { + const BSONObj& collation, + const boost::optional<BSONObj>& letParameters, + const boost::optional<LegacyRuntimeConstants>& runtimeConstants) { + auto expCtx = makeExpressionContextWithDefaultsForTargeter( + opCtx, nss, collation, boost::none /*explainVerbosity*/, letParameters, runtimeConstants); std::vector<std::pair<ShardId, BSONObj>> requests; auto ars_requests = buildVersionedRequestsForTargetedShards( - opCtx, nss, cm, {} /* shardsToSkip */, cmdObj, query, collation); + expCtx, nss, cm, {} /* shardsToSkip */, cmdObj, query, collation); std::transform(std::make_move_iterator(ars_requests.begin()), std::make_move_iterator(ars_requests.end()), std::back_inserter(requests), diff --git a/src/mongo/s/cluster_commands_helpers.h b/src/mongo/s/cluster_commands_helpers.h index f342b7f799e..086c00b03f8 100644 --- a/src/mongo/s/cluster_commands_helpers.h +++ b/src/mongo/s/cluster_commands_helpers.h @@ -86,7 +86,8 @@ boost::intrusive_ptr<ExpressionContext> makeExpressionContextWithDefaultsForTarg * Dispatches all the specified requests in parallel and waits until all complete, returning a * vector of the same size and positions as that of 'requests'. * - * Throws StaleConfigException if any remote returns a stale shardVersion error. + * Throws StaleConfig if any of the remotes returns that error, regardless of what the other errors + * are. */ std::vector<AsyncRequestsSender::Response> gatherResponses( OperationContext* opCtx, @@ -166,7 +167,7 @@ std::vector<AsyncRequestsSender::Response> scatterGatherUnversionedTargetAllShar * Utility for dispatching versioned commands on a namespace, deciding which shards to * target by applying the passed-in query and collation to the local routing table cache. * - * Does not retry on StaleConfigException. + * Does not retry on StaleConfig errors. */ std::vector<AsyncRequestsSender::Response> scatterGatherVersionedTargetByRoutingTable( OperationContext* opCtx, @@ -177,8 +178,24 @@ std::vector<AsyncRequestsSender::Response> scatterGatherVersionedTargetByRouting const ReadPreferenceSetting& readPref, Shard::RetryPolicy retryPolicy, const BSONObj& query, - const BSONObj& collation); + const BSONObj& collation, + const boost::optional<BSONObj>& letParameters, + const boost::optional<LegacyRuntimeConstants>& runtimeConstants); +/** + * This overload is for callers which already have a fully initialized 'ExpressionContext' (e.g. + * callers from the aggregation framework). Most callers should prefer the overload above. + */ +[[nodiscard]] std::vector<AsyncRequestsSender::Response> scatterGatherVersionedTargetByRoutingTable( + boost::intrusive_ptr<ExpressionContext> expCtx, + StringData dbName, + const NamespaceString& nss, + const ChunkManager& cm, + const BSONObj& cmdObj, + const ReadPreferenceSetting& readPref, + Shard::RetryPolicy retryPolicy, + const BSONObj& query, + const BSONObj& collation); /** * Utility for dispatching versioned commands on a namespace, deciding which shards to @@ -186,7 +203,7 @@ std::vector<AsyncRequestsSender::Response> scatterGatherVersionedTargetByRouting * * Callers can specify shards to skip, even if these shards would be otherwise targeted. * - * Allows StaleConfigException errors to append to the response list. + * Allows StaleConfig errors to append to the response list. */ std::vector<AsyncRequestsSender::Response> scatterGatherVersionedTargetByRoutingTableNoThrowOnStaleShardVersionErrors( @@ -199,7 +216,9 @@ scatterGatherVersionedTargetByRoutingTableNoThrowOnStaleShardVersionErrors( const ReadPreferenceSetting& readPref, Shard::RetryPolicy retryPolicy, const BSONObj& query, - const BSONObj& collation); + const BSONObj& collation, + const boost::optional<BSONObj>& letParameters, + const boost::optional<LegacyRuntimeConstants>& runtimeConstants); /** * Utility for dispatching commands against the primary of a database and attaching the appropriate @@ -217,7 +236,7 @@ AsyncRequestsSender::Response executeCommandAgainstDatabasePrimary( * Utility for dispatching commands against the shard with the MinKey chunk for the namespace and * attaching the appropriate shard version. * - * Does not retry on StaleConfigException. + * Does not retry on StaleConfig errors. */ AsyncRequestsSender::Response executeCommandAgainstShardWithMinKeyChunk( OperationContext* opCtx, @@ -291,7 +310,9 @@ std::vector<std::pair<ShardId, BSONObj>> getVersionedRequestsForTargetedShards( const ChunkManager& cm, const BSONObj& cmdObj, const BSONObj& query, - const BSONObj& collation); + const BSONObj& collation, + const boost::optional<BSONObj>& letParameters, + const boost::optional<LegacyRuntimeConstants>& runtimeConstants); /** * If the command is running in a transaction, returns the proper routing table to use for targeting diff --git a/src/mongo/s/commands/SConscript b/src/mongo/s/commands/SConscript index be0eb7d5203..916d1448fe3 100644 --- a/src/mongo/s/commands/SConscript +++ b/src/mongo/s/commands/SConscript @@ -84,7 +84,6 @@ env.Library( 'cluster_set_allow_migrations_cmd.cpp', 'cluster_set_cluster_parameter_cmd.cpp', 'cluster_set_feature_compatibility_version_cmd.cpp', - 'cluster_set_free_monitoring_cmd.cpp' if get_option("enable-free-mon") == 'on' else [], 'cluster_set_index_commit_quorum_cmd.cpp', 'cluster_set_user_write_block_mode_command.cpp', 'cluster_shard_collection_cmd.cpp', @@ -102,6 +101,7 @@ env.Library( LIBDEPS_PRIVATE=[ '$BUILD_DIR/mongo/db/api_parameters', '$BUILD_DIR/mongo/db/auth/auth_checks', + '$BUILD_DIR/mongo/db/catalog/collection_uuid_mismatch_info', '$BUILD_DIR/mongo/db/change_stream_options_manager', '$BUILD_DIR/mongo/db/commands/cluster_server_parameter_cmds_idl', '$BUILD_DIR/mongo/db/commands/core', diff --git a/src/mongo/s/commands/cluster_coll_stats_cmd.cpp b/src/mongo/s/commands/cluster_coll_stats_cmd.cpp index f85ae09aeb2..d3067b4769e 100644 --- a/src/mongo/s/commands/cluster_coll_stats_cmd.cpp +++ b/src/mongo/s/commands/cluster_coll_stats_cmd.cpp @@ -232,8 +232,10 @@ public: opCtx, this, CommandHelpers::filterCommandRequestForPassthrough(cmdObjToSend)), ReadPreferenceSetting::get(opCtx), Shard::RetryPolicy::kIdempotent, - {}, - {}); + {} /*query*/, + {} /*collation*/, + boost::none /*letParameters*/, + boost::none /*runtimeConstants*/); BSONObjBuilder shardStats; std::map<std::string, long long> counts; diff --git a/src/mongo/s/commands/cluster_collection_mod_cmd.cpp b/src/mongo/s/commands/cluster_collection_mod_cmd.cpp index d295ca18c71..42e318aaa57 100644 --- a/src/mongo/s/commands/cluster_collection_mod_cmd.cpp +++ b/src/mongo/s/commands/cluster_collection_mod_cmd.cpp @@ -33,6 +33,7 @@ #include "mongo/db/auth/authorization_checks.h" #include "mongo/db/auth/authorization_session.h" +#include "mongo/db/catalog/collection_uuid_mismatch_info.h" #include "mongo/db/coll_mod_gen.h" #include "mongo/db/coll_mod_reply_validation.h" #include "mongo/db/commands.h" @@ -93,8 +94,17 @@ public: "namespace"_attr = nss, "command"_attr = redact(cmdObj)); - const auto dbInfo = - uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, cmd.getDbName())); + auto swDbInfo = Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, cmd.getDbName()); + if (swDbInfo == ErrorCodes::NamespaceNotFound) { + uassert(CollectionUUIDMismatchInfo(cmd.getDbName().toString(), + *cmd.getCollectionUUID(), + nss.coll().toString(), + boost::none), + "Database does not exist", + !cmd.getCollectionUUID()); + } + const auto dbInfo = uassertStatusOK(swDbInfo); + ShardsvrCollMod collModCommand(nss); collModCommand.setCollModRequest(cmd.getCollModRequest()); auto cmdResponse = diff --git a/src/mongo/s/commands/cluster_convert_to_capped_cmd.cpp b/src/mongo/s/commands/cluster_convert_to_capped_cmd.cpp index 8c657bc453f..c6206bce5be 100644 --- a/src/mongo/s/commands/cluster_convert_to_capped_cmd.cpp +++ b/src/mongo/s/commands/cluster_convert_to_capped_cmd.cpp @@ -51,8 +51,17 @@ bool nonShardedCollectionCommandPassthrough(OperationContext* opCtx, str::stream() << "Can't do command: " << cmdName << " on a sharded collection", !cm.isSharded()); - auto responses = scatterGatherVersionedTargetByRoutingTable( - opCtx, dbName, nss, cm, cmdObj, ReadPreferenceSetting::get(opCtx), retryPolicy, {}, {}); + auto responses = scatterGatherVersionedTargetByRoutingTable(opCtx, + dbName, + nss, + cm, + cmdObj, + ReadPreferenceSetting::get(opCtx), + retryPolicy, + {} /*query*/, + {} /*collation*/, + boost::none /*letParameters*/, + boost::none /*runtimeConstants*/); invariant(responses.size() == 1); const auto cmdResponse = uassertStatusOK(std::move(responses.front().swResponse)); diff --git a/src/mongo/s/commands/cluster_count_cmd.cpp b/src/mongo/s/commands/cluster_count_cmd.cpp index 48545c7c0d7..94e6c7c0eee 100644 --- a/src/mongo/s/commands/cluster_count_cmd.cpp +++ b/src/mongo/s/commands/cluster_count_cmd.cpp @@ -131,7 +131,9 @@ public: ReadPreferenceSetting::get(opCtx), Shard::RetryPolicy::kIdempotent, countRequest.getQuery(), - collation); + collation, + boost::none /*letParameters*/, + boost::none /*runtimeConstants*/); } catch (const ExceptionFor<ErrorCodes::CommandOnShardedViewNotSupportedOnMongod>& ex) { // Rewrite the count command as an aggregation. auto countRequest = CountCommandRequest::parse(IDLParserErrorContext("count"), cmdObj); @@ -235,7 +237,9 @@ public: ReadPreferenceSetting::get(opCtx), Shard::RetryPolicy::kIdempotent, targetingQuery, - targetingCollation); + targetingCollation, + boost::none /*letParameters*/, + boost::none /*runtimeConstants*/); } catch (const ExceptionFor<ErrorCodes::CommandOnShardedViewNotSupportedOnMongod>& ex) { CountCommandRequest countRequest(NamespaceStringOrUUID(NamespaceString{})); try { @@ -284,14 +288,14 @@ private: BSONElement l = cmd["limit"]; if (s.isNumber()) { - num = num - s.numberLong(); + num = num - s.safeNumberLong(); if (num < 0) { num = 0; } } if (l.isNumber()) { - long long limit = l.numberLong(); + auto limit = l.safeNumberLong(); if (limit < 0) { limit = -limit; } diff --git a/src/mongo/s/commands/cluster_create_indexes_cmd.cpp b/src/mongo/s/commands/cluster_create_indexes_cmd.cpp index 8c4c8af8784..cadbe748d0f 100644 --- a/src/mongo/s/commands/cluster_create_indexes_cmd.cpp +++ b/src/mongo/s/commands/cluster_create_indexes_cmd.cpp @@ -108,8 +108,10 @@ public: applyReadWriteConcern(opCtx, this, cmdToBeSent)), ReadPreferenceSetting(ReadPreference::PrimaryOnly), Shard::RetryPolicy::kNoRetry, - BSONObj() /* query */, - BSONObj() /* collation */); + BSONObj() /*query*/, + BSONObj() /*collation*/, + boost::none /*letParameters*/, + boost::none /*runtimeConstants*/); std::string errmsg; const bool ok = diff --git a/src/mongo/s/commands/cluster_data_size_cmd.cpp b/src/mongo/s/commands/cluster_data_size_cmd.cpp index 7d645d1a51c..1c87c6a98f0 100644 --- a/src/mongo/s/commands/cluster_data_size_cmd.cpp +++ b/src/mongo/s/commands/cluster_data_size_cmd.cpp @@ -85,8 +85,10 @@ public: opCtx, this, CommandHelpers::filterCommandRequestForPassthrough(cmdObj)), ReadPreferenceSetting::get(opCtx), Shard::RetryPolicy::kIdempotent, - {}, - {}); + {} /*query*/, + {} /*collation*/, + boost::none /*letParameters*/, + boost::none /*runtimeConstants*/); // yes these are doubles... double size = 0; diff --git a/src/mongo/s/commands/cluster_distinct_cmd.cpp b/src/mongo/s/commands/cluster_distinct_cmd.cpp index 70ae5f4b671..f34377258e4 100644 --- a/src/mongo/s/commands/cluster_distinct_cmd.cpp +++ b/src/mongo/s/commands/cluster_distinct_cmd.cpp @@ -132,7 +132,9 @@ public: ReadPreferenceSetting::get(opCtx), Shard::RetryPolicy::kIdempotent, targetingQuery, - targetingCollation); + targetingCollation, + boost::none /*letParameters*/, + boost::none /*runtimeConstants*/); } catch (const ExceptionFor<ErrorCodes::CommandOnShardedViewNotSupportedOnMongod>& ex) { auto parsedDistinct = ParsedDistinct::parse( opCtx, ex->getNamespace(), cmdObj, ExtensionsCallbackNoop(), true); @@ -217,7 +219,9 @@ public: ReadPreferenceSetting::get(opCtx), Shard::RetryPolicy::kIdempotent, query, - collation); + collation, + boost::none /*letParameters*/, + boost::none /*runtimeConstants*/); } catch (const ExceptionFor<ErrorCodes::CommandOnShardedViewNotSupportedOnMongod>& ex) { auto parsedDistinct = ParsedDistinct::parse( opCtx, ex->getNamespace(), cmdObj, ExtensionsCallbackNoop(), true); diff --git a/src/mongo/s/commands/cluster_drop_collection_cmd.cpp b/src/mongo/s/commands/cluster_drop_collection_cmd.cpp index e16f473ab3c..40e9196371f 100644 --- a/src/mongo/s/commands/cluster_drop_collection_cmd.cpp +++ b/src/mongo/s/commands/cluster_drop_collection_cmd.cpp @@ -33,6 +33,7 @@ #include "mongo/base/status.h" #include "mongo/db/auth/authorization_session.h" +#include "mongo/db/catalog/collection_uuid_mismatch_info.h" #include "mongo/db/commands.h" #include "mongo/db/drop_gen.h" #include "mongo/db/operation_context.h" @@ -116,6 +117,13 @@ public: // Ensure our reply conforms to the IDL-defined reply structure. return DropReply::parse({"drop"}, resultObj); } catch (const ExceptionFor<ErrorCodes::NamespaceNotFound>&) { + uassert(CollectionUUIDMismatchInfo(request().getDbName().toString(), + *request().getCollectionUUID(), + request().getNamespace().coll().toString(), + boost::none), + "Database does not exist", + !request().getCollectionUUID()); + // If the namespace isn't found, treat the drop as a success but inform about the // failure. DropReply reply; diff --git a/src/mongo/s/commands/cluster_filemd5_cmd.cpp b/src/mongo/s/commands/cluster_filemd5_cmd.cpp index d351a7be7a3..3248d9ce0ba 100644 --- a/src/mongo/s/commands/cluster_filemd5_cmd.cpp +++ b/src/mongo/s/commands/cluster_filemd5_cmd.cpp @@ -98,7 +98,9 @@ public: ReadPreferenceSetting::get(opCtx), Shard::RetryPolicy::kIdempotent, routingQuery, - CollationSpec::kSimpleSpec); + CollationSpec::kSimpleSpec, + boost::none /*letParameters*/, + boost::none /*runtimeConstants*/); invariant(shardResults.size() == 1); const auto shardResponse = uassertStatusOK(std::move(shardResults[0].swResponse)); uassertStatusOK(shardResponse.status); diff --git a/src/mongo/s/commands/cluster_find_and_modify_cmd.cpp b/src/mongo/s/commands/cluster_find_and_modify_cmd.cpp index 740d1561487..ff7e8029c67 100644 --- a/src/mongo/s/commands/cluster_find_and_modify_cmd.cpp +++ b/src/mongo/s/commands/cluster_find_and_modify_cmd.cpp @@ -305,6 +305,27 @@ void handleWouldChangeOwningShardErrorTransactionLegacy(OperationContext* opCtx, } } +ShardId targetSingleShard(boost::intrusive_ptr<ExpressionContext> expCtx, + const ChunkManager& cm, + const BSONObj& query, + const BSONObj& collation) { + std::set<ShardId> shardIds; + + // For now, set bypassIsFieldHashedCheck to be true in order to skip the + // isFieldHashedCheck in the special case where _id is hashed and used as the shard + // key. This means that we always assume that a findAndModify request using _id is + // targetable to a single shard. + cm.getShardIdsForQuery(expCtx, query, collation, &shardIds, true); + + uassert(ErrorCodes::ShardKeyNotFound, + str::stream() << "Query with sharded findAndModify expected to only target one " + "shard, but the query targeted " + << shardIds.size() << " shard(s)", + shardIds.size() == 1); + + return *shardIds.begin(); +} + class FindAndModifyCmd : public BasicCommand { public: FindAndModifyCmd() @@ -392,12 +413,14 @@ public: const BSONObj collation = getCollation(cmdObj); const auto let = getLet(cmdObj); const auto rc = getLegacyRuntimeConstants(cmdObj); - const BSONObj shardKey = - getShardKey(opCtx, cm, nss, query, collation, verbosity, let, rc); - const auto chunk = cm.findIntersectingChunk(shardKey, collation); + ShardId shardId; + { + auto expCtx = makeExpressionContextWithDefaultsForTargeter( + opCtx, nss, collation, verbosity, let, rc); + shardId = targetSingleShard(expCtx, cm, query, collation); + } - shard = uassertStatusOK( - Grid::get(opCtx)->shardRegistry()->getShard(opCtx, chunk.getShardId())); + shard = uassertStatusOK(Grid::get(opCtx)->shardRegistry()->getShard(opCtx, shardId)); } else { shard = uassertStatusOK(Grid::get(opCtx)->shardRegistry()->getShard(opCtx, cm.dbPrimary())); @@ -472,18 +495,15 @@ public: const BSONObj collation = getCollation(cmdObjForShard); const auto let = getLet(cmdObjForShard); const auto rc = getLegacyRuntimeConstants(cmdObjForShard); - const BSONObj shardKey = - getShardKey(opCtx, cm, nss, query, collation, boost::none, let, rc); - - // For now, set bypassIsFieldHashedCheck to be true in order to skip the - // isFieldHashedCheck in the special case where _id is hashed and used as the shard key. - // This means that we always assume that a findAndModify request using _id is targetable - // to a single shard. - auto chunk = cm.findIntersectingChunk(shardKey, collation, true); - + ShardId shardId; + { + auto expCtx = makeExpressionContextWithDefaultsForTargeter( + opCtx, nss, collation, boost::none, let, rc); + shardId = targetSingleShard(expCtx, cm, query, collation); + } _runCommand(opCtx, - chunk.getShardId(), - cm.getVersion(chunk.getShardId()), + shardId, + cm.getVersion(shardId), boost::none, nss, applyReadWriteConcern(opCtx, this, cmdObjForShard), diff --git a/src/mongo/s/commands/cluster_find_cmd.h b/src/mongo/s/commands/cluster_find_cmd.h index 87d25934d0c..7d58f7ab0c4 100644 --- a/src/mongo/s/commands/cluster_find_cmd.h +++ b/src/mongo/s/commands/cluster_find_cmd.h @@ -152,7 +152,9 @@ public: ReadPreferenceSetting::get(opCtx), Shard::RetryPolicy::kIdempotent, findCommand->getFilter(), - findCommand->getCollation()); + findCommand->getCollation(), + findCommand->getLet(), + findCommand->getLegacyRuntimeConstants()); millisElapsed = timer.millis(); const char* mongosStageName = @@ -199,11 +201,6 @@ public: Impl::checkCanRunHere(opCtx); - ON_BLOCK_EXIT([opCtx] { - Grid::get(opCtx)->catalogCache()->checkAndRecordOperationBlockedByRefresh( - opCtx, mongo::LogicalOp::opQuery); - }); - auto findCommand = _parseCmdObjectToFindCommandRequest(opCtx, ns(), _request.body); const boost::intrusive_ptr<ExpressionContext> expCtx; diff --git a/src/mongo/s/commands/cluster_get_cluster_parameter_cmd.cpp b/src/mongo/s/commands/cluster_get_cluster_parameter_cmd.cpp index 1058e6ebafd..4b6d49ebae1 100644 --- a/src/mongo/s/commands/cluster_get_cluster_parameter_cmd.cpp +++ b/src/mongo/s/commands/cluster_get_cluster_parameter_cmd.cpp @@ -66,6 +66,11 @@ public: using InvocationBase::InvocationBase; Reply typedRun(OperationContext* opCtx) { + uassert(ErrorCodes::UnknownFeatureCompatibilityVersion, + "FCV is not yet initialized, retry the command after FCV initialization has " + "completed", + serverGlobalParams.featureCompatibility.isVersionInitialized()); + uassert( ErrorCodes::IllegalOperation, "featureFlagClusterWideConfig not enabled", diff --git a/src/mongo/s/commands/cluster_index_filter_cmd.cpp b/src/mongo/s/commands/cluster_index_filter_cmd.cpp index f6c5bd37777..48aec39e5cb 100644 --- a/src/mongo/s/commands/cluster_index_filter_cmd.cpp +++ b/src/mongo/s/commands/cluster_index_filter_cmd.cpp @@ -104,7 +104,9 @@ public: ReadPreferenceSetting::get(opCtx), Shard::RetryPolicy::kIdempotent, query, - CollationSpec::kSimpleSpec); + CollationSpec::kSimpleSpec, + boost::none /*letParameters*/, + boost::none /*runtimeConstants*/); // Sort shard responses by shard id. std::sort(shardResponses.begin(), diff --git a/src/mongo/s/commands/cluster_map_reduce_agg.cpp b/src/mongo/s/commands/cluster_map_reduce_agg.cpp index 52a61ab9325..746ba39c7bc 100644 --- a/src/mongo/s/commands/cluster_map_reduce_agg.cpp +++ b/src/mongo/s/commands/cluster_map_reduce_agg.cpp @@ -187,6 +187,7 @@ bool runAggregationMapReduce(OperationContext* opCtx, cm, involvedNamespaces, false, // hasChangeStream + false, // startsWithDocuments true, // allowedToPassthrough false); // perShardCursor try { @@ -230,14 +231,15 @@ bool runAggregationMapReduce(OperationContext* opCtx, namespaces, privileges, &tempResults, - false)); // hasChangeStream + false, // hasChangeStream + false)); // startsWithDocuments break; } case cluster_aggregation_planner::AggregationTargeter::TargetingPolicy:: kSpecificShardOnly: { // It should not be possible to pass $_passthroughToShard to a map reduce command. - MONGO_UNREACHABLE_TASSERT(6273803); + MONGO_UNREACHABLE_TASSERT(6273805); } } } catch (DBException& e) { diff --git a/src/mongo/s/commands/cluster_move_range_cmd.cpp b/src/mongo/s/commands/cluster_move_range_cmd.cpp index c06ca4a5c47..fd2f77ea9f5 100644 --- a/src/mongo/s/commands/cluster_move_range_cmd.cpp +++ b/src/mongo/s/commands/cluster_move_range_cmd.cpp @@ -100,7 +100,7 @@ public: uassert(ErrorCodes::Unauthorized, "Unauthorized", AuthorizationSession::get(opCtx->getClient()) - ->isAuthorizedForActionsOnResource(ResourcePattern::forClusterResource(), + ->isAuthorizedForActionsOnResource(ResourcePattern::forExactNamespace(ns()), ActionType::moveChunk)); } }; diff --git a/src/mongo/s/commands/cluster_plan_cache_clear_cmd.cpp b/src/mongo/s/commands/cluster_plan_cache_clear_cmd.cpp index 96499720ea6..e3411fbda2c 100644 --- a/src/mongo/s/commands/cluster_plan_cache_clear_cmd.cpp +++ b/src/mongo/s/commands/cluster_plan_cache_clear_cmd.cpp @@ -110,7 +110,9 @@ bool ClusterPlanCacheClearCmd::run(OperationContext* opCtx, ReadPreferenceSetting::get(opCtx), Shard::RetryPolicy::kIdempotent, query, - CollationSpec::kSimpleSpec); + CollationSpec::kSimpleSpec, + boost::none /*letParameters*/, + boost::none /*runtimeConstants*/); // Sort shard responses by shard id. std::sort(shardResponses.begin(), diff --git a/src/mongo/s/commands/cluster_profile_cmd.cpp b/src/mongo/s/commands/cluster_profile_cmd.cpp index c2564c19b1d..2c2b09792da 100644 --- a/src/mongo/s/commands/cluster_profile_cmd.cpp +++ b/src/mongo/s/commands/cluster_profile_cmd.cpp @@ -33,6 +33,7 @@ #include "mongo/db/commands.h" #include "mongo/db/commands/profile_common.h" #include "mongo/db/commands/profile_gen.h" +#include "mongo/db/commands/set_profiling_filter_globally_cmd.h" #include "mongo/db/profile_filter_impl.h" namespace mongo { @@ -85,5 +86,7 @@ protected: } profileCmd; +SetProfilingFilterGloballyCmd setProfilingFilterGloballyCmd; + } // namespace } // namespace mongo diff --git a/src/mongo/s/commands/cluster_rename_collection_cmd.cpp b/src/mongo/s/commands/cluster_rename_collection_cmd.cpp index 4d4d88907a6..a5c3d9d385c 100644 --- a/src/mongo/s/commands/cluster_rename_collection_cmd.cpp +++ b/src/mongo/s/commands/cluster_rename_collection_cmd.cpp @@ -32,6 +32,7 @@ #include "mongo/platform/basic.h" #include "mongo/db/auth/authorization_session.h" +#include "mongo/db/catalog/collection_uuid_mismatch_info.h" #include "mongo/db/commands.h" #include "mongo/db/commands/rename_collection_common.h" #include "mongo/db/commands/rename_collection_gen.h" @@ -74,6 +75,21 @@ public: "Can't rename a collection to itself", fromNss != toNss); + if (fromNss.isTimeseriesBucketsCollection()) { + uassert( + ErrorCodes::IllegalOperation, + "Renaming system.buckets collections is not allowed", + AuthorizationSession::get(opCtx->getClient()) + ->isAuthorizedForActionsOnResource(ResourcePattern::forClusterResource(), + ActionType::setUserWriteBlockMode)); + + uassert(ErrorCodes::IllegalOperation, + str::stream() + << "Cannot rename time-series buckets collection {" << fromNss.ns() + << "} to a non-time-series buckets namespace {" << toNss.ns() << "}", + toNss.isTimeseriesBucketsCollection()); + } + RenameCollectionRequest renameCollReq(request().getTo()); renameCollReq.setStayTemp(request().getStayTemp()); renameCollReq.setExpectedSourceUUID(request().getCollectionUUID()); @@ -90,9 +106,22 @@ public: ShardsvrRenameCollection renameCollRequest(fromNss); renameCollRequest.setDbName(fromNss.db()); renameCollRequest.setRenameCollectionRequest(renameCollReq); + renameCollRequest.setAllowEncryptedCollectionRename( + AuthorizationSession::get(opCtx->getClient()) + ->isAuthorizedForActionsOnResource(ResourcePattern::forClusterResource(), + ActionType::setUserWriteBlockMode)); auto catalogCache = Grid::get(opCtx)->catalogCache(); - const auto dbInfo = uassertStatusOK(catalogCache->getDatabase(opCtx, fromNss.db())); + auto swDbInfo = Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, fromNss.db()); + if (swDbInfo == ErrorCodes::NamespaceNotFound) { + uassert(CollectionUUIDMismatchInfo(fromNss.db().toString(), + *request().getCollectionUUID(), + fromNss.coll().toString(), + boost::none), + "Database does not exist", + !request().getCollectionUUID()); + } + const auto dbInfo = uassertStatusOK(swDbInfo); auto cri = uassertStatusOK(catalogCache->getCollectionRoutingInfo(opCtx, fromNss)); auto shard = uassertStatusOK( diff --git a/src/mongo/s/commands/cluster_set_free_monitoring_cmd.cpp b/src/mongo/s/commands/cluster_set_free_monitoring_cmd.cpp deleted file mode 100644 index d54cd80b219..00000000000 --- a/src/mongo/s/commands/cluster_set_free_monitoring_cmd.cpp +++ /dev/null @@ -1,75 +0,0 @@ -/** - * Copyright (C) 2018-present MongoDB, Inc. - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the Server Side Public License, version 1, - * as published by MongoDB, Inc. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * Server Side Public License for more details. - * - * You should have received a copy of the Server Side Public License - * along with this program. If not, see - * <http://www.mongodb.com/licensing/server-side-public-license>. - * - * As a special exception, the copyright holders give permission to link the - * code of portions of this program with the OpenSSL library under certain - * conditions as described in each individual source file and distribute - * linked combinations including the program with the OpenSSL library. You - * must comply with the Server Side Public License in all respects for - * all of the code used other than as permitted herein. If you modify file(s) - * with this exception, you may extend this exception to your version of the - * file(s), but you are not obligated to do so. If you do not wish to do so, - * delete this exception statement from your version. If you delete this - * exception statement from all source files in the program, then also delete - * it in the license file. - */ - -#include "mongo/platform/basic.h" - -#include "mongo/db/auth/authorization_session.h" -#include "mongo/db/commands.h" - -namespace mongo { -namespace { - -class ClusterSetFreeMonitoring : public BasicCommand { -public: - ClusterSetFreeMonitoring() : BasicCommand("setFreeMonitoring") {} - - AllowedOnSecondary secondaryAllowed(ServiceContext*) const final { - return AllowedOnSecondary::kNever; - } - - bool supportsWriteConcern(const BSONObj& cmd) const final { - return false; - } - - std::string help() const final { - return "setFreeMonitoring command must be run against mongod instances"; - } - - Status checkAuthForCommand(Client* client, - const std::string& dbname, - const BSONObj& cmdObj) const final { - if (!AuthorizationSession::get(client)->isAuthorizedForActionsOnResource( - ResourcePattern::forClusterResource(), ActionType::setFreeMonitoring)) { - return Status(ErrorCodes::Unauthorized, "Unauthorized"); - } - return Status::OK(); - } - - bool run(OperationContext* opCtx, - const std::string& dbname, - const BSONObj& cmdObj, - BSONObjBuilder& result) final { - uasserted(ErrorCodes::CommandFailed, help()); - return true; - } - -} clusterSetFreeMonitoring; - -} // namespace -} // namespace mongo diff --git a/src/mongo/s/commands/cluster_set_index_commit_quorum_cmd.cpp b/src/mongo/s/commands/cluster_set_index_commit_quorum_cmd.cpp index 8727a9da078..f494e41a6a6 100644 --- a/src/mongo/s/commands/cluster_set_index_commit_quorum_cmd.cpp +++ b/src/mongo/s/commands/cluster_set_index_commit_quorum_cmd.cpp @@ -117,8 +117,10 @@ public: opCtx, this, CommandHelpers::filterCommandRequestForPassthrough(cmdObj)), ReadPreferenceSetting::get(opCtx), Shard::RetryPolicy::kNotIdempotent, - BSONObj() /* query */, - BSONObj() /* collation */); + BSONObj() /*query*/, + BSONObj() /*collation*/, + boost::none /*letParameters*/, + boost::none /*runtimeConstants*/); std::string errmsg; const bool ok = diff --git a/src/mongo/s/commands/cluster_validate_cmd.cpp b/src/mongo/s/commands/cluster_validate_cmd.cpp index 52285704516..285d23fc29d 100644 --- a/src/mongo/s/commands/cluster_validate_cmd.cpp +++ b/src/mongo/s/commands/cluster_validate_cmd.cpp @@ -84,8 +84,10 @@ public: opCtx, this, CommandHelpers::filterCommandRequestForPassthrough(cmdObj)), ReadPreferenceSetting::get(opCtx), Shard::RetryPolicy::kIdempotent, - {}, - {}); + {} /*query*/, + {} /*collation*/, + boost::none /*letParameters*/, + boost::none /*runtimeConstants*/); Status firstFailedShardStatus = Status::OK(); bool isValid = true; diff --git a/src/mongo/s/commands/cluster_write_cmd.cpp b/src/mongo/s/commands/cluster_write_cmd.cpp index 10d04ba0b82..4a7982cf24f 100644 --- a/src/mongo/s/commands/cluster_write_cmd.cpp +++ b/src/mongo/s/commands/cluster_write_cmd.cpp @@ -555,22 +555,17 @@ bool ClusterWriteCmd::InvocationBase::runImpl(OperationContext* opCtx, // TODO: increase opcounters by more than one auto& debug = CurOp::get(opCtx)->debug(); - auto catalogCache = Grid::get(opCtx)->catalogCache(); switch (_batchedRequest.getBatchType()) { case BatchedCommandRequest::BatchType_Insert: for (size_t i = 0; i < numAttempts; ++i) { globalOpCounters.gotInsert(); } - catalogCache->checkAndRecordOperationBlockedByRefresh(opCtx, - mongo::LogicalOp::opInsert); debug.additiveMetrics.ninserted = response.getN(); break; case BatchedCommandRequest::BatchType_Update: for (size_t i = 0; i < numAttempts; ++i) { globalOpCounters.gotUpdate(); } - catalogCache->checkAndRecordOperationBlockedByRefresh(opCtx, - mongo::LogicalOp::opUpdate); // The response.getN() count is the sum of documents matched and upserted. if (response.isUpsertDetailsSet()) { @@ -603,8 +598,6 @@ bool ClusterWriteCmd::InvocationBase::runImpl(OperationContext* opCtx, for (size_t i = 0; i < numAttempts; ++i) { globalOpCounters.gotDelete(); } - catalogCache->checkAndRecordOperationBlockedByRefresh(opCtx, - mongo::LogicalOp::opDelete); debug.additiveMetrics.ndeleted = response.getN(); break; } diff --git a/src/mongo/s/commands/strategy.cpp b/src/mongo/s/commands/strategy.cpp index 13d061e9e6d..542815eac98 100644 --- a/src/mongo/s/commands/strategy.cpp +++ b/src/mongo/s/commands/strategy.cpp @@ -426,14 +426,6 @@ public: explicit RunInvocation(ParseAndRunCommand* parc) : _parc(parc) {} - ~RunInvocation() { - if (!_shouldAffectCommandCounter) - return; - auto opCtx = _parc->_rec->getOpCtx(); - Grid::get(opCtx)->catalogCache()->checkAndRecordOperationBlockedByRefresh( - opCtx, mongo::LogicalOp::opCommand); - } - Future<void> run(); private: @@ -442,7 +434,6 @@ private: ParseAndRunCommand* const _parc; boost::optional<RouterOperationContextSession> _routerSession; - bool _shouldAffectCommandCounter = false; }; /* @@ -898,7 +889,6 @@ Status ParseAndRunCommand::RunInvocation::_setup() { if (command->shouldAffectCommandCounter()) { globalOpCounters.gotCommand(); - _shouldAffectCommandCounter = true; } return Status::OK(); diff --git a/src/mongo/s/commands/strategy.h b/src/mongo/s/commands/strategy.h index 1e04130f657..73916229f7f 100644 --- a/src/mongo/s/commands/strategy.h +++ b/src/mongo/s/commands/strategy.h @@ -41,8 +41,8 @@ public: /** * Executes a command from either OP_QUERY or OP_MSG wire protocols. * - * Catches StaleConfigException errors and retries the command automatically after refreshing - * the metadata for the failing namespace. + * Catches StaleConfig errors and retries the command automatically after refreshing the + * metadata for the failing namespace. */ static Future<DbResponse> clientCommand(std::shared_ptr<RequestExecutionContext> rec); }; diff --git a/src/mongo/s/concurrency/locker_mongos.h b/src/mongo/s/concurrency/locker_mongos.h index f4c3d72894e..4db71e9149b 100644 --- a/src/mongo/s/concurrency/locker_mongos.h +++ b/src/mongo/s/concurrency/locker_mongos.h @@ -115,7 +115,10 @@ public: MONGO_UNREACHABLE; } - void lockRSTLComplete(OperationContext* opCtx, LockMode mode, Date_t deadline) override { + void lockRSTLComplete(OperationContext* opCtx, + LockMode mode, + Date_t deadline, + const LockTimeoutCallback& onTimeout) override { MONGO_UNREACHABLE; } diff --git a/src/mongo/s/config_server_catalog_cache_loader.cpp b/src/mongo/s/config_server_catalog_cache_loader.cpp index 3910e1e4c88..84cc73b987a 100644 --- a/src/mongo/s/config_server_catalog_cache_loader.cpp +++ b/src/mongo/s/config_server_catalog_cache_loader.cpp @@ -114,6 +114,10 @@ void ConfigServerCatalogCacheLoader::onStepUp() { MONGO_UNREACHABLE; } +void ConfigServerCatalogCacheLoader::onReplicationRollback() { + MONGO_UNREACHABLE; +} + void ConfigServerCatalogCacheLoader::shutDown() { _executor->shutdown(); _executor->join(); diff --git a/src/mongo/s/config_server_catalog_cache_loader.h b/src/mongo/s/config_server_catalog_cache_loader.h index 8c1384946f8..86af7a6b14b 100644 --- a/src/mongo/s/config_server_catalog_cache_loader.h +++ b/src/mongo/s/config_server_catalog_cache_loader.h @@ -46,6 +46,7 @@ public: void initializeReplicaSetRole(bool isPrimary) override; void onStepDown() override; void onStepUp() override; + void onReplicationRollback() override; void shutDown() override; void notifyOfCollectionVersionUpdate(const NamespaceString& nss) override; void waitForCollectionFlush(OperationContext* opCtx, const NamespaceString& nss) override; diff --git a/src/mongo/s/mongod_and_mongos_server_parameters.idl b/src/mongo/s/mongod_and_mongos_server_parameters.idl index 9d9929174e0..0a760d967d8 100644 --- a/src/mongo/s/mongod_and_mongos_server_parameters.idl +++ b/src/mongo/s/mongod_and_mongos_server_parameters.idl @@ -69,3 +69,13 @@ server_parameters: cpp_vartype: bool cpp_varname: "gEnableFinerGrainedCatalogCacheRefresh" default: false + + routingTableCacheChunkBucketSize: + description: >- + Size of the routing table cache buckets used to implement chunk grouping optimization. + set_at: [ startup ] + cpp_vartype: long long + cpp_varname: "gRoutingTableCacheChunkBucketSize" + default: 500 + validator: + gte: 0 diff --git a/src/mongo/s/mongos_main.cpp b/src/mongo/s/mongos_main.cpp index 0504b714aec..441e60a7f1a 100644 --- a/src/mongo/s/mongos_main.cpp +++ b/src/mongo/s/mongos_main.cpp @@ -341,6 +341,7 @@ void cleanupTask(const ShutdownTaskArgs& shutdownArgs) { } if (auto pool = Grid::get(opCtx)->getExecutorPool()) { + LOGV2_OPTIONS(7698300, {LogComponent::kSharding}, "Shutting down the ExecutorPool"); pool->shutdownAndJoin(); } @@ -349,6 +350,13 @@ void cleanupTask(const ShutdownTaskArgs& shutdownArgs) { } if (Grid::get(serviceContext)->isShardingInitialized()) { + // The CatalogCache must be shuted down before shutting down the CatalogCacheLoader as + // the CatalogCache may try to schedule work on CatalogCacheLoader and fail. + LOGV2_OPTIONS(7698301, {LogComponent::kSharding}, "Shutting down the CatalogCache"); + Grid::get(serviceContext)->catalogCache()->shutDownAndJoin(); + + LOGV2_OPTIONS( + 7698302, {LogComponent::kSharding}, "Shutting down the CatalogCacheLoader"); CatalogCacheLoader::get(serviceContext).shutDown(); } diff --git a/src/mongo/s/query/SConscript b/src/mongo/s/query/SConscript index f6f1ac53b05..94bfbde4878 100644 --- a/src/mongo/s/query/SConscript +++ b/src/mongo/s/query/SConscript @@ -43,6 +43,7 @@ env.Library( 'cluster_query', ], LIBDEPS_PRIVATE=[ + '$BUILD_DIR/mongo/db/catalog/collection_uuid_mismatch_info', '$BUILD_DIR/mongo/db/timeseries/timeseries_options', ] ) diff --git a/src/mongo/s/query/async_results_merger.cpp b/src/mongo/s/query/async_results_merger.cpp index 363e151fbfd..50fb6310888 100644 --- a/src/mongo/s/query/async_results_merger.cpp +++ b/src/mongo/s/query/async_results_merger.cpp @@ -846,7 +846,8 @@ void AsyncResultsMerger::_scheduleKillCursors(WithLock, OperationContext* opCtx) invariant(_killCompleteInfo); for (const auto& remote : _remotes) { - if (remote.status.isOK() && remote.cursorId && !remote.exhausted()) { + if ((remote.status.isOK() || remote.status == ErrorCodes::MaxTimeMSExpired) && + remote.cursorId && !remote.exhausted()) { BSONObj cmdObj = KillCursorsCommandRequest(_params.getNss(), {remote.cursorId}).toBSON(BSONObj{}); diff --git a/src/mongo/s/query/async_results_merger_params.idl b/src/mongo/s/query/async_results_merger_params.idl index 2b5857e6c9b..e3c4d03bdd3 100644 --- a/src/mongo/s/query/async_results_merger_params.idl +++ b/src/mongo/s/query/async_results_merger_params.idl @@ -95,4 +95,7 @@ structs: recordRemoteOpWaitTime: type: bool default: false - description: If set, records the total time spent waiting for remote operations to complete. + description: >- + This parameter is not used anymore but should stay for a while for backward + compatibility. + TODO SERVER-73120 Remove this parameter when releasing 8.0. diff --git a/src/mongo/s/query/blocking_results_merger.cpp b/src/mongo/s/query/blocking_results_merger.cpp index c1a957a311f..fc56a0d9e3b 100644 --- a/src/mongo/s/query/blocking_results_merger.cpp +++ b/src/mongo/s/query/blocking_results_merger.cpp @@ -42,7 +42,6 @@ BlockingResultsMerger::BlockingResultsMerger(OperationContext* opCtx, std::shared_ptr<executor::TaskExecutor> executor, std::unique_ptr<ResourceYielder> resourceYielder) : _tailableMode(armParams.getTailableMode().value_or(TailableModeEnum::kNormal)), - _recordRemoteOpWaitTime(armParams.getRecordRemoteOpWaitTime()), _executor(executor), _arm(opCtx, std::move(executor), std::move(armParams)), _resourceYielder(std::move(resourceYielder)) {} @@ -146,9 +145,7 @@ StatusWith<ClusterQueryResult> BlockingResultsMerger::blockUntilNext(OperationCo return _arm.nextReady(); } StatusWith<ClusterQueryResult> BlockingResultsMerger::next(OperationContext* opCtx) { - if (_recordRemoteOpWaitTime) { - CurOp::get(opCtx)->enableRecordRemoteOpWait(); - } + CurOp::get(opCtx)->ensureRecordRemoteOpWait(); // Non-tailable and tailable non-awaitData cursors always block until ready(). AwaitData // cursors wait for ready() only until a specified time limit is exceeded. diff --git a/src/mongo/s/query/blocking_results_merger.h b/src/mongo/s/query/blocking_results_merger.h index 4a368d9471a..c05cecc5da8 100644 --- a/src/mongo/s/query/blocking_results_merger.h +++ b/src/mongo/s/query/blocking_results_merger.h @@ -118,7 +118,6 @@ private: const std::function<StatusWith<stdx::cv_status>()>& waitFn) noexcept; TailableModeEnum _tailableMode; - bool _recordRemoteOpWaitTime; std::shared_ptr<executor::TaskExecutor> _executor; // In a case where we have a tailable, awaitData cursor, a call to 'next()' will block waiting diff --git a/src/mongo/s/query/cluster_aggregate.cpp b/src/mongo/s/query/cluster_aggregate.cpp index b3df71582da..7bd6808798c 100644 --- a/src/mongo/s/query/cluster_aggregate.cpp +++ b/src/mongo/s/query/cluster_aggregate.cpp @@ -37,6 +37,7 @@ #include "mongo/db/api_parameters.h" #include "mongo/db/auth/authorization_session.h" +#include "mongo/db/catalog/collection_uuid_mismatch_info.h" #include "mongo/db/client.h" #include "mongo/db/commands.h" #include "mongo/db/curop.h" @@ -309,6 +310,7 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, auto hasChangeStream = liteParsedPipeline.hasChangeStream(); auto involvedNamespaces = liteParsedPipeline.getInvolvedNamespaces(); auto shouldDoFLERewrite = ::mongo::shouldDoFLERewrite(request); + auto startsWithDocuments = liteParsedPipeline.startsWithDocuments(); // If the routing table is not already taken by the higher level, fill it now. if (!cm) { @@ -323,6 +325,14 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, sharded_agg_helpers::getExecutionNsRoutingInfo(opCtx, namespaces.executionNss); if (!executionNsRoutingInfoStatus.isOK()) { + uassert(CollectionUUIDMismatchInfo(request.getDbName().toString(), + *request.getCollectionUUID(), + request.getNamespace().coll().toString(), + boost::none), + "Database does not exist", + executionNsRoutingInfoStatus != ErrorCodes::NamespaceNotFound || + !request.getCollectionUUID()); + if (liteParsedPipeline.startsWithCollStats()) { uassertStatusOKWithContext(executionNsRoutingInfoStatus, "Unable to retrieve information for $collStats stage"); @@ -331,7 +341,7 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, if (executionNsRoutingInfoStatus.isOK()) { cm = std::move(executionNsRoutingInfoStatus.getValue()); - } else if (!(hasChangeStream && + } else if (!((hasChangeStream || startsWithDocuments) && executionNsRoutingInfoStatus == ErrorCodes::NamespaceNotFound)) { appendEmptyResultSetWithStatus( opCtx, namespaces.requestedNss, executionNsRoutingInfoStatus.getStatus(), result); @@ -366,6 +376,13 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, resolveInvolvedNamespaces(involvedNamespaces), hasChangeStream); + // A pipeline with $changeStreamSplitLargeEvent requires the use of resume token format v2, + // since the 'fragmentNum' field only exists in this version and later. + if (hasChangeStream && liteParsedPipeline.endsWithChangeStreamSplitLargeEvent()) { + expCtx->changeStreamTokenVersion = 2; + expCtx->ignoreTokenVersionOnResume = true; + } + // Parse and optimize the full pipeline. auto pipeline = Pipeline::parse(request.getPipeline(), expCtx); @@ -381,6 +398,11 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, } pipeline->optimizePipeline(); + + // Validate the pipeline post-optimization. + const bool alreadyOptimized = true; + pipeline->validateCommon(alreadyOptimized); + return pipeline; }; @@ -395,6 +417,7 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, cm, involvedNamespaces, hasChangeStream, + startsWithDocuments, allowedToPassthrough, request.getPassthroughToShard().has_value()); @@ -469,7 +492,8 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx, namespaces, privileges, result, - hasChangeStream); + hasChangeStream, + startsWithDocuments); } case cluster_aggregation_planner::AggregationTargeter::TargetingPolicy:: kSpecificShardOnly: { diff --git a/src/mongo/s/query/cluster_aggregation_planner.cpp b/src/mongo/s/query/cluster_aggregation_planner.cpp index 3f970f2f8c1..e124b63c4f9 100644 --- a/src/mongo/s/query/cluster_aggregation_planner.cpp +++ b/src/mongo/s/query/cluster_aggregation_planner.cpp @@ -65,6 +65,7 @@ namespace cluster_aggregation_planner { MONGO_FAIL_POINT_DEFINE(shardedAggregateFailToDispatchExchangeConsumerPipeline); MONGO_FAIL_POINT_DEFINE(shardedAggregateFailToEstablishMergingShardCursor); +MONGO_FAIL_POINT_DEFINE(shardedAggregateHangBeforeDispatchMergingPipeline); using sharded_agg_helpers::DispatchShardPipelineResults; using sharded_agg_helpers::SplitPipeline; @@ -171,9 +172,13 @@ Status dispatchMergingPipeline(const boost::intrusive_ptr<ExpressionContext>& ex const PrivilegeVector& privileges, bool hasChangeStream) { // We should never be in a situation where we call this function on a non-merge pipeline. - invariant(shardDispatchResults.splitPipeline); + tassert(6525900, + "tried to dispatch merge pipeline but the pipeline was not split", + shardDispatchResults.splitPipeline); auto* mergePipeline = shardDispatchResults.splitPipeline->mergePipeline.get(); - invariant(mergePipeline); + tassert(6525901, + "tried to dispatch merge pipeline but there was no merge portion of the split pipeline", + mergePipeline); auto* opCtx = expCtx->opCtx; std::vector<ShardId> targetedShards; @@ -232,9 +237,13 @@ Status dispatchMergingPipeline(const boost::intrusive_ptr<ExpressionContext>& ex privileges, expCtx->tailableMode)); - // Ownership for the shard cursors has been transferred to the merging shard. Dismiss the - // ownership in the current merging pipeline such that when it goes out of scope it does not - // attempt to kill the cursors. + // If the mergingShard returned an error and did not accept ownership it is our responsibility + // to kill the cursors. + uassertStatusOK(getStatusFromCommandResult(mergeResponse.swResponse.getValue().data)); + + // If we didn't get an error from the merging shard, ownership for the shard cursors has been + // transferred to the merging shard. Dismiss the ownership in the current merging pipeline such + // that when it goes out of scope it does not attempt to kill the cursors. auto mergeCursors = static_cast<DocumentSourceMergeCursors*>(mergePipeline->peekFront()); mergeCursors->dismissCursorOwnership(); @@ -374,6 +383,9 @@ DispatchShardPipelineResults dispatchExchangeConsumerPipeline( const NamespaceString& executionNss, Document serializedCommand, DispatchShardPipelineResults* shardDispatchResults) { + tassert(7163600, + "dispatchExchangeConsumerPipeline() must not be called for explain operation", + !expCtx->explain); auto opCtx = expCtx->opCtx; if (MONGO_unlikely(shardedAggregateFailToDispatchExchangeConsumerPipeline.shouldFail())) { @@ -410,7 +422,8 @@ DispatchShardPipelineResults dispatchExchangeConsumerPipeline( serializedCommand, consumerPipelines.back(), boost::none, /* exchangeSpec */ - false /* needsMerge */); + false /* needsMerge */, + boost::none /* explain */); requests.emplace_back(shardDispatchResults->exchangeSpec->consumerShards[idx], consumerCmdObj); @@ -435,8 +448,11 @@ DispatchShardPipelineResults dispatchExchangeConsumerPipeline( SplitPipeline splitPipeline{nullptr, std::move(mergePipeline), boost::none}; - // Relinquish ownership of the local consumer pipelines' cursors as each shard is now - // responsible for its own producer cursors. + // Relinquish ownership of the consumer pipelines' cursors. These cursors are now set up to be + // merged by a set of $mergeCursors pipelines that we just dispatched to the shards above. Now + // that we've established those pipelines on the shards, we are no longer responsible for + // ensuring they are cleaned up. If there was a problem establishing the cursors then + // establishCursors() would have thrown and mongos would kill all the consumer cursors itself. for (const auto& pipeline : consumerPipelines) { const auto& mergeCursors = static_cast<DocumentSourceMergeCursors*>(pipeline.shardsPipeline->peekFront()); @@ -563,6 +579,7 @@ AggregationTargeter AggregationTargeter::make( boost::optional<ChunkManager> cm, stdx::unordered_set<NamespaceString> involvedNamespaces, bool hasChangeStream, + bool startsWithDocuments, bool allowedToPassthrough, bool perShardCursor) { if (perShardCursor) { @@ -583,10 +600,11 @@ AggregationTargeter AggregationTargeter::make( // Determine whether this aggregation must be dispatched to all shards in the cluster. const bool mustRunOnAll = - sharded_agg_helpers::mustRunOnAllShards(executionNss, hasChangeStream); + sharded_agg_helpers::mustRunOnAllShards(executionNss, hasChangeStream, startsWithDocuments); - // If we don't have a routing table, then this is a $changeStream which must run on all shards. - invariant(cm || (mustRunOnAll && hasChangeStream)); + // If we don't have a routing table, then this is either a $changeStream which must run on all + // shards or a $documents stage which must not. + invariant(cm || (mustRunOnAll && hasChangeStream) || (startsWithDocuments && !mustRunOnAll)); // A pipeline is allowed to passthrough to the primary shard iff the following conditions are // met: @@ -663,11 +681,16 @@ Status dispatchPipelineAndMerge(OperationContext* opCtx, const ClusterAggregate::Namespaces& namespaces, const PrivilegeVector& privileges, BSONObjBuilder* result, - bool hasChangeStream) { + bool hasChangeStream, + bool startsWithDocuments) { auto expCtx = targeter.pipeline->getContext(); // If not, split the pipeline as necessary and dispatch to the relevant shards. - auto shardDispatchResults = sharded_agg_helpers::dispatchShardPipeline( - serializedCommand, hasChangeStream, std::move(targeter.pipeline)); + auto shardDispatchResults = + sharded_agg_helpers::dispatchShardPipeline(serializedCommand, + hasChangeStream, + startsWithDocuments, + std::move(targeter.pipeline), + expCtx->explain); // If the operation is an explain, then we verify that it succeeded on all targeted // shards, write the results to the output builder, and return immediately. @@ -701,6 +724,8 @@ Status dispatchPipelineAndMerge(OperationContext* opCtx, expCtx, namespaces.executionNss, serializedCommand, &shardDispatchResults); } + shardedAggregateHangBeforeDispatchMergingPipeline.pauseWhileSet(); + // If we reach here, we have a merge pipeline to dispatch. return dispatchMergingPipeline(expCtx, namespaces, diff --git a/src/mongo/s/query/cluster_aggregation_planner.h b/src/mongo/s/query/cluster_aggregation_planner.h index 046d8eab6ca..78c919192db 100644 --- a/src/mongo/s/query/cluster_aggregation_planner.h +++ b/src/mongo/s/query/cluster_aggregation_planner.h @@ -82,6 +82,7 @@ struct AggregationTargeter { boost::optional<ChunkManager> cm, stdx::unordered_set<NamespaceString> involvedNamespaces, bool hasChangeStream, + bool startsWithDocuments, bool allowedToPassthrough, bool perShardCursor); @@ -125,7 +126,8 @@ Status dispatchPipelineAndMerge(OperationContext* opCtx, const ClusterAggregate::Namespaces& namespaces, const PrivilegeVector& privileges, BSONObjBuilder* result, - bool hasChangeStream); + bool hasChangeStream, + bool startsWithDocuments); /** * Similar to runPipelineOnPrimaryShard but allows $changeStreams. Intended for use by per shard diff --git a/src/mongo/s/query/cluster_find.cpp b/src/mongo/s/query/cluster_find.cpp index bd186639684..e27d4174e99 100644 --- a/src/mongo/s/query/cluster_find.cpp +++ b/src/mongo/s/query/cluster_find.cpp @@ -61,6 +61,7 @@ #include "mongo/s/client/num_hosts_targeted_metrics.h" #include "mongo/s/client/shard_registry.h" #include "mongo/s/cluster_commands_helpers.h" +#include "mongo/s/collection_uuid_mismatch.h" #include "mongo/s/grid.h" #include "mongo/s/query/async_results_merger.h" #include "mongo/s/query/cluster_client_cursor_impl.h" @@ -82,11 +83,6 @@ static const BSONObj kSortKeyMetaProjection = BSON("$meta" << "sortKey"); static const BSONObj kGeoNearDistanceMetaProjection = BSON("$meta" << "geoNearDistance"); -// We must allow some amount of overhead per result document, since when we make a cursor response -// the documents are elements of a BSONArray. The overhead is 1 byte/doc for the type + 1 byte/doc -// for the field name's null terminator + 1 byte per digit in the array index. The index can be no -// more than 8 decimal digits since the response is at most 16MB, and 16 * 1024 * 1024 < 1 * 10^8. -static const int kPerDocumentOverheadBytesUpperBound = 10; const char kFindCmdName[] = "find"; @@ -303,12 +299,9 @@ CursorId runQueryWithoutRetrying(OperationContext* opCtx, if (ex.code() == ErrorCodes::CollectionUUIDMismatch && !ex.extraInfo<CollectionUUIDMismatchInfo>()->actualCollection() && !shardIds.count(cm.dbPrimary())) { - // We received CollectionUUIDMismatchInfo but it does not contain the actual - // namespace, and we did not attempt to establish a cursor on the primary shard. - // Attempt to do so now in case the collection corresponding to the provided UUID is - // unsharded. This should throw CollectionUUIDMismatchInfo, StaleShardVersion, or - // StaleDbVersion. - establishCursorsOnShards({cm.dbPrimary()}); + // We received CollectionUUIDMismatch but it does not contain the actual namespace, and + // we did not attempt to establish a cursor on the primary shard. + uassertStatusOK(populateCollectionUUIDMismatch(opCtx, ex.toStatus())); MONGO_UNREACHABLE; } @@ -341,7 +334,7 @@ CursorId runQueryWithoutRetrying(OperationContext* opCtx, FindCommon::waitInFindBeforeMakingBatch(opCtx, query); auto cursorState = ClusterCursorManager::CursorState::NotExhausted; - size_t bytesBuffered = 0; + FindCommon::BSONArrayResponseSizeTracker responseSizeTracker; // This loop will not result in actually calling getMore against shards, but just loading // results from the initial batches (that were obtained while establishing cursors) into @@ -364,14 +357,13 @@ CursorId runQueryWithoutRetrying(OperationContext* opCtx, // If adding this object will cause us to exceed the message size limit, then we stash it // for later. - if (!FindCommon::haveSpaceForNext(nextObj, results->size(), bytesBuffered)) { + if (!responseSizeTracker.haveSpaceForNext(nextObj)) { ccc->queueResult(nextObj); break; } - // Add doc to the batch. Account for the space overhead associated with returning this doc - // inside a BSON array. - bytesBuffered += (nextObj.objsize() + kPerDocumentOverheadBytesUpperBound); + // Add doc to the batch. + responseSizeTracker.add(nextObj); results->push_back(std::move(nextObj)); } @@ -514,6 +506,13 @@ CursorId ClusterFind::runQuery(OperationContext* opCtx, for (size_t retries = 1; retries <= kMaxRetries; ++retries) { auto swCM = getCollectionRoutingInfoForTxnCmd(opCtx, query.nss()); if (swCM == ErrorCodes::NamespaceNotFound) { + uassert(CollectionUUIDMismatchInfo(query.nss().db().toString(), + *findCommand.getCollectionUUID(), + query.nss().coll().toString(), + boost::none), + "Database does not exist", + !findCommand.getCollectionUUID()); + // If the database doesn't exist, we successfully return an empty result set without // creating a cursor. return CursorId(0); @@ -763,7 +762,7 @@ StatusWith<CursorResponse> ClusterFind::runGetMore(OperationContext* opCtx, } std::vector<BSONObj> batch; - size_t bytesBuffered = 0; + FindCommon::BSONArrayResponseSizeTracker responseSizeTracker; long long batchSize = cmd.getBatchSize().value_or(0); auto cursorState = ClusterCursorManager::CursorState::NotExhausted; BSONObj postBatchResumeToken; @@ -816,8 +815,7 @@ StatusWith<CursorResponse> ClusterFind::runGetMore(OperationContext* opCtx, break; } - if (!FindCommon::haveSpaceForNext( - *next.getValue().getResult(), batch.size(), bytesBuffered)) { + if (!responseSizeTracker.haveSpaceForNext(*next.getValue().getResult())) { pinnedCursor.getValue()->queueResult(*next.getValue().getResult()); stashedResult = true; break; @@ -826,10 +824,8 @@ StatusWith<CursorResponse> ClusterFind::runGetMore(OperationContext* opCtx, // As soon as we get a result, this operation no longer waits. awaitDataState(opCtx).shouldWaitForInserts = false; - // Add doc to the batch. Account for the space overhead associated with returning this doc - // inside a BSON array. - bytesBuffered += - (next.getValue().getResult()->objsize() + kPerDocumentOverheadBytesUpperBound); + // Add doc to the batch. + responseSizeTracker.add(*next.getValue().getResult()); batch.push_back(std::move(*next.getValue().getResult())); // Update the postBatchResumeToken. For non-$changeStream aggregations, this will be empty. diff --git a/src/mongo/s/query/document_source_merge_cursors.cpp b/src/mongo/s/query/document_source_merge_cursors.cpp index c6f8f3fbaf6..f3af2bf0d99 100644 --- a/src/mongo/s/query/document_source_merge_cursors.cpp +++ b/src/mongo/s/query/document_source_merge_cursors.cpp @@ -52,7 +52,6 @@ DocumentSourceMergeCursors::DocumentSourceMergeCursors( : DocumentSource(kStageName, expCtx), _armParamsObj(std::move(ownedParamsSpec)), _armParams(std::move(armParams)) { - _armParams->setRecordRemoteOpWaitTime(true); // Populate the shard ids from the 'RemoteCursor'. recordRemoteCursorShardIds(_armParams->getRemotes()); @@ -83,7 +82,6 @@ bool DocumentSourceMergeCursors::remotesExhausted() const { void DocumentSourceMergeCursors::populateMerger() { invariant(!_blockingResultsMerger); invariant(_armParams); - invariant(_armParams->getRecordRemoteOpWaitTime()); _blockingResultsMerger.emplace( pExpCtx->opCtx, diff --git a/src/mongo/s/query/router_stage_remove_metadata_fields.cpp b/src/mongo/s/query/router_stage_remove_metadata_fields.cpp index 320c441ef46..7f3d89beef1 100644 --- a/src/mongo/s/query/router_stage_remove_metadata_fields.cpp +++ b/src/mongo/s/query/router_stage_remove_metadata_fields.cpp @@ -53,9 +53,14 @@ StatusWith<ClusterQueryResult> RouterStageRemoveMetadataFields::next() { } BSONObjIterator iterator(*childResult.getValue().getResult()); + // Find the first field that we need to remove. - while (iterator.more() && (*iterator).fieldName()[0] != '$') { - ++iterator; + for (; iterator.more(); ++iterator) { + // To save some time, we ensure that the current field name starts with a $ + // before checking if it's actually a metadata field in the map. + if ((*iterator).fieldName()[0] == '$' && _metaFields.contains((*iterator).fieldName())) { + break; + } } if (!iterator.more()) { diff --git a/src/mongo/s/query/router_stage_remove_metadata_fields_test.cpp b/src/mongo/s/query/router_stage_remove_metadata_fields_test.cpp index a18aa0cbb31..9fa199a2ac9 100644 --- a/src/mongo/s/query/router_stage_remove_metadata_fields_test.cpp +++ b/src/mongo/s/query/router_stage_remove_metadata_fields_test.cpp @@ -194,6 +194,60 @@ TEST(RouterStageRemoveMetadataFieldsTest, ForwardsAwaitDataTimeout) { ASSERT_EQ(789, durationCount<Milliseconds>(awaitDataTimeout.getValue())); } +// Grabs the next document from a stage and ensure it matches expectedDoc. +// Assumes that remotes are exhausted after one document. +void verifyNextDocument(RouterExecStage* stage, const BSONObj& expectedDoc) { + auto result = stage->next(); + ASSERT_OK(result.getStatus()); + ASSERT(result.getValue().getResult()); + ASSERT_BSONOBJ_EQ(*result.getValue().getResult(), expectedDoc); + ASSERT_TRUE(stage->remotesExhausted()); +} + +TEST(RouterStageRemoveMetadataFieldsTest, AllowsNonMetaDataDollars) { + auto mockStage = std::make_unique<RouterStageMock>(opCtx); + mockStage->queueResult(BSON("$a" << 1 << "$sortKey" << 1 << "b" << 1)); + mockStage->queueResult(BSON("a" << 2 << "$sortKey" << 1 << "$b" << 2)); + mockStage->markRemotesExhausted(); + + auto sortKeyStage = std::make_unique<RouterStageRemoveMetadataFields>( + opCtx, std::move(mockStage), StringDataSet{"$sortKey"_sd}); + ASSERT_TRUE(sortKeyStage->remotesExhausted()); + + verifyNextDocument(sortKeyStage.get(), BSON("$a" << 1 << "b" << 1)); + verifyNextDocument(sortKeyStage.get(), BSON("a" << 2 << "$b" << 2)); + + auto endResult = sortKeyStage->next(); + ASSERT_OK(endResult.getStatus()); + ASSERT(endResult.getValue().isEOF()); + ASSERT_TRUE(sortKeyStage->remotesExhausted()); +} + +// For every keyword, ensure it's removed if it's in the first, middle, or +// last position, and that the remainder of the document is undisturbed. +TEST(RouterStageRemoveMetadataFieldsTest, RemovesAllMetaDataDollars) { + for (auto& keyword : Document::allMetadataFieldNames) { + auto mockStage = std::make_unique<RouterStageMock>(opCtx); + mockStage->queueResult(BSON(keyword << 1 << "$a" << 1)); + mockStage->queueResult(BSON("$a" << 1 << keyword << 1)); + mockStage->queueResult(BSON("$a" << 1 << keyword << 1 << "$b" << 1)); + mockStage->markRemotesExhausted(); + + auto sortKeyStage = std::make_unique<RouterStageRemoveMetadataFields>( + opCtx, std::move(mockStage), Document::allMetadataFieldNames); + ASSERT_TRUE(sortKeyStage->remotesExhausted()); + + verifyNextDocument(sortKeyStage.get(), BSON("$a" << 1)); + verifyNextDocument(sortKeyStage.get(), BSON("$a" << 1)); + verifyNextDocument(sortKeyStage.get(), BSON("$a" << 1 << "$b" << 1)); + + auto endResult = sortKeyStage->next(); + ASSERT_OK(endResult.getStatus()); + ASSERT(endResult.getValue().isEOF()); + ASSERT_TRUE(sortKeyStage->remotesExhausted()); + } +} + } // namespace } // namespace mongo diff --git a/src/mongo/s/request_types/auto_split_vector.idl b/src/mongo/s/request_types/auto_split_vector.idl index abcbece2019..3e55eb01d8e 100644 --- a/src/mongo/s/request_types/auto_split_vector.idl +++ b/src/mongo/s/request_types/auto_split_vector.idl @@ -75,3 +75,7 @@ commands: maxChunkSizeBytes: type: safeInt64 description: "Max chunk size of the collection expressed in bytes" + limit: + type: int + description: "Max number of split points to look for" + optional: true diff --git a/src/mongo/s/request_types/balancer_collection_status.idl b/src/mongo/s/request_types/balancer_collection_status.idl index 53e86886562..a821222f933 100644 --- a/src/mongo/s/request_types/balancer_collection_status.idl +++ b/src/mongo/s/request_types/balancer_collection_status.idl @@ -40,7 +40,7 @@ structs: strict: false fields: chunkSize: - type: safeInt64 + type: safeDouble description: "Configured chunk size in MiB for this collection" balancerCompliant: type: bool diff --git a/src/mongo/s/request_types/commit_chunk_migration_request_type.h b/src/mongo/s/request_types/commit_chunk_migration_request_type.h index 16d5f0ef8ce..073ef9404f6 100644 --- a/src/mongo/s/request_types/commit_chunk_migration_request_type.h +++ b/src/mongo/s/request_types/commit_chunk_migration_request_type.h @@ -81,6 +81,11 @@ public: const Timestamp& getCollectionTimestamp() { return _collectionTimestamp; } + /* + * NOTE: v7.0 is deprecating the _validAfter field of CommitChunkMigrationRequest + * (see SERVER-74469 for details). + * This method should not be used to develop new functionality. + */ const boost::optional<Timestamp>& getValidAfter() { return _validAfter; } diff --git a/src/mongo/s/request_types/get_stats_for_balancing.idl b/src/mongo/s/request_types/get_stats_for_balancing.idl new file mode 100644 index 00000000000..7ee96da4eaa --- /dev/null +++ b/src/mongo/s/request_types/get_stats_for_balancing.idl @@ -0,0 +1,84 @@ +# Copyright (C) 2022-present MongoDB, Inc. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the Server Side Public License, version 1, +# as published by MongoDB, Inc. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# Server Side Public License for more details. +# +# You should have received a copy of the Server Side Public License +# along with this program. If not, see +# <http://www.mongodb.com/licensing/server-side-public-license>. +# +# As a special exception, the copyright holders give permission to link the +# code of portions of this program with the OpenSSL library under certain +# conditions as described in each individual source file and distribute +# linked combinations including the program with the OpenSSL library. You +# must comply with the Server Side Public License in all respects for +# all of the code used other than as permitted herein. If you modify file(s) +# with this exception, you may extend this exception to your version of the +# file(s), but you are not obligated to do so. If you do not wish to do so, +# delete this exception statement from your version. If you delete this +# exception statement from all source files in the program, then also delete +# it in the license file. +# + +global: + cpp_namespace: "mongo" + +imports: + - "mongo/idl/basic_types.idl" + +structs: + NamespaceWithOptionalUUID: + description: 'Namespace with an optional collection UUID' + strict: false + fields: + ns: + description: 'Namespace of the collection' + type: namespacestring + UUID: + description: 'Collection UUID' + type: uuid + optional: true # optional because the caller may not attach the collection UUID + + CollStatsForBalancing: + description: 'Collection stats for a specific collection' + strict: false + fields: + namespace: + description: 'Namespace of the collection' + type: namespacestring + cpp_name: ns + collSize: + description: 'size of data currently owned by this shard for this collection' + type: safeInt64 + + ShardsvrGetStatsForBalancingReply: + description: 'Response for ShardsvrGetStatsForBalancing command' + strict: false + fields: + stats: + description: 'List of stats for each of the requested collection' + type: array<CollStatsForBalancing> + +commands: + _shardsvrGetStatsForBalancing: + command_name: _shardsvrGetStatsForBalancing + cpp_name: ShardsvrGetStatsForBalancing + description: 'Internal command used by the balancer to retrieve stats for balancing.' + namespace: ignored + api_version: '' + strict: false + reply_type: ShardsvrGetStatsForBalancingReply + fields: + collections: + description: 'List of namespaces to retrieve statistic for' + type: array<NamespaceWithOptionalUUID> + scaleFactor: + description: 'Scale factor for data size units. If omitted 1048576 (MiB) will be used' + type: exactInt64 + optional: true diff --git a/src/mongo/s/request_types/merge_chunks_request_test.cpp b/src/mongo/s/request_types/merge_chunks_request_test.cpp index b631ca1dffa..f7e77f1adaf 100644 --- a/src/mongo/s/request_types/merge_chunks_request_test.cpp +++ b/src/mongo/s/request_types/merge_chunks_request_test.cpp @@ -60,12 +60,12 @@ TEST(ConfigSvrMergeChunks, BasicValidConfigCommand) { TEST(ConfigSvrMergeChunks, ConfigCommandtoBSON) { auto collUUID = UUID::gen(); - BSONObj serializedRequest = BSON("_configsvrCommitChunksMerge" - << "TestDB.TestColl" - << "shard" - << "shard0000" - << "collUUID" << collUUID.toBSON() << "chunkRange" - << chunkRange.toBSON() << "validAfter" << Timestamp{100}); + BSONObj serializedRequest = + BSON("_configsvrCommitChunksMerge" + << "TestDB.TestColl" + << "shard" + << "shard0000" + << "collUUID" << collUUID.toBSON() << "chunkRange" << chunkRange.toBSON()); BSONObj writeConcernObj = BSON("w" << "majority"); diff --git a/src/mongo/s/request_types/move_primary.idl b/src/mongo/s/request_types/move_primary.idl index 51d4a095b8a..4691542c06b 100644 --- a/src/mongo/s/request_types/move_primary.idl +++ b/src/mongo/s/request_types/move_primary.idl @@ -33,6 +33,7 @@ global: imports: - "mongo/idl/basic_types.idl" + - "mongo/s/sharding_types.idl" structs: movePrimary: @@ -62,3 +63,20 @@ structs: to: type: string description: "The shard serving as the destination for un-sharded collections." + +commands: + _configsvrCommitMovePrimary: + command_name: _configsvrCommitMovePrimary + description: Reassign a new primary shard for the given database on the config server + cpp_name: ConfigsvrCommitMovePrimary + namespace: type + type: string + api_version: + strict: false + fields: + expectedDatabaseVersion: + type: database_version + description: Database version known by the current primary shard + to: + type: shard_id + description: Shard serving as the destination for un-sharded collections diff --git a/src/mongo/s/request_types/sharded_ddl_commands.idl b/src/mongo/s/request_types/sharded_ddl_commands.idl index 2c86a3d199a..2791c5a1b65 100644 --- a/src/mongo/s/request_types/sharded_ddl_commands.idl +++ b/src/mongo/s/request_types/sharded_ddl_commands.idl @@ -315,6 +315,11 @@ commands: api_version: "" cpp_name: ShardsvrDropCollectionParticipant strict: false + fields: + fromMigrate: + type: bool + description: "Whether the drop comes as a result of an interrupted migration process." + optional: true _shardsvrRenameCollection: command_name: _shardsvrRenameCollection @@ -325,6 +330,12 @@ commands: api_version: "" chained_structs: RenameCollectionRequest: RenameCollectionRequest + fields: + allowEncryptedCollectionRename: + description: "Encrypted Collection renames are usually disallowed to minimize user error. + C2C needs to do the renames to replicate create collection." + type: bool + optional: true _shardsvrSetAllowMigrations: command_name: _shardsvrSetAllowMigrations @@ -370,7 +381,7 @@ commands: namespace: concatenate_with_db api_version: "" strict: false - chained_structs: + chained_structs: RefineCollectionShardKeyRequest: RefineCollectionShardKeyRequest _configsvrRefineCollectionShardKey: @@ -403,7 +414,7 @@ commands: namespace: concatenate_with_db api_version: "" strict: false - chained_structs: + chained_structs: DropIndexesRequest: DropIndexesRequest _configsvrCreateDatabase: diff --git a/src/mongo/s/router.cpp b/src/mongo/s/router_role.cpp index dba40ad1137..6fa31d6a447 100644 --- a/src/mongo/s/router.cpp +++ b/src/mongo/s/router_role.cpp @@ -29,7 +29,7 @@ #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding -#include "mongo/s/router.h" +#include "mongo/s/router_role.h" #include "mongo/logv2/log.h" #include "mongo/s/grid.h" @@ -69,32 +69,34 @@ CachedDatabaseInfo DBPrimaryRouter::_getRoutingInfo(OperationContext* opCtx) con } void DBPrimaryRouter::_onException(RouteContext* context, Status s) { - if (++context->numAttempts > kMaxNumStaleVersionRetries) { - uassertStatusOKWithContext( - s, - str::stream() << "Exceeded maximum number of " << kMaxNumStaleVersionRetries - << " retries attempting \'" << context->comment << "\'"); - } else { - LOGV2_DEBUG(637590, - 3, - "Retrying {description}. Got error: {status}", - "description"_attr = context->comment, - "status"_attr = s); - } - auto catalogCache = Grid::get(_service)->catalogCache(); if (s == ErrorCodes::StaleDbVersion) { auto si = s.extraInfo<StaleDbRoutingVersion>(); - invariant(si); - invariant(si->getDb() == _db, - str::stream() << "StaleDbVersion on unexpected database. Expected " << _db - << ", received " << si->getDb()); + tassert(6375900, "StaleDbVersion must have extraInfo", si); + tassert(6375901, + str::stream() << "StaleDbVersion on unexpected database. Expected " << _db + << ", received " << si->getDb(), + si->getDb() == _db); catalogCache->onStaleDatabaseVersion(si->getDb(), si->getVersionWanted()); } else { uassertStatusOK(s); } + + if (++context->numAttempts > kMaxNumStaleVersionRetries) { + uassertStatusOKWithContext( + s, + str::stream() << "Exceeded maximum number of " << kMaxNumStaleVersionRetries + << " retries attempting \'" << context->comment << "\'"); + } else { + LOGV2_DEBUG(6375902, + 3, + "Retrying database primary routing operation", + "attempt"_attr = context->numAttempts, + "comment"_attr = context->comment, + "status"_attr = s); + } } CollectionRouter::CollectionRouter(ServiceContext* service, NamespaceString nss) @@ -122,42 +124,39 @@ ChunkManager CollectionRouter::_getRoutingInfo(OperationContext* opCtx) const { } void CollectionRouter::_onException(RouteContext* context, Status s) { + auto catalogCache = Grid::get(_service)->catalogCache(); + + if (s == ErrorCodes::StaleDbVersion) { + auto si = s.extraInfo<StaleDbRoutingVersion>(); + tassert(6375903, "StaleDbVersion must have extraInfo", si); + catalogCache->onStaleDatabaseVersion(si->getDb(), si->getVersionWanted()); + } else if (s == ErrorCodes::StaleConfig) { + auto si = s.extraInfo<StaleConfigInfo>(); + tassert(6375904, "StaleConfig must have extraInfo", si); + catalogCache->invalidateShardOrEntireCollectionEntryForShardedCollection( + si->getNss(), si->getVersionWanted(), si->getShardId()); + } else if (s == ErrorCodes::StaleEpoch) { + if (auto si = s.extraInfo<StaleEpochInfo>()) { + catalogCache->invalidateShardOrEntireCollectionEntryForShardedCollection( + si->getNss(), si->getVersionWanted(), ShardId()); + } + } else { + uassertStatusOK(s); + } + if (++context->numAttempts > kMaxNumStaleVersionRetries) { uassertStatusOKWithContext( s, str::stream() << "Exceeded maximum number of " << kMaxNumStaleVersionRetries << " retries attempting \'" << context->comment << "\'"); } else { - LOGV2_DEBUG(637591, + LOGV2_DEBUG(6375906, 3, - "Retrying {description}. Got error: {status}", - "description"_attr = context->comment, + "Retrying collection routing operation", + "attempt"_attr = context->numAttempts, + "comment"_attr = context->comment, "status"_attr = s); } - - auto catalogCache = Grid::get(_service)->catalogCache(); - - if (s.isA<ErrorCategory::StaleShardVersionError>()) { - if (auto si = s.extraInfo<StaleConfigInfo>()) { - invariant(si->getNss() == _nss, - str::stream() << "StaleConfig on unexpected namespace. Expected " << _nss - << ", received " << si->getNss()); - catalogCache->invalidateShardOrEntireCollectionEntryForShardedCollection( - _nss, si->getVersionWanted(), si->getShardId()); - } else { - catalogCache->invalidateCollectionEntry_LINEARIZABLE(_nss); - } - } else if (s == ErrorCodes::StaleDbVersion) { - auto si = s.extraInfo<StaleDbRoutingVersion>(); - invariant(si); - invariant(si->getDb() == _nss.db(), - str::stream() << "StaleDbVersion on unexpected database. Expected " << _nss.db() - << ", received " << si->getDb()); - - catalogCache->onStaleDatabaseVersion(si->getDb(), si->getVersionWanted()); - } else { - uassertStatusOK(s); - } } } // namespace router diff --git a/src/mongo/s/router.h b/src/mongo/s/router_role.h index 6cfd1d9e546..6cfd1d9e546 100644 --- a/src/mongo/s/router.h +++ b/src/mongo/s/router_role.h diff --git a/src/mongo/s/routing_table_history_test.cpp b/src/mongo/s/routing_table_history_test.cpp index 9651911ee64..008c15f26ed 100644 --- a/src/mongo/s/routing_table_history_test.cpp +++ b/src/mongo/s/routing_table_history_test.cpp @@ -27,20 +27,38 @@ * it in the license file. */ +#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kTest + #include "mongo/platform/basic.h" #include "mongo/bson/bsonobjbuilder.h" #include "mongo/db/namespace_string.h" #include "mongo/db/service_context.h" +#include "mongo/idl/server_parameter_test_util.h" +#include "mongo/logv2/log.h" #include "mongo/s/catalog/type_chunk.h" #include "mongo/s/chunk_manager.h" #include "mongo/s/chunk_writes_tracker.h" +#include "mongo/s/chunks_test_util.h" #include "mongo/unittest/death_test.h" #include "mongo/unittest/unittest.h" + namespace mongo { + +using chunks_test_util::assertEqualChunkInfo; +using chunks_test_util::calculateCollVersion; +using chunks_test_util::calculateIntermediateShardKey; +using chunks_test_util::calculateShardVersions; +using chunks_test_util::genChunkVector; +using chunks_test_util::genRandomChunkVector; +using chunks_test_util::getShardId; +using chunks_test_util::performRandomChunkOperations; + namespace { +PseudoRandom _random{SecureRandom().nextInt64()}; + const ShardId kThisShard("thisShard"); const NamespaceString kNss("TestDB", "TestColl"); @@ -144,64 +162,57 @@ void assertCorrectBytesWritten(const RoutingTableHistory& rt, }); } -/** - * Test fixture for tests that need to start with a fresh routing table with - * only a single chunk in it, with bytes already written to that chunk object. - */ class RoutingTableHistoryTest : public unittest::Test { public: - void setUp() override { - const UUID uuid = UUID::gen(); - const OID epoch = OID::gen(); - const Timestamp timestamp(1); - ChunkVersion version{1, 0, epoch, timestamp}; - - auto initChunk = - ChunkType{uuid, - ChunkRange{_shardKeyPattern.globalMin(), _shardKeyPattern.globalMax()}, - version, - kThisShard}; - - _rt.emplace(RoutingTableHistory::makeNew(kNss, - uuid, - _shardKeyPattern, - nullptr, - false, - epoch, - timestamp, - boost::none /* timeseriesFields */, - boost::none, - boost::none /* chunkSizeBytes */, - true, - {initChunk})); - ASSERT_EQ(_rt->numChunks(), 1ull); - - // Should only be one - _rt->forEachChunk([&](const auto& chunkInfo) { - auto writesTracker = chunkInfo->getWritesTracker(); - writesTracker->addBytesWritten(_bytesInOriginalChunk); - return true; - }); - } - const KeyPattern& getShardKeyPattern() const { return _shardKeyPattern; } - uint64_t getBytesInOriginalChunk() const { - return _bytesInOriginalChunk; + const OID& collEpoch() const { + return _epoch; } - const RoutingTableHistory& getInitialRoutingTable() const { - return *_rt; + const Timestamp& collTimestamp() const { + return _collTimestamp; } -private: - uint64_t _bytesInOriginalChunk{4ull}; + const UUID& collUUID() const { + return _collUUID; + } - boost::optional<RoutingTableHistory> _rt; + std::vector<ChunkType> genRandomChunkVector(size_t minNumChunks = 1, + size_t maxNumChunks = 30) const { + return chunks_test_util::genRandomChunkVector( + _collUUID, _epoch, _collTimestamp, maxNumChunks, minNumChunks); + } - KeyPattern _shardKeyPattern{BSON("a" << 1)}; + RoutingTableHistory makeNewRt(const std::vector<ChunkType>& chunks) const { + const auto chunkBucketSize = llround(_random.nextInt64(chunks.size() * 1.2)) + 1; + LOGV2(7162710, + "Creating new RoutingTable", + "chunkBucketSize"_attr = chunkBucketSize, + "numChunks"_attr = chunks.size()); + RAIIServerParameterControllerForTest chunkBucketSizeParameter( + "routingTableCacheChunkBucketSize", chunkBucketSize); + return RoutingTableHistory::makeNew(kNss, + _collUUID, + _shardKeyPattern, + nullptr, + false, + _epoch, + _collTimestamp, + boost::none /* timeseriesFields */, + boost::none /* reshardingFields */, + boost::none /* maxChunkSizeBytes */, + true, + chunks); + } + +protected: + KeyPattern _shardKeyPattern{chunks_test_util::kShardKeyPattern}; + const OID _epoch{OID::gen()}; + const Timestamp _collTimestamp{1, 1}; + const UUID _collUUID{UUID::gen()}; }; /** @@ -212,13 +223,28 @@ class RoutingTableHistoryTestThreeInitialChunks : public RoutingTableHistoryTest public: void setUp() override { RoutingTableHistoryTest::setUp(); + _initialChunkBoundaryPoints = {getShardKeyPattern().globalMin(), BSON("a" << 10), BSON("a" << 20), getShardKeyPattern().globalMax()}; - _rt.emplace(splitChunk(RoutingTableHistoryTest::getInitialRoutingTable(), - _initialChunkBoundaryPoints)); + ChunkVersion version{1, 0, collEpoch(), collTimestamp()}; + auto chunks = + genChunkVector(collUUID(), _initialChunkBoundaryPoints, version, 1 /* numShards */); + + _rt.emplace(makeNewRt(chunks)); + ASSERT_EQ(_rt->numChunks(), 3ull); + + _rt->forEachChunk([&](const auto& chunkInfo) { + auto writesTracker = chunkInfo->getWritesTracker(); + writesTracker->addBytesWritten(_bytesInOriginalChunk); + return true; + }); + } + + uint64_t getBytesInOriginalChunk() const { + return _bytesInOriginalChunk; } const RoutingTableHistory& getInitialRoutingTable() const { @@ -230,24 +256,384 @@ public: } private: + uint64_t _bytesInOriginalChunk{4ull}; + boost::optional<RoutingTableHistory> _rt; std::vector<BSONObj> _initialChunkBoundaryPoints; }; +/* + * Test creation of a Routing Table with randomly generated chunks + */ +TEST_F(RoutingTableHistoryTest, RandomCreateBasic) { + const auto chunks = genRandomChunkVector(); + const auto expectedShardVersions = calculateShardVersions(chunks); + const auto expectedCollVersion = calculateCollVersion(expectedShardVersions); + + // Create a new routing table from the randomly generated chunks + auto rt = makeNewRt(chunks); + + // Checks basic getter of routing table return correct values + ASSERT_EQ(kNss, rt.nss()); + ASSERT_EQ(ShardKeyPattern(getShardKeyPattern()).toString(), rt.getShardKeyPattern().toString()); + ASSERT_EQ(chunks.size(), rt.numChunks()); + + // Check that chunks have correct info + size_t i = 0; + rt.forEachChunk([&](const auto& chunkInfo) { + assertEqualChunkInfo(ChunkInfo{chunks[i++]}, *chunkInfo); + return true; + }); + ASSERT_EQ(i, chunks.size()); + + // Checks collection version is correct + ASSERT_EQ(expectedCollVersion, rt.getVersion()); + + // Checks version for each chunk + for (const auto& [shardId, shardVersion] : expectedShardVersions) { + ASSERT_EQ(shardVersion, rt.getVersion(shardId)); + } + + ASSERT_EQ(expectedShardVersions.size(), rt.getNShardsOwningChunks()); + + std::set<ShardId> expectedShardIds; + for (const auto& [shardId, shardVersion] : expectedShardVersions) { + expectedShardIds.insert(shardId); + } + std::set<ShardId> shardIds; + rt.getAllShardIds(&shardIds); + ASSERT(expectedShardIds == shardIds); +} + +/* + * Test that creation of Routing Table with chunks that do not cover the entire shard key space + * fails. + * + * The gap is produced by removing a random chunks from the randomly generated chunk list. Thus it + * also cover the case for which min/max key is missing. + */ +TEST_F(RoutingTableHistoryTest, RandomCreateWithMissingChunkFail) { + auto chunks = genRandomChunkVector(2 /*minNumChunks*/); + + // Remove one random chunk to simulate a gap in the shardkey + chunks.erase(chunks.begin() + _random.nextInt64(chunks.size())); + + // Create a new routing table from the randomly generated chunks + ASSERT_THROWS_CODE(makeNewRt(chunks), DBException, ErrorCodes::ConflictingOperationInProgress); +} + +/* + * Test that creation of Routing Table with chunks that do not cover the entire shard key space + * fails. + * + * The gap is produced by shrinking the range of a random chunk. + */ +TEST_F(RoutingTableHistoryTest, RandomCreateWithChunkGapFail) { + auto chunks = genRandomChunkVector(2 /*minNumChunks*/); + + auto& shrinkedChunk = chunks.at(_random.nextInt64(chunks.size())); + auto intermediateKey = + calculateIntermediateShardKey(shrinkedChunk.getMin(), shrinkedChunk.getMax()); + if (_random.nextInt64(2)) { + // Shrink right bound + shrinkedChunk.setMax(intermediateKey); + } else { + // Shrink left bound + shrinkedChunk.setMin(intermediateKey); + } + + // Create a new routing table from the randomly generated chunks + ASSERT_THROWS_CODE(makeNewRt(chunks), DBException, ErrorCodes::ConflictingOperationInProgress); +} + +/* + * Updating ChunkMap with gaps must fail + */ +TEST_F(RoutingTableHistoryTest, RandomUpdateWithChunkGapFail) { + auto chunks = genRandomChunkVector(); + + // Create a new routing table from the randomly generated chunks + auto rt = makeNewRt(chunks); + auto collVersion = rt.getVersion(); + + auto shrinkedChunk = chunks.at(_random.nextInt64(chunks.size())); + auto intermediateKey = + calculateIntermediateShardKey(shrinkedChunk.getMin(), shrinkedChunk.getMax()); + if (_random.nextInt64(2)) { + // Shrink right bound + shrinkedChunk.setMax(intermediateKey); + } else { + // Shrink left bound + shrinkedChunk.setMin(intermediateKey); + } + + // Bump chunk version + collVersion.incMajor(); + shrinkedChunk.setVersion(collVersion); + + ASSERT_THROWS_CODE(rt.makeUpdated(boost::none /* timeseriesFields */, + boost::none /* reshardingFields */, + boost::none /* maxChunkSizeBytes */, + true, + {std::move(shrinkedChunk)}), + AssertionException, + ErrorCodes::ConflictingOperationInProgress); +} + +/* + * Creating a Routing Table with overlapping chunks must fail. + */ +TEST_F(RoutingTableHistoryTest, RandomCreateWithChunkOverlapFail) { + auto chunks = genRandomChunkVector(2 /* minNumChunks */); + + auto chunkToExtendIt = chunks.begin() + _random.nextInt64(chunks.size()); + + const auto canExtendLeft = chunkToExtendIt > chunks.begin(); + const auto extendRight = + !canExtendLeft || ((chunkToExtendIt < std::prev(chunks.end())) && _random.nextInt64(2)); + const auto extendLeft = !extendRight; + if (extendRight) { + // extend right bound + chunkToExtendIt->setMax(calculateIntermediateShardKey(chunkToExtendIt->getMax(), + std::next(chunkToExtendIt)->getMax(), + 0.0 /* minKeyProb */, + 0.1 /* maxKeyProb */)); + auto newVersion = chunkToExtendIt->getVersion(); + newVersion.incMajor(); + std::next(chunkToExtendIt)->setVersion(newVersion); + } + + if (extendLeft) { + invariant(canExtendLeft); + // extend left bound + chunkToExtendIt->setMin(calculateIntermediateShardKey(std::prev(chunkToExtendIt)->getMin(), + chunkToExtendIt->getMin(), + 0.1 /* minKeyProb */, + 0.0 /* maxKeyProb */)); + auto newVersion = chunkToExtendIt->getVersion(); + newVersion.incMajor(); + std::prev(chunkToExtendIt)->setVersion(newVersion); + } + + // Create a new routing table from the randomly generated chunks + ASSERT_THROWS_CODE(makeNewRt(chunks), DBException, ErrorCodes::ConflictingOperationInProgress); +} + +/* + * Updating a ChunkMap with overlapping chunks must fail. + */ +TEST_F(RoutingTableHistoryTest, RandomUpdateWithChunkOverlapFail) { + auto chunks = genRandomChunkVector(2 /* minNumChunks */); + + // Create a new routing table from the randomly generated chunks + auto rt = makeNewRt(chunks); + auto collVersion = rt.getVersion(); + + auto chunkToExtendIt = chunks.begin() + _random.nextInt64(chunks.size()); + + const auto canExtendLeft = chunkToExtendIt > chunks.begin(); + const auto extendRight = + !canExtendLeft || (chunkToExtendIt < std::prev(chunks.end()) && _random.nextInt64(2)); + const auto extendLeft = !extendRight; + if (extendRight) { + // extend right bound + chunkToExtendIt->setMax(calculateIntermediateShardKey( + chunkToExtendIt->getMax(), std::next(chunkToExtendIt)->getMax())); + } + + if (extendLeft) { + invariant(canExtendLeft); + // extend left bound + chunkToExtendIt->setMin(calculateIntermediateShardKey(std::prev(chunkToExtendIt)->getMin(), + chunkToExtendIt->getMin())); + } + + // Bump chunk version + collVersion.incMajor(); + chunkToExtendIt->setVersion(collVersion); + + ASSERT_THROWS_CODE(rt.makeUpdated(boost::none /* timeseriesFields */, + boost::none /* reshardingFields */, + boost::none /* maxChunkSizeBytes */, + true, + {*chunkToExtendIt}), + AssertionException, + ErrorCodes::ConflictingOperationInProgress); +} + +/* + * Creating a Routing Table with wrong min key must fail. + */ +TEST_F(RoutingTableHistoryTest, RandomCreateWrongMinFail) { + auto chunks = genRandomChunkVector(); + + chunks.begin()->setMin(BSON("a" << std::numeric_limits<int64_t>::min())); + + // Create a new routing table from the randomly generated chunks + ASSERT_THROWS_CODE(makeNewRt(chunks), DBException, ErrorCodes::ConflictingOperationInProgress); +} + +/* + * Creating a Routing Table with wrong max key must fail. + */ +TEST_F(RoutingTableHistoryTest, RandomCreateWrongMaxFail) { + auto chunks = genRandomChunkVector(); + + chunks.begin()->setMax(BSON("a" << std::numeric_limits<int64_t>::max())); + + // Create a new routing table from the randomly generated chunks + ASSERT_THROWS_CODE(makeNewRt(chunks), DBException, ErrorCodes::ConflictingOperationInProgress); +} + +/* + * Creating a Routing Table with mismatching epoch must fail. + */ +TEST_F(RoutingTableHistoryTest, RandomCreateMismatchingTimestampFail) { + auto chunks = genRandomChunkVector(); + + // Change epoch on a random chunk + auto chunkIt = chunks.begin() + _random.nextInt64(chunks.size()); + const auto& oldVersion = chunkIt->getVersion(); + + const Timestamp wrongTimestamp{Date_t::now()}; + ChunkVersion newVersion{ + oldVersion.majorVersion(), oldVersion.minorVersion(), collEpoch(), wrongTimestamp}; + chunkIt->setVersion(newVersion); + + // Create a new routing table from the randomly generated chunks + ASSERT_THROWS_CODE(makeNewRt(chunks), DBException, ErrorCodes::ConflictingOperationInProgress); +} + +/* + * Updating a Routing Table with mismatching Timestamp must fail. + */ +TEST_F(RoutingTableHistoryTest, RandomUpdateMismatchingTimestampFail) { + auto chunks = genRandomChunkVector(); + + // Create a new routing table from the randomly generated chunks + auto rt = makeNewRt(chunks); + + // Change epoch on a random chunk + auto chunkIt = chunks.begin() + _random.nextInt64(chunks.size()); + const auto& oldVersion = chunkIt->getVersion(); + const Timestamp wrongTimestamp{Date_t::now()}; + ChunkVersion newVersion{ + oldVersion.majorVersion(), oldVersion.minorVersion(), collEpoch(), wrongTimestamp}; + chunkIt->setVersion(newVersion); + + ASSERT_THROWS_CODE(rt.makeUpdated(boost::none /* timeseriesFields */, + boost::none /* reshardingFields */, + boost::none /* maxChunkSizeBytes */, + true, + {*chunkIt}), + AssertionException, + ErrorCodes::ConflictingOperationInProgress); +} + + +/* + * Test update of the Routing Table with randomly generated changed chunks. + */ +TEST_F(RoutingTableHistoryTest, RandomUpdate) { + auto initialChunks = genRandomChunkVector(); + + const auto initialShardVersions = calculateShardVersions(initialChunks); + const auto initialCollVersion = calculateCollVersion(initialShardVersions); + + // Create a new routing table from the randomly generated initialChunks + auto initialRt = makeNewRt(initialChunks); + + auto chunks = initialChunks; + const auto maxNumChunkOps = 2 * initialChunks.size(); + const auto numChunkOps = _random.nextInt32(maxNumChunkOps); + + performRandomChunkOperations(&chunks, numChunkOps); + + std::vector<ChunkType> updatedChunks; + for (const auto& chunk : chunks) { + if (!chunk.getVersion().isOlderOrEqualThan(initialCollVersion)) { + updatedChunks.push_back(chunk); + } + } + + const auto expectedShardVersions = calculateShardVersions(chunks); + const auto expectedCollVersion = calculateCollVersion(expectedShardVersions); + + auto rt = initialRt.makeUpdated(boost::none /* timeseriesFields */, + boost::none /* reshardingFields */, + boost::none /* maxChunkSizeBytes */, + true, + updatedChunks); + + // Checks basic getter of routing table return correct values + ASSERT_EQ(kNss, rt.nss()); + ASSERT_EQ(ShardKeyPattern(getShardKeyPattern()).toString(), rt.getShardKeyPattern().toString()); + ASSERT_EQ(chunks.size(), rt.numChunks()); + + // Check that chunks have correct info + size_t i = 0; + rt.forEachChunk([&](const auto& chunkInfo) { + assertEqualChunkInfo(ChunkInfo{chunks[i++]}, *chunkInfo); + return true; + }); + ASSERT_EQ(i, chunks.size()); + + // Checks collection version is correct + ASSERT_EQ(expectedCollVersion, rt.getVersion()); + + // Checks version for each shard + for (const auto& [shardId, shardVersion] : expectedShardVersions) { + ASSERT_EQ(shardVersion, rt.getVersion(shardId)); + } + + ASSERT_EQ(expectedShardVersions.size(), rt.getNShardsOwningChunks()); + + std::set<ShardId> expectedShardIds; + for (const auto& [shardId, shardVersion] : expectedShardVersions) { + expectedShardIds.insert(shardId); + } + std::set<ShardId> shardIds; + rt.getAllShardIds(&shardIds); + ASSERT(expectedShardIds == shardIds); +} + TEST_F(RoutingTableHistoryTest, SplittingOnlyChunkCopiesBytesWrittenToAllSubchunks) { - auto minKey = BSON("a" << 10); - auto maxKey = BSON("a" << 20); - auto newChunkBoundaryPoints = { - getShardKeyPattern().globalMin(), minKey, maxKey, getShardKeyPattern().globalMax()}; + ChunkVersion version{1, 0, collEpoch(), collTimestamp()}; - auto rt = splitChunk(getInitialRoutingTable(), newChunkBoundaryPoints); - ASSERT_EQ(rt.numChunks(), 3ull); + const ChunkType initialChunk{ + collUUID(), + ChunkRange{getShardKeyPattern().globalMin(), getShardKeyPattern().globalMax()}, + version, + kThisShard}; + auto rt = makeNewRt({initialChunk}); + ASSERT_EQ(1, rt.numChunks()); + + // Set the 4 written bytes in each chunk + const size_t bytesInOriginalChunk{4}; rt.forEachChunk([&](const auto& chunkInfo) { - auto writesTracker = chunkInfo->getWritesTracker(); - auto bytesWritten = writesTracker->getBytesWritten(); - ASSERT_EQ(bytesWritten, getBytesInOriginalChunk()); + chunkInfo->getWritesTracker()->addBytesWritten(bytesInOriginalChunk); + return true; + }); + + version.incMinor(); + auto newChunks = genChunkVector(collUUID(), + {getShardKeyPattern().globalMin(), + BSON("a" << 10), + BSON("a" << 20), + getShardKeyPattern().globalMax()}, + version, + 1 /*numShards*/); + auto newRt = rt.makeUpdated(boost::none /* timeseriesFields */, + boost::none /* reshardingFields */, + boost::none /* maxChunkSizeBytes */, + true, + newChunks); + ASSERT_EQ(3, newRt.numChunks()); + + rt.forEachChunk([&](const auto& chunkInfo) { + ASSERT_EQ(bytesInOriginalChunk, chunkInfo->getWritesTracker()->getBytesWritten()); return true; }); } @@ -328,104 +714,124 @@ TEST_F(RoutingTableHistoryTestThreeInitialChunks, expectedBytesInChunksNotSplit); } +TEST_F(RoutingTableHistoryTest, AllowMigrationFlag) { + auto chunks = genRandomChunkVector(); + + auto makeUpdatedChunk = [&](const ChunkVersion& oldVersion) { + auto updatedChunk = chunks[_random.nextInt64(chunks.size())]; + updatedChunk.setVersion({oldVersion.majorVersion() + 1, + oldVersion.minorVersion(), + collEpoch(), + collTimestamp()}); + return updatedChunk; + }; + + for (auto initialAllowMigrationsValue : std::vector<bool>{false, true}) { + + auto allowMigrationsValue = initialAllowMigrationsValue; + auto rt = RoutingTableHistory::makeNew(kNss, + collUUID(), + getShardKeyPattern(), + nullptr, + false, + collEpoch(), + collTimestamp(), + boost::none /* timeseriesFields */, + boost::none /* reshardingFields */, + boost::none /* maxChunkSizeBytes */, + allowMigrationsValue, + chunks); + ASSERT_EQ(allowMigrationsValue, rt.allowMigrations()); + + // Create an updated routing table with flipped allowMigration flag + allowMigrationsValue = !allowMigrationsValue; + rt = rt.makeUpdated(boost::none /* timeseriesFields */, + boost::none /* reshardingFields */, + boost::none /* maxChunkSizeBytes */, + allowMigrationsValue, + {makeUpdatedChunk(rt.getVersion())}); + ASSERT_EQ(allowMigrationsValue, rt.allowMigrations()); + + // Change back the allow migration flag to the original value + allowMigrationsValue = !allowMigrationsValue; + rt = rt.makeUpdated(boost::none /* timeseriesFields */, + boost::none /* reshardingFields */, + boost::none /* maxChunkSizeBytes */, + allowMigrationsValue, + {makeUpdatedChunk(rt.getVersion())}); + ASSERT_EQ(allowMigrationsValue, rt.allowMigrations()); + } +} + TEST_F(RoutingTableHistoryTest, TestSplits) { - const UUID uuid = UUID::gen(); - const OID epoch = OID::gen(); - const Timestamp timestamp(1); - ChunkVersion version{1, 0, epoch, timestamp}; + ChunkVersion version{1, 0, collEpoch(), collTimestamp()}; auto chunkAll = - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), getShardKeyPattern().globalMax()}, version, kThisShard}; - auto rt = RoutingTableHistory::makeNew(kNss, - uuid, - getShardKeyPattern(), - nullptr, - false, - epoch, - timestamp, - boost::none /* timeseriesFields */, - boost::none, - boost::none /* chunkSizeBytes */, - true, - {chunkAll}); + auto rt = makeNewRt({chunkAll}); std::vector<ChunkType> chunks1 = { - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), BSON("a" << 0)}, - ChunkVersion{2, 1, epoch, timestamp}, + ChunkVersion{2, 1, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << 0), getShardKeyPattern().globalMax()}, - ChunkVersion{2, 2, epoch, timestamp}, + ChunkVersion{2, 2, collEpoch(), collTimestamp()}, kThisShard}}; auto rt1 = rt.makeUpdated(boost::none /* timeseriesFields */, boost::none, boost::none, true, chunks1); - auto v1 = ChunkVersion{2, 2, epoch, timestamp}; + auto v1 = ChunkVersion{2, 2, collEpoch(), collTimestamp()}; ASSERT_EQ(v1, rt1.getVersion(kThisShard)); std::vector<ChunkType> chunks2 = { - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << 0), getShardKeyPattern().globalMax()}, - ChunkVersion{2, 2, epoch, timestamp}, + ChunkVersion{2, 2, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), BSON("a" << -1)}, - ChunkVersion{3, 1, epoch, timestamp}, + ChunkVersion{3, 1, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << -1), BSON("a" << 0)}, - ChunkVersion{3, 2, epoch, timestamp}, + ChunkVersion{3, 2, collEpoch(), collTimestamp()}, kThisShard}}; auto rt2 = rt1.makeUpdated( boost::none /* timeseriesFields */, boost::none, boost::none, true, chunks2); - auto v2 = ChunkVersion{3, 2, epoch, timestamp}; + auto v2 = ChunkVersion{3, 2, collEpoch(), collTimestamp()}; ASSERT_EQ(v2, rt2.getVersion(kThisShard)); } TEST_F(RoutingTableHistoryTest, TestReplaceEmptyChunk) { - const UUID uuid = UUID::gen(); - const OID epoch = OID::gen(); - const Timestamp timestamp(1); - std::vector<ChunkType> initialChunks = { - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), getShardKeyPattern().globalMax()}, - ChunkVersion{1, 0, epoch, timestamp}, + ChunkVersion{1, 0, collEpoch(), collTimestamp()}, kThisShard}}; - auto rt = RoutingTableHistory::makeNew(kNss, - uuid, - getShardKeyPattern(), - nullptr, - false, - epoch, - timestamp, - boost::none /* timeseriesFields */, - boost::none, - boost::none /* chunkSizeBytes */, - true, - initialChunks); + auto rt = makeNewRt(initialChunks); ASSERT_EQ(rt.numChunks(), 1); std::vector<ChunkType> changedChunks = { - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), BSON("a" << 0)}, - ChunkVersion{2, 1, epoch, timestamp}, + ChunkVersion{2, 1, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << 0), getShardKeyPattern().globalMax()}, - ChunkVersion{2, 2, epoch, timestamp}, + ChunkVersion{2, 2, collEpoch(), collTimestamp()}, kThisShard}}; auto rt1 = rt.makeUpdated( boost::none /* timeseriesFields */, boost::none, boost::none, true, changedChunks); - auto v1 = ChunkVersion{2, 2, epoch, timestamp}; + auto v1 = ChunkVersion{2, 2, collEpoch(), collTimestamp()}; ASSERT_EQ(v1, rt1.getVersion(kThisShard)); ASSERT_EQ(rt1.numChunks(), 2); @@ -444,258 +850,184 @@ TEST_F(RoutingTableHistoryTest, TestReplaceEmptyChunk) { } TEST_F(RoutingTableHistoryTest, TestUseLatestVersions) { - const UUID uuid = UUID::gen(); - const OID epoch = OID::gen(); - const Timestamp timestamp(1); - std::vector<ChunkType> initialChunks = { - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), getShardKeyPattern().globalMax()}, - ChunkVersion{1, 0, epoch, timestamp}, + ChunkVersion{1, 0, collEpoch(), collTimestamp()}, kThisShard}}; - auto rt = RoutingTableHistory::makeNew(kNss, - uuid, - getShardKeyPattern(), - nullptr, - false, - epoch, - timestamp, - boost::none /* timeseriesFields */, - boost::none, - boost::none /* chunkSizeBytes */, - true, - initialChunks); + auto rt = makeNewRt(initialChunks); ASSERT_EQ(rt.numChunks(), 1); std::vector<ChunkType> changedChunks = { - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), getShardKeyPattern().globalMax()}, - ChunkVersion{1, 0, epoch, timestamp}, + ChunkVersion{1, 0, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), BSON("a" << 0)}, - ChunkVersion{2, 1, epoch, timestamp}, + ChunkVersion{2, 1, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << 0), getShardKeyPattern().globalMax()}, - ChunkVersion{2, 2, epoch, timestamp}, + ChunkVersion{2, 2, collEpoch(), collTimestamp()}, kThisShard}}; auto rt1 = rt.makeUpdated( boost::none /* timeseriesFields */, boost::none, boost::none, true, changedChunks); - auto v1 = ChunkVersion{2, 2, epoch, timestamp}; + auto v1 = ChunkVersion{2, 2, collEpoch(), collTimestamp()}; ASSERT_EQ(v1, rt1.getVersion(kThisShard)); ASSERT_EQ(rt1.numChunks(), 2); } TEST_F(RoutingTableHistoryTest, TestOutOfOrderVersion) { - const UUID uuid = UUID::gen(); - const OID epoch = OID::gen(); - const Timestamp timestamp(1); - std::vector<ChunkType> initialChunks = { - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), BSON("a" << 0)}, - ChunkVersion{2, 1, epoch, timestamp}, + ChunkVersion{2, 1, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << 0), getShardKeyPattern().globalMax()}, - ChunkVersion{2, 2, epoch, timestamp}, + ChunkVersion{2, 2, collEpoch(), collTimestamp()}, kThisShard}}; - auto rt = RoutingTableHistory::makeNew(kNss, - uuid, - getShardKeyPattern(), - nullptr, - false, - epoch, - timestamp, - boost::none /* timeseriesFields */, - boost::none, - boost::none /* chunkSizeBytes */, - true, - initialChunks); + auto rt = makeNewRt(initialChunks); ASSERT_EQ(rt.numChunks(), 2); std::vector<ChunkType> changedChunks = { - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << 0), getShardKeyPattern().globalMax()}, - ChunkVersion{3, 0, epoch, timestamp}, + ChunkVersion{3, 0, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), BSON("a" << 0)}, - ChunkVersion{3, 1, epoch, timestamp}, + ChunkVersion{3, 1, collEpoch(), collTimestamp()}, kThisShard}}; auto rt1 = rt.makeUpdated( boost::none /* timeseriesFields */, boost::none, boost::none, true, changedChunks); - auto v1 = ChunkVersion{3, 1, epoch, timestamp}; + auto v1 = ChunkVersion{3, 1, collEpoch(), collTimestamp()}; ASSERT_EQ(v1, rt1.getVersion(kThisShard)); ASSERT_EQ(rt1.numChunks(), 2); auto chunk1 = rt1.findIntersectingChunk(BSON("a" << 0)); - ASSERT_EQ(chunk1->getLastmod(), ChunkVersion(3, 0, epoch, timestamp)); + ASSERT_EQ(chunk1->getLastmod(), ChunkVersion(3, 0, collEpoch(), collTimestamp())); ASSERT_EQ(chunk1->getMin().woCompare(BSON("a" << 0)), 0); ASSERT_EQ(chunk1->getMax().woCompare(getShardKeyPattern().globalMax()), 0); } TEST_F(RoutingTableHistoryTest, TestMergeChunks) { - const UUID uuid = UUID::gen(); - const OID epoch = OID::gen(); - const Timestamp timestamp(1); - std::vector<ChunkType> initialChunks = { - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << 0), BSON("a" << 10)}, - ChunkVersion{2, 0, epoch, timestamp}, + ChunkVersion{2, 0, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), BSON("a" << 0)}, - ChunkVersion{2, 1, epoch, timestamp}, + ChunkVersion{2, 1, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << 10), getShardKeyPattern().globalMax()}, - ChunkVersion{2, 2, epoch, timestamp}, + ChunkVersion{2, 2, collEpoch(), collTimestamp()}, kThisShard}}; - auto rt = RoutingTableHistory::makeNew(kNss, - uuid, - getShardKeyPattern(), - nullptr, - false, - epoch, - timestamp, - boost::none, - boost::none /* timeseriesFields */, - boost::none /* chunkSizeBytes */, - true, - initialChunks); + auto rt = makeNewRt(initialChunks); + ASSERT_EQ(rt.numChunks(), 3); - ASSERT_EQ(rt.getVersion(), ChunkVersion(2, 2, epoch, timestamp)); + ASSERT_EQ(rt.getVersion(), ChunkVersion(2, 2, collEpoch(), collTimestamp())); std::vector<ChunkType> changedChunks = { - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << 10), getShardKeyPattern().globalMax()}, - ChunkVersion{3, 0, epoch, timestamp}, + ChunkVersion{3, 0, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), BSON("a" << 10)}, - ChunkVersion{3, 1, epoch, timestamp}, + ChunkVersion{3, 1, collEpoch(), collTimestamp()}, kThisShard}}; auto rt1 = rt.makeUpdated( boost::none /* timeseriesFields */, boost::none, boost::none, true, changedChunks); - auto v1 = ChunkVersion{3, 1, epoch, timestamp}; + auto v1 = ChunkVersion{3, 1, collEpoch(), collTimestamp()}; ASSERT_EQ(v1, rt1.getVersion(kThisShard)); ASSERT_EQ(rt1.numChunks(), 2); } TEST_F(RoutingTableHistoryTest, TestMergeChunksOrdering) { - const UUID uuid = UUID::gen(); - const OID epoch = OID::gen(); - const Timestamp timestamp(1); - std::vector<ChunkType> initialChunks = { - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << -10), getShardKeyPattern().globalMax()}, - ChunkVersion{2, 0, epoch, timestamp}, + ChunkVersion{2, 0, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), BSON("a" << -500)}, - ChunkVersion{2, 1, epoch, timestamp}, + ChunkVersion{2, 1, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << -500), BSON("a" << -10)}, - ChunkVersion{2, 2, epoch, timestamp}, + ChunkVersion{2, 2, collEpoch(), collTimestamp()}, kThisShard}}; - auto rt = RoutingTableHistory::makeNew(kNss, - uuid, - getShardKeyPattern(), - nullptr, - false, - epoch, - timestamp, - boost::none /* timeseriesFields */, - boost::none, - boost::none /* chunkSizeBytes */, - true, - initialChunks); + auto rt = makeNewRt(initialChunks); ASSERT_EQ(rt.numChunks(), 3); - ASSERT_EQ(rt.getVersion(), ChunkVersion(2, 2, epoch, timestamp)); + ASSERT_EQ(rt.getVersion(), ChunkVersion(2, 2, collEpoch(), collTimestamp())); std::vector<ChunkType> changedChunks = { - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << -500), BSON("a" << -10)}, - ChunkVersion{2, 2, epoch, timestamp}, + ChunkVersion{2, 2, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), BSON("a" << -10)}, - ChunkVersion{3, 1, epoch, timestamp}, + ChunkVersion{3, 1, collEpoch(), collTimestamp()}, kThisShard}}; auto rt1 = rt.makeUpdated( boost::none /* timeseriesFields */, boost::none, boost::none, true, changedChunks); - auto v1 = ChunkVersion{3, 1, epoch, timestamp}; + auto v1 = ChunkVersion{3, 1, collEpoch(), collTimestamp()}; ASSERT_EQ(v1, rt1.getVersion(kThisShard)); ASSERT_EQ(rt1.numChunks(), 2); auto chunk1 = rt1.findIntersectingChunk(BSON("a" << -500)); - ASSERT_EQ(chunk1->getLastmod(), ChunkVersion(3, 1, epoch, timestamp)); + ASSERT_EQ(chunk1->getLastmod(), ChunkVersion(3, 1, collEpoch(), collTimestamp())); ASSERT_EQ(chunk1->getMin().woCompare(getShardKeyPattern().globalMin()), 0); ASSERT_EQ(chunk1->getMax().woCompare(BSON("a" << -10)), 0); } TEST_F(RoutingTableHistoryTest, TestFlatten) { - const UUID uuid = UUID::gen(); - const OID epoch = OID::gen(); - const Timestamp timestamp(1); - std::vector<ChunkType> initialChunks = { - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), BSON("a" << 10)}, - ChunkVersion{2, 0, epoch, timestamp}, + ChunkVersion{2, 0, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << 10), BSON("a" << 20)}, - ChunkVersion{2, 1, epoch, timestamp}, + ChunkVersion{2, 1, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << 20), getShardKeyPattern().globalMax()}, - ChunkVersion{2, 2, epoch, timestamp}, + ChunkVersion{2, 2, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), getShardKeyPattern().globalMax()}, - ChunkVersion{3, 0, epoch, timestamp}, + ChunkVersion{3, 0, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{getShardKeyPattern().globalMin(), BSON("a" << 10)}, - ChunkVersion{4, 0, epoch, timestamp}, + ChunkVersion{4, 0, collEpoch(), collTimestamp()}, kThisShard}, - ChunkType{uuid, + ChunkType{collUUID(), ChunkRange{BSON("a" << 10), getShardKeyPattern().globalMax()}, - ChunkVersion{4, 1, epoch, timestamp}, + ChunkVersion{4, 1, collEpoch(), collTimestamp()}, kThisShard}, }; - auto rt = RoutingTableHistory::makeNew(kNss, - UUID::gen(), - getShardKeyPattern(), - nullptr, - false, - epoch, - timestamp, - boost::none /* timeseriesFields */, - boost::none, - boost::none /* chunkSizeBytes */, - true, - initialChunks); + auto rt = makeNewRt(initialChunks); ASSERT_EQ(rt.numChunks(), 2); - ASSERT_EQ(rt.getVersion(), ChunkVersion(4, 1, epoch, timestamp)); + ASSERT_EQ(rt.getVersion(), ChunkVersion(4, 1, collEpoch(), collTimestamp())); auto chunk1 = rt.findIntersectingChunk(BSON("a" << 0)); - ASSERT_EQ(chunk1->getLastmod(), ChunkVersion(4, 0, epoch, timestamp)); + ASSERT_EQ(chunk1->getLastmod(), ChunkVersion(4, 0, collEpoch(), collTimestamp())); ASSERT_EQ(chunk1->getMin().woCompare(getShardKeyPattern().globalMin()), 0); ASSERT_EQ(chunk1->getMax().woCompare(BSON("a" << 10)), 0); } diff --git a/src/mongo/s/shard_id_test.cpp b/src/mongo/s/shard_id_test.cpp index 08765eeff91..1182d429e26 100644 --- a/src/mongo/s/shard_id_test.cpp +++ b/src/mongo/s/shard_id_test.cpp @@ -88,8 +88,8 @@ TEST(ShardId, Compare) { ShardId sa1(a1); ShardId sb(b); ASSERT_EQUALS(sa.compare(sa1), a.compare(a1)); - ASSERT_EQUALS(sb.compare(sa1), b.compare(a1)); - ASSERT_EQUALS(sa.compare(sb), a.compare(b)); + ASSERT_EQUALS(sb.compare(sa1) > 0, b.compare(a1) > 0); + ASSERT_EQUALS(sa.compare(sb) < 0, a.compare(b) < 0); } TEST(ShardId, Equals) { diff --git a/src/mongo/s/shard_key_pattern.cpp b/src/mongo/s/shard_key_pattern.cpp index c8fb1b7a84c..9e6f3244dcd 100644 --- a/src/mongo/s/shard_key_pattern.cpp +++ b/src/mongo/s/shard_key_pattern.cpp @@ -33,6 +33,7 @@ #include <vector> +#include "mongo/bson/simple_bsonelement_comparator.h" #include "mongo/db/field_ref.h" #include "mongo/db/field_ref_set.h" #include "mongo/db/hasher.h" @@ -85,10 +86,20 @@ std::vector<std::unique_ptr<FieldRef>> parseShardKeyPattern(const BSONObj& keyPa // Empty parts of the path, ".."? for (size_t i = 0; i < newFieldRef->numParts(); ++i) { + const StringData part = newFieldRef->getPart(i); + uassert(ErrorCodes::BadValue, str::stream() << "Field " << patternEl.fieldNameStringData() << " contains empty parts", - !newFieldRef->getPart(i).empty()); + !part.empty()); + + // Reject a shard key that has a field name that starts with '$' or contains parts that + // start with '$' unless the part is a DBRef (i.e. is equal to '$id', '$db' or '$ref'). + uassert(ErrorCodes::BadValue, + str::stream() << "Field " << patternEl.fieldNameStringData() + << " contains parts that start with '$'", + !part.startsWith("$") || + (i != 0 && (part == "$db" || part == "$id" || part == "$ref"))); } // Numeric and ascending (1.0), or "hashed" with exactly hashed field. @@ -288,7 +299,7 @@ bool ShardKeyPattern::isShardKey(const BSONObj& shardKey) const { } bool ShardKeyPattern::isExtendedBy(const ShardKeyPattern& newShardKeyPattern) const { - return toBSON().isFieldNamePrefixOf(newShardKeyPattern.toBSON()); + return toBSON().isPrefixOf(newShardKeyPattern.toBSON(), SimpleBSONElementComparator::kInstance); } BSONObj ShardKeyPattern::normalizeShardKey(const BSONObj& shardKey) const { @@ -544,12 +555,12 @@ BSONObj ShardKeyPattern::extractShardKeyFromQuery(const CanonicalQuery& query) c return keyBuilder.obj(); } -bool ShardKeyPattern::isUniqueIndexCompatible(const BSONObj& uniqueIndexPattern) const { - if (!uniqueIndexPattern.isEmpty() && uniqueIndexPattern.firstElementFieldName() == kIdField) { +bool ShardKeyPattern::isIndexUniquenessCompatible(const BSONObj& indexPattern) const { + if (!indexPattern.isEmpty() && indexPattern.firstElementFieldName() == kIdField) { return true; } - return _keyPattern.toBSON().isFieldNamePrefixOf(uniqueIndexPattern); + return _keyPattern.toBSON().isFieldNamePrefixOf(indexPattern); } BoundList ShardKeyPattern::flattenBounds(const IndexBounds& indexBounds) const { diff --git a/src/mongo/s/shard_key_pattern.h b/src/mongo/s/shard_key_pattern.h index 545d1906e31..70f5aff336d 100644 --- a/src/mongo/s/shard_key_pattern.h +++ b/src/mongo/s/shard_key_pattern.h @@ -274,36 +274,37 @@ public: BSONObj extractShardKeyFromQuery(const CanonicalQuery& query) const; /** - * Returns true if the shard key pattern can ensure that the unique index pattern is - * respected across all shards. + * Returns true if the shard key pattern can ensure that the index uniqueness is respected + * across all shards. * * Primarily this just checks whether the shard key pattern field names are equal to or a - * prefix of the unique index pattern field names. Since documents with the same fields in - * the shard key pattern are guaranteed to go to the same shard, and all documents must - * contain the full shard key, a unique index with a shard key pattern prefix can be sure - * when resolving duplicates that documents on other shards will have different shard keys, - * and so are not duplicates. + * prefix of the 'unique' or 'prepareUnique' index pattern field names. Since documents with the + * same fields in the shard key pattern are guaranteed to go to the same shard, and all + * documents must contain the full shard key, an index with {unique: true} or {prepareUnique: + * true} and a shard key pattern prefix can be sure when resolving duplicates that documents on + * other shards will have different shard keys, and so are not duplicates. * * Hashed shard key patterns are similar to ordinary patterns in that they guarantee similar * shard keys go to the same shard. * * Examples: - * shard key {a : 1} is compatible with a unique index on {_id : 1} - * shard key {a : 1} is compatible with a unique index on {a : 1 , b : 1} - * shard key {a : 1} is compatible with a unique index on {a : -1 , b : 1 } - * shard key {a : "hashed"} is compatible with a unique index on {a : 1} - * shard key {a : 1} is not compatible with a unique index on {b : 1} - * shard key {a : "hashed" , b : 1 } is not compatible with unique index on { b : 1 } + * shard key {a : 1} is compatible with a unique/prepareUnique index on {_id : 1} + * shard key {a : 1} is compatible with a unique/prepareUnique index on {a : 1, b : 1} + * shard key {a : 1} is compatible with a unique/prepareUnique index on {a : -1, b : 1} + * shard key {a : "hashed"} is compatible with a unique/prepareUnique index on {a : 1} + * shard key {a : 1} is not compatible with a unique/prepareUnique index on {b : 1} + * shard key {a : "hashed", b : 1} is not compatible with unique/prepareUnique index on + * {b : 1} * * All unique index patterns starting with _id are assumed to be enforceable by the fact * that _ids must be unique, and so all unique _id prefixed indexes are compatible with * any shard key pattern. * - * NOTE: We assume 'uniqueIndexPattern' is a valid unique index pattern - a pattern like - * { k : "hashed" } is not capable of being a unique index and is an invalid argument to - * this method. + * NOTE: We assume 'indexPattern' is a valid unique/prepareUnique index pattern - a pattern like + * { k : "hashed" } is not capable of being a unique/prepareUnique index and is an invalid + * argument to this method. */ - bool isUniqueIndexCompatible(const BSONObj& uniqueIndexPattern) const; + bool isIndexUniquenessCompatible(const BSONObj& indexPattern) const; /** * Return an ordered list of bounds generated using this KeyPattern and the diff --git a/src/mongo/s/shard_key_pattern_test.cpp b/src/mongo/s/shard_key_pattern_test.cpp index 2965b5d6222..2c1a3d92287 100644 --- a/src/mongo/s/shard_key_pattern_test.cpp +++ b/src/mongo/s/shard_key_pattern_test.cpp @@ -99,6 +99,7 @@ TEST_F(ShardKeyPatternTest, SingleFieldShardKeyPatternsValidityCheck) { ShardKeyPattern s3(BSON("a" << (long long)1L)); ShardKeyPattern s4(BSON("a" << "hashed")); + ShardKeyPattern s5(BSON("a$" << 1)); ASSERT_THROWS(ShardKeyPattern(BSONObj()), DBException); ASSERT_THROWS(ShardKeyPattern(BSON("a" << -1)), DBException); @@ -111,12 +112,21 @@ TEST_F(ShardKeyPatternTest, SingleFieldShardKeyPatternsValidityCheck) { DBException); ASSERT_THROWS(ShardKeyPattern(BSON("" << 1)), DBException); ASSERT_THROWS(ShardKeyPattern(BSON("." << 1)), DBException); + ASSERT_THROWS(ShardKeyPattern(BSON("$" << 1)), DBException); + ASSERT_THROWS(ShardKeyPattern(BSON("$a" << 1)), DBException); + ASSERT_THROWS(ShardKeyPattern(BSON("$**" << 1)), DBException); + ASSERT_THROWS(ShardKeyPattern(BSON("$id" << 1)), DBException); + ASSERT_THROWS(ShardKeyPattern(BSON("$db" << 1)), DBException); + ASSERT_THROWS(ShardKeyPattern(BSON("$ref" << 1)), DBException); } TEST_F(ShardKeyPatternTest, CompositeShardKeyPatternsValidityCheck) { ShardKeyPattern s1(BSON("a" << 1 << "b" << 1)); ShardKeyPattern s2(BSON("a" << 1.0f << "b" << 1.0)); ShardKeyPattern s3(BSON("a" << 1 << "b" << 1.0 << "c" << 1.0f)); + ShardKeyPattern s4(BSON("a.$id" << 1)); + ShardKeyPattern s5(BSON("a.$db" << 1)); + ShardKeyPattern s6(BSON("a.$ref" << 1)); ASSERT_THROWS(ShardKeyPattern(BSON("a" << 1 << "b" << -1)), DBException); ASSERT_THROWS(ShardKeyPattern(BSON("a" << 1 << "b" @@ -124,6 +134,9 @@ TEST_F(ShardKeyPatternTest, CompositeShardKeyPatternsValidityCheck) { DBException); ASSERT_THROWS(ShardKeyPattern(BSON("a" << 1 << "b." << 1.0)), DBException); ASSERT_THROWS(ShardKeyPattern(BSON("a" << 1 << "" << 1.0)), DBException); + ASSERT_THROWS(ShardKeyPattern(BSON("a" << 1 << "$" << 1.0)), DBException); + ASSERT_THROWS(ShardKeyPattern(BSON("a" << 1 << "$b" << 1.0)), DBException); + ASSERT_THROWS(ShardKeyPattern(BSON("a" << 1 << "$**" << 1.0)), DBException); } TEST_F(ShardKeyPatternTest, NestedShardKeyPatternsValidtyCheck) { @@ -131,6 +144,7 @@ TEST_F(ShardKeyPatternTest, NestedShardKeyPatternsValidtyCheck) { ShardKeyPattern s2(BSON("a.b.c.d" << 1.0)); ShardKeyPattern s3(BSON("a" << 1 << "c.d" << 1.0 << "e.f.g" << 1.0f)); ShardKeyPattern s4(BSON("a" << 1 << "a.b" << 1.0 << "a.b.c" << 1.0f)); + ShardKeyPattern s6(BSON("a.b$" << 1)); ASSERT_THROWS(ShardKeyPattern(BSON("a.b" << -1)), DBException); ASSERT_THROWS(ShardKeyPattern(BSON("a" << BSON("b" << 1))), DBException); @@ -139,6 +153,9 @@ TEST_F(ShardKeyPatternTest, NestedShardKeyPatternsValidtyCheck) { ASSERT_THROWS(ShardKeyPattern(BSON("a..b" << 1)), DBException); ASSERT_THROWS(ShardKeyPattern(BSON("a" << 1 << "a.b." << 1.0)), DBException); ASSERT_THROWS(ShardKeyPattern(BSON("a" << BSON("b" << 1) << "c.d" << 1.0)), DBException); + ASSERT_THROWS(ShardKeyPattern(BSON("a.$" << 1)), DBException); + ASSERT_THROWS(ShardKeyPattern(BSON("a.$b" << 1)), DBException); + ASSERT_THROWS(ShardKeyPattern(BSON("a.$**" << 1)), DBException); } TEST_F(ShardKeyPatternTest, IsShardKey) { @@ -608,7 +625,7 @@ TEST_F(ShardKeyPatternTest, ExtractQueryShardKeyHashed) { } static bool indexComp(const ShardKeyPattern& pattern, const BSONObj& indexPattern) { - return pattern.isUniqueIndexCompatible(indexPattern); + return pattern.isIndexUniquenessCompatible(indexPattern); } TEST_F(ShardKeyPatternTest, UniqueIndexCompatibleSingle) { @@ -1016,5 +1033,83 @@ TEST_F(ShardKeyPatternTest, ExtractShardKeyFromDocumentKey_Hashed) { BSONObj()); } +TEST_F(ShardKeyPatternTest, IsExtendedBy) { + + // NumberOfFields + ShardKeyPattern shardKeyPattern1(BSON("a" << 1)); + ShardKeyPattern shardKeyPattern2(BSON("a" << 1 << "b" << 1)); + ShardKeyPattern shardKeyPattern3(BSON("a" << 1 << "b" << 1 << "c" << 1)); + + // NumberOfFields_PositionOfHash + ShardKeyPattern shardKeyPatternHashed1_0(BSON("a" + << "hashed")); + ShardKeyPattern shardKeyPatternHashed2_0(BSON("a" + << "hashed" + << "b" << 1)); + ShardKeyPattern shardKeyPatternHashed2_1(BSON("a" << 1 << "b" + << "hashed")); + ShardKeyPattern shardKeyPatternHashed3_0(BSON("a" + << "hashed" + << "b" << 1 << "c" << 1)); + ShardKeyPattern shardKeyPatternHashed3_1(BSON("a" << 1 << "b" + << "hashed" + << "c" << 1)); + ShardKeyPattern shardKeyPatternHashed3_2(BSON("a" << 1 << "b" << 1 << "c" + << "hashed")); + + // same pattern, always true + ASSERT_TRUE(shardKeyPattern1.isExtendedBy(shardKeyPattern1)); + ASSERT_TRUE(shardKeyPattern2.isExtendedBy(shardKeyPattern2)); + ASSERT_TRUE(shardKeyPattern3.isExtendedBy(shardKeyPattern3)); + ASSERT_TRUE(shardKeyPatternHashed1_0.isExtendedBy(shardKeyPatternHashed1_0)); + ASSERT_TRUE(shardKeyPatternHashed2_0.isExtendedBy(shardKeyPatternHashed2_0)); + ASSERT_TRUE(shardKeyPatternHashed2_1.isExtendedBy(shardKeyPatternHashed2_1)); + ASSERT_TRUE(shardKeyPatternHashed3_0.isExtendedBy(shardKeyPatternHashed3_0)); + ASSERT_TRUE(shardKeyPatternHashed3_1.isExtendedBy(shardKeyPatternHashed3_1)); + ASSERT_TRUE(shardKeyPatternHashed3_2.isExtendedBy(shardKeyPatternHashed3_2)); + + // different number of fields, same values + ASSERT_TRUE(shardKeyPattern1.isExtendedBy(shardKeyPattern2)); + ASSERT_TRUE(shardKeyPattern2.isExtendedBy(shardKeyPattern3)); + ASSERT_TRUE(shardKeyPattern1.isExtendedBy(shardKeyPattern3)); + + ASSERT_FALSE(shardKeyPattern2.isExtendedBy(shardKeyPattern1)); + ASSERT_FALSE(shardKeyPattern3.isExtendedBy(shardKeyPattern2)); + ASSERT_FALSE(shardKeyPattern3.isExtendedBy(shardKeyPattern1)); + + // different number of fields, different values + // { a : 1 } is not extended by { a : "hashed" } and viceversa + ASSERT_FALSE(shardKeyPattern1.isExtendedBy(shardKeyPatternHashed1_0)); + ASSERT_FALSE(shardKeyPatternHashed1_0.isExtendedBy(shardKeyPattern1)); + + // { a : 1, b : 1 } is not extended by { a : 1, b : "hashed" } and viceversa + ASSERT_FALSE(shardKeyPattern2.isExtendedBy(shardKeyPatternHashed2_1)); + ASSERT_FALSE(shardKeyPatternHashed2_1.isExtendedBy(shardKeyPattern2)); + + // { a : 1 } is extended by { a : 1, b : "hashed" } but not viceversa + ASSERT_TRUE(shardKeyPattern1.isExtendedBy(shardKeyPatternHashed2_1)); + ASSERT_FALSE(shardKeyPatternHashed2_1.isExtendedBy(shardKeyPattern1)); + + // { a : 1, b : 1 } is extended by { a : 1, b : 1, c : "hashed" } but not viceversa + ASSERT_TRUE(shardKeyPattern2.isExtendedBy(shardKeyPatternHashed3_2)); + ASSERT_FALSE(shardKeyPatternHashed3_2.isExtendedBy(shardKeyPattern2)); + + // { a : 1, b : 1, c : 1 } is not extended by { a : 1, b : 1, c : "hashed" } and viceversa + ASSERT_FALSE(shardKeyPattern3.isExtendedBy(shardKeyPatternHashed3_2)); + ASSERT_FALSE(shardKeyPatternHashed3_2.isExtendedBy(shardKeyPattern3)); + + // { a: "hashed", b : 1 } is not extended by { a : 1, b : 1, c : "hashed" } and viceversa + ASSERT_FALSE(shardKeyPatternHashed2_1.isExtendedBy(shardKeyPatternHashed3_2)); + ASSERT_FALSE(shardKeyPatternHashed3_2.isExtendedBy(shardKeyPatternHashed2_1)); + + // { a : "hashed", b : 1 } is extended by { a : "hashed", b : 1, c : "1" } but not viceversa + ASSERT_TRUE(shardKeyPatternHashed2_0.isExtendedBy(shardKeyPatternHashed3_0)); + ASSERT_FALSE(shardKeyPatternHashed3_0.isExtendedBy(shardKeyPatternHashed2_0)); + + // { a : 1, b : "hashed " } is extended by { a : 1, b : "hashed", c : "1" } but not viceversa + ASSERT_TRUE(shardKeyPatternHashed2_1.isExtendedBy(shardKeyPatternHashed3_1)); + ASSERT_FALSE(shardKeyPatternHashed3_1.isExtendedBy(shardKeyPatternHashed2_1)); +} + } // namespace } // namespace mongo diff --git a/src/mongo/s/shard_util.cpp b/src/mongo/s/shard_util.cpp index ae5e1aa9a1b..62ee76524ea 100644 --- a/src/mongo/s/shard_util.cpp +++ b/src/mongo/s/shard_util.cpp @@ -132,7 +132,8 @@ StatusWith<std::vector<BSONObj>> selectChunkSplitPoints(OperationContext* opCtx, const NamespaceString& nss, const ShardKeyPattern& shardKeyPattern, const ChunkRange& chunkRange, - long long chunkSizeBytes) { + long long chunkSizeBytes, + boost::optional<int> limit) { auto shardStatus = Grid::get(opCtx)->shardRegistry()->getShard(opCtx, shardId); if (!shardStatus.isOK()) { return shardStatus.getStatus(); @@ -147,8 +148,9 @@ StatusWith<std::vector<BSONObj>> selectChunkSplitPoints(OperationContext* opCtx, Shard::RetryPolicy::kIdempotent); }; - const AutoSplitVectorRequest req( + AutoSplitVectorRequest req( nss, shardKeyPattern.toBSON(), chunkRange.getMin(), chunkRange.getMax(), chunkSizeBytes); + req.setLimit(limit); auto cmdStatus = invokeSplitCommand(req.toBSON({}), nss.db()); diff --git a/src/mongo/s/shard_util.h b/src/mongo/s/shard_util.h index 997dcc1e4c9..18eb780eb1e 100644 --- a/src/mongo/s/shard_util.h +++ b/src/mongo/s/shard_util.h @@ -79,20 +79,20 @@ StatusWith<long long> retrieveCollectionShardSize(OperationContext* opCtx, /** * Ask the specified shard to figure out the split points for a given chunk. * - * shardId The shard id to query. - * nss Namespace, which owns the chunk. - * shardKeyPattern The shard key which corresponds to this sharded namespace. - * chunkRange Bounds of the chunk to be split. - * chunkSize Chunk size to target in bytes. - * maxObjs Limits the number of objects in each chunk. Zero means max, unspecified means use the - * server default. + * - shardId: the shard id to query. + * - nss: namespace, which owns the chunk. + * - shardKeyPattern: the shard key which corresponds to this sharded namespace. + * - chunkRange: bounds of the chunk to search for split points on. + * - chunkSizeBytes: chunk size to target in bytes. + * - limit: limits the number of split points to search. Unspecified means no limit */ StatusWith<std::vector<BSONObj>> selectChunkSplitPoints(OperationContext* opCtx, const ShardId& shardId, const NamespaceString& nss, const ShardKeyPattern& shardKeyPattern, const ChunkRange& chunkRange, - long long chunkSizeBytes); + long long chunkSizeBytes, + boost::optional<int> limit = boost::none); /** * Asks the specified shard to split the chunk described by min/maxKey into the respective split diff --git a/src/mongo/s/sharding_feature_flags.idl b/src/mongo/s/sharding_feature_flags.idl index 2eea856a286..aaeb4365106 100644 --- a/src/mongo/s/sharding_feature_flags.idl +++ b/src/mongo/s/sharding_feature_flags.idl @@ -35,11 +35,13 @@ feature_flags: featureFlagNoMoreAutoSplitter: description: "Guarding code for the no more auto-splitter project" cpp_varname: feature_flags::gNoMoreAutoSplitter - default: false + default: true + version: 6.0 featureFlagBalanceAccordingToDataSize: description: Balancer taking decisions based on the data size if enabled, based on number of chunks if disabled cpp_varname: feature_flags::gBalanceAccordingToDataSize - default: false + default: true + version: 6.0 featureFlagShardingDataTransformMetrics: description: Feature flag for enabling the new metrics for global indexes and resharding. cpp_varname: feature_flags::gFeatureFlagShardingDataTransformMetrics @@ -55,3 +57,8 @@ feature_flags: cpp_varname: feature_flags::gOrphanTracking default: true version: 6.0 + featureFlagConcurrencyInChunkMigration: + description: "Feature flag for enabling concurrency within a chunk migration" + cpp_varname: feature_flags::gConcurrencyInChunkMigration + default: true + version: 6.0 diff --git a/src/mongo/s/sharding_initialization.cpp b/src/mongo/s/sharding_initialization.cpp index 98129648e05..aa183dd6177 100644 --- a/src/mongo/s/sharding_initialization.cpp +++ b/src/mongo/s/sharding_initialization.cpp @@ -196,6 +196,7 @@ Status initializeGlobalShardingState(OperationContext* opCtx, // The shard registry must be started once the grid is initialized grid->shardRegistry()->startupPeriodicReloader(opCtx); + // Start up the cluster time keys manager with a sharded keys client. auto keysCollectionClient = std::make_unique<KeysCollectionClientSharded>(grid->catalogClient()); auto keyManager = diff --git a/src/mongo/s/sharding_task_executor_pool_controller.cpp b/src/mongo/s/sharding_task_executor_pool_controller.cpp index d95a76a168f..c73967e6c50 100644 --- a/src/mongo/s/sharding_task_executor_pool_controller.cpp +++ b/src/mongo/s/sharding_task_executor_pool_controller.cpp @@ -253,7 +253,7 @@ auto ShardingTaskExecutorPoolController::updateHost(PoolId id, const HostState& "maxConns"_attr = maxConns); // Update the target for just the pool first - poolData.target = stats.requests + stats.active; + poolData.target = stats.requests + stats.active + stats.leased; if (poolData.target < minConns) { poolData.target = minConns; diff --git a/src/mongo/s/stale_exception.cpp b/src/mongo/s/stale_exception.cpp index ac026a45ea2..a04c6c27937 100644 --- a/src/mongo/s/stale_exception.cpp +++ b/src/mongo/s/stale_exception.cpp @@ -39,6 +39,18 @@ MONGO_INIT_REGISTER_ERROR_EXTRA_INFO(StaleConfigInfo); MONGO_INIT_REGISTER_ERROR_EXTRA_INFO(StaleEpochInfo); MONGO_INIT_REGISTER_ERROR_EXTRA_INFO(StaleDbRoutingVersion); +boost::optional<ChunkVersion> extractOptionalChunkVersion(const BSONObj& obj, StringData field) { + try { + return ChunkVersion::fromBSONLegacyOrNewerFormat(obj, field); + } catch (const DBException& ex) { + auto status = ex.toStatus(); + if (status != ErrorCodes::NoSuchKey) { + throw; + } + } + return boost::none; +} + } // namespace void StaleConfigInfo::serialize(BSONObjBuilder* bob) const { @@ -56,31 +68,35 @@ std::shared_ptr<const ErrorExtraInfo> StaleConfigInfo::parse(const BSONObj& obj) const auto shardId = obj["shardId"].String(); uassert(ErrorCodes::NoSuchKey, "The shardId field is missing", !shardId.empty()); - auto extractOptionalChunkVersion = [&obj](StringData field) -> boost::optional<ChunkVersion> { - try { - return ChunkVersion::fromBSONLegacyOrNewerFormat(obj, field); - } catch (const DBException& ex) { - auto status = ex.toStatus(); - if (status != ErrorCodes::NoSuchKey) { - throw; - } - } - return boost::none; - }; - return std::make_shared<StaleConfigInfo>( NamespaceString(obj["ns"].String()), ChunkVersion::fromBSONLegacyOrNewerFormat(obj, "vReceived"), - extractOptionalChunkVersion("vWanted"), + extractOptionalChunkVersion(obj, "vWanted"), ShardId(shardId)); } void StaleEpochInfo::serialize(BSONObjBuilder* bob) const { bob->append("ns", _nss.ns()); + if (_received) + _received->appendLegacyWithField(bob, "vReceived"); + if (_wanted) + _wanted->appendLegacyWithField(bob, "vWanted"); } std::shared_ptr<const ErrorExtraInfo> StaleEpochInfo::parse(const BSONObj& obj) { - return std::make_shared<StaleEpochInfo>(NamespaceString(obj["ns"].String())); + auto received = extractOptionalChunkVersion(obj, "vReceived"); + auto wanted = extractOptionalChunkVersion(obj, "vWanted"); + + uassert(6375907, + str::stream() << "Either both vReceived (" << received << ")" + << " and vWanted (" << wanted << ") must be present or none", + received.is_initialized() == wanted.is_initialized()); + + if (received) + return std::make_shared<StaleEpochInfo>( + NamespaceString(obj["ns"].String()), *received, *wanted); + else + return std::make_shared<StaleEpochInfo>(NamespaceString(obj["ns"].String())); } void StaleDbRoutingVersion::serialize(BSONObjBuilder* bob) const { diff --git a/src/mongo/s/stale_exception.h b/src/mongo/s/stale_exception.h index d82b18bab52..1778fe689a4 100644 --- a/src/mongo/s/stale_exception.h +++ b/src/mongo/s/stale_exception.h @@ -75,7 +75,7 @@ public: void serialize(BSONObjBuilder* bob) const; static std::shared_ptr<const ErrorExtraInfo> parse(const BSONObj& obj); -protected: +private: NamespaceString _nss; ChunkVersion _received; boost::optional<ChunkVersion> _wanted; @@ -85,24 +85,41 @@ protected: boost::optional<SharedSemiFuture<void>> _criticalSectionSignal; }; +// TODO (SERVER-74380): Rename the StaleEpoch code to StaleDownstreamRouter and the info to +// StaleDownstreamRouterInfo class StaleEpochInfo final : public ErrorExtraInfo { public: static constexpr auto code = ErrorCodes::StaleEpoch; + StaleEpochInfo(NamespaceString nss, ChunkVersion received, ChunkVersion wanted) + : _nss(std::move(nss)), _received(received), _wanted(wanted) {} + + // TODO (SERVER-74380): Remove this constructor StaleEpochInfo(NamespaceString nss) : _nss(std::move(nss)) {} const auto& getNss() const { return _nss; } + const auto& getVersionReceived() const { + return _received; + } + + const auto& getVersionWanted() const { + return _wanted; + } + void serialize(BSONObjBuilder* bob) const; static std::shared_ptr<const ErrorExtraInfo> parse(const BSONObj& obj); private: NamespaceString _nss; -}; -using StaleConfigException = ExceptionFor<ErrorCodes::StaleConfig>; + // TODO (SERVER-74380): These two fields are boost::optional for backwards compatibility. Either + // both of them are boost::none or both are set. + boost::optional<ChunkVersion> _received; + boost::optional<ChunkVersion> _wanted; +}; class StaleDbRoutingVersion final : public ErrorExtraInfo { public: diff --git a/src/mongo/s/stale_exception_test.cpp b/src/mongo/s/stale_exception_test.cpp index 57cb2b89062..98a2b897af5 100644 --- a/src/mongo/s/stale_exception_test.cpp +++ b/src/mongo/s/stale_exception_test.cpp @@ -55,7 +55,7 @@ TEST(StaleExceptionTest, StaleConfigInfoSerializationTest) { ASSERT_EQUALS(deserializedInfo->getShardId(), kShardId); } -TEST(StaleExceptionTest, StaleEpochInfoSerializationTest) { +TEST(StaleExceptionTest, StaleEpochInfoLegacySerializationTest) { StaleEpochInfo info(kNss); // Serialize @@ -67,6 +67,24 @@ TEST(StaleExceptionTest, StaleEpochInfoSerializationTest) { std::static_pointer_cast<const StaleEpochInfo>(StaleEpochInfo::parse(bob.obj())); ASSERT_EQUALS(deserializedInfo->getNss(), kNss); + ASSERT(!deserializedInfo->getVersionReceived()); + ASSERT(!deserializedInfo->getVersionWanted()); +} + +TEST(StaleExceptionTest, StaleEpochInfoSerializationTest) { + StaleEpochInfo info(kNss, ChunkVersion::UNSHARDED(), ChunkVersion::UNSHARDED()); + + // Serialize + BSONObjBuilder bob; + info.serialize(&bob); + + // Deserialize + auto deserializedInfo = + std::static_pointer_cast<const StaleEpochInfo>(StaleEpochInfo::parse(bob.obj())); + + ASSERT_EQ(deserializedInfo->getNss(), kNss); + ASSERT_EQ(*deserializedInfo->getVersionReceived(), ChunkVersion::UNSHARDED()); + ASSERT_EQ(*deserializedInfo->getVersionWanted(), ChunkVersion::UNSHARDED()); } } // namespace diff --git a/src/mongo/s/write_ops/batch_write_exec_test.cpp b/src/mongo/s/write_ops/batch_write_exec_test.cpp index 3b4079b4c84..1381756da21 100644 --- a/src/mongo/s/write_ops/batch_write_exec_test.cpp +++ b/src/mongo/s/write_ops/batch_write_exec_test.cpp @@ -33,6 +33,7 @@ #include "mongo/client/remote_command_targeter_factory_mock.h" #include "mongo/client/remote_command_targeter_mock.h" #include "mongo/db/commands.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/logical_session_id.h" #include "mongo/db/vector_clock.h" #include "mongo/s/catalog/type_shard.h" @@ -2413,11 +2414,9 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_TransientTxnError) { auto future = launchAsync([&] { BatchedCommandResponse response; BatchWriteExecStats stats; - ASSERT_THROWS_CODE( - BatchWriteExec::executeBatch( - operationContext(), singleShardNSTargeter, request, &response, &stats), - AssertionException, - ErrorCodes::WriteConflict); + ASSERT_THROWS(BatchWriteExec::executeBatch( + operationContext(), singleShardNSTargeter, request, &response, &stats), + WriteConflictException); }); expectInsertsReturnTransientTxnErrors({BSON("x" << 1), BSON("x" << 2)}); diff --git a/src/mongo/s/write_ops/batch_write_op.cpp b/src/mongo/s/write_ops/batch_write_op.cpp index a61ee3dd4bf..86b2f4c4fab 100644 --- a/src/mongo/s/write_ops/batch_write_op.cpp +++ b/src/mongo/s/write_ops/batch_write_op.cpp @@ -59,7 +59,6 @@ struct WriteErrorComp { // batches before serializing. // // TODO: Revisit when we revisit command limits in general -const int kEstUpdateOverheadBytes = (BSONObjMaxInternalSize - BSONObjMaxUserSize) / 100; const int kEstDeleteOverheadBytes = (BSONObjMaxInternalSize - BSONObjMaxUserSize) / 100; /** @@ -159,51 +158,17 @@ int getWriteSizeBytes(const WriteOp& writeOp) { return item.getDocument().objsize(); } else if (batchType == BatchedCommandRequest::BatchType_Update) { // Note: Be conservative here - it's okay if we send slightly too many batches. - auto estSize = static_cast<int>(BSONObj::kMinBSONLength); - static const auto boolSize = 1; - - // Add the size of the 'collation' field, if present. - estSize += !item.getUpdate().getCollation() ? 0 - : (UpdateOpEntry::kCollationFieldName.size() + - item.getUpdate().getCollation()->objsize()); - - // Add the size of the 'arrayFilters' field, if present. - estSize += !item.getUpdate().getArrayFilters() ? 0 : ([&item]() { - auto size = BSONObj::kMinBSONLength + UpdateOpEntry::kArrayFiltersFieldName.size(); - for (auto&& filter : *item.getUpdate().getArrayFilters()) { - size += filter.objsize(); - } - return size; - })(); - - // Add the sizes of the 'multi' and 'upsert' fields. - estSize += UpdateOpEntry::kUpsertFieldName.size() + boolSize; - estSize += UpdateOpEntry::kMultiFieldName.size() + boolSize; - - // Add the size of 'upsertSupplied' field if present. - if (auto upsertSupplied = item.getUpdate().getUpsertSupplied()) { - estSize += UpdateOpEntry::kUpsertSuppliedFieldName.size() + boolSize; - } - - // Add the sizes of the 'q' and 'u' fields. - estSize += (UpdateOpEntry::kQFieldName.size() + item.getUpdate().getQ().objsize() + - UpdateOpEntry::kUFieldName.size() + item.getUpdate().getU().objsize()); - - // Add the size of the 'c' field if present. - if (auto constants = item.getUpdate().getC()) { - estSize += UpdateOpEntry::kCFieldName.size() + item.getUpdate().getC()->objsize(); - } - - // Add the size of 'hint' field if present. - if (auto hint = item.getUpdate().getHint(); !hint.isEmpty()) { - estSize += UpdateOpEntry::kHintFieldName.size() + hint.objsize(); - } - - // Finally, add the constant updateOp overhead size. - estSize += kEstUpdateOverheadBytes; + const auto& update = item.getUpdate(); + auto estSize = write_ops::getUpdateSizeEstimate(update.getQ(), + update.getU(), + update.getC(), + update.getUpsertSupplied().has_value(), + update.getCollation(), + update.getArrayFilters(), + update.getHint()); // When running a debug build, verify that estSize is at least the BSON serialization size. - dassert(estSize >= item.getUpdate().toBSON().objsize()); + dassert(estSize >= update.toBSON().objsize()); return estSize; } else if (batchType == BatchedCommandRequest::BatchType_Delete) { // Note: Be conservative here - it's okay if we send slightly too many batches. diff --git a/src/mongo/s/write_ops/batched_command_request.h b/src/mongo/s/write_ops/batched_command_request.h index eea7f7bbe11..2caf9b07a4a 100644 --- a/src/mongo/s/write_ops/batched_command_request.h +++ b/src/mongo/s/write_ops/batched_command_request.h @@ -52,15 +52,25 @@ public: : _batchType(BatchType_Insert), _insertReq(std::make_unique<write_ops::InsertCommandRequest>(std::move(insertOp))) {} + BatchedCommandRequest(std::unique_ptr<write_ops::InsertCommandRequest> insertOp) + : _batchType(BatchType_Insert), _insertReq(std::move(insertOp)) {} + BatchedCommandRequest(write_ops::UpdateCommandRequest updateOp) : _batchType(BatchType_Update), _updateReq(std::make_unique<write_ops::UpdateCommandRequest>(std::move(updateOp))) {} + BatchedCommandRequest(std::unique_ptr<write_ops::UpdateCommandRequest> updateOp) + : _batchType(BatchType_Update), _updateReq(std::move(updateOp)) {} + BatchedCommandRequest(write_ops::DeleteCommandRequest deleteOp) : _batchType(BatchType_Delete), _deleteReq(std::make_unique<write_ops::DeleteCommandRequest>(std::move(deleteOp))) {} + BatchedCommandRequest(std::unique_ptr<write_ops::DeleteCommandRequest> deleteOp) + : _batchType(BatchType_Delete), _deleteReq(std::move(deleteOp)) {} + BatchedCommandRequest(BatchedCommandRequest&&) = default; + BatchedCommandRequest& operator=(BatchedCommandRequest&&) = default; static BatchedCommandRequest parseInsert(const OpMsgRequest& request); static BatchedCommandRequest parseUpdate(const OpMsgRequest& request); @@ -91,6 +101,18 @@ public: return *_deleteReq; } + std::unique_ptr<write_ops::InsertCommandRequest> extractInsertRequest() { + return std::move(_insertReq); + } + + std::unique_ptr<write_ops::UpdateCommandRequest> extractUpdateRequest() { + return std::move(_updateReq); + } + + std::unique_ptr<write_ops::DeleteCommandRequest> extractDeleteRequest() { + return std::move(_deleteReq); + } + std::size_t sizeWriteOps() const; void setWriteConcern(const BSONObj& writeConcern) { |