diff options
Diffstat (limited to 'src/mongo/db/commands')
48 files changed, 671 insertions, 1454 deletions
diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript index d5210d9603c..143c5eae3ef 100644 --- a/src/mongo/db/commands/SConscript +++ b/src/mongo/db/commands/SConscript @@ -231,9 +231,8 @@ env.Library( '$BUILD_DIR/mongo/db/auth/auth', '$BUILD_DIR/mongo/db/auth/authprivilege', '$BUILD_DIR/mongo/db/commands', - '$BUILD_DIR/mongo/db/concurrency/exception_util', - '$BUILD_DIR/mongo/db/dbdirectclient', - '$BUILD_DIR/mongo/db/query/query_stats/query_stats', + '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception', + '$BUILD_DIR/mongo/db/curop', '$BUILD_DIR/mongo/db/storage/backup_cursor_hooks', 'fsync_locked', ] @@ -373,8 +372,8 @@ env.Library( '$BUILD_DIR/mongo/db/catalog/multi_index_block', '$BUILD_DIR/mongo/db/command_can_run_here', '$BUILD_DIR/mongo/db/commands', - '$BUILD_DIR/mongo/db/concurrency/exception_util', '$BUILD_DIR/mongo/db/concurrency/lock_manager', + '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception', '$BUILD_DIR/mongo/db/curop_failpoint_helpers', '$BUILD_DIR/mongo/db/exec/sbe/query_sbe_abt', '$BUILD_DIR/mongo/db/fle_crud_mongod', @@ -387,8 +386,7 @@ env.Library( '$BUILD_DIR/mongo/db/query/ce/query_ce', '$BUILD_DIR/mongo/db/query/command_request_response', '$BUILD_DIR/mongo/db/query/cursor_response_idl', - '$BUILD_DIR/mongo/db/query/query_shape/query_shape', - '$BUILD_DIR/mongo/db/query/query_stats/query_stats', + '$BUILD_DIR/mongo/db/query/optimizer/optimizer', '$BUILD_DIR/mongo/db/query_exec', '$BUILD_DIR/mongo/db/repl/replica_set_messages', '$BUILD_DIR/mongo/db/repl/tenant_migration_access_blocker', @@ -403,7 +401,6 @@ env.Library( '$BUILD_DIR/mongo/db/timeseries/catalog_helper', '$BUILD_DIR/mongo/db/timeseries/timeseries_collmod', '$BUILD_DIR/mongo/db/timeseries/timeseries_conversion_util', - '$BUILD_DIR/mongo/db/timeseries/timeseries_extended_range', '$BUILD_DIR/mongo/db/timeseries/timeseries_options', '$BUILD_DIR/mongo/db/timeseries/timeseries_stats', '$BUILD_DIR/mongo/db/transaction', @@ -497,7 +494,6 @@ env.Library( 'shutdown.idl', ], LIBDEPS_PRIVATE=[ - '$BUILD_DIR/mongo/bson/bson_validate', '$BUILD_DIR/mongo/idl/idl_parser', '$BUILD_DIR/mongo/util/fail_point', ], @@ -579,7 +575,6 @@ env.Library( '$BUILD_DIR/mongo/db/repl/tenant_migration_donor_service', '$BUILD_DIR/mongo/db/repl/tenant_migration_recipient_service', '$BUILD_DIR/mongo/db/rw_concern_d', - '$BUILD_DIR/mongo/db/s/balancer_stats_registry', '$BUILD_DIR/mongo/db/s/sharding_api_d', '$BUILD_DIR/mongo/db/s/sharding_catalog_manager', '$BUILD_DIR/mongo/db/s/sharding_commands_d', @@ -666,7 +661,6 @@ env.Library( 'profile_common.cpp', 'profile.idl', '$BUILD_DIR/mongo/db/profile_filter_impl.cpp', - 'set_profiling_filter_globally_cmd.cpp', ], LIBDEPS=[ '$BUILD_DIR/mongo/db/commands', @@ -674,8 +668,8 @@ env.Library( LIBDEPS_PRIVATE=[ '$BUILD_DIR/mongo/db/auth/auth', '$BUILD_DIR/mongo/db/auth/authprivilege', + '$BUILD_DIR/mongo/db/curop', '$BUILD_DIR/mongo/db/matcher/expressions', - '$BUILD_DIR/mongo/db/query/query_stats/query_stats', '$BUILD_DIR/mongo/db/server_options_core', ], ) @@ -742,7 +736,7 @@ env.Library( LIBDEPS=[ '$BUILD_DIR/mongo/db/commands/servers', '$BUILD_DIR/mongo/db/db_raii', - '$BUILD_DIR/mongo/db/index/index_access_method', + '$BUILD_DIR/mongo/db/index/index_access_methods', '$BUILD_DIR/mongo/db/pipeline/process_interface/mongo_process_interface', '$BUILD_DIR/mongo/db/pipeline/process_interface/mongod_process_interface_factory', '$BUILD_DIR/mongo/db/query/map_reduce_output_format', @@ -815,7 +809,6 @@ env.CppUnitTest( "$BUILD_DIR/mongo/db/auth/authmocks", "$BUILD_DIR/mongo/db/catalog/collection", "$BUILD_DIR/mongo/db/commands/list_collections_filter", - "$BUILD_DIR/mongo/db/concurrency/exception_util", "$BUILD_DIR/mongo/db/dbdirectclient", "$BUILD_DIR/mongo/db/fle_crud", "$BUILD_DIR/mongo/db/fle_mocks", diff --git a/src/mongo/db/commands/apply_ops_cmd.cpp b/src/mongo/db/commands/apply_ops_cmd.cpp index 1a31cb0d8f8..2b379ee085e 100644 --- a/src/mongo/db/commands/apply_ops_cmd.cpp +++ b/src/mongo/db/commands/apply_ops_cmd.cpp @@ -39,6 +39,7 @@ #include "mongo/db/client.h" #include "mongo/db/commands.h" #include "mongo/db/commands/oplog_application_checks.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/jsobj.h" diff --git a/src/mongo/db/commands/authentication_commands.cpp b/src/mongo/db/commands/authentication_commands.cpp index 81a844dacba..0fef67553d0 100644 --- a/src/mongo/db/commands/authentication_commands.cpp +++ b/src/mongo/db/commands/authentication_commands.cpp @@ -240,7 +240,7 @@ void _authenticateX509(OperationContext* opCtx, AuthenticationSession* session) auto user = [&] { if (session->getUserName().empty()) { auto user = UserName(clientName.toString(), session->getDatabase().toString()); - session->updateUserName(user, true /* isMechX509 */); + session->updateUserName(user); return user; } else { uassert(ErrorCodes::AuthenticationFailed, @@ -258,6 +258,10 @@ void _authenticateX509(OperationContext* opCtx, AuthenticationSession* session) auto sslConfiguration = opCtx->getClient()->session()->getSSLConfiguration(); + uassert(ErrorCodes::AuthenticationFailed, + "Unable to verify x.509 certificate, as no CA has been provided.", + sslConfiguration->hasCA); + uassert(ErrorCodes::ProtocolError, "X.509 authentication must always use the $external database.", user.getDB() == kExternalDB); @@ -337,9 +341,9 @@ AuthenticateReply authCommand(OperationContext* opCtx, // Allows authenticating as the internal user against the admin database. This is to // support the auth passthrough test framework on mongos (since you can't use the local // database on a mongos, so you can't auth as the internal user without this). - session->updateUserName(internalSecurityUser, mechanism == auth::kMechanismMongoX509); + session->updateUserName(internalSecurityUser); } else { - session->updateUserName(UserName{user, dbname}, mechanism == auth::kMechanismMongoX509); + session->updateUserName(UserName{user, dbname}); } if (mechanism.empty()) { diff --git a/src/mongo/db/commands/authentication_commands.h b/src/mongo/db/commands/authentication_commands.h index 814a2ab17d2..2e82168e050 100644 --- a/src/mongo/db/commands/authentication_commands.h +++ b/src/mongo/db/commands/authentication_commands.h @@ -42,6 +42,6 @@ constexpr StringData kX509AuthMechanism = "MONGODB-X509"_sd; void disableX509Auth(ServiceContext* svcCtx); bool isX509AuthDisabled(ServiceContext* svcCtx); -void doSpeculativeAuthenticate(OperationContext* opCtx, BSONObj helloCmd, BSONObjBuilder* result); +void doSpeculativeAuthenticate(OperationContext* opCtx, BSONObj isMaster, BSONObjBuilder* result); } // namespace mongo diff --git a/src/mongo/db/commands/command_mirroring_test.cpp b/src/mongo/db/commands/command_mirroring_test.cpp index 8222cc9b74e..286d20813a2 100644 --- a/src/mongo/db/commands/command_mirroring_test.cpp +++ b/src/mongo/db/commands/command_mirroring_test.cpp @@ -38,7 +38,6 @@ #include "mongo/unittest/unittest.h" namespace mongo { - namespace { class CommandMirroringTest : public unittest::Test { @@ -109,7 +108,6 @@ public: void setUp() override { CommandMirroringTest::setUp(); shardVersion = boost::none; - databaseVersion = boost::none; } std::string commandName() override { @@ -121,10 +119,6 @@ public: if (shardVersion) { args.push_back(shardVersion.get()); } - if (databaseVersion) { - args.push_back(databaseVersion.value()); - } - auto request = CommandMirroringTest::makeCommand(coll, args); // Directly add `updates` to `OpMsg::sequences` to emulate `OpMsg::parse()` behavior. @@ -140,7 +134,6 @@ public: } boost::optional<BSONObj> shardVersion; - boost::optional<BSONObj> databaseVersion; }; TEST_F(UpdateCommandTest, NoQuery) { @@ -199,26 +192,21 @@ TEST_F(UpdateCommandTest, MultipleQueries) { ASSERT_EQ(mirroredObj["batchSize"].Int(), 1); } -TEST_F(UpdateCommandTest, ValidateShardVersionAndDatabaseVersion) { +TEST_F(UpdateCommandTest, ValidateShardVersion) { auto update = BSON("q" << BSONObj() << "u" << BSON("$set" << BSON("_id" << 1))); { auto mirroredObj = createCommandAndGetMirrored(kCollection, {update}); ASSERT_FALSE(mirroredObj.hasField("shardVersion")); - ASSERT_FALSE(mirroredObj.hasField("databaseVersion")); } const auto kShardVersion = 123; - const auto kDatabaseVersion = 456; shardVersion = BSON("shardVersion" << kShardVersion); - databaseVersion = BSON("databaseVersion" << kDatabaseVersion); { auto mirroredObj = createCommandAndGetMirrored(kCollection, {update}); ASSERT_TRUE(mirroredObj.hasField("shardVersion")); - ASSERT_TRUE(mirroredObj.hasField("databaseVersion")); ASSERT_EQ(mirroredObj["shardVersion"].Int(), kShardVersion); - ASSERT_EQ(mirroredObj["databaseVersion"].Int(), kDatabaseVersion); } } @@ -240,8 +228,7 @@ public: "max", "batchSize", "singleBatch", - "shardVersion", - "databaseVersion"}; + "shardVersion"}; } void checkFieldNamesAreAllowed(BSONObj& mirroredObj) { @@ -276,8 +263,7 @@ TEST_F(FindCommandTest, MirrorableKeys) { BSON("awaitData" << true), BSON("allowPartialResults" << true), BSON("collation" << BSONObj()), - BSON("shardVersion" << BSONObj()), - BSON("databaseVersion" << BSONObj())}; + BSON("shardVersion" << BSONObj())}; auto mirroredObj = createCommandAndGetMirrored(kCollection, findArgs); checkFieldNamesAreAllowed(mirroredObj); @@ -305,8 +291,7 @@ TEST_F(FindCommandTest, ValidateMirroredQuery) { const auto min = BSONObj(); const auto max = BSONObj(); - const auto shardVersion = BSON("v" << 123); - const auto databaseVersion = BSON("v" << 456); + const auto shardVersion = BSONObj(); auto findArgs = {BSON("filter" << filter), BSON("skip" << skip), @@ -316,8 +301,7 @@ TEST_F(FindCommandTest, ValidateMirroredQuery) { BSON("collation" << collation), BSON("min" << min), BSON("max" << max), - BSON("shardVersion" << shardVersion), - BSON("databaseVersion" << databaseVersion)}; + BSON("shardVersion" << shardVersion)}; auto mirroredObj = createCommandAndGetMirrored(kCollection, findArgs); @@ -331,27 +315,21 @@ TEST_F(FindCommandTest, ValidateMirroredQuery) { ASSERT(compareBSONObjs(mirroredObj["min"].Obj(), min)); ASSERT(compareBSONObjs(mirroredObj["max"].Obj(), max)); ASSERT(compareBSONObjs(mirroredObj["shardVersion"].Obj(), shardVersion)); - ASSERT(compareBSONObjs(mirroredObj["databaseVersion"].Obj(), databaseVersion)); } -TEST_F(FindCommandTest, ValidateShardVersionAndDatabaseVersion) { +TEST_F(FindCommandTest, ValidateShardVersion) { std::vector<BSONObj> findArgs = {BSON("filter" << BSONObj())}; { auto mirroredObj = createCommandAndGetMirrored(kCollection, findArgs); ASSERT_FALSE(mirroredObj.hasField("shardVersion")); - ASSERT_FALSE(mirroredObj.hasField("databaseVersion")); } const auto kShardVersion = 123; - const auto kDatabaseVersion = 456; findArgs.push_back(BSON("shardVersion" << kShardVersion)); - findArgs.push_back(BSON("databaseVersion" << kDatabaseVersion)); { auto mirroredObj = createCommandAndGetMirrored(kCollection, findArgs); ASSERT_TRUE(mirroredObj.hasField("shardVersion")); - ASSERT_TRUE(mirroredObj.hasField("databaseVersion")); ASSERT_EQ(mirroredObj["shardVersion"].Int(), kShardVersion); - ASSERT_EQ(mirroredObj["databaseVersion"].Int(), kDatabaseVersion); } } @@ -362,14 +340,7 @@ public: } std::vector<std::string> getAllowedKeys() const override { - return {"sort", - "collation", - "find", - "filter", - "batchSize", - "singleBatch", - "shardVersion", - "databaseVersion"}; + return {"sort", "collation", "find", "filter", "batchSize", "singleBatch", "shardVersion"}; } }; @@ -385,9 +356,7 @@ TEST_F(FindAndModifyCommandTest, MirrorableKeys) { BSON("writeConcern" << BSONObj()), BSON("maxTimeMS" << 100), BSON("collation" << BSONObj()), - BSON("arrayFilters" << BSONArray()), - BSON("shardVersion" << 123), - BSON("databaseVersion" << 456)}; + BSON("arrayFilters" << BSONArray())}; auto mirroredObj = createCommandAndGetMirrored(kCollection, findAndModifyArgs); checkFieldNamesAreAllowed(mirroredObj); @@ -412,16 +381,12 @@ TEST_F(FindAndModifyCommandTest, ValidateMirroredQuery) { constexpr auto upsert = true; const auto collation = BSON("locale" << "\"fr\""); - const auto shardVersion = BSON("v" << 123); - const auto databaseVersion = BSON("v" << 456); auto findAndModifyArgs = {BSON("query" << query), BSON("sort" << sortObj), BSON("update" << update), BSON("upsert" << upsert), - BSON("collation" << collation), - BSON("shardVersion" << shardVersion), - BSON("databaseVersion" << databaseVersion)}; + BSON("collation" << collation)}; auto mirroredObj = createCommandAndGetMirrored(kCollection, findAndModifyArgs); @@ -430,11 +395,9 @@ TEST_F(FindAndModifyCommandTest, ValidateMirroredQuery) { ASSERT(compareBSONObjs(mirroredObj["filter"].Obj(), query)); ASSERT(compareBSONObjs(mirroredObj["sort"].Obj(), sortObj)); ASSERT(compareBSONObjs(mirroredObj["collation"].Obj(), collation)); - ASSERT(compareBSONObjs(mirroredObj["shardVersion"].Obj(), shardVersion)); - ASSERT(compareBSONObjs(mirroredObj["databaseVersion"].Obj(), databaseVersion)); } -TEST_F(FindAndModifyCommandTest, ValidateShardVersionAndDatabaseVersion) { +TEST_F(FindAndModifyCommandTest, ValidateShardVersion) { std::vector<BSONObj> findAndModifyArgs = {BSON("query" << BSON("name" << "Andy")), BSON("update" << BSON("$inc" << BSON("score" << 1)))}; @@ -442,19 +405,14 @@ TEST_F(FindAndModifyCommandTest, ValidateShardVersionAndDatabaseVersion) { { auto mirroredObj = createCommandAndGetMirrored(kCollection, findAndModifyArgs); ASSERT_FALSE(mirroredObj.hasField("shardVersion")); - ASSERT_FALSE(mirroredObj.hasField("databaseVersion")); } const auto kShardVersion = 123; - const auto kDatabaseVersion = 456; findAndModifyArgs.push_back(BSON("shardVersion" << kShardVersion)); - findAndModifyArgs.push_back(BSON("databaseVersion" << kDatabaseVersion)); { auto mirroredObj = createCommandAndGetMirrored(kCollection, findAndModifyArgs); ASSERT_TRUE(mirroredObj.hasField("shardVersion")); - ASSERT_TRUE(mirroredObj.hasField("databaseVersion")); ASSERT_EQ(mirroredObj["shardVersion"].Int(), kShardVersion); - ASSERT_EQ(mirroredObj["databaseVersion"].Int(), kDatabaseVersion); } } @@ -465,7 +423,7 @@ public: } std::vector<std::string> getAllowedKeys() const override { - return {"distinct", "key", "query", "collation", "shardVersion", "databaseVersion"}; + return {"distinct", "key", "query", "collation", "shardVersion"}; } }; @@ -475,8 +433,7 @@ TEST_F(DistinctCommandTest, MirrorableKeys) { BSON("query" << BSONObj()), BSON("readConcern" << BSONObj()), BSON("collation" << BSONObj()), - BSON("shardVersion" << BSONObj()), - BSON("databaseVersion" << BSONObj())}; + BSON("shardVersion" << BSONObj())}; auto mirroredObj = createCommandAndGetMirrored(kCollection, distinctArgs); checkFieldNamesAreAllowed(mirroredObj); @@ -489,15 +446,13 @@ TEST_F(DistinctCommandTest, ValidateMirroredQuery) { const auto readConcern = BSON("level" << "majority"); const auto collation = BSON("strength" << 1); - const auto shardVersion = BSON("v" << 123); - const auto databaseVersion = BSON("v" << 456); + const auto shardVersion = BSONObj(); auto distinctArgs = {BSON("key" << key), BSON("query" << query), BSON("readConcern" << readConcern), BSON("collation" << collation), - BSON("shardVersion" << shardVersion), - BSON("databaseVersion" << databaseVersion)}; + BSON("shardVersion" << shardVersion)}; auto mirroredObj = createCommandAndGetMirrored(kCollection, distinctArgs); @@ -507,27 +462,23 @@ TEST_F(DistinctCommandTest, ValidateMirroredQuery) { ASSERT(compareBSONObjs(mirroredObj["query"].Obj(), query)); ASSERT(compareBSONObjs(mirroredObj["collation"].Obj(), collation)); ASSERT(compareBSONObjs(mirroredObj["shardVersion"].Obj(), shardVersion)); - ASSERT(compareBSONObjs(mirroredObj["databaseVersion"].Obj(), databaseVersion)); } -TEST_F(DistinctCommandTest, ValidateShardVersionAndDatabaseVersion) { +TEST_F(DistinctCommandTest, ValidateShardVersion) { + const auto kCollection = "test"; + std::vector<BSONObj> distinctArgs = {BSON("distinct" << BSONObj())}; { auto mirroredObj = createCommandAndGetMirrored(kCollection, distinctArgs); ASSERT_FALSE(mirroredObj.hasField("shardVersion")); - ASSERT_FALSE(mirroredObj.hasField("databaseVersion")); } const auto kShardVersion = 123; - const auto kDatabaseVersion = 456; distinctArgs.push_back(BSON("shardVersion" << kShardVersion)); - distinctArgs.push_back(BSON("databaseVersion" << kDatabaseVersion)); { auto mirroredObj = createCommandAndGetMirrored(kCollection, distinctArgs); ASSERT_TRUE(mirroredObj.hasField("shardVersion")); - ASSERT_TRUE(mirroredObj.hasField("databaseVersion")); ASSERT_EQ(mirroredObj["shardVersion"].Int(), kShardVersion); - ASSERT_EQ(mirroredObj["databaseVersion"].Int(), kDatabaseVersion); } } @@ -538,14 +489,7 @@ public: } std::vector<std::string> getAllowedKeys() const override { - return {"count", - "query", - "skip", - "limit", - "hint", - "collation", - "shardVersion", - "databaseVersion"}; + return {"count", "query", "skip", "limit", "hint", "collation", "shardVersion"}; } }; @@ -556,8 +500,7 @@ TEST_F(CountCommandTest, MirrorableKeys) { BSON("hint" << BSONObj()), BSON("readConcern" << BSONObj()), BSON("collation" << BSONObj()), - BSON("shardVersion" << BSONObj()), - BSON("databaseVersion" << BSONObj())}; + BSON("shardVersion" << BSONObj())}; auto mirroredObj = createCommandAndGetMirrored(kCollection, countArgs); checkFieldNamesAreAllowed(mirroredObj); @@ -568,14 +511,12 @@ TEST_F(CountCommandTest, ValidateMirroredQuery) { << "Delivered"); const auto hint = BSON("status" << 1); constexpr auto limit = 1000; - const auto shardVersion = BSON("v" << 123); - const auto databaseVersion = BSON("v" << 456); + const auto shardVersion = BSONObj(); auto countArgs = {BSON("query" << query), BSON("hint" << hint), BSON("limit" << limit), - BSON("shardVersion" << shardVersion), - BSON("databaseVersion" << databaseVersion)}; + BSON("shardVersion" << shardVersion)}; auto mirroredObj = createCommandAndGetMirrored(kCollection, countArgs); ASSERT_EQ(mirroredObj["count"].String(), kCollection); @@ -585,27 +526,21 @@ TEST_F(CountCommandTest, ValidateMirroredQuery) { ASSERT(compareBSONObjs(mirroredObj["hint"].Obj(), hint)); ASSERT_EQ(mirroredObj["limit"].Int(), limit); ASSERT(compareBSONObjs(mirroredObj["shardVersion"].Obj(), shardVersion)); - ASSERT(compareBSONObjs(mirroredObj["databaseVersion"].Obj(), databaseVersion)); } -TEST_F(CountCommandTest, ValidateShardVersionAndDatabaseVersion) { +TEST_F(CountCommandTest, ValidateShardVersion) { std::vector<BSONObj> countArgs = {BSON("count" << BSONObj())}; { auto mirroredObj = createCommandAndGetMirrored(kCollection, countArgs); ASSERT_FALSE(mirroredObj.hasField("shardVersion")); - ASSERT_FALSE(mirroredObj.hasField("databaseVersion")); } const auto kShardVersion = 123; - const auto kDatabaseVersion = 456; countArgs.push_back(BSON("shardVersion" << kShardVersion)); - countArgs.push_back(BSON("databaseVersion" << kDatabaseVersion)); { auto mirroredObj = createCommandAndGetMirrored(kCollection, countArgs); ASSERT_TRUE(mirroredObj.hasField("shardVersion")); - ASSERT_TRUE(mirroredObj.hasField("databaseVersion")); ASSERT_EQ(mirroredObj["shardVersion"].Int(), kShardVersion); - ASSERT_EQ(mirroredObj["databaseVersion"].Int(), kDatabaseVersion); } } diff --git a/src/mongo/db/commands/compact.cpp b/src/mongo/db/commands/compact.cpp index 26ee8ab5d4b..6aa009ea216 100644 --- a/src/mongo/db/commands/compact.cpp +++ b/src/mongo/db/commands/compact.cpp @@ -95,9 +95,6 @@ public: return false; } - // This command is internal to the storage engine and should not block oplog application. - ShouldNotConflictWithSecondaryBatchApplicationBlock noPBWMBlock(opCtx->lockState()); - StatusWith<int64_t> status = compactCollection(opCtx, nss); uassertStatusOK(status.getStatus()); diff --git a/src/mongo/db/commands/count_cmd.cpp b/src/mongo/db/commands/count_cmd.cpp index 57291c7b782..c548bf3ef1c 100644 --- a/src/mongo/db/commands/count_cmd.cpp +++ b/src/mongo/db/commands/count_cmd.cpp @@ -110,7 +110,7 @@ public: Status::OK()}; } - bool shouldAffectReadOptionCounters() const override { + bool shouldAffectReadConcernCounter() const override { return true; } @@ -182,8 +182,12 @@ public: // An empty PrivilegeVector is acceptable because these privileges are only checked on // getMore and explain will not open a cursor. - return runAggregate( - opCtx, viewAggRequest, viewAggregation.getValue(), PrivilegeVector(), result); + return runAggregate(opCtx, + viewAggRequest.getNamespace(), + viewAggRequest, + viewAggregation.getValue(), + PrivilegeVector(), + result); } const auto& collection = ctx->getCollection(); @@ -232,8 +236,6 @@ public: &hangBeforeCollectionCount, opCtx, "hangBeforeCollectionCount", []() {}, nss); auto request = CountCommandRequest::parse(IDLParserErrorContext("count"), cmdObj); - auto curOp = CurOp::get(opCtx); - curOp->beginQueryPlanningTimer(); if (shouldDoFLERewrite(request)) { processFLECountD(opCtx, nss, &request); } @@ -283,6 +285,7 @@ public: auto exec = std::move(statusWithPlanExecutor.getValue()); // Store the plan summary string in CurOp. + auto curOp = CurOp::get(opCtx); { stdx::lock_guard<Client> lk(*opCtx->getClient()); curOp->setPlanSummary_inlock(exec->getPlanExplainer().getPlanSummary()); @@ -319,7 +322,6 @@ public: keyBob.append("hint", 1); keyBob.append("collation", 1); keyBob.append("shardVersion", 1); - keyBob.append("databaseVersion", 1); return keyBob.obj(); }(); diff --git a/src/mongo/db/commands/cqf/cqf_aggregate.cpp b/src/mongo/db/commands/cqf/cqf_aggregate.cpp index 30cdc8c3846..100da0be582 100644 --- a/src/mongo/db/commands/cqf/cqf_aggregate.cpp +++ b/src/mongo/db/commands/cqf/cqf_aggregate.cpp @@ -29,7 +29,6 @@ #include "mongo/db/commands/cqf/cqf_aggregate.h" -#include "mongo/db/curop.h" #include "mongo/db/exec/sbe/abt/abt_lower.h" #include "mongo/db/pipeline/abt/abt_document_source_visitor.h" #include "mongo/db/pipeline/abt/match_expression_visitor.h" @@ -63,8 +62,7 @@ static opt::unordered_map<std::string, optimizer::IndexDefinition> buildIndexSpe const IndexCatalog& indexCatalog = *collection->getIndexCatalog(); opt::unordered_map<std::string, IndexDefinition> result; - auto indexIterator = - indexCatalog.getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady); + auto indexIterator = indexCatalog.getIndexIterator(opCtx, false /*includeUnfinished*/); while (indexIterator->more()) { const IndexCatalogEntry& catalogEntry = *indexIterator->next(); @@ -257,8 +255,7 @@ static std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> optimizeAndCreateExe } auto yieldPolicy = - std::make_unique<PlanYieldPolicySBE>(opCtx, - PlanYieldPolicy::YieldPolicy::YIELD_AUTO, + std::make_unique<PlanYieldPolicySBE>(PlanYieldPolicy::YieldPolicy::YIELD_AUTO, opCtx->getServiceContext()->getFastClockSource(), internalQueryExecYieldIterations.load(), Milliseconds{internalQueryExecYieldPeriodMS.load()}, @@ -338,9 +335,6 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> getSBEExecutorViaCascadesOp const std::string collNameStr = nss.coll().toString(); const std::string scanDefName = collNameStr + "_" + uuidStr; - auto curOp = CurOp::get(opCtx); - curOp->debug().cqfUsed = true; - QueryHints queryHints = getHintsFromQueryKnobs(); PrefixId prefixId; diff --git a/src/mongo/db/commands/create_indexes.cpp b/src/mongo/db/commands/create_indexes.cpp index f6684487a21..c743a405714 100644 --- a/src/mongo/db/commands/create_indexes.cpp +++ b/src/mongo/db/commands/create_indexes.cpp @@ -50,7 +50,7 @@ #include "mongo/db/catalog/uncommitted_catalog_updates.h" #include "mongo/db/commands.h" #include "mongo/db/concurrency/d_concurrency.h" -#include "mongo/db/concurrency/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/create_indexes_gen.h" #include "mongo/db/db_raii.h" #include "mongo/db/index/index_descriptor.h" @@ -321,12 +321,11 @@ bool indexesAlreadyExist(OperationContext* opCtx, * Checks database sharding state. Throws exception on error. */ void checkDatabaseShardingState(OperationContext* opCtx, const NamespaceString& ns) { - Lock::CollectionLock collLock(opCtx, ns, MODE_IS); - auto dss = DatabaseShardingState::get(opCtx, ns.db()); auto dssLock = DatabaseShardingState::DSSLock::lockShared(opCtx, dss); dss->checkDbVersion(opCtx, dssLock); + Lock::CollectionLock collLock(opCtx, ns, MODE_IS); try { const auto collDesc = CollectionShardingState::get(opCtx, ns)->getCollectionDescription(opCtx); @@ -448,7 +447,7 @@ CreateIndexesReply runCreateIndexesOnNewCollection( } bool isCreatingInternalConfigTxnsPartialIndex(const CreateIndexesCommand& cmd) { - if (cmd.getIndexes().size() != 1) { + if (cmd.getIndexes().size() > 1) { return false; } const auto& index = cmd.getIndexes()[0]; diff --git a/src/mongo/db/commands/current_op.cpp b/src/mongo/db/commands/current_op.cpp index b4a16bc40cb..78fec805202 100644 --- a/src/mongo/db/commands/current_op.cpp +++ b/src/mongo/db/commands/current_op.cpp @@ -77,7 +77,12 @@ public: privileges = {Privilege(ResourcePattern::forClusterResource(), ActionType::inprog)}; } - auto status = runAggregate(opCtx, request, std::move(aggCmdObj), privileges, &replyBuilder); + auto status = runAggregate(opCtx, + request.getNamespace(), + request, + std::move(aggCmdObj), + privileges, + &replyBuilder); if (!status.isOK()) { return status; diff --git a/src/mongo/db/commands/dbcheck.cpp b/src/mongo/db/commands/dbcheck.cpp index 5d4579691dd..c26b09d8831 100644 --- a/src/mongo/db/commands/dbcheck.cpp +++ b/src/mongo/db/commands/dbcheck.cpp @@ -36,10 +36,10 @@ #include "mongo/db/catalog/collection_catalog.h" #include "mongo/db/catalog/collection_catalog_helper.h" #include "mongo/db/catalog/database.h" -#include "mongo/db/catalog/health_log_interface.h" +#include "mongo/db/catalog/health_log.h" #include "mongo/db/commands.h" #include "mongo/db/commands/test_commands_enabled.h" -#include "mongo/db/concurrency/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/db_raii.h" #include "mongo/db/jsobj.h" #include "mongo/db/operation_context.h" @@ -52,10 +52,6 @@ #include "mongo/logv2/log.h" -MONGO_FAIL_POINT_DEFINE(sleepAfterExtraIndexKeysHashing); -MONGO_FAIL_POINT_DEFINE(hangBeforeProcessingDbCheckRun); -MONGO_FAIL_POINT_DEFINE(hangBeforeAddingDBCheckBatchToOplog); - namespace mongo { namespace { @@ -97,7 +93,7 @@ public: OplogEntriesEnum::Start, boost::none /*data*/ ); - HealthLogInterface::get(_opCtx->getServiceContext())->log(*healthLogEntry); + HealthLog::get(_opCtx->getServiceContext()).log(*healthLogEntry); DbCheckOplogStartStop oplogEntry; const auto nss = NamespaceString("admin.$cmd"); @@ -123,7 +119,7 @@ public: OplogEntriesEnum::Stop, boost::none /*data*/ ); - HealthLogInterface::get(_opCtx->getServiceContext())->log(*healthLogEntry); + HealthLog::get(_opCtx->getServiceContext()).log(*healthLogEntry); } catch (const DBException&) { LOGV2(6202201, "Could not log stop event"); } @@ -290,18 +286,13 @@ protected: DbCheckStartAndStopLogger startStop(opCtx); - if (MONGO_unlikely(hangBeforeProcessingDbCheckRun.shouldFail())) { - LOGV2(7949000, "Hanging dbcheck due to failpoint 'hangBeforeProcessingDbCheckRun'"); - hangBeforeProcessingDbCheckRun.pauseWhileSet(); - } - for (const auto& coll : *_run) { try { _doCollection(opCtx, coll); } catch (const DBException& e) { auto logEntry = dbCheckErrorHealthLogEntry( coll.nss, "dbCheck failed", OplogEntriesEnum::Batch, e.toStatus()); - HealthLogInterface::get(Client::getCurrent()->getServiceContext())->log(*logEntry); + HealthLog::get(Client::getCurrent()->getServiceContext()).log(*logEntry); return; } @@ -332,7 +323,7 @@ private: "abandoning dbCheck batch because collection no longer exists", OplogEntriesEnum::Batch, Status(ErrorCodes::NamespaceNotFound, "collection not found")); - HealthLogInterface::get(Client::getCurrent()->getServiceContext())->log(*entry); + HealthLog::get(Client::getCurrent()->getServiceContext()).log(*entry); return; } } @@ -410,7 +401,7 @@ private: OplogEntriesEnum::Batch, result.getStatus()); } - HealthLogInterface::get(opCtx)->log(*entry); + HealthLog::get(opCtx).log(*entry); if (retryable) { continue; } @@ -433,28 +424,17 @@ private: (_batchesProcessed % gDbCheckHealthLogEveryNBatches.load() == 0)) { // On debug builds, health-log every batch result; on release builds, health-log // every N batches. - HealthLogInterface::get(opCtx)->log(*entry); - } - - if (MONGO_unlikely(sleepAfterExtraIndexKeysHashing.shouldFail())) { - LOGV2_DEBUG( - 3083201, - 3, - "Sleeping for 1 second due to sleepAfterExtraIndexKeysHashing failpoint"); - opCtx->sleepFor(Milliseconds(1000)); + HealthLog::get(opCtx).log(*entry); } WriteConcernResult unused; auto status = waitForWriteConcern(opCtx, stats.time, info.writeConcern, &unused); if (!status.isOK()) { - // TODO SERVER-89817: Add context with batch ID and lastKey once those are - // backported. - auto entry = dbCheckErrorHealthLogEntry(info.nss, - "dbCheck failed waiting for writeConcern", - OplogEntriesEnum::Batch, - status); - HealthLogInterface::get(opCtx)->log(*entry); - return; + auto entry = dbCheckWarningHealthLogEntry(info.nss, + "dbCheck failed waiting for writeConcern", + OplogEntriesEnum::Batch, + status); + HealthLog::get(opCtx).log(*entry); } start = stats.lastKey; @@ -585,12 +565,6 @@ private: batch.setMaxKey(BSONKey(hasher->lastKey())); batch.setReadTimestamp(readTimestamp); - if (MONGO_unlikely(hangBeforeAddingDBCheckBatchToOplog.shouldFail())) { - LOGV2(8589000, - "Hanging dbCheck due to failpoint 'hangBeforeAddingDBCheckBatchToOplog'"); - hangBeforeAddingDBCheckBatchToOplog.pauseWhileSet(); - } - // Send information on this batch over the oplog. result.time = _logOp(opCtx, info.nss, collection->uuid(), batch.toBSON()); result.readTimestamp = readTimestamp; diff --git a/src/mongo/db/commands/dbcommands.cpp b/src/mongo/db/commands/dbcommands.cpp index 54574139484..0e88f993f93 100644 --- a/src/mongo/db/commands/dbcommands.cpp +++ b/src/mongo/db/commands/dbcommands.cpp @@ -58,6 +58,7 @@ #include "mongo/db/commands.h" #include "mongo/db/commands/feature_compatibility_version.h" #include "mongo/db/commands/server_status.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/curop_failpoint_helpers.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" @@ -168,6 +169,41 @@ public: }; } cmdDropDatabase; +static const char* repairRemovedMessage = + "This command has been removed. If you would like to compact your data, use the 'compact' " + "command. If you would like to rebuild indexes, use the 'reIndex' command. If you need to " + "recover data, please see the documentation for repairing your database offline: " + "http://dochub.mongodb.org/core/repair"; + +class CmdRepairDatabase : public ErrmsgCommandDeprecated { +public: + AllowedOnSecondary secondaryAllowed(ServiceContext*) const override { + return AllowedOnSecondary::kAlways; + } + virtual bool maintenanceMode() const { + return false; + } + + std::string help() const override { + return repairRemovedMessage; + } + virtual bool supportsWriteConcern(const BSONObj& cmd) const override { + return false; + } + + CmdRepairDatabase() : ErrmsgCommandDeprecated("repairDatabase") {} + + bool errmsgRun(OperationContext* opCtx, + const std::string& dbname, + const BSONObj& cmdObj, + std::string& errmsg, + BSONObjBuilder& result) { + + uasserted(ErrorCodes::CommandNotFound, repairRemovedMessage); + return false; + } +} cmdRepairDatabase; + /* drop collection */ class CmdDrop : public DropCmdVersion1Gen<CmdDrop> { public: @@ -294,19 +330,13 @@ public: bool estimate = jsobj["estimate"].trueValue(); const NamespaceString nss(ns); - AutoGetCollectionForReadCommand autoColl(opCtx, nss); - const auto& collection = autoColl.getCollection(); + AutoGetCollectionForReadCommand collection(opCtx, nss); - if (!collection) { - // Collection does not exist - result.appendNumber("size", 0); - result.appendNumber("numObjects", 0); - result.append("millis", timer.millis()); - return true; - } + const auto collDesc = + CollectionShardingState::get(opCtx, nss)->getCollectionDescription(opCtx); - if (collection.isSharded()) { - const ShardKeyPattern shardKeyPattern(collection.getShardKeyPattern()); + if (collDesc.isSharded()) { + const ShardKeyPattern shardKeyPattern(collDesc.getKeyPattern()); uassert(ErrorCodes::BadValue, "keyPattern must be empty or must be an object that equals the shard key", keyPattern.isEmpty() || @@ -324,7 +354,10 @@ public: max = shardKeyPattern.normalizeShardKey(max); } - const long long numRecords = collection->numRecords(opCtx); + long long numRecords = 0; + if (collection) { + numRecords = collection->numRecords(opCtx); + } if (numRecords == 0) { result.appendNumber("size", 0); @@ -344,7 +377,7 @@ public: return 1; } exec = InternalPlanner::collectionScan( - opCtx, &collection, PlanYieldPolicy::YieldPolicy::YIELD_AUTO); + opCtx, &collection.getCollection(), PlanYieldPolicy::YieldPolicy::INTERRUPT_ONLY); } else if (min.isEmpty() || max.isEmpty()) { errmsg = "only one of min or max specified"; return false; @@ -355,7 +388,7 @@ public: } auto shardKeyIdx = findShardKeyPrefixedIndex(opCtx, - collection, + *collection, collection->getIndexCatalog(), keyPattern, /*requireSingleKey=*/true); @@ -370,12 +403,12 @@ public: max = Helpers::toKeyFormat(kp.extendRangeBound(max, false)); exec = InternalPlanner::shardKeyIndexScan(opCtx, - &collection, + &collection.getCollection(), *shardKeyIdx, min, max, BoundInclusion::kIncludeStartKeyOnly, - PlanYieldPolicy::YieldPolicy::YIELD_AUTO); + PlanYieldPolicy::YieldPolicy::INTERRUPT_ONLY); } CurOpFailpointHelpers::waitWhileFailPointEnabled( diff --git a/src/mongo/db/commands/dbcommands_d.cpp b/src/mongo/db/commands/dbcommands_d.cpp index 339d442bcaf..41969950a43 100644 --- a/src/mongo/db/commands/dbcommands_d.cpp +++ b/src/mongo/db/commands/dbcommands_d.cpp @@ -57,8 +57,7 @@ #include "mongo/db/commands/profile_common.h" #include "mongo/db/commands/profile_gen.h" #include "mongo/db/commands/server_status.h" -#include "mongo/db/commands/set_profiling_filter_globally_cmd.h" -#include "mongo/db/concurrency/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/curop_failpoint_helpers.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" @@ -205,8 +204,6 @@ protected: } cmdProfile; -SetProfilingFilterGloballyCmd cmdSetProfilingFilterGlobally; - class CmdFileMD5 : public BasicCommand { public: CmdFileMD5() : BasicCommand("filemd5") {} @@ -382,7 +379,7 @@ public: try { // RELOCKED ctx.reset(new AutoGetCollectionForReadCommand(opCtx, nss)); - } catch (const ExceptionFor<ErrorCodes::StaleConfig>&) { + } catch (const StaleConfigException&) { LOGV2_DEBUG( 20453, 1, diff --git a/src/mongo/db/commands/distinct.cpp b/src/mongo/db/commands/distinct.cpp index ef08e46aee7..57585a88401 100644 --- a/src/mongo/db/commands/distinct.cpp +++ b/src/mongo/db/commands/distinct.cpp @@ -100,7 +100,7 @@ public: return ReadConcernSupportResult::allSupportedAndDefaultPermitted(); } - bool shouldAffectReadOptionCounters() const override { + bool shouldAffectReadConcernCounter() const override { return true; } @@ -172,7 +172,7 @@ public: // An empty PrivilegeVector is acceptable because these privileges are only checked on // getMore and explain will not open a cursor. return runAggregate( - opCtx, viewAggRequest, viewAggregation.getValue(), PrivilegeVector(), result); + opCtx, nss, viewAggRequest, viewAggregation.getValue(), PrivilegeVector(), result); } const auto& collection = ctx->getCollection(); @@ -296,6 +296,8 @@ public: auto&& [stats, _] = explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats); LOGV2_WARNING(23797, + "Plan executor error during distinct command: {error}, " + "stats: {stats}, cmd: {cmd}", "Plan executor error during distinct command", "error"_attr = exception.toStatus(), "stats"_attr = redact(stats), @@ -346,7 +348,6 @@ public: keyBob.append("query", 1); keyBob.append("collation", 1); keyBob.append("shardVersion", 1); - keyBob.append("databaseVersion", 1); return keyBob.obj(); }(); diff --git a/src/mongo/db/commands/drop_indexes.cpp b/src/mongo/db/commands/drop_indexes.cpp index a2c773af89d..f4595688706 100644 --- a/src/mongo/db/commands/drop_indexes.cpp +++ b/src/mongo/db/commands/drop_indexes.cpp @@ -44,7 +44,7 @@ #include "mongo/db/catalog/multi_index_block.h" #include "mongo/db/client.h" #include "mongo/db/commands.h" -#include "mongo/db/concurrency/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/curop.h" #include "mongo/db/db_raii.h" #include "mongo/db/drop_indexes_gen.h" @@ -236,8 +236,8 @@ public: collection.getWritableCollection()->getIndexCatalog()->dropAllIndexes( opCtx, collection.getWritableCollection(), true, {}); - swIndexesToRebuild = indexer->init( - opCtx, collection, all, MultiIndexBlock::kNoopOnInitFn, /*forRecovery=*/false); + swIndexesToRebuild = + indexer->init(opCtx, collection, all, MultiIndexBlock::kNoopOnInitFn); uassertStatusOK(swIndexesToRebuild.getStatus()); wunit.commit(); }); diff --git a/src/mongo/db/commands/find_and_modify.cpp b/src/mongo/db/commands/find_and_modify.cpp index d94346896da..abbc0d834fd 100644 --- a/src/mongo/db/commands/find_and_modify.cpp +++ b/src/mongo/db/commands/find_and_modify.cpp @@ -41,7 +41,7 @@ #include "mongo/db/client.h" #include "mongo/db/commands.h" #include "mongo/db/commands/update_metrics.h" -#include "mongo/db/concurrency/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/curop_failpoint_helpers.h" #include "mongo/db/db_raii.h" #include "mongo/db/exec/update_stage.h" @@ -95,17 +95,16 @@ boost::optional<BSONObj> advanceExecutor(OperationContext* opCtx, PlanExecutor::ExecState state; try { state = exec->getNext(&value, nullptr); - } catch (const WriteConflictException&) { - // Propagate the WCE to be retried at a higher-level without logging. - throw; } catch (DBException& exception) { auto&& explainer = exec->getPlanExplainer(); auto&& [stats, _] = explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats); - LOGV2_WARNING(23802, - "Plan executor error during findAndModify", - "error"_attr = exception.toStatus(), - "stats"_attr = redact(stats), - "cmd"_attr = request.toBSON(BSONObj() /* commandPassthroughFields */)); + LOGV2_WARNING( + 23802, + "Plan executor error during findAndModify: {error}, stats: {stats}, cmd: {cmd}", + "Plan executor error during findAndModify", + "error"_attr = exception.toStatus(), + "stats"_attr = redact(stats), + "cmd"_attr = request.toBSON(BSONObj() /* commandPassthroughFields */)); exception.addContext("Plan executor error during findAndModify"); throw; @@ -170,8 +169,9 @@ void makeUpdateRequest(OperationContext* opCtx, requestOut->setMulti(false); requestOut->setExplain(explain); - requestOut->setYieldPolicy(PlanYieldPolicy::YieldPolicy::YIELD_AUTO); - requestOut->setBypassEmptyTsReplacement(request.getBypassEmptyTsReplacement()); + requestOut->setYieldPolicy(opCtx->inMultiDocumentTransaction() + ? PlanYieldPolicy::YieldPolicy::INTERRUPT_ONLY + : PlanYieldPolicy::YieldPolicy::YIELD_AUTO); } void makeDeleteRequest(OperationContext* opCtx, @@ -190,7 +190,9 @@ void makeDeleteRequest(OperationContext* opCtx, requestOut->setReturnDeleted(true); // Always return the old value. requestOut->setIsExplain(explain); - requestOut->setYieldPolicy(PlanYieldPolicy::YieldPolicy::YIELD_AUTO); + requestOut->setYieldPolicy(opCtx->inMultiDocumentTransaction() + ? PlanYieldPolicy::YieldPolicy::INTERRUPT_ONLY + : PlanYieldPolicy::YieldPolicy::YIELD_AUTO); } write_ops::FindAndModifyCommandReply buildResponse(const PlanExecutor* exec, @@ -761,9 +763,6 @@ void CmdFindAndModify::Invocation::appendMirrorableRequest(BSONObjBuilder* bob) if (const auto& shardVersion = rawCmd.getField("shardVersion"); !shardVersion.eoo()) { bob->append(shardVersion); } - if (const auto& databaseVersion = rawCmd.getField("databaseVersion"); !databaseVersion.eoo()) { - bob->append(databaseVersion); - } // Prevent the find from returning multiple documents since we can bob->append("batchSize", 1); diff --git a/src/mongo/db/commands/find_cmd.cpp b/src/mongo/db/commands/find_cmd.cpp index df96e09e2ce..f7ac781404f 100644 --- a/src/mongo/db/commands/find_cmd.cpp +++ b/src/mongo/db/commands/find_cmd.cpp @@ -36,7 +36,6 @@ #include "mongo/db/catalog/collection_uuid_mismatch.h" #include "mongo/db/client.h" #include "mongo/db/clientcursor.h" -#include "mongo/db/collection_type.h" #include "mongo/db/commands.h" #include "mongo/db/commands/run_aggregate.h" #include "mongo/db/commands/test_commands_enabled.h" @@ -55,10 +54,6 @@ #include "mongo/db/query/find_common.h" #include "mongo/db/query/get_executor.h" #include "mongo/db/query/query_knobs_gen.h" -#include "mongo/db/query/query_shape/query_shape.h" -#include "mongo/db/query/query_stats/find_key.h" -#include "mongo/db/query/query_stats/key.h" -#include "mongo/db/query/query_stats/query_stats.h" #include "mongo/db/repl/replication_coordinator.h" #include "mongo/db/service_context.h" #include "mongo/db/stats/counters.h" @@ -116,28 +111,66 @@ std::unique_ptr<FindCommandRequest> translateNtoReturnToLimitOrBatchSize( return findCmd; } +// Parses the command object to a FindCommandRequest. If the client request did not specify any +// runtime constants, make them available to the query here. +std::unique_ptr<FindCommandRequest> parseCmdObjectToFindCommandRequest(OperationContext* opCtx, + NamespaceString nss, + BSONObj cmdObj) { + auto findCommand = query_request_helper::makeFromFindCommand( + std::move(cmdObj), + std::move(nss), + APIParameters::get(opCtx).getAPIStrict().value_or(false)); + + // Rewrite any FLE find payloads that exist in the query if this is a FLE 2 query. + if (shouldDoFLERewrite(findCommand)) { + invariant(findCommand->getNamespaceOrUUID().nss()); + processFLEFindD(opCtx, findCommand->getNamespaceOrUUID().nss().get(), findCommand.get()); + } + + return translateNtoReturnToLimitOrBatchSize(std::move(findCommand)); +} + boost::intrusive_ptr<ExpressionContext> makeExpressionContext( OperationContext* opCtx, const FindCommandRequest& findCommand, - const CollectionPtr& collPtr, boost::optional<ExplainOptions::Verbosity> verbosity) { std::unique_ptr<CollatorInterface> collator; if (!findCommand.getCollation().isEmpty()) { collator = uassertStatusOK(CollatorFactoryInterface::get(opCtx->getServiceContext()) ->makeFromBSON(findCommand.getCollation())); - } else if (collPtr && collPtr->getDefaultCollator()) { - // The 'collPtr' will be null for views, but we don't need to worry about views here. The - // views will get rewritten into aggregate command and will regenerate the - // ExpressionContext. - collator = collPtr->getDefaultCollator()->clone(); } - auto expCtx = - make_intrusive<ExpressionContext>(opCtx, - findCommand, - std::move(collator), - CurOp::get(opCtx)->dbProfileLevel() > 0, // mayDbProfile - verbosity, - allowDiskUseByDefault.load()); + + // Although both 'find' and 'aggregate' commands have an ExpressionContext, some of the data + // members in the ExpressionContext are used exclusively by the aggregation subsystem. This + // includes the following fields which here we simply initialize to some meaningless default + // value: + // - explain + // - fromMongos + // - needsMerge + // - bypassDocumentValidation + // - mongoProcessInterface + // - resolvedNamespaces + // - uuid + // + // As we change the code to make the find and agg systems more tightly coupled, it would make + // sense to start initializing these fields for find operations as well. + auto expCtx = make_intrusive<ExpressionContext>( + opCtx, + verbosity, + false, // fromMongos + false, // needsMerge + findCommand.getAllowDiskUse().value_or(allowDiskUseByDefault.load()), + false, // bypassDocumentValidation + false, // isMapReduceCommand + findCommand.getNamespaceOrUUID().nss().value_or(NamespaceString()), + findCommand.getLegacyRuntimeConstants(), + std::move(collator), + nullptr, // mongoProcessInterface + StringMap<ExpressionContext::ResolvedNamespace>{}, + boost::none, // uuid + findCommand.getLet(), // let + CurOp::get(opCtx)->dbProfileLevel() > 0 // mayDbProfile + ); if (storageGlobalParams.readOnly) { // Disallow disk use if in read-only mode. expCtx->allowDiskUse = false; @@ -159,45 +192,6 @@ void beginQueryOp(OperationContext* opCtx, const NamespaceString& nss, const BSO } /** - * Parses the grammar elements like 'filter', 'sort', and 'projection' from the raw - * 'FindCommandRequest', and tracks internal state like begining the operation's timer and recording - * query shape stats (if enabled). - */ -std::unique_ptr<CanonicalQuery> parseQueryAndBeginOperation( - OperationContext* opCtx, - const AutoGetCollectionForReadCommandMaybeLockFree& ctx, - const NamespaceString& nss, - BSONObj requestBody, - std::unique_ptr<FindCommandRequest> findCommand, - const CollectionPtr& collection) { - // Fill out curop information. - beginQueryOp(opCtx, nss, requestBody); - // Finish the parsing step by using the FindCommandRequest to create a CanonicalQuery. - const ExtensionsCallbackReal extensionsCallback(opCtx, &nss); - - auto expCtx = - makeExpressionContext(opCtx, *findCommand, collection, boost::none /* verbosity */); - - auto parsedRequest = uassertStatusOK( - parsed_find_command::parse(expCtx, - std::move(findCommand), - extensionsCallback, - MatchExpressionParser::kAllowAllSpecialFeatures)); - - // Register query stats collection. Exclude queries against collections with encrypted fields. - // It is important to do this before canonicalizing and optimizing the query, each of which - // would alter the query shape. - if (!(collection && collection.get()->getCollectionOptions().encryptedFieldConfig)) { - query_stats::registerRequest(opCtx, nss, [&]() { - return std::make_unique<query_stats::FindKey>( - expCtx, *parsedRequest, ctx.getCollectionType()); - }); - } - - return uassertStatusOK( - CanonicalQuery::canonicalize(std::move(expCtx), std::move(parsedRequest))); -} -/** * A command for running .find() queries. */ class FindCmd final : public Command { @@ -250,7 +244,7 @@ public: return false; } - bool shouldAffectReadOptionCounters() const override { + bool shouldAffectReadConcernCounter() const override { return true; } @@ -322,20 +316,11 @@ public: const auto nss = ctx->getNss(); // Parse the command BSON to a FindCommandRequest. - auto findCommand = _parseCmdObjectToFindCommandRequest(opCtx, nss, _request.body); + auto findCommand = parseCmdObjectToFindCommandRequest(opCtx, nss, _request.body); // Finish the parsing step by using the FindCommandRequest to create a CanonicalQuery. const ExtensionsCallbackReal extensionsCallback(opCtx, &nss); - - // The collection may be NULL. If so, getExecutor() should handle it by returning an - // execution tree with an EOFStage. - const auto& collection = ctx->getCollection(); - if (!ctx->getView()) { - const bool isClusteredCollection = collection && collection->isClustered(); - uassertStatusOK(query_request_helper::validateResumeAfter( - findCommand->getResumeAfter(), isClusteredCollection)); - } - auto expCtx = makeExpressionContext(opCtx, *findCommand, collection, verbosity); + auto expCtx = makeExpressionContext(opCtx, *findCommand, verbosity); const bool isExplain = true; auto cq = uassertStatusOK( CanonicalQuery::canonicalize(opCtx, @@ -372,8 +357,8 @@ public: try { // An empty PrivilegeVector is acceptable because these privileges are only // checked on getMore and explain will not open a cursor. - uassertStatusOK( - runAggregate(opCtx, aggRequest, viewAggCmd, PrivilegeVector(), result)); + uassertStatusOK(runAggregate( + opCtx, nss, aggRequest, viewAggCmd, PrivilegeVector(), result)); } catch (DBException& error) { if (error.code() == ErrorCodes::InvalidPipelineOperator) { uasserted(ErrorCodes::InvalidPipelineOperator, @@ -385,6 +370,10 @@ public: return; } + // The collection may be NULL. If so, getExecutor() should handle it by returning an + // execution tree with an EOFStage. + const auto& collection = ctx->getCollection(); + // Get the execution plan for the query. bool permitYield = true; auto exec = @@ -419,10 +408,10 @@ public: // Parse the command BSON to a FindCommandRequest. Pass in the parsedNss in case cmdObj // does not have a UUID. auto parsedNss = NamespaceString{CommandHelpers::parseNsFromCommand(_dbName, cmdObj)}; + const bool isExplain = false; const bool isOplogNss = (parsedNss == NamespaceString::kRsOplogNamespace); auto findCommand = - _parseCmdObjectToFindCommandRequest(opCtx, std::move(parsedNss), cmdObj); - CurOp::get(opCtx)->beginQueryPlanningTimer(); + parseCmdObjectToFindCommandRequest(opCtx, std::move(parsedNss), cmdObj); // Only allow speculative majority for internal commands that specify the correct flag. uassert(ErrorCodes::ReadConcernMajorityNotEnabled, @@ -516,16 +505,14 @@ public: } // Tailing a replicated capped clustered collection requires majority read concern. - const auto& collection = ctx->getCollection(); - - bool isClusteredCollection = false; - if (collection) { + const auto coll = ctx->getCollection().get(); + if (coll) { const bool isTailable = findCommand->getTailable(); const bool isMajorityReadConcern = repl::ReadConcernArgs::get(opCtx).getLevel() == repl::ReadConcernLevel::kMajorityReadConcern; - isClusteredCollection = collection->isClustered(); - const bool isCapped = collection->isCapped(); - const bool isReplicated = collection->ns().isReplicated(); + const bool isClusteredCollection = coll->isClustered(); + const bool isCapped = coll->isCapped(); + const bool isReplicated = coll->ns().isReplicated(); if (isClusteredCollection && isCapped && isReplicated && isTailable) { uassert(ErrorCodes::Error(6049203), "A tailable cursor on a capped clustered collection requires majority " @@ -534,16 +521,19 @@ public: } } - // Views use the aggregation system and the $_resumeAfter parameter is not allowed. A - // more descriptive error will be raised later, but we want to validate this parameter - // before beginning the operation. - if (!ctx->getView()) { - uassertStatusOK(query_request_helper::validateResumeAfter( - findCommand->getResumeAfter(), isClusteredCollection)); - } + // Fill out curop information. + beginQueryOp(opCtx, nss, _request.body); - auto cq = parseQueryAndBeginOperation( - opCtx, *ctx, nss, _request.body, std::move(findCommand), collection); + // Finish the parsing step by using the FindCommandRequest to create a CanonicalQuery. + const ExtensionsCallbackReal extensionsCallback(opCtx, &nss); + auto expCtx = makeExpressionContext(opCtx, *findCommand, boost::none /* verbosity */); + auto cq = uassertStatusOK( + CanonicalQuery::canonicalize(opCtx, + std::move(findCommand), + isExplain, + std::move(expCtx), + extensionsCallback, + MatchExpressionParser::kAllowAllSpecialFeatures)); // If we are running a query against a view, or if we are trying to test the new // optimizer, redirect this query through the aggregation system. @@ -560,9 +550,6 @@ public: auto viewAggregationCommand = uassertStatusOK(query_request_helper::asAggregationCommand(findCommand)); - // This doesn't directly call 'runAggregate()' so it doesn't need to adapt to the - // new API on v6.0. @Alyssa this suggests we should look into view performance more - // carefully on v6.0. The perf of this code path may have different characteristics? BSONObj aggResult = CommandHelpers::runCommandDirectly( opCtx, OpMsgRequest::fromDBAndBody(_dbName, std::move(viewAggregationCommand))); auto status = getStatusFromCommandResult(aggResult); @@ -583,6 +570,8 @@ public: uassertStatusOK(replCoord->checkCanServeReadsFor( opCtx, nss, ReadPreferenceSetting::get(opCtx).canRunOnSecondary())); + const auto& collection = ctx->getCollection(); + if (cq->getFindCommandRequest().getReadOnce()) { // The readOnce option causes any storage-layer cursors created during plan // execution to assume read data will not be needed again and need not be cached. @@ -616,7 +605,7 @@ public: // there is no ClientCursor id, and then return. const long long numResults = 0; const CursorId cursorId = 0; - endQueryOp(opCtx, collection, *exec, numResults, boost::none, cmdObj); + endQueryOp(opCtx, collection, *exec, numResults, cursorId); auto bodyBuilder = result->getBodyBuilder(); appendCursorResponseObject( cursorId, nss.ns(), BSONArray(), boost::none, &bodyBuilder); @@ -667,6 +656,8 @@ public: auto&& [stats, _] = explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats); LOGV2_WARNING(23798, + "Plan executor error during find command: {error}, " + "stats: {stats}, cmd: {cmd}", "Plan executor error during find command", "error"_attr = exception.toStatus(), "stats"_attr = redact(stats), @@ -715,24 +706,26 @@ public: pinnedCursor.getCursor()->setLeftoverMaxTimeMicros( opCtx->getRemainingMaxTimeMicros()); } + pinnedCursor.getCursor()->setNReturnedSoFar(numResults); + pinnedCursor.getCursor()->incNBatches(); // Fill out curop based on the results. - endQueryOp(opCtx, collection, *cursorExec, numResults, pinnedCursor, cmdObj); + endQueryOp(opCtx, collection, *cursorExec, numResults, cursorId); if (stashResourcesForGetMore) { // Collect storage stats now before we stash the recovery unit. These stats are // normally collected in the service entry point layer just before a command - // ends, but they must be collected before stashing the RecoveryUnit. Otherwise, - // the service entry point layer will collect the stats from the new - // RecoveryUnit, which wasn't actually used for the query. + // ends, but they must be collected before stashing the + // RecoveryUnit. Otherwise, the service entry point layer will collect the + // stats from the new RecoveryUnit, which wasn't actually used for the query. // // The stats collected here will not get overwritten, as the service entry // point layer will only set these stats when they're not empty. CurOp::get(opCtx)->debug().storageStats = - opCtx->recoveryUnit()->computeOperationStatisticsSinceLastCall(); + opCtx->recoveryUnit()->getOperationStatistics(); } } else { - endQueryOp(opCtx, collection, *exec, numResults, boost::none, cmdObj); + endQueryOp(opCtx, collection, *exec, numResults, cursorId); } // Generate the response object to send to the client. @@ -759,7 +752,6 @@ public: keyBob.append("min", 1); keyBob.append("max", 1); keyBob.append("shardVersion", 1); - keyBob.append("databaseVersion", 1); return keyBob.obj(); }(); @@ -773,25 +765,6 @@ public: private: const OpMsgRequest _request; const StringData _dbName; - - // Parses the command object to a FindCommandRequest. If the client request did not specify - // any runtime constants, make them available to the query here. - std::unique_ptr<FindCommandRequest> _parseCmdObjectToFindCommandRequest( - OperationContext* opCtx, NamespaceString nss, BSONObj cmdObj) { - auto findCommand = query_request_helper::makeFromFindCommand( - std::move(cmdObj), - std::move(nss), - APIParameters::get(opCtx).getAPIStrict().value_or(false)); - - // Rewrite any FLE find payloads that exist in the query if this is a FLE 2 query. - if (shouldDoFLERewrite(findCommand)) { - invariant(findCommand->getNamespaceOrUUID().nss()); - processFLEFindD( - opCtx, findCommand->getNamespaceOrUUID().nss().value(), findCommand.get()); - } - - return translateNtoReturnToLimitOrBatchSize(std::move(findCommand)); - } }; } findCmd; diff --git a/src/mongo/db/commands/fsync.cpp b/src/mongo/db/commands/fsync.cpp index 35ae6323188..f1d92aa636a 100644 --- a/src/mongo/db/commands/fsync.cpp +++ b/src/mongo/db/commands/fsync.cpp @@ -47,13 +47,11 @@ #include "mongo/db/commands.h" #include "mongo/db/commands/fsync_locked.h" #include "mongo/db/concurrency/d_concurrency.h" -#include "mongo/db/concurrency/exception_util.h" -#include "mongo/db/dbdirectclient.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/service_context.h" #include "mongo/db/storage/backup_cursor_hooks.h" #include "mongo/db/storage/storage_engine.h" #include "mongo/logv2/log.h" -#include "mongo/s/sharding_feature_flags_gen.h" #include "mongo/stdx/condition_variable.h" #include "mongo/util/assert_util.h" #include "mongo/util/background.h" @@ -71,13 +69,10 @@ Lock::ResourceMutex commandMutex("fsyncCommandMutex"); */ class FSyncLockThread : public BackgroundJob { public: - FSyncLockThread(ServiceContext* serviceContext, - bool allowFsyncFailure, - const Milliseconds deadline) + FSyncLockThread(ServiceContext* serviceContext, bool allowFsyncFailure) : BackgroundJob(false), _serviceContext(serviceContext), - _allowFsyncFailure(allowFsyncFailure), - _deadline(deadline) {} + _allowFsyncFailure(allowFsyncFailure) {} std::string name() const override { return "FSyncLockThread"; @@ -89,7 +84,6 @@ private: ServiceContext* const _serviceContext; bool _allowFsyncFailure; static bool _shutdownTaskRegistered; - const Milliseconds _deadline; }; class FSyncCommand : public ErrmsgCommandDeprecated { @@ -131,20 +125,6 @@ public: actions.addAction(ActionType::fsync); out->push_back(Privilege(ResourcePattern::forClusterResource(), actions)); } - - virtual void checkForInProgressDDLOperations(OperationContext* opCtx) { - DBDirectClient client(opCtx); - const auto numDDLDocuments = - client.count(NamespaceString::kShardingDDLCoordinatorsNamespace); - - if (numDDLDocuments != 0) { - LOGV2_WARNING(781541, "Cannot take lock while DDL operations is in progress"); - releaseLock(); - uasserted(ErrorCodes::IllegalOperation, - "Cannot take lock while DDL operation is in progress"); - } - } - virtual bool errmsgRun(OperationContext* opCtx, const std::string& dbname, const BSONObj& cmdObj, @@ -156,12 +136,7 @@ public: } const bool lock = cmdObj["lock"].trueValue(); - const bool forBackup = cmdObj["forBackup"].trueValue(); - LOGV2(20461, - "CMD fsync: lock:{lock}", - "CMD fsync", - "lock"_attr = lock, - "forBackup"_attr = forBackup); + LOGV2(20461, "CMD fsync: lock:{lock}", "CMD fsync", "lock"_attr = lock); // fsync + lock is sometimes used to block writes out of the system and does not care if // the `BackupCursorService::fsyncLock` call succeeds. @@ -193,23 +168,8 @@ public: stdx::unique_lock<Latch> lk(lockStateMutex); threadStatus = Status::OK(); threadStarted = false; - - Milliseconds deadline = Milliseconds::max(); - if (forBackup) { - // Set a default deadline of 90s for the fsyncLock to be acquired. - deadline = Milliseconds(90000); - // Parse the cmdObj and update the deadline if - // "fsyncLockAcquisitionTimeoutMillis" exists. - for (const auto& elem : cmdObj) { - if (elem.fieldNameStringData() == "fsyncLockAcquisitionTimeoutMillis") { - deadline = Milliseconds{uassertStatusOK(parseMaxTimeMS(elem))}; - } - } - } - - _lockThread = std::make_unique<FSyncLockThread>( - opCtx->getServiceContext(), allowFsyncFailure, deadline); - + _lockThread = std::make_unique<FSyncLockThread>(opCtx->getServiceContext(), + allowFsyncFailure); _lockThread->go(); while (!threadStarted && threadStatus.isOK()) { @@ -229,13 +189,6 @@ public: } } - if (forBackup) { - // The check must be performed only if the fsync+lock command has been issued for backup - // purposes (through monogs). There are valid cases where fsync+lock can be invoked on - // the mongod while DDLs are in progress. - checkForInProgressDDLOperations(opCtx); - } - LOGV2(20462, "mongod is locked and no writes are allowed. db.fsyncUnlock() to unlock, " "lock count is {lockCount}, for more info see {seeAlso}", @@ -396,18 +349,7 @@ void FSyncLockThread::run() { try { const ServiceContext::UniqueOperationContext opCtxPtr = cc().makeOperationContext(); OperationContext& opCtx = *opCtxPtr; - - // If the deadline exists, set it on the opCtx and GlobalRead lock. - Date_t lockDeadline = Date_t::max(); - if (_deadline < Milliseconds::max()) { - lockDeadline = Date_t::now() + _deadline; - } - - opCtx.setDeadlineAfterNowBy(Milliseconds(_deadline), ErrorCodes::ExceededTimeLimit); - Lock::GlobalRead global( - &opCtx, - lockDeadline, - Lock::InterruptBehavior::kThrow); // Block any writes in order to flush the files. + Lock::GlobalRead global(&opCtx); // Block any writes in order to flush the files. StorageEngine* storageEngine = _serviceContext->getStorageEngine(); @@ -497,16 +439,7 @@ void FSyncLockThread::run() { storageEngine->endBackup(&opCtx); } } - } catch (const ExceptionForCat<ErrorCategory::ExceededTimeLimitError>&) { - LOGV2_ERROR(204739, "Fsync timed out with ExceededTimeLimitError"); - fsyncCmd.threadStatus = Status(ErrorCodes::Error::LockTimeout, "Fsync lock timed out"); - fsyncCmd.acquireFsyncLockSyncCV.notify_one(); - return; - } catch (const ExceptionFor<ErrorCodes::LockTimeout>&) { - LOGV2_ERROR(204740, "Fsync timed out with LockTimeout"); - fsyncCmd.threadStatus = Status(ErrorCodes::Error::LockTimeout, "Fsync lock timed out"); - fsyncCmd.acquireFsyncLockSyncCV.notify_one(); - return; + } catch (const std::exception& e) { LOGV2_FATAL(40350, "FSyncLockThread exception: {error}", diff --git a/src/mongo/db/commands/generic_servers.cpp b/src/mongo/db/commands/generic_servers.cpp index 7a6b87c13fe..41448e90dde 100644 --- a/src/mongo/db/commands/generic_servers.cpp +++ b/src/mongo/db/commands/generic_servers.cpp @@ -156,11 +156,8 @@ HostInfoReply HostInfoCmd::Invocation::typedRun(OperationContext*) { system.setMemSizeMB(static_cast<long>(p.getSystemMemSizeMB())); system.setMemLimitMB(static_cast<long>(p.getMemSizeMB())); system.setNumCores(static_cast<int>(p.getNumAvailableCores())); - system.setNumPhysicalCores(static_cast<int>(p.getNumPhysicalCores())); - system.setNumCpuSockets(static_cast<int>(p.getNumCpuSockets())); system.setCpuArch(p.getArch()); system.setNumaEnabled(p.hasNumaEnabled()); - system.setNumNumaNodes(static_cast<int>(p.getNumNumaNodes())); HostInfoOsReply os; os.setType(p.getOsType()); diff --git a/src/mongo/db/commands/generic_servers.idl b/src/mongo/db/commands/generic_servers.idl index c1f626a7f07..598ce277333 100644 --- a/src/mongo/db/commands/generic_servers.idl +++ b/src/mongo/db/commands/generic_servers.idl @@ -64,11 +64,8 @@ structs: memSizeMB: long memLimitMB: long numCores: int - numPhysicalCores: int - numCpuSockets: int cpuArch: string numaEnabled: bool - numNumaNodes: int hostInfoOsReply: description: "hostInfo.os reply fields" diff --git a/src/mongo/db/commands/get_cluster_parameter_command.cpp b/src/mongo/db/commands/get_cluster_parameter_command.cpp index 6f87839e88d..35060c90ed4 100644 --- a/src/mongo/db/commands/get_cluster_parameter_command.cpp +++ b/src/mongo/db/commands/get_cluster_parameter_command.cpp @@ -65,11 +65,6 @@ 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/db/commands/getmore_cmd.cpp b/src/mongo/db/commands/getmore_cmd.cpp index 8d7510a7e2f..b3c00996ec6 100644 --- a/src/mongo/db/commands/getmore_cmd.cpp +++ b/src/mongo/db/commands/getmore_cmd.cpp @@ -431,6 +431,7 @@ public: auto&& [stats, _] = explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats); LOGV2_WARNING(20478, + "getMore command executor error: {error}, stats: {stats}, cmd: {cmd}", "getMore command executor error", "error"_attr = exception.toStatus(), "stats"_attr = redact(stats), @@ -616,6 +617,7 @@ public: options.atClusterTime = repl::ReadConcernArgs::get(opCtx).getArgsAtClusterTime(); } CursorResponseBuilder nextBatch(reply, options); + BSONObj obj; std::uint64_t numResults = 0; ResourceConsumption::DocumentUnitCounter docUnitsReturned; @@ -632,7 +634,9 @@ public: // Use the commit point of the last batch for exhaust cursors. lastKnownCommittedOpTime = cursorPin->getLastKnownCommittedOpTime(); } - clientsLastKnownCommittedOpTime(opCtx) = lastKnownCommittedOpTime; + if (lastKnownCommittedOpTime) { + clientsLastKnownCommittedOpTime(opCtx) = lastKnownCommittedOpTime.get(); + } awaitDataState(opCtx).shouldWaitForInserts = true; } @@ -695,19 +699,10 @@ public: cursorPin->setLeftoverMaxTimeMicros(opCtx->getRemainingMaxTimeMicros()); - if (opCtx->isExhaust() && clientsLastKnownCommittedOpTime(opCtx)) { - // Update the cursor's lastKnownCommittedOpTime to the current - // lastCommittedOpTime. The lastCommittedOpTime now may be staler than the - // actual lastCommittedOpTime returned in the metadata of this latest batch (see - // appendReplyMetadata). As a result, we may sometimes return more empty - // batches than we need to. But it is fine to be conservative in this. + if (opCtx->isExhaust() && !clientsLastKnownCommittedOpTime(opCtx).isNull()) { + // Set the commit point of the latest batch. auto replCoord = repl::ReplicationCoordinator::get(opCtx); - auto myLastCommittedOpTime = replCoord->getLastCommittedOpTime(); - auto clientsLastKnownCommittedOpTime = cursorPin->getLastKnownCommittedOpTime(); - if (!clientsLastKnownCommittedOpTime.has_value() || - clientsLastKnownCommittedOpTime.value() < myLastCommittedOpTime) { - cursorPin->setLastKnownCommittedOpTime(myLastCommittedOpTime); - } + cursorPin->setLastKnownCommittedOpTime(replCoord->getLastCommittedOpTime()); } } else { curOp->debug().cursorExhausted = true; @@ -719,9 +714,12 @@ public: // documents. auto& metricsCollector = ResourceConsumption::MetricsCollector::get(opCtx); metricsCollector.incrementDocUnitsReturned(docUnitsReturned); - curOp->debug().additiveMetrics.nBatches = 1; - curOp->setEndOfOpMetrics(numResults); - collectQueryStatsMongod(opCtx, cursorPin); + cursorPin->incNReturnedSoFar(numResults); + cursorPin->incNBatches(); + + // Ensure log and profiler include the number of results returned in this getMore's + // response batch. + curOp->debug().nreturned = numResults; if (respondWithId) { cursorDeleter.dismiss(); diff --git a/src/mongo/db/commands/internal_rename_if_options_and_indexes_match_cmd.cpp b/src/mongo/db/commands/internal_rename_if_options_and_indexes_match_cmd.cpp index ffb4b0c8555..f9648db8105 100644 --- a/src/mongo/db/commands/internal_rename_if_options_and_indexes_match_cmd.cpp +++ b/src/mongo/db/commands/internal_rename_if_options_and_indexes_match_cmd.cpp @@ -112,9 +112,8 @@ public: RenameCollectionOptions options; options.dropTarget = true; options.stayTemp = false; - options.originalIndexes = indexList; - options.originalCollectionOptions = collectionOptions; - doLocalRenameIfOptionsAndIndexesHaveNotChanged(opCtx, fromNss, toNss, options); + doLocalRenameIfOptionsAndIndexesHaveNotChanged( + opCtx, fromNss, toNss, options, std::move(indexList), collectionOptions); } NamespaceString ns() const override { diff --git a/src/mongo/db/commands/killoperations_common.h b/src/mongo/db/commands/killoperations_common.h index bae129a4bbe..b9dd9eb79a9 100644 --- a/src/mongo/db/commands/killoperations_common.h +++ b/src/mongo/db/commands/killoperations_common.h @@ -64,8 +64,7 @@ public: auto opKeys = Base::request().getOperationKeys(); for (auto& opKey : opKeys) { - LOGV2_DEBUG( - 4615602, 2, "Attempting to kill operation", "operationKey"_attr = opKey); + LOGV2(4615602, "Attempting to kill operation", "operationKey"_attr = opKey); opKiller.killOperation(OperationKey(opKey)); } Derived::killCursors(opCtx, opKeys); diff --git a/src/mongo/db/commands/list_collections.cpp b/src/mongo/db/commands/list_collections.cpp index ec45ee6356b..f78cfda6308 100644 --- a/src/mongo/db/commands/list_collections.cpp +++ b/src/mongo/db/commands/list_collections.cpp @@ -167,7 +167,9 @@ BSONObj buildViewBson(const ViewDefinition& view, bool nameOnly) { return b.obj(); } -BSONObj buildTimeseriesBson(const CollectionPtr& collection, bool nameOnly) { +BSONObj buildTimeseriesBson(OperationContext* opCtx, + const CollectionPtr& collection, + bool nameOnly) { invariant(collection); BSONObjBuilder builder; @@ -375,7 +377,8 @@ public: if (auto bucketsCollection = CollectionCatalog::get(opCtx) ->lookupCollectionByNamespace( opCtx, view->viewOn())) { - return buildTimeseriesBson(bucketsCollection, nameOnly); + return buildTimeseriesBson( + opCtx, bucketsCollection, nameOnly); } else { // The buckets collection does not exist, so the time-series // view will be appended when we iterate through the view @@ -394,22 +397,21 @@ public: } else { auto perCollectionWork = [&](const CollectionPtr& collection) { if (collection && collection->getTimeseriesOptions() && - !collection->ns().isDropPendingNamespace()) { - auto viewNss = collection->ns().getTimeseriesViewNamespace(); - auto view = - catalog->lookupViewWithoutValidatingDurable(opCtx, viewNss); - if (view && view->timeseries() && - (!authorizedCollections || - as->isAuthorizedForAnyActionOnResource( - ResourcePattern::forExactNamespace(viewNss)))) { - // The time-series view for this buckets namespace exists, so - // add it here while we have the collection options. - _addWorkingSetMember(opCtx, - buildTimeseriesBson(collection, nameOnly), - matcher.get(), - ws.get(), - root.get()); - } + !collection->ns().isDropPendingNamespace() && + catalog->lookupViewWithoutValidatingDurable( + opCtx, collection->ns().getTimeseriesViewNamespace()) && + (!authorizedCollections || + as->isAuthorizedForAnyActionOnResource( + ResourcePattern::forExactNamespace( + collection->ns().getTimeseriesViewNamespace())))) { + // The time-series view for this buckets namespace exists, so add it + // here while we have the collection options. + _addWorkingSetMember( + opCtx, + buildTimeseriesBson(opCtx, collection, nameOnly), + matcher.get(), + ws.get(), + root.get()); } if (authorizedCollections && @@ -433,8 +435,10 @@ public: // needing to yield as we don't take any locks. if (opCtx->isLockFreeReadsOp()) { auto collectionCatalog = CollectionCatalog::get(opCtx); - for (auto&& coll : collectionCatalog->range(tenantDbName)) { - perCollectionWork(coll); + for (auto it = collectionCatalog->begin(opCtx, tenantDbName); + it != collectionCatalog->end(opCtx); + ++it) { + perCollectionWork(*it); } } else { mongo::catalog::forEachCollectionFromDb( @@ -497,7 +501,7 @@ public: batchSize = *listCollRequest.getCursor()->getBatchSize(); } - FindCommon::BSONArrayResponseSizeTracker responseSizeTracker; + size_t bytesBuffered = 0; for (long long objCount = 0; objCount < batchSize; objCount++) { BSONObj nextDoc; PlanExecutor::ExecState state = exec->getNext(&nextDoc, nullptr); @@ -508,7 +512,7 @@ public: // If we can't fit this result inside the current batch, then we stash it for // later. - if (!responseSizeTracker.haveSpaceForNext(nextDoc)) { + if (!FindCommon::haveSpaceForNext(nextDoc, objCount, bytesBuffered)) { exec->stashResult(nextDoc); break; } @@ -524,7 +528,7 @@ public: "error"_attr = exc); fassertFailed(5254301); } - responseSizeTracker.add(nextDoc); + bytesBuffered += nextDoc.objsize(); } if (exec->isEOF()) { return createListCollectionsCursorReply( diff --git a/src/mongo/db/commands/list_databases.cpp b/src/mongo/db/commands/list_databases.cpp index fd60d2b26b5..4b59a41f64e 100644 --- a/src/mongo/db/commands/list_databases.cpp +++ b/src/mongo/db/commands/list_databases.cpp @@ -35,7 +35,7 @@ #include "mongo/db/client.h" #include "mongo/db/commands.h" #include "mongo/db/commands/list_databases_gen.h" -#include "mongo/db/concurrency/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/curop_failpoint_helpers.h" #include "mongo/db/db_raii.h" #include "mongo/db/matcher/expression.h" diff --git a/src/mongo/db/commands/list_indexes.cpp b/src/mongo/db/commands/list_indexes.cpp index 22cbb4b11df..6f4f4f461b9 100644 --- a/src/mongo/db/commands/list_indexes.cpp +++ b/src/mongo/db/commands/list_indexes.cpp @@ -40,6 +40,7 @@ #include "mongo/db/catalog/list_indexes.h" #include "mongo/db/clientcursor.h" #include "mongo/db/commands.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/curop.h" #include "mongo/db/curop_failpoint_helpers.h" #include "mongo/db/cursor_manager.h" @@ -64,40 +65,38 @@ namespace mongo { namespace { // The allowed fields have to be in sync with those defined in 'src/mongo/db/list_indexes.idl'. -static std::map<StringData, std::set<IndexType>> allowedFieldNames = { - {ListIndexesReplyItem::k2dsphereIndexVersionFieldName, - {IndexType::INDEX_2DSPHERE, IndexType::INDEX_2DSPHERE_BUCKET}}, - {ListIndexesReplyItem::kBackgroundFieldName, {}}, - {ListIndexesReplyItem::kBitsFieldName, {IndexType::INDEX_2D}}, - {ListIndexesReplyItem::kBucketSizeFieldName, {}}, - {ListIndexesReplyItem::kBuildUUIDFieldName, {}}, - {ListIndexesReplyItem::kClusteredFieldName, {}}, - {ListIndexesReplyItem::kCoarsestIndexedLevelFieldName, {IndexType::INDEX_2DSPHERE}}, - {ListIndexesReplyItem::kCollationFieldName, {}}, - {ListIndexesReplyItem::kDefault_languageFieldName, {}}, - {ListIndexesReplyItem::kDropDupsFieldName, {}}, - {ListIndexesReplyItem::kExpireAfterSecondsFieldName, {}}, - {ListIndexesReplyItem::kFinestIndexedLevelFieldName, {IndexType::INDEX_2DSPHERE}}, - {ListIndexesReplyItem::kHiddenFieldName, {}}, - {ListIndexesReplyItem::kIndexBuildInfoFieldName, {}}, - {ListIndexesReplyItem::kKeyFieldName, {}}, - {ListIndexesReplyItem::kLanguage_overrideFieldName, {}}, - {ListIndexesReplyItem::kMaxFieldName, {IndexType::INDEX_2D}}, - {ListIndexesReplyItem::kMinFieldName, {IndexType::INDEX_2D}}, - {ListIndexesReplyItem::kNameFieldName, {}}, - {ListIndexesReplyItem::kNsFieldName, {}}, - {ListIndexesReplyItem::kOriginalSpecFieldName, {}}, - {ListIndexesReplyItem::kPartialFilterExpressionFieldName, {}}, - {ListIndexesReplyItem::kPrepareUniqueFieldName, {}}, - {ListIndexesReplyItem::kSparseFieldName, {}}, - {ListIndexesReplyItem::kSpecFieldName, {}}, - {ListIndexesReplyItem::kStorageEngineFieldName, {}}, - {ListIndexesReplyItem::kTextIndexVersionFieldName, {IndexType::INDEX_TEXT}}, - {ListIndexesReplyItem::kUniqueFieldName, {}}, - {ListIndexesReplyItem::kVFieldName, {}}, - {ListIndexesReplyItem::kWeightsFieldName, {IndexType::INDEX_TEXT}}, - {ListIndexesReplyItem::kWildcardProjectionFieldName, {IndexType::INDEX_WILDCARD}}, -}; +static std::set<StringData> allowedFieldNames = { + ListIndexesReplyItem::k2dsphereIndexVersionFieldName, + ListIndexesReplyItem::kBackgroundFieldName, + ListIndexesReplyItem::kBitsFieldName, + ListIndexesReplyItem::kBucketSizeFieldName, + ListIndexesReplyItem::kBuildUUIDFieldName, + ListIndexesReplyItem::kClusteredFieldName, + ListIndexesReplyItem::kCoarsestIndexedLevelFieldName, + ListIndexesReplyItem::kCollationFieldName, + ListIndexesReplyItem::kDefault_languageFieldName, + ListIndexesReplyItem::kDropDupsFieldName, + ListIndexesReplyItem::kExpireAfterSecondsFieldName, + ListIndexesReplyItem::kFinestIndexedLevelFieldName, + ListIndexesReplyItem::kHiddenFieldName, + ListIndexesReplyItem::kIndexBuildInfoFieldName, + ListIndexesReplyItem::kKeyFieldName, + ListIndexesReplyItem::kLanguage_overrideFieldName, + ListIndexesReplyItem::kMaxFieldName, + ListIndexesReplyItem::kMinFieldName, + ListIndexesReplyItem::kNameFieldName, + ListIndexesReplyItem::kNsFieldName, + ListIndexesReplyItem::kOriginalSpecFieldName, + ListIndexesReplyItem::kPartialFilterExpressionFieldName, + ListIndexesReplyItem::kPrepareUniqueFieldName, + ListIndexesReplyItem::kSparseFieldName, + ListIndexesReplyItem::kSpecFieldName, + ListIndexesReplyItem::kStorageEngineFieldName, + ListIndexesReplyItem::kTextIndexVersionFieldName, + ListIndexesReplyItem::kUniqueFieldName, + ListIndexesReplyItem::kVFieldName, + ListIndexesReplyItem::kWeightsFieldName, + ListIndexesReplyItem::kWildcardProjectionFieldName}; /** * Returns index specs, with resolved namespace, from the catalog for this listIndexes request. @@ -307,7 +306,7 @@ public: nss)); std::vector<mongo::ListIndexesReplyItem> firstBatch; - FindCommon::BSONArrayResponseSizeTracker responseSizeTracker; + size_t bytesBuffered = 0; for (long long objCount = 0; objCount < batchSize; objCount++) { BSONObj nextDoc; PlanExecutor::ExecState state = exec->getNext(&nextDoc, nullptr); @@ -319,7 +318,7 @@ public: // If we can't fit this result inside the current batch, then we stash it for // later. - if (!responseSizeTracker.haveSpaceForNext(nextDoc)) { + if (!FindCommon::haveSpaceForNext(nextDoc, objCount, bytesBuffered)) { exec->stashResult(nextDoc); break; } @@ -338,7 +337,7 @@ public: nextDoc.toString(), exc.toString())); } - responseSizeTracker.add(nextDoc); + bytesBuffered += nextDoc.objsize(); } if (exec->isEOF()) { diff --git a/src/mongo/db/commands/map_reduce_agg.cpp b/src/mongo/db/commands/map_reduce_agg.cpp index 91a27c93bf4..eb1d432f5b5 100644 --- a/src/mongo/db/commands/map_reduce_agg.cpp +++ b/src/mongo/db/commands/map_reduce_agg.cpp @@ -138,8 +138,6 @@ bool runAggregationMapReduce(OperationContext* opCtx, Timer cmdTimer; auto parsedMr = MapReduceCommandRequest::parse(IDLParserErrorContext("mapReduce"), cmd); - auto curop = CurOp::get(opCtx); - curop->beginQueryPlanningTimer(); auto expCtx = makeExpressionContext(opCtx, parsedMr, verbosity); auto runnablePipeline = [&]() { auto pipeline = map_reduce_common::translateFromMR(parsedMr, expCtx); @@ -148,10 +146,10 @@ bool runAggregationMapReduce(OperationContext* opCtx, }(); auto exec = plan_executor_factory::make(expCtx, std::move(runnablePipeline)); auto&& explainer = exec->getPlanExplainer(); - // Store the plan summary string in CurOp. + { stdx::lock_guard<Client> lk(*opCtx->getClient()); - curop->setPlanSummary_inlock(explainer.getPlanSummary()); + CurOp::get(opCtx)->setPlanSummary_inlock(explainer.getPlanSummary()); } try { diff --git a/src/mongo/db/commands/map_reduce_agg_test.cpp b/src/mongo/db/commands/map_reduce_agg_test.cpp index b47b2bfa0e4..52ee67c416b 100644 --- a/src/mongo/db/commands/map_reduce_agg_test.cpp +++ b/src/mongo/db/commands/map_reduce_agg_test.cpp @@ -46,6 +46,15 @@ #include "mongo/db/pipeline/expression_context_for_test.h" #include "mongo/unittest/unittest.h" +#define ASSERT_DOES_NOT_THROW(EXPRESSION) \ + try { \ + EXPRESSION; \ + } catch (const AssertionException& e) { \ + str::stream err; \ + err << "Threw an exception incorrectly: " << e.toString(); \ + ::mongo::unittest::TestAssertionFailure(__FILE__, __LINE__, err).stream(); \ + } + namespace mongo { namespace { diff --git a/src/mongo/db/commands/map_reduce_command_base.h b/src/mongo/db/commands/map_reduce_command_base.h index c0d19d6372a..f3806c40766 100644 --- a/src/mongo/db/commands/map_reduce_command_base.h +++ b/src/mongo/db/commands/map_reduce_command_base.h @@ -63,7 +63,7 @@ public: {kDefaultReadConcernNotPermitted}}; } - bool shouldAffectReadOptionCounters() const override { + bool shouldAffectReadConcernCounter() const override { return true; } diff --git a/src/mongo/db/commands/mr_test.cpp b/src/mongo/db/commands/mr_test.cpp index 95fe13b68ee..745d883c8e1 100644 --- a/src/mongo/db/commands/mr_test.cpp +++ b/src/mongo/db/commands/mr_test.cpp @@ -42,7 +42,7 @@ #include "mongo/db/commands.h" #include "mongo/db/commands/map_reduce_gen.h" #include "mongo/db/commands/mr_common.h" -#include "mongo/db/concurrency/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/json.h" #include "mongo/db/op_observer_noop.h" diff --git a/src/mongo/db/commands/oplog_note.cpp b/src/mongo/db/commands/oplog_note.cpp index 9f93f116815..7af5c111ca3 100644 --- a/src/mongo/db/commands/oplog_note.cpp +++ b/src/mongo/db/commands/oplog_note.cpp @@ -39,7 +39,7 @@ #include "mongo/db/auth/authorization_session.h" #include "mongo/db/auth/resource_pattern.h" #include "mongo/db/concurrency/d_concurrency.h" -#include "mongo/db/concurrency/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/curop.h" #include "mongo/db/jsobj.h" #include "mongo/db/namespace_string.h" diff --git a/src/mongo/db/commands/pipeline_command.cpp b/src/mongo/db/commands/pipeline_command.cpp index 2042914e16b..eb5f74d4387 100644 --- a/src/mongo/db/commands/pipeline_command.cpp +++ b/src/mongo/db/commands/pipeline_command.cpp @@ -89,7 +89,7 @@ public: this, opMsgRequest, std::move(aggregationRequest), std::move(privileges)); } - bool shouldAffectReadOptionCounters() const override { + bool shouldAffectReadConcernCounter() const override { return true; } @@ -142,6 +142,7 @@ public: opCtx, !Pipeline::aggHasWriteStage(_request.body)); uassertStatusOK(runAggregate(opCtx, + _aggregationRequest.getNamespace(), _aggregationRequest, _liteParsedPipeline, _request.body, @@ -164,6 +165,7 @@ public: rpc::ReplyBuilderInterface* result) override { uassertStatusOK(runAggregate(opCtx, + _aggregationRequest.getNamespace(), _aggregationRequest, _liteParsedPipeline, _request.body, diff --git a/src/mongo/db/commands/profile.idl b/src/mongo/db/commands/profile.idl index f1c56a88a05..cd4a295a950 100644 --- a/src/mongo/db/commands/profile.idl +++ b/src/mongo/db/commands/profile.idl @@ -66,16 +66,3 @@ commands: an alternative to slowms and sampleRate. The special value 'unset' removes the filter." optional: true - - setProfilingFilterGlobally: - description: "Parser for the 'setProfilingFilterGlobally' command." - command_name: "setProfilingFilterGlobally" - cpp_name: SetProfilingFilterGloballyCmdRequest - strict: true - namespace: ignored - api_version: "" - fields: - filter: - type: ObjectOrUnset - description: "A query predicate that determines which ops are logged/profiled on a global - level. The special value 'unset' removes the filter." diff --git a/src/mongo/db/commands/resize_oplog.cpp b/src/mongo/db/commands/resize_oplog.cpp index 4433edae662..48dc95ade4e 100644 --- a/src/mongo/db/commands/resize_oplog.cpp +++ b/src/mongo/db/commands/resize_oplog.cpp @@ -37,7 +37,7 @@ #include "mongo/db/catalog/collection.h" #include "mongo/db/commands.h" #include "mongo/db/commands/resize_oplog_gen.h" -#include "mongo/db/concurrency/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/db_raii.h" #include "mongo/db/jsobj.h" #include "mongo/db/operation_context.h" diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp index 66d68f4413c..9172103b493 100644 --- a/src/mongo/db/commands/run_aggregate.cpp +++ b/src/mongo/db/commands/run_aggregate.cpp @@ -74,9 +74,6 @@ #include "mongo/db/query/query_feature_flags_gen.h" #include "mongo/db/query/query_knobs_gen.h" #include "mongo/db/query/query_planner_common.h" -#include "mongo/db/query/query_stats/agg_key.h" -#include "mongo/db/query/query_stats/key.h" -#include "mongo/db/query/query_stats/query_stats.h" #include "mongo/db/read_concern.h" #include "mongo/db/repl/oplog.h" #include "mongo/db/repl/read_concern_args.h" @@ -107,8 +104,6 @@ namespace { ServerStatusMetricField<Counter64> allowDiskUseMetric{"query.allowDiskUseFalse", &allowDiskUseFalseCounter}; -MONGO_FAIL_POINT_DEFINE(hangAfterCreatingAggregationPlan); - /** * If a pipeline is empty (assuming that a $cursor stage hasn't been created yet), it could mean * that we were able to absorb all pipeline stages and pull them into a single PlanExecutor. So, @@ -226,6 +221,7 @@ bool handleCursorCommand(OperationContext* opCtx, auto&& [stats, _] = explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats); LOGV2_WARNING(23799, + "Aggregate command executor error: {error}, stats: {stats}, cmd: {cmd}", "Aggregate command executor error", "error"_attr = exception.toStatus(), "stats"_attr = redact(stats), @@ -630,6 +626,7 @@ std::vector<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createLegacyEx getSearchHelpers(expCtx->opCtx->getServiceContext()) ->injectSearchShardFiltererIfNeeded(pipeline.get()); + // Complete creation of the initial $cursor stage, if needed. PipelineD::attachInnerQueryExecutorToPipeline(collections, attachExecutorCallback.first, @@ -642,6 +639,7 @@ std::vector<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createLegacyEx // There are separate ExpressionContexts for each exchange pipeline, so make sure to // pass the pipeline's ExpressionContext to the plan executor factory. auto pipelineExpCtx = pipelineIt->getContext(); + execs.emplace_back( plan_executor_factory::make(std::move(pipelineExpCtx), std::move(pipelineIt), @@ -659,222 +657,25 @@ std::vector<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createLegacyEx return execs; } -Status runAggregateOnView(OperationContext* opCtx, - const NamespaceString& origNss, - const AggregateCommandRequest& request, - const MultipleCollectionAccessor& collections, - boost::optional<std::unique_ptr<CollatorInterface>> collatorToUse, - const ViewDefinition* view, - std::shared_ptr<const CollectionCatalog> catalog, - const PrivilegeVector& privileges, - rpc::ReplyBuilderInterface* result, - const std::function<void(void)>& resetContextFn) { - auto nss = request.getNamespace(); - - uassert(ErrorCodes::CommandNotSupportedOnView, - "mapReduce on a view is not supported", - !request.getIsMapReduceCommand()); - - // Check that the default collation of 'view' is compatible with the operation's - // collation. The check is skipped if the request did not specify a collation. - if (!request.getCollation().get_value_or(BSONObj()).isEmpty()) { - invariant(collatorToUse); // Should already be resolved at this point. - if (!CollatorInterface::collatorsMatch(view->defaultCollator(), collatorToUse->get()) && - !view->timeseries()) { - - return {ErrorCodes::OptionNotSupportedOnView, - "Cannot override a view's default collation"}; - } - } - - // Queries on timeseries views may specify non-default collation whereas queries - // on all other types of views must match the default collator (the collation use - // to originally create that collections). Thus in the case of operations on TS - // views, we use the request's collation. - auto timeSeriesCollator = view->timeseries() ? request.getCollation() : boost::none; - - auto resolvedView = - uassertStatusOK(view_catalog_helpers::resolveView(opCtx, catalog, nss, timeSeriesCollator)); - - // With the view & collation resolved, we can relinquish locks. - resetContextFn(); - - // Set this operation's shard version for the underlying collection to unsharded. - // This is prerequisite for future shard versioning checks. - ScopedSetShardRole scopedSetShardRole(opCtx, - resolvedView.getNamespace(), - ChunkVersion::UNSHARDED() /* shardVersion */, - boost::none /* databaseVersion */); - - uassert(std::move(resolvedView), - "Explain of a resolved view must be executed by mongos", - !ShardingState::get(opCtx)->enabled() || !request.getExplain()); - - // Parse the resolved view into a new aggregation request. - auto newRequest = resolvedView.asExpandedViewAggregation(request); - auto newCmd = aggregation_request_helper::serializeToCommandObj(newRequest); - - auto status{Status::OK()}; - try { - status = runAggregate(opCtx, newRequest, newCmd, privileges, result, resolvedView, request); - } catch (const ExceptionForCat<ErrorCategory::StaleShardVersionError>& ex) { - // Since we expect the view to be UNSHARDED, if we reached to this point there are - // two possibilities: - // 1. The shard doesn't know what its shard version/state is and needs to recover - // it (in which case we throw so that the shard can run recovery) - // 2. The collection references by the view is actually SHARDED, in which case the - // router must execute it - if (const auto staleInfo{ex.extraInfo<StaleConfigInfo>()}) { - uassert(std::move(resolvedView), - "Resolved views on sharded collections must be executed by mongos", - !staleInfo->getVersionWanted()); - } - throw; - } - - { - // Set the namespace of the curop back to the view namespace so ctx records - // stats on this view namespace on destruction. - stdx::lock_guard<Client> lk(*opCtx->getClient()); - CurOp::get(opCtx)->setNS_inlock(nss.ns()); - } - - return status; -} - -/** - * Determines the collection type of the query by precedence of various configurations. The order - * of these checks is critical since there may be overlap (e.g., a view over a virtual collection - * is classified as a view). - */ -query_shape::CollectionType determineCollectionType( - const boost::optional<AutoGetCollectionForReadCommandMaybeLockFree>& ctx, - boost::optional<const ResolvedView&> resolvedView, - bool hasChangeStream, - bool isCollectionless) { - if (resolvedView.has_value()) { - if (resolvedView->timeseries()) { - return query_shape::CollectionType::kTimeseries; - } - return query_shape::CollectionType::kView; - } - if (isCollectionless) { - return query_shape::CollectionType::kVirtual; - } - if (hasChangeStream) { - return query_shape::CollectionType::kChangeStream; - } - return ctx ? ctx->getCollectionType() : query_shape::CollectionType::kUnknown; -} - -std::unique_ptr<Pipeline, PipelineDeleter> parsePipelineAndRegisterQueryStats( - OperationContext* opCtx, - const NamespaceString& origNss, - const AggregateCommandRequest& request, - const boost::optional<AutoGetCollectionForReadCommandMaybeLockFree>& ctx, - std::unique_ptr<CollatorInterface> collator, - boost::optional<UUID> uuid, - ExpressionContext::CollationMatchesDefault collationMatchesDefault, - const MultipleCollectionAccessor& collections, - stdx::unordered_set<NamespaceString> pipelineInvolvedNamespaces, - const LiteParsedPipeline& liteParsedPipeline, - bool isCollectionless, - boost::optional<const ResolvedView&> resolvedView, - boost::optional<const AggregateCommandRequest&> origRequest) { - // If we're operating over a view, we first parse just the original user-given request - // for the sake of registering query stats. Then, we'll parse the view pipeline and stitch - // the two pipelines together below. - auto expCtx = - makeExpressionContext(opCtx, request, std::move(collator), uuid, collationMatchesDefault); - // If any involved collection contains extended-range data, set a flag which individual - // DocumentSource parsers can check. - collections.forEach([&](const CollectionPtr& coll) { - if (coll->getRequiresTimeseriesExtendedRangeSupport()) - expCtx->setRequiresTimeseriesExtendedRangeSupport(true); - }); - - const bool hasChangeStream = liteParsedPipeline.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; - } - - auto requestForQueryStats = origRequest.has_value() ? *origRequest : request; - expCtx->startExpressionCounters(); - auto pipeline = Pipeline::parse(requestForQueryStats.getPipeline(), expCtx); - expCtx->stopExpressionCounters(); - - // Register query stats with the pre-optimized pipeline. Exclude queries against collections - // with encrypted fields. We still collect query stats on collection-less aggregations. - bool hasEncryptedFields = ctx && ctx->getCollection() && - ctx->getCollection()->getCollectionOptions().encryptedFieldConfig; - if (!hasEncryptedFields) { - // If this is a query over a resolved view, we want to register query stats with the - // original user-given request and pipeline, rather than the new request generated when - // resolving the view. - auto collectionType = - determineCollectionType(ctx, resolvedView, hasChangeStream, isCollectionless); - - query_stats::registerRequest(opCtx, - origNss, - [&]() { - return std::make_unique<query_stats::AggKey>( - requestForQueryStats, - *pipeline, - expCtx, - pipelineInvolvedNamespaces, - origNss, - collectionType); - }, - hasChangeStream); - } - - if (resolvedView.has_value()) { - expCtx->startExpressionCounters(); - - if (resolvedView->timeseries()) { - // For timeseries, there may have been rewrites done on the raw BSON pipeline - // during view resolution. We must parse the request's full resolved pipeline - // which will account for those rewrites. - // TODO SERVER-82101 Re-organize timeseries rewrites so timeseries can follow the - // same pattern here as other views - pipeline = Pipeline::parse(request.getPipeline(), expCtx); - } else { - // Parse the view pipeline, then stitch the user pipeline and view pipeline together - // to build the total aggregation pipeline. - auto userPipeline = std::move(pipeline); - pipeline = Pipeline::parse(resolvedView->getPipeline(), expCtx); - pipeline->appendPipeline(std::move(userPipeline)); - } - - expCtx->stopExpressionCounters(); - } - - return pipeline; -} } // namespace Status runAggregate(OperationContext* opCtx, + const NamespaceString& nss, AggregateCommandRequest& request, const BSONObj& cmdObj, const PrivilegeVector& privileges, - rpc::ReplyBuilderInterface* result, - boost::optional<const ResolvedView&> resolvedView, - boost::optional<const AggregateCommandRequest&> origRequest) { - return runAggregate( - opCtx, request, {request}, cmdObj, privileges, result, resolvedView, origRequest); + rpc::ReplyBuilderInterface* result) { + return runAggregate(opCtx, nss, request, {request}, cmdObj, privileges, result); } Status runAggregate(OperationContext* opCtx, + const NamespaceString& origNss, AggregateCommandRequest& request, const LiteParsedPipeline& liteParsedPipeline, const BSONObj& cmdObj, const PrivilegeVector& privileges, - rpc::ReplyBuilderInterface* result, - boost::optional<const ResolvedView&> resolvedView, - boost::optional<const AggregateCommandRequest&> origRequest) { - auto origNss = origRequest.has_value() ? origRequest->getNamespace() : request.getNamespace(); + rpc::ReplyBuilderInterface* result) { + // Perform some validations on the LiteParsedPipeline and request before continuing with the // aggregation command. performValidationChecks(opCtx, request, liteParsedPipeline); @@ -944,8 +745,7 @@ Status runAggregate(OperationContext* opCtx, boost::optional<AutoStatsTracker> statsTracker; // If this is a change stream, perform special checks and change the execution namespace. - const auto hasChangeStream = liteParsedPipeline.hasChangeStream(); - if (hasChangeStream) { + if (liteParsedPipeline.hasChangeStream()) { uassert(4928900, str::stream() << AggregateCommandRequest::kCollectionUUIDFieldName << " is not supported for a change stream", @@ -959,6 +759,7 @@ Status runAggregate(OperationContext* opCtx, // Raise an error if 'origNss' is a view. We do not need to check this if we are opening // a stream on an entire db or across the cluster. + const TenantDatabaseName origTenantDbName(boost::none, origNss.db()); if (!origNss.isCollectionlessAggregateNS()) { auto view = catalog->lookupView(opCtx, origNss); uassert(ErrorCodes::CommandNotSupportedOnView, @@ -992,7 +793,7 @@ Status runAggregate(OperationContext* opCtx, nss, Top::LockType::NotLocked, AutoStatsTracker::LogMode::kUpdateTopAndCurOp, - catalog->getDatabaseProfileLevel(nss.db())); + 0); auto [collator, match] = PipelineD::resolveCollator( opCtx, request.getCollation().get_value_or(BSONObj()), nullptr); collatorToUse.emplace(std::move(collator)); @@ -1014,53 +815,108 @@ Status runAggregate(OperationContext* opCtx, } } - // If collectionUUID was provided, verify the collection exists and has the expected UUID. - checkCollectionUUIDMismatch(opCtx, - nss, - collections.getMainCollection(), - request.getCollectionUUID(), - false /* checkFeatureFlag */); - // If this is a view, resolve it by finding the underlying collection and stitching view // pipelines and this request's pipeline together. We then release our locks before // recursively calling runAggregate(), which will re-acquire locks on the underlying // collection. (The lock must be released because recursively acquiring locks on the // database will prohibit yielding.) - // We do not need to expand the view pipeline when there is a $collStats stage, as - // $collStats is supported on a view namespace. For a time-series collection, however, the - // view is abstracted out for the users, so we needed to resolve the namespace to get the - // underlying bucket collection. - if (ctx && ctx->getView() && - (!liteParsedPipeline.startsWithCollStats() || ctx->getView()->timeseries())) { - return runAggregateOnView(opCtx, - origNss, - request, - collections, - std::move(collatorToUse), - ctx->getView(), - catalog, - privileges, - result, - resetContext); + if (ctx && ctx->getView() && !liteParsedPipeline.startsWithCollStats()) { + invariant(nss != NamespaceString::kRsOplogNamespace); + invariant(!nss.isCollectionlessAggregateNS()); + + checkCollectionUUIDMismatch(opCtx, + nss, + collections.getMainCollection(), + request.getCollectionUUID(), + false /* checkFeatureFlag */); + + uassert(ErrorCodes::CommandNotSupportedOnView, + "mapReduce on a view is not supported", + !request.getIsMapReduceCommand()); + + // Check that the default collation of 'view' is compatible with the operation's + // collation. The check is skipped if the request did not specify a collation. + if (!request.getCollation().get_value_or(BSONObj()).isEmpty()) { + invariant(collatorToUse); // Should already be resolved at this point. + if (!CollatorInterface::collatorsMatch(ctx->getView()->defaultCollator(), + collatorToUse->get()) && + !ctx->getView()->timeseries()) { + + return {ErrorCodes::OptionNotSupportedOnView, + "Cannot override a view's default collation"}; + } + } + + // Queries on timeseries views may specify non-default collation whereas queries + // on all other types of views must match the default collator (the collation use + // to originally create that collections). Thus in the case of operations on TS + // views, we use the request's collation. + auto timeSeriesCollator = + ctx->getView()->timeseries() ? request.getCollation() : boost::none; + + auto resolvedView = uassertStatusOK( + view_catalog_helpers::resolveView(opCtx, catalog, nss, timeSeriesCollator)); + + // With the view & collation resolved, we can relinquish locks. + resetContext(); + + // Set this operation's shard version for the underlying collection to unsharded. + // This is prerequisite for future shard versioning checks. + ScopedSetShardRole scopedSetShardRole(opCtx, + resolvedView.getNamespace(), + ChunkVersion::UNSHARDED() /* shardVersion */, + boost::none /* databaseVersion */); + + uassert(std::move(resolvedView), + "Explain of a resolved view must be executed by mongos", + !ShardingState::get(opCtx)->enabled() || !request.getExplain()); + + // Parse the resolved view into a new aggregation request. + auto newRequest = resolvedView.asExpandedViewAggregation(request); + auto newCmd = aggregation_request_helper::serializeToCommandObj(newRequest); + + auto status{Status::OK()}; + try { + status = runAggregate(opCtx, origNss, newRequest, newCmd, privileges, result); + } catch (const ExceptionForCat<ErrorCategory::StaleShardVersionError>& ex) { + // Since we expect the view to be UNSHARDED, if we reached to this point there are + // two possibilities: + // 1. The shard doesn't know what its shard version/state is and needs to recover + // it (in which case we throw so that the shard can run recovery) + // 2. The collection references by the view is actually SHARDED, in which case the + // router must execute it + if (const auto staleInfo{ex.extraInfo<StaleConfigInfo>()}) { + uassert(std::move(resolvedView), + "Resolved views on sharded collections must be executed by mongos", + !staleInfo->getVersionWanted()); + } + throw; + } + + { + // Set the namespace of the curop back to the view namespace so ctx records + // stats on this view namespace on destruction. + stdx::lock_guard<Client> lk(*opCtx->getClient()); + curOp->setNS_inlock(nss.ns()); + } + + return status; } + // If collectionUUID was provided, verify the collection exists and has the expected UUID. + checkCollectionUUIDMismatch(opCtx, + nss, + collections.getMainCollection(), + request.getCollectionUUID(), + false /* checkFeatureFlag */); + invariant(collatorToUse); - auto pipeline = parsePipelineAndRegisterQueryStats(opCtx, - origNss, - request, - ctx, - std::move(*collatorToUse), - uuid, - collatorToUseMatchesDefault, - collections, - pipelineInvolvedNamespaces, - liteParsedPipeline, - nss.isCollectionlessAggregateNS(), - resolvedView, - origRequest); - expCtx = pipeline->getContext(); - - CurOp::get(opCtx)->beginQueryPlanningTimer(); + expCtx = makeExpressionContext( + opCtx, request, std::move(*collatorToUse), uuid, collatorToUseMatchesDefault); + + expCtx->startExpressionCounters(); + auto pipeline = Pipeline::parse(request.getPipeline(), expCtx); + expCtx->stopExpressionCounters(); if (!request.getAllowDiskUse().value_or(true)) { allowDiskUseFalseCounter.increment(); @@ -1126,9 +982,6 @@ Status runAggregate(OperationContext* opCtx, // cursor manager. The global cursor manager does not deliver invalidations or kill // notifications; the underlying PlanExecutor(s) used by the pipeline will be receiving // invalidations and kill notifications themselves, not the cursor we create here. - hangAfterCreatingAggregationPlan.executeIf( - [](const auto&) { hangAfterCreatingAggregationPlan.pauseWhileSet(); }, - [&](const BSONObj& data) { return uuid && UUID::parse(data["uuid"]) == *uuid; }); std::vector<ClientCursorPin> pins; std::vector<ClientCursor*> cursors; @@ -1139,7 +992,6 @@ Status runAggregate(OperationContext* opCtx, } }); for (auto&& exec : execs) { - // TODO SERVER-79373: Do not create a cursor if results can fit in a single batch. ClientCursorParams cursorParams( std::move(exec), origNss, @@ -1183,7 +1035,6 @@ Status runAggregate(OperationContext* opCtx, cmdObj, &bodyBuilder); } - collectQueryStatsMongod(opCtx, std::move(curOp->debug().queryStatsInfo.key)); } else { // Cursor must be specified, if explain is not. const bool keepCursor = handleCursorCommand( @@ -1196,15 +1047,13 @@ Status runAggregate(OperationContext* opCtx, PlanSummaryStats stats; planExplainer.getSummaryStats(&stats); curOp->debug().setPlanSummaryMetrics(stats); - curOp->setEndOfOpMetrics(stats.nReturned); + curOp->debug().nreturned = stats.nReturned; - collectQueryStatsMongod(opCtx, pins[0]); - - // For an optimized away pipeline, signal the cache that a query operation has - // completed. For normal pipelines this is done in DocumentSourceCursor. + // For an optimized away pipeline, signal the cache that a query operation has completed. + // For normal pipelines this is done in DocumentSourceCursor. if (ctx) { - // Due to yielding, the collection pointers saved in MultipleCollectionAccessor - // might have become invalid. We will need to refresh them here. + // Due to yielding, the collection pointers saved in MultipleCollectionAccessor might + // have become invalid. We will need to refresh them here. collections = MultipleCollectionAccessor(opCtx, &ctx->getCollection(), ctx->getNss(), @@ -1227,11 +1076,10 @@ Status runAggregate(OperationContext* opCtx, } } - // The aggregation pipeline may change the namespace of the curop and we need to set it back - // to the original namespace to correctly report command stats. One example when the - // namespace can be changed is when the pipeline contains an $out stage, which executes an - // internal command to create a temp collection, changing the curop namespace to the name of - // this temp collection. + // The aggregation pipeline may change the namespace of the curop and we need to set it back to + // the original namespace to correctly report command stats. One example when the namespace can + // be changed is when the pipeline contains an $out stage, which executes an internal command to + // create a temp collection, changing the curop namespace to the name of this temp collection. { stdx::lock_guard<Client> lk(*opCtx->getClient()); curOp->setNS_inlock(origNss.ns()); @@ -1239,4 +1087,5 @@ Status runAggregate(OperationContext* opCtx, return Status::OK(); } + } // namespace mongo diff --git a/src/mongo/db/commands/run_aggregate.h b/src/mongo/db/commands/run_aggregate.h index ba73245bdbf..b61538fb93d 100644 --- a/src/mongo/db/commands/run_aggregate.h +++ b/src/mongo/db/commands/run_aggregate.h @@ -49,30 +49,25 @@ namespace mongo { * 'privileges' contains the privileges that were required to run this aggregation, to be used later * for re-checking privileges for getMore commands. * - * If the query over a view that's already been resolved, the resolved view and the original - * user-provided request both must be provided. - * * On success, fills out 'result' with the command response. */ Status runAggregate(OperationContext* opCtx, + const NamespaceString& nss, AggregateCommandRequest& request, const LiteParsedPipeline& liteParsedPipeline, const BSONObj& cmdObj, const PrivilegeVector& privileges, - rpc::ReplyBuilderInterface* result, - boost::optional<const ResolvedView&> resolvedView = boost::none, - boost::optional<const AggregateCommandRequest&> origRequest = boost::none); + rpc::ReplyBuilderInterface* result); /** * Convenience version that internally constructs the LiteParsedPipeline. */ Status runAggregate(OperationContext* opCtx, + const NamespaceString& nss, AggregateCommandRequest& request, const BSONObj& cmdObj, const PrivilegeVector& privileges, - rpc::ReplyBuilderInterface* result, - boost::optional<const ResolvedView&> resolvedView = boost::none, - boost::optional<const AggregateCommandRequest&> origRequest = boost::none); + rpc::ReplyBuilderInterface* result); /** * Tracks explicit use of allowDiskUse:false with find and aggregate commands. diff --git a/src/mongo/db/commands/server_status_metric.h b/src/mongo/db/commands/server_status_metric.h index 423c49e2f5f..ff546527a8d 100644 --- a/src/mongo/db/commands/server_status_metric.h +++ b/src/mongo/db/commands/server_status_metric.h @@ -32,13 +32,8 @@ #include <string> #include "mongo/db/jsobj.h" -#include "mongo/platform/atomic_word.h" namespace mongo { -class Atomic64Metric; - -template <> -struct BSONObjAppendFormat<Atomic64Metric> : FormatKind<NumberLong> {}; class ServerStatusMetric { public: @@ -89,36 +84,4 @@ public: private: const T* _t; }; - -/** - * Atomic wrapper for long long type for Metrics. This is for values which are set rather than - * just incremented or decremented; if you want a counter, use Counter64. - */ -class Atomic64Metric { -public: - /** Set _value to the max of the current or newMax. */ - void setIfMax(long long newMax) { - /* Note: compareAndSwap will load into val most recent value. */ - for (long long val = _value.load(); val < newMax && !_value.compareAndSwap(&val, newMax);) { - } - } - - /** store val into value. */ - void set(long long val) { - _value.storeRelaxed(val); - } - - /** Return the current value. */ - long long get() const { - return _value.loadRelaxed(); - } - - /** TODO: SERVER-73806 Avoid implicit conversion to long long */ - operator long long() const { - return get(); - } - -private: - mongo::AtomicWord<long long> _value; -}; } // namespace mongo diff --git a/src/mongo/db/commands/set_cluster_parameter_command.cpp b/src/mongo/db/commands/set_cluster_parameter_command.cpp index 2049b3043e5..005f17d0c06 100644 --- a/src/mongo/db/commands/set_cluster_parameter_command.cpp +++ b/src/mongo/db/commands/set_cluster_parameter_command.cpp @@ -74,11 +74,6 @@ public: (serverGlobalParams.clusterRole == ClusterRole::None)); FixedFCVRegion fcvRegion(opCtx); - uassert(ErrorCodes::UnknownFeatureCompatibilityVersion, - "FCV is not yet initialized, retry the command after FCV initialization has " - "completed", - serverGlobalParams.featureCompatibility.isVersionInitialized()); - uassert( ErrorCodes::IllegalOperation, "Cannot set cluster parameter, gFeatureFlagClusterWideConfig is not enabled", diff --git a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp index 7462b287ed8..73933d1abe2 100644 --- a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp +++ b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp @@ -84,7 +84,6 @@ #include "mongo/db/session_catalog.h" #include "mongo/db/session_catalog_mongod.h" #include "mongo/db/session_txn_record_gen.h" -#include "mongo/db/storage/storage_parameters_gen.h" #include "mongo/db/timeseries/timeseries_index_schema_conversion_functions.h" #include "mongo/db/vector_clock.h" #include "mongo/idl/cluster_server_parameter_gen.h" @@ -466,7 +465,7 @@ public: if (actualVersion > requestedVersion && !feature_flags::gOrphanTracking.isEnabledOnVersion(requestedVersion)) { BalancerStatsRegistry::get(opCtx)->terminate(); - ScopedRangeDeleterLock rangeDeleterLock(opCtx, LockMode::MODE_X); + ScopedRangeDeleterLock rangeDeleterLock(opCtx); clearOrphanCountersFromRangeDeletionTasks(opCtx); } @@ -631,6 +630,14 @@ private: tenantDbName, MODE_X, [&](const CollectionPtr& collection) { + if (collection->getTimeseriesBucketsMayHaveMixedSchemaData()) { + // The catalog entry flag has already been added. This can happen if the + // upgrade process was interrupted and is being run again, or if there + // was a time-series collection created during the upgrade. The upgrade + // process cannot be aborted at this point. + return true; + } + NamespaceStringOrUUID nsOrUUID(dbName, collection->uuid()); CollMod collModCmd(collection->ns()); BSONObjBuilder unusedBuilder; @@ -743,46 +750,16 @@ private: const auto& dbName = tenantDbName.dbName(); Lock::DBLock dbLock(opCtx, dbName, MODE_IX); catalog::forEachCollectionFromDb( - opCtx, tenantDbName, MODE_X, [&](const CollectionPtr& collection) { - const auto collNs = collection->getTimeseriesOptions() - ? collection->ns().getTimeseriesViewNamespace() - : collection->ns(); - auto indexCatalog = collection->getIndexCatalog(); + opCtx, + tenantDbName, + MODE_X, + [&](const CollectionPtr& collection) { + invariant(collection->getTimeseriesOptions()); + auto indexCatalog = collection->getIndexCatalog(); auto indexIt = indexCatalog->getIndexIterator( - opCtx, - IndexCatalog::InclusionPolicy::kReady | - IndexCatalog::InclusionPolicy::kUnfinished | - IndexCatalog::InclusionPolicy::kFrozen); - while (indexIt->more()) { - auto indexEntry = indexIt->next(); - if (auto filter = indexEntry->getFilterExpression()) { - auto status = IndexCatalogImpl::checkValidFilterExpressions( - filter, - /*timeseriesMetricIndexesFeatureFlagEnabled*/ false); - uassert(ErrorCodes::CannotDowngrade, - str::stream() - << "Cannot downgrade the cluster when there are " - "secondary indexes with partial filter expressions " - "that contain $in/$or/$geoWithin or an $and that is " - "not top level. Drop all indexes containing these " - "partial filter elements before downgrading. First " - "detected incompatible index name: '" - << indexEntry->descriptor()->indexName() - << "' on collection: '" << collNs << "'", - status.isOK()); - } - } + opCtx, /*includeUnfinishedIndexes=*/true); - if (!collection->getTimeseriesOptions()) { - return true; - } - - indexIt = indexCatalog->getIndexIterator( - opCtx, - IndexCatalog::InclusionPolicy::kReady | - IndexCatalog::InclusionPolicy::kUnfinished | - IndexCatalog::InclusionPolicy::kFrozen); while (indexIt->more()) { auto indexEntry = indexIt->next(); // Secondary indexes on time-series measurements are only supported @@ -793,17 +770,34 @@ private: ErrorCodes::CannotDowngrade, str::stream() << "Cannot downgrade the cluster when there are secondary " - "indexes on time-series measurements present, or when " - "there are partial indexes on a time-series collection. " - "Drop " - "all secondary indexes on time-series measurements, and all " + "indexes on time-series measurements present, or when there " + "are partial indexes on a time-series collection. Drop all " + "secondary indexes on time-series measurements, and all " "partial indexes on time-series collections, before " "downgrading. First detected incompatible index name: '" << indexEntry->descriptor()->indexName() << "' on collection: '" - << collNs << "'", + << collection->ns().getTimeseriesViewNamespace() << "'", timeseries::isBucketsIndexSpecCompatibleForDowngrade( *collection->getTimeseriesOptions(), indexEntry->descriptor()->infoObj())); + + if (auto filter = indexEntry->getFilterExpression()) { + auto status = IndexCatalogImpl::checkValidFilterExpressions( + filter, + /*timeseriesMetricIndexesFeatureFlagEnabled*/ false); + uassert(ErrorCodes::CannotDowngrade, + str::stream() + << "Cannot downgrade the cluster when there are " + "secondary indexes with partial filter expressions " + "that contain $in/$or/$geoWithin or an $and that is " + "not top level. Drop all indexes containing these " + "partial filter elements before downgrading. First " + "detected incompatible index name: '" + << indexEntry->descriptor()->indexName() + << "' on collection: '" + << collection->ns().getTimeseriesViewNamespace() << "'", + status.isOK()); + } } if (!collection->getTimeseriesBucketsMayHaveMixedSchemaData()) { @@ -830,6 +824,9 @@ private: } return true; + }, + [&](const CollectionPtr& collection) { + return collection->getTimeseriesOptions() != boost::none; }); } } @@ -871,10 +868,7 @@ private: opCtx, tenantDbName, MODE_X, [&](const CollectionPtr& collection) { auto indexCatalog = collection->getIndexCatalog(); auto indexIt = indexCatalog->getIndexIterator( - opCtx, - IndexCatalog::InclusionPolicy::kReady | - IndexCatalog::InclusionPolicy::kUnfinished | - IndexCatalog::InclusionPolicy::kFrozen); + opCtx, true /* includeUnfinishedIndexes */); while (indexIt->more()) { auto indexEntry = indexIt->next(); uassert( diff --git a/src/mongo/db/commands/set_profiling_filter_globally_cmd.cpp b/src/mongo/db/commands/set_profiling_filter_globally_cmd.cpp deleted file mode 100644 index 287b4331831..00000000000 --- a/src/mongo/db/commands/set_profiling_filter_globally_cmd.cpp +++ /dev/null @@ -1,112 +0,0 @@ -/** - * 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::kCommand - -#include "mongo/db/commands/set_profiling_filter_globally_cmd.h" -#include "mongo/db/auth/authorization_session.h" -#include "mongo/db/catalog/collection_catalog.h" -#include "mongo/db/commands/profile_gen.h" -#include "mongo/db/profile_filter_impl.h" -#include "mongo/logv2/log.h" - -namespace mongo { - -Status SetProfilingFilterGloballyCmd::checkAuthForCommand(Client* client, - const std::string& dbName, - const BSONObj& cmdObj) const { - AuthorizationSession* authSession = AuthorizationSession::get(client); - return authSession->isAuthorizedForActionsOnResource(ResourcePattern::forAnyNormalResource(), - ActionType::enableProfiler) - ? Status::OK() - : Status(ErrorCodes::Unauthorized, "unauthorized"); -} - -bool SetProfilingFilterGloballyCmd::run(OperationContext* opCtx, - const std::string& dbName, - const BSONObj& cmdObj, - BSONObjBuilder& result) { - uassert(7283301, - str::stream() << "setProfilingFilterGlobally command requires query knob to be enabled", - internalQueryGlobalProfilingFilter.load()); - - auto request = SetProfilingFilterGloballyCmdRequest::parse( - IDLParserErrorContext("setProfilingFilterGlobally"), cmdObj); - - // Save off the old global default setting so that we can log it and return in the result. - auto oldDefault = ProfileFilter::getDefault(); - auto newDefault = [&request] { - const auto& filterOrUnset = request.getFilter(); - if (auto filter = filterOrUnset.obj) { - return std::make_shared<ProfileFilterImpl>(*filter); - } - return std::shared_ptr<ProfileFilterImpl>(nullptr); - }(); - - // Update the global default. - // Note that since this is not done atomically with the collection catalog write, there is a - // minor race condition where queries on some databases see the new global default while queries - // on other databases see old database-specific settings. This is a temporary state and - // shouldn't impact much in practice. We also don't have to worry about races with database - // creation, since the global default gets picked up dynamically by queries instead of being - // explicitly stored for new databases. - ProfileFilter::setDefault(newDefault); - - // Writing to the CollectionCatalog requires holding the Global lock to avoid concurrent races - // with BatchedCollectionCatalogWriter. - boost::optional<Lock::GlobalLock> lk; - if (!opCtx->lockState()->isNoop()) { - // Taking the lock is only meaningful if we're in a mongod. Other mongos do not have a - // notion of collections. - lk.emplace(opCtx, MODE_IX); - } - - // Update all existing database settings. - CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) { - catalog.setAllDatabaseProfileFilters(newDefault); - }); - - // Capture the old setting in the result object. - if (oldDefault) { - result.append("was", oldDefault->serialize()); - } else { - result.append("was", "none"); - } - - // Log the change made to server's global profiling settings. - LOGV2(72832, - "Profiler settings changed globally", - "from"_attr = oldDefault ? BSON("filter" << oldDefault->serialize()) - : BSON("filter" - << "none"), - "to"_attr = newDefault ? BSON("filter" << newDefault->serialize()) - : BSON("filter" - << "none")); - return true; -} -} // namespace mongo diff --git a/src/mongo/db/commands/set_profiling_filter_globally_cmd.h b/src/mongo/db/commands/set_profiling_filter_globally_cmd.h deleted file mode 100644 index 15c36f0a9cc..00000000000 --- a/src/mongo/db/commands/set_profiling_filter_globally_cmd.h +++ /dev/null @@ -1,70 +0,0 @@ -/** - * 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. - */ - -#pragma once - -#include "mongo/base/status.h" -#include "mongo/db/catalog/collection_catalog.h" -#include "mongo/db/commands.h" - -namespace mongo { - -class SetProfilingFilterGloballyCmdRequest; - -/** - * Command class implementing functionality for both the mongoD and mongoS - * 'setProfilingFilterGlobally' command. - */ -class SetProfilingFilterGloballyCmd : public BasicCommand { -public: - SetProfilingFilterGloballyCmd() : BasicCommand("setProfilingFilterGlobally") {} - - AllowedOnSecondary secondaryAllowed(ServiceContext*) const final { - return AllowedOnSecondary::kAlways; - } - - std::string help() const final { - return "updates a global filter that determines which operations are eligible for " - "logging/profiling"; - } - - bool supportsWriteConcern(const BSONObj& cmd) const final { - return false; - } - - Status checkAuthForCommand(Client* client, - const std::string& dbname, - const BSONObj& cmdObj) const final; - - bool run(OperationContext* opCtx, - const std::string& dbName, - const BSONObj& cmdObj, - BSONObjBuilder& result) final; -}; -} // namespace mongo diff --git a/src/mongo/db/commands/tenant_migration_donor_cmds.cpp b/src/mongo/db/commands/tenant_migration_donor_cmds.cpp index 14fa2dcf36b..e94f865d1cb 100644 --- a/src/mongo/db/commands/tenant_migration_donor_cmds.cpp +++ b/src/mongo/db/commands/tenant_migration_donor_cmds.cpp @@ -187,7 +187,7 @@ public: auto donorService = repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext()) ->lookupServiceByName(TenantMigrationDonorService::kServiceName); - auto [optionalDonor, _] = TenantMigrationDonorService::Instance::lookup( + auto optionalDonor = TenantMigrationDonorService::Instance::lookup( opCtx, donorService, BSON("_id" << cmd.getMigrationId())); uassert(ErrorCodes::NoSuchTenantMigration, str::stream() << "Could not find tenant migration with id " @@ -261,7 +261,7 @@ public: auto donorService = repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext()) ->lookupServiceByName(TenantMigrationDonorService::kServiceName); - auto [optionalDonor, _] = TenantMigrationDonorService::Instance::lookup( + auto optionalDonor = TenantMigrationDonorService::Instance::lookup( opCtx, donorService, BSON("_id" << cmd.getMigrationId())); // If there is NoSuchTenantMigration, perform a noop write and wait for it to be diff --git a/src/mongo/db/commands/tenant_migration_recipient_cmds.cpp b/src/mongo/db/commands/tenant_migration_recipient_cmds.cpp index 505a5f73f68..e826e51dacc 100644 --- a/src/mongo/db/commands/tenant_migration_recipient_cmds.cpp +++ b/src/mongo/db/commands/tenant_migration_recipient_cmds.cpp @@ -201,7 +201,7 @@ public: repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext()) ->lookupServiceByName(repl::TenantMigrationRecipientService:: kTenantMigrationRecipientServiceName); - auto [instance, _] = repl::TenantMigrationRecipientService::Instance::lookup( + auto instance = repl::TenantMigrationRecipientService::Instance::lookup( opCtx, recipientService, BSON("_id" << cmd.getMigrationId())); uassert(8423340, "Unknown migrationId", instance); (*instance)->onMemberImportedFiles(cmd.getFrom(), cmd.getSuccess(), cmd.getReason()); diff --git a/src/mongo/db/commands/user_management_commands.cpp b/src/mongo/db/commands/user_management_commands.cpp index ff0af0d0f0d..13c5650bc98 100644 --- a/src/mongo/db/commands/user_management_commands.cpp +++ b/src/mongo/db/commands/user_management_commands.cpp @@ -1450,6 +1450,7 @@ UsersInfoReply CmdUMCTyped<UsersInfoCommand, UMCInfoParams>::Invocation::typedRu std::move(pipeline)); // Impose no cursor privilege requirements, as cursor is drained internally uassertStatusOK(runAggregate(opCtx, + AuthorizationManager::usersCollectionNamespace, aggRequest, aggregation_request_helper::serializeToCommandObj(aggRequest), PrivilegeVector(), diff --git a/src/mongo/db/commands/validate.cpp b/src/mongo/db/commands/validate.cpp index 1ca15aca557..c6724076b9f 100644 --- a/src/mongo/db/commands/validate.cpp +++ b/src/mongo/db/commands/validate.cpp @@ -31,29 +31,15 @@ #include "mongo/platform/basic.h" - -#include "mongo/base/error_codes.h" -#include "mongo/base/status.h" -#include "mongo/base/string_data.h" -#include "mongo/bson/bson_validate_gen.h" -#include "mongo/bson/bsonelement.h" -#include "mongo/bson/bsonmisc.h" -#include "mongo/bson/bsonobj.h" -#include "mongo/bson/bsonobjbuilder.h" -#include "mongo/db/auth/action_type.h" -#include "mongo/db/auth/authorization_session.h" #include "mongo/db/catalog/collection.h" #include "mongo/db/catalog/collection_validation.h" #include "mongo/db/client.h" #include "mongo/db/commands.h" -#include "mongo/db/commands/test_commands_enabled.h" -#include "mongo/db/dbdirectclient.h" #include "mongo/db/query/internal_plans.h" #include "mongo/db/storage/record_store.h" #include "mongo/logv2/log.h" #include "mongo/util/fail_point.h" #include "mongo/util/scopeguard.h" -#include "mongo/util/testing_proctor.h" namespace mongo { @@ -75,79 +61,6 @@ std::set<std::string> _validationsInProgress; // finishes on any namespace. stdx::condition_variable _validationNotifier; -/** - * Creates an aggregation command with a $collStats pipeline that fetches 'storageStats' and - * 'count'. - */ -BSONObj makeCollStatsCommand(StringData collectionNameOnly) { - BSONArrayBuilder pipelineBuilder; - pipelineBuilder << BSON("$collStats" - << BSON("storageStats" << BSONObj() << "count" << BSONObj())); - return BSON("aggregate" << collectionNameOnly << "pipeline" << pipelineBuilder.arr() << "cursor" - << BSONObj()); -} - -/** - * $collStats never returns more than a single document. If that ever changes in future, validate - * must invariant so that the handling can be updated, but only invariant in testing environments, - * never invariant because of debug logging in production situations. - */ -void verifyCommandResponse(const BSONObj& collStatsResult) { - if (TestingProctor::instance().isEnabled()) { - invariant( - !collStatsResult.getObjectField("cursor").isEmpty() && - !collStatsResult.getObjectField("cursor").getObjectField("firstBatch").isEmpty(), - str::stream() << "Expected a cursor to be present in the $collStats results: " - << collStatsResult.toString()); - invariant(collStatsResult.getObjectField("cursor").getIntField("id") == 0, - str::stream() << "Expected cursor ID to be 0: " << collStatsResult.toString()); - } else { - uassert( - 7463202, - str::stream() << "Expected a cursor to be present in the $collStats results: " - << collStatsResult.toString(), - !collStatsResult.getObjectField("cursor").isEmpty() && - !collStatsResult.getObjectField("cursor").getObjectField("firstBatch").isEmpty()); - uassert(7463203, - str::stream() << "Expected cursor ID to be 0: " << collStatsResult.toString(), - collStatsResult.getObjectField("cursor").getIntField("id") == 0); - } -} - -/** - * Log the $collStats results for 'nss' to provide additional debug information for validation - * failures. - */ -void logCollStats(OperationContext* opCtx, const NamespaceString& nss) { - DBDirectClient client(opCtx); - - BSONObj collStatsResult; - try { - // Run $collStats via aggregation. - client.runCommand(nss.db().toString(), - makeCollStatsCommand(nss.coll()), - collStatsResult /* command return results */); - // Logging $collStats information is best effort. If the collection doesn't exist, for - // example, then the $collStats query will fail and the failure reason will be logged. - uassertStatusOK(getStatusFromWriteCommandReply(collStatsResult)); - verifyCommandResponse(collStatsResult); - - LOGV2_OPTIONS(7463200, - logv2::LogTruncation::Disabled, - "Corrupt namespace $collStats results", - "namespace"_attr = nss, - "collStats"_attr = - collStatsResult.getObjectField("cursor").getObjectField("firstBatch")); - } catch (const DBException& ex) { - // Catch the error so that the validate error does not get overwritten by the attempt to add - // debug logging. - LOGV2_WARNING(7463201, - "Failed to fetch $collStats for validation error", - "namespace"_attr = nss, - "error"_attr = ex.toStatus()); - } -} - } // namespace /** @@ -210,7 +123,6 @@ public: const NamespaceString nss(CommandHelpers::parseNsCollectionRequired(dbname, cmdObj)); bool background = cmdObj["background"].trueValue(); - bool logDiagnostics = cmdObj["logDiagnostics"].trueValue(); // Background validation is not supported on the ephemeralForTest storage engine due to its // lack of support for timestamps. Switch the mode to foreground validation instead. @@ -342,20 +254,9 @@ public: PrepareConflictBehavior::kIgnoreConflictsAllowWrites); } - CollectionValidation::AdditionalOptions additionalOptions; - additionalOptions.validationVersion = getTestCommandsEnabled() - ? (ValidationVersion)bsonTestValidationVersion - : currentValidationVersion; - ValidateResults validateResults; - Status status = CollectionValidation::validate(opCtx, - nss, - mode, - repairMode, - additionalOptions, - &validateResults, - &result, - logDiagnostics); + Status status = + CollectionValidation::validate(opCtx, nss, mode, repairMode, &validateResults, &result); if (!status.isOK()) { return CommandHelpers::appendCommandStatusNoThrow(result, status); } @@ -366,7 +267,6 @@ public: result.append("advice", "A corrupt namespace has been detected. See " "http://dochub.mongodb.org/core/data-recovery for recovery steps."); - logCollStats(opCtx, nss); } return true; diff --git a/src/mongo/db/commands/validate_db_metadata_cmd.cpp b/src/mongo/db/commands/validate_db_metadata_cmd.cpp index 737bec2a1e7..c3a00ff74e6 100644 --- a/src/mongo/db/commands/validate_db_metadata_cmd.cpp +++ b/src/mongo/db/commands/validate_db_metadata_cmd.cpp @@ -149,10 +149,12 @@ public: return _validateView(opCtx, view); }); - for (auto&& coll : collectionCatalog->range(tenantDbName)) { + for (auto collIt = collectionCatalog->begin(opCtx, tenantDbName); + collIt != collectionCatalog->end(opCtx); + ++collIt) { if (!_validateNamespace( opCtx, - collectionCatalog->lookupNSSByUUID(opCtx, coll->uuid()).value())) { + collectionCatalog->lookupNSSByUUID(opCtx, collIt.uuid().get()).get())) { return; } } @@ -213,10 +215,8 @@ public: // Ensure there are no unstable indexes. const auto* indexCatalog = collection->getIndexCatalog(); - auto ii = indexCatalog->getIndexIterator( - opCtx, - IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished | - IndexCatalog::InclusionPolicy::kFrozen); + std::unique_ptr<IndexCatalog::IndexIterator> ii = + indexCatalog->getIndexIterator(opCtx, true /* includeUnfinishedIndexes */); while (ii->more()) { // Check if the index is allowed in API version 1. const IndexDescriptor* desc = ii->next()->descriptor(); diff --git a/src/mongo/db/commands/write_commands.cpp b/src/mongo/db/commands/write_commands.cpp index 7196c905f1b..0254baca47d 100644 --- a/src/mongo/db/commands/write_commands.cpp +++ b/src/mongo/db/commands/write_commands.cpp @@ -30,13 +30,11 @@ #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kDefault #include "mongo/base/checked_cast.h" -#include "mongo/base/error_codes.h" #include "mongo/bson/bsonobjbuilder.h" #include "mongo/bson/mutable/document.h" #include "mongo/bson/mutable/element.h" #include "mongo/db/catalog/collection_catalog.h" #include "mongo/db/catalog/collection_operation_source.h" -#include "mongo/db/catalog/collection_uuid_mismatch.h" #include "mongo/db/catalog/database_holder.h" #include "mongo/db/catalog/document_validation.h" #include "mongo/db/client.h" @@ -75,7 +73,6 @@ #include "mongo/db/timeseries/bucket_catalog.h" #include "mongo/db/timeseries/bucket_compression.h" #include "mongo/db/timeseries/timeseries_constants.h" -#include "mongo/db/timeseries/timeseries_extended_range.h" #include "mongo/db/timeseries/timeseries_options.h" #include "mongo/db/timeseries/timeseries_stats.h" #include "mongo/db/transaction_participant.h" @@ -139,26 +136,9 @@ bool isTimeseries(OperationContext* opCtx, const Request& request) { // collection does not yet exist, this check may return false unnecessarily. As a result, an // insert attempt into the time-series namespace will either succeed or fail, depending on who // wins the race. - // Hold reference to the catalog for collection lookup without locks to be safe. - auto catalog = CollectionCatalog::get(opCtx); - auto coll = catalog->lookupCollectionByNamespace(opCtx, bucketNss); - if (!coll) { - return false; - } - - if (auto options = coll->getTimeseriesOptions()) { - uassert(ErrorCodes::InvalidOptions, - "Time-series buckets collection is not clustered", - coll->isClustered()); - - uassert(ErrorCodes::InvalidOptions, - "Time-series buckets collection is missing bucketMaxSpanSeconds", - options->getBucketMaxSpanSeconds()); - - return true; - } - - return false; + return CollectionCatalog::get(opCtx) + ->lookupCollectionByNamespaceForRead(opCtx, bucketNss) + .get(); } NamespaceString makeTimeseriesBucketsNamespace(const NamespaceString& nss) { @@ -344,22 +324,6 @@ boost::optional<std::pair<Status, bool>> checkFailUnorderedTimeseriesInsertFailP return boost::none; } -boost::optional<write_ops::WriteError> generateErrorNoTenantMigration(OperationContext* opCtx, - const Status& status, - int index, - size_t numErrors) { - constexpr size_t kMaxErrorReasonsToReport = 1; - constexpr size_t kMaxErrorSizeToReportAfterMaxReasonsReached = 1024 * 1024; - - if (numErrors > kMaxErrorReasonsToReport) { - size_t errorSize = status.reason().size(); - if (errorSize > kMaxErrorSizeToReportAfterMaxReasonsReached) - return write_ops::WriteError(index, status.withReason("")); - } - - return write_ops::WriteError(index, status); -} - boost::optional<write_ops::WriteError> generateError(OperationContext* opCtx, const Status& status, int index, @@ -368,16 +332,18 @@ boost::optional<write_ops::WriteError> generateError(OperationContext* opCtx, return boost::none; } + boost::optional<Status> overwrittenStatus; + if (status == ErrorCodes::TenantMigrationConflict) { hangWriteBeforeWaitingForMigrationDecision.pauseWhileSet(opCtx); - Status overwrittenStatus = - tenant_migration_access_blocker::handleTenantMigrationConflict(opCtx, status); + overwrittenStatus.emplace( + tenant_migration_access_blocker::handleTenantMigrationConflict(opCtx, status)); // Interruption errors encountered during batch execution fail the entire batch, so throw on // such errors here for consistency. - if (ErrorCodes::isInterruption(overwrittenStatus)) { - uassertStatusOK(overwrittenStatus); + if (ErrorCodes::isInterruption(*overwrittenStatus)) { + uassertStatusOK(*overwrittenStatus); } // Tenant migration errors, similarly to migration errors consume too much space in the @@ -385,13 +351,25 @@ boost::optional<write_ops::WriteError> generateError(OperationContext* opCtx, // 'handleTenantMigrationConflict' above replaces the original status, we need to manually // truncate the new reason if the original 'status' was also truncated. if (status.reason().empty()) { - overwrittenStatus = overwrittenStatus.withReason(""); + overwrittenStatus = overwrittenStatus->withReason(""); } + } - return generateErrorNoTenantMigration(opCtx, overwrittenStatus, index, numErrors); + constexpr size_t kMaxErrorReasonsToReport = 1; + constexpr size_t kMaxErrorSizeToReportAfterMaxReasonsReached = 1024 * 1024; + + if (numErrors > kMaxErrorReasonsToReport) { + size_t errorSize = + overwrittenStatus ? overwrittenStatus->reason().size() : status.reason().size(); + if (errorSize > kMaxErrorSizeToReportAfterMaxReasonsReached) + overwrittenStatus = + overwrittenStatus ? overwrittenStatus->withReason("") : status.withReason(""); } - return generateErrorNoTenantMigration(opCtx, status, index, numErrors); + if (overwrittenStatus) + return write_ops::WriteError(index, std::move(*overwrittenStatus)); + else + return write_ops::WriteError(index, status); } template <typename T> @@ -435,7 +413,6 @@ void populateReply(OperationContext* opCtx, const auto& lastResult = result.results.back(); if (lastResult == ErrorCodes::StaleDbVersion || - lastResult == ErrorCodes::ShardCannotRefreshDueToLocksHeld || ErrorCodes::isStaleShardVersionError(lastResult.getStatus()) || ErrorCodes::isTenantMigrationError(lastResult.getStatus())) { // For ordered:false commands we need to duplicate these error results for all ops @@ -550,11 +527,6 @@ public: } write_ops::InsertCommandReply typedRun(OperationContext* opCtx) final try { - // On debug builds, verify that the estimated size of the insert command is at least as - // large as the size of the actual, serialized insert command. This ensures that the - // logic which estimates the size of insert commands is correct. - dassert(write_ops::verifySizeEstimate(request(), &unparsedRequest())); - transactionChecks(opCtx, ns()); if (request().getEncryptionInformation().has_value() && @@ -726,16 +698,16 @@ public: OperationSource::kTimeseriesInsert)); } - void _performTimeseriesBucketCompression( + TimeseriesSingleWriteResult _performTimeseriesBucketCompression( OperationContext* opCtx, const BucketCatalog::ClosedBucket& closedBucket) const { if (!feature_flags::gTimeseriesBucketCompression.isEnabled( serverGlobalParams.featureCompatibility)) { - return; + return {SingleWriteResult(), true}; } // Buckets with just a single measurement is not worth compressing. if (closedBucket.numMeasurements <= 1) { - return; + return {SingleWriteResult(), true}; } bool validateCompression = gValidateTimeseriesCompression.load(); @@ -773,8 +745,8 @@ public: auto compressionOp = _makeTimeseriesCompressionOp(opCtx, closedBucket.bucketId, bucketCompressionFunc); - auto result = _getTimeseriesSingleWriteResult(write_ops_exec::performUpdates( - opCtx, compressionOp, OperationSource::kTimeseriesBucketCompression)); + auto result = _getTimeseriesSingleWriteResult( + write_ops_exec::performUpdates(opCtx, compressionOp, OperationSource::kStandard)); // Report stats, if we fail before running the transform function then just skip // reporting. @@ -789,6 +761,8 @@ public: stats.onBucketClosed(*beforeSize, compressionStats); } } + + return result; } /** @@ -802,8 +776,7 @@ public: std::vector<write_ops::WriteError>* errors, boost::optional<repl::OpTime>* opTime, boost::optional<OID>* electionId, - std::vector<size_t>* docsToRetry, - absl::flat_hash_map<int, int>& retryAttemptsForDup) const try { + std::vector<size_t>* docsToRetry) const try { auto& bucketCatalog = BucketCatalog::get(opCtx); auto metadata = bucketCatalog.getMetadata(batch->bucket()); @@ -823,18 +796,9 @@ public: _performTimeseriesInsert(opCtx, batch, metadata, std::move(stmtIds)); if (auto error = generateError(opCtx, output.result, start + index, errors->size())) { - bool canContinue = output.canContinue; - // Automatically attempts to retry on DuplicateKey error. - if (error->getStatus().code() == ErrorCodes::DuplicateKey && - retryAttemptsForDup[index]++ < - gTimeseriesInsertMaxRetriesOnDuplicates.load()) { - docsToRetry->push_back(index); - canContinue = true; - } else { - errors->emplace_back(std::move(*error)); - } - BucketCatalog::get(opCtx).abort(batch, output.result.getStatus()); - return canContinue; + errors->emplace_back(std::move(*error)); + bucketCatalog.abort(batch, output.result.getStatus()); + return output.canContinue; } invariant(output.result.getValue().getN() == 1, @@ -864,7 +828,12 @@ public: if (closedBucket) { // If this write closed a bucket, compress the bucket - _performTimeseriesBucketCompression(opCtx, *closedBucket); + auto output = _performTimeseriesBucketCompression(opCtx, *closedBucket); + if (auto error = + generateError(opCtx, output.result, start + index, errors->size())) { + errors->emplace_back(std::move(*error)); + return output.canContinue; + } } return true; } catch (const DBException& ex) { @@ -872,12 +841,19 @@ public: throw; } - bool _commitTimeseriesBucketsAtomically(OperationContext* opCtx, - TimeseriesBatches* batches, - TimeseriesStmtIds&& stmtIds, - std::vector<write_ops::WriteError>* errors, - boost::optional<repl::OpTime>* opTime, - boost::optional<OID>* electionId) const { + enum struct TimeseriesAtomicWriteResult { + kSuccess, + kContinuableError, + kNonContinuableError, + }; + + TimeseriesAtomicWriteResult _commitTimeseriesBucketsAtomically( + OperationContext* opCtx, + TimeseriesBatches* batches, + TimeseriesStmtIds&& stmtIds, + std::vector<write_ops::WriteError>* errors, + boost::optional<repl::OpTime>* opTime, + boost::optional<OID>* electionId) const { auto& bucketCatalog = BucketCatalog::get(opCtx); std::vector<std::reference_wrapper<std::shared_ptr<BucketCatalog::WriteBatch>>> @@ -890,7 +866,7 @@ public: } if (batchesToCommit.empty()) { - return true; + return TimeseriesAtomicWriteResult::kSuccess; } // Sort by bucket so that preparing the commit for each batch cannot deadlock. @@ -916,7 +892,7 @@ public: auto prepareCommitStatus = bucketCatalog.prepareCommit(batch); if (!prepareCommitStatus.isOK()) { abortStatus = prepareCommitStatus; - return false; + return TimeseriesAtomicWriteResult::kContinuableError; } if (batch.get()->numPreviouslyCommittedMeasurements() == 0) { @@ -933,26 +909,33 @@ public: auto result = write_ops_exec::performAtomicTimeseriesWrites(opCtx, insertOps, updateOps); if (!result.isOK()) { - if (result.code() == ErrorCodes::DuplicateKey) { - BucketCatalog::get(opCtx).resetBucketOIDCounter(); - } abortStatus = result; - return false; + return TimeseriesAtomicWriteResult::kContinuableError; } getOpTimeAndElectionId(opCtx, opTime, electionId); + bool compressClosedBuckets = true; for (auto batch : batchesToCommit) { auto closedBucket = bucketCatalog.finish( batch, BucketCatalog::CommitInfo{*opTime, *electionId}); batch.get().reset(); - if (!closedBucket) { + if (!closedBucket || !compressClosedBuckets) { continue; } // If this write closed a bucket, compress the bucket - _performTimeseriesBucketCompression(opCtx, *closedBucket); + auto ret = _performTimeseriesBucketCompression(opCtx, *closedBucket); + if (!ret.result.isOK()) { + // Don't try to compress any other buckets if we fail. We're not allowed to + // do more write operations. + compressClosedBuckets = false; + } + if (!ret.canContinue) { + abortStatus = ret.result.getStatus(); + return TimeseriesAtomicWriteResult::kNonContinuableError; + } } } catch (const DBException& ex) { abortStatus = ex.toStatus(); @@ -960,7 +943,7 @@ public: } batchGuard.dismiss(); - return true; + return TimeseriesAtomicWriteResult::kSuccess; } // For sharded time-series collections, we need to use the granularity from the config @@ -985,7 +968,10 @@ public: } } - std::tuple<TimeseriesBatches, TimeseriesStmtIds, size_t /* numInserted */> + std::tuple<TimeseriesBatches, + TimeseriesStmtIds, + size_t /* numInserted */, + bool /* canContinue */> _insertIntoBucketCatalog(OperationContext* opCtx, size_t start, size_t numDocs, @@ -1025,6 +1011,7 @@ public: TimeseriesBatches batches; TimeseriesStmtIds stmtIds; + bool canContinue = true; auto insert = [&](size_t index) { invariant(start + index < request().getDocuments().size()); @@ -1070,63 +1057,58 @@ public: // If this insert closed buckets, rewrite to be a compressed column. If we cannot // perform write operations at this point the bucket will be left uncompressed. for (const auto& closedBucket : result.getValue().closedBuckets) { + if (!canContinue) { + break; + } + // If this write closed a bucket, compress the bucket - _performTimeseriesBucketCompression(opCtx, closedBucket); + auto ret = _performTimeseriesBucketCompression(opCtx, closedBucket); + if (auto error = + generateError(opCtx, ret.result, start + index, errors->size())) { + // Bucket compression only fail when we may not try to perform any other + // write operation. When handleError() inside write_ops_exec.cpp return + // false. + errors->emplace_back(std::move(*error)); + canContinue = false; + return false; + } + canContinue = ret.canContinue; } return true; }; - try { - if (!indices.empty()) { - std::for_each(indices.begin(), indices.end(), insert); - } else { - for (size_t i = 0; i < numDocs; i++) { - if (!insert(i) && request().getOrdered()) { - return {std::move(batches), std::move(stmtIds), i}; - } + if (!indices.empty()) { + std::for_each(indices.begin(), indices.end(), insert); + } else { + for (size_t i = 0; i < numDocs; i++) { + if (!insert(i) && request().getOrdered()) { + return {std::move(batches), std::move(stmtIds), i, canContinue}; } } - } catch (const DBException& ex) { - // Exception insert into bucket catalog, append error and wait for all batches that - // we've already managed to write into to commit or abort. We need to wait here as - // pointers to memory owned by this command is stored in the WriteBatch(es). This - // ensures that no other thread may try to access this memory after this command has - // been torn down due to the exception. - - boost::optional<repl::OpTime> opTime; - boost::optional<OID> electionId; - std::vector<size_t> docsToRetry; - errors->emplace_back( - *generateErrorNoTenantMigration(opCtx, ex.toStatus(), 0, errors->size())); - - _getTimeseriesBatchResultsNoTenantMigration( - opCtx, batches, 0, -1, false, errors, &opTime, &electionId, &docsToRetry); - throw; } - - return {std::move(batches), std::move(stmtIds), request().getDocuments().size()}; + return {std::move(batches), + std::move(stmtIds), + request().getDocuments().size(), + canContinue}; } - template <typename ErrorGenerator> - void _getTimeseriesBatchResultsBase(ErrorGenerator&& errorGenerator, - OperationContext* opCtx, - const TimeseriesBatches& batches, - int64_t start, - int64_t indexOfLastProcessedBatch, - bool canContinue, - std::vector<write_ops::WriteError>* errors, - boost::optional<repl::OpTime>* opTime, - boost::optional<OID>* electionId, - std::vector<size_t>* docsToRetry = nullptr) const { + void _getTimeseriesBatchResults(OperationContext* opCtx, + const TimeseriesBatches& batches, + size_t start, + size_t indexOfLastProcessedBatch, + bool canContinue, + std::vector<write_ops::WriteError>* errors, + boost::optional<repl::OpTime>* opTime, + boost::optional<OID>* electionId, + std::vector<size_t>* docsToRetry = nullptr) const { boost::optional<write_ops::WriteError> lastError; if (!errors->empty()) { lastError = errors->back(); } - invariant(indexOfLastProcessedBatch == (int64_t)batches.size() || lastError); - for (int64_t itr = 0, size = batches.size(); itr < size; ++itr) { + for (size_t itr = 0; itr < batches.size(); ++itr) { const auto& [batch, index] = batches[itr]; if (!batch) { continue; @@ -1142,11 +1124,11 @@ public: auto swCommitInfo = batch->getResult(); if (swCommitInfo.getStatus() == ErrorCodes::TimeseriesBucketCleared) { - invariant(docsToRetry, "the 'docsToRetry' cannot be null"); + tassert(6023102, "the 'docsToRetry' cannot be null", docsToRetry); docsToRetry->push_back(index); continue; } - if (auto error = errorGenerator( + if (auto error = generateError( opCtx, swCommitInfo.getStatus(), start + index, errors->size())) { errors->emplace_back(std::move(*error)); continue; @@ -1171,76 +1153,30 @@ public: } } - void _getTimeseriesBatchResults(OperationContext* opCtx, - const TimeseriesBatches& batches, - int64_t start, - int64_t indexOfLastProcessedBatch, - bool canContinue, - std::vector<write_ops::WriteError>* errors, - boost::optional<repl::OpTime>* opTime, - boost::optional<OID>* electionId, - std::vector<size_t>* docsToRetry = nullptr) const { - auto errorGenerator = - [](OperationContext* opCtx, const Status& status, int index, size_t numErrors) { - return generateError(opCtx, status, index, numErrors); - }; - _getTimeseriesBatchResultsBase(errorGenerator, - opCtx, - batches, - start, - indexOfLastProcessedBatch, - canContinue, - errors, - opTime, - electionId, - docsToRetry); - } - - void _getTimeseriesBatchResultsNoTenantMigration( + TimeseriesAtomicWriteResult _performOrderedTimeseriesWritesAtomically( OperationContext* opCtx, - const TimeseriesBatches& batches, - int64_t start, - int64_t indexOfLastProcessedBatch, - bool canContinue, std::vector<write_ops::WriteError>* errors, boost::optional<repl::OpTime>* opTime, boost::optional<OID>* electionId, - std::vector<size_t>* docsToRetry = nullptr) const { - auto errorGenerator = - [](OperationContext* opCtx, const Status& status, int index, size_t numErrors) { - return generateErrorNoTenantMigration(opCtx, status, index, numErrors); - }; - _getTimeseriesBatchResultsBase(errorGenerator, - opCtx, - batches, - start, - indexOfLastProcessedBatch, - canContinue, - errors, - opTime, - electionId, - docsToRetry); - } - - bool _performOrderedTimeseriesWritesAtomically(OperationContext* opCtx, - std::vector<write_ops::WriteError>* errors, - boost::optional<repl::OpTime>* opTime, - boost::optional<OID>* electionId, - bool* containsRetry) const { - auto [batches, stmtIds, numInserted] = _insertIntoBucketCatalog( + bool* containsRetry) const { + auto [batches, stmtIds, numInserted, canContinue] = _insertIntoBucketCatalog( opCtx, 0, request().getDocuments().size(), {}, errors, containsRetry); + if (!canContinue) { + return TimeseriesAtomicWriteResult::kNonContinuableError; + } hangTimeseriesInsertBeforeCommit.pauseWhileSet(); - if (!_commitTimeseriesBucketsAtomically( - opCtx, &batches, std::move(stmtIds), errors, opTime, electionId)) { - return false; + auto result = _commitTimeseriesBucketsAtomically( + opCtx, &batches, std::move(stmtIds), errors, opTime, electionId); + if (result != TimeseriesAtomicWriteResult::kSuccess) { + return result; } _getTimeseriesBatchResults( opCtx, batches, 0, batches.size(), true, errors, opTime, electionId); - return true; + return TimeseriesAtomicWriteResult::kSuccess; } /** @@ -1251,9 +1187,19 @@ public: boost::optional<repl::OpTime>* opTime, boost::optional<OID>* electionId, bool* containsRetry) const { - if (_performOrderedTimeseriesWritesAtomically( - opCtx, errors, opTime, electionId, containsRetry)) { - return request().getDocuments().size(); + auto result = _performOrderedTimeseriesWritesAtomically( + opCtx, errors, opTime, electionId, containsRetry); + switch (result) { + case TimeseriesAtomicWriteResult::kSuccess: + return request().getDocuments().size(); + case TimeseriesAtomicWriteResult::kNonContinuableError: + // If we can't continue, we know that 0 were inserted since this function should + // guarantee that the inserts are atomic. + return 0; + case TimeseriesAtomicWriteResult::kContinuableError: + break; + default: + MONGO_UNREACHABLE; } for (size_t i = 0; i < request().getDocuments().size(); ++i) { @@ -1272,8 +1218,6 @@ public: * which were attempted in an update operation, but found no bucket to update. These indices * can be passed as the 'indices' parameter in a subsequent call to this function, in order * to to be retried. - * In rare cases due to collision from OID generation, we will also retry inserting those - * bucket * documents for a limited number of times. */ std::vector<size_t> _performUnorderedTimeseriesWrites( OperationContext* opCtx, @@ -1283,73 +1227,39 @@ public: std::vector<write_ops::WriteError>* errors, boost::optional<repl::OpTime>* opTime, boost::optional<OID>* electionId, - bool* containsRetry, - absl::flat_hash_map<int, int>& retryAttemptsForDup) const { - auto [batches, bucketStmtIds, _] = + bool* containsRetry) const { + auto [batches, bucketStmtIds, _, canContinue] = _insertIntoBucketCatalog(opCtx, start, numDocs, indices, errors, containsRetry); hangTimeseriesInsertBeforeCommit.pauseWhileSet(); std::vector<size_t> docsToRetry; - bool canContinue = true; - - - stdx::unordered_set<BucketCatalog::WriteBatch*> handledHere; - int64_t handledElsewhere = 0; - auto guard = ScopeGuard([this, &handledElsewhere, opCtx]() { - if (handledElsewhere > 0) { - auto& bucketCatalog = BucketCatalog::get(opCtx); - bucketCatalog.reportMeasurementsGroupCommitted(request().getNamespace(), - handledElsewhere); - } - }); + if (!canContinue) { + return docsToRetry; + } size_t itr = 0; for (; itr < batches.size(); ++itr) { auto& [batch, index] = batches[itr]; if (batch->claimCommitRights()) { - handledHere.insert(batch.get()); auto stmtIds = isTimeseriesWriteRetryable(opCtx) ? std::move(bucketStmtIds[batch->bucket().id]) : std::vector<StmtId>{}; - try { - canContinue = _commitTimeseriesBucket(opCtx, - batch, - start, - index, - std::move(stmtIds), - errors, - opTime, - electionId, - &docsToRetry, - retryAttemptsForDup); - } catch (const DBException& ex) { - // Exception during commit, append error and wait for all our batches to - // commit or - // abort. We need to wait here as pointers to memory owned by this command - // is stored in the WriteBatch(es). This ensures that no other thread may - // try to access this memory after this command has been torn down due to - // the exception. - errors->emplace_back(*generateErrorNoTenantMigration( - opCtx, ex.toStatus(), start + index, errors->size())); - _getTimeseriesBatchResultsNoTenantMigration(opCtx, - batches, - 0, - itr, - canContinue, - errors, - opTime, - electionId, - &docsToRetry); - throw; - } + + canContinue = _commitTimeseriesBucket(opCtx, + batch, + start, + index, + std::move(stmtIds), + errors, + opTime, + electionId, + &docsToRetry); batch.reset(); if (!canContinue) { break; } - } else if (!handledHere.contains(batch.get())) { - ++handledElsewhere; } } @@ -1370,20 +1280,9 @@ public: boost::optional<OID>* electionId, bool* containsRetry) const { std::vector<size_t> docsToRetry; - absl::flat_hash_map<int, int> retryAttemptsForDup; do { - docsToRetry = _performUnorderedTimeseriesWrites(opCtx, - start, - numDocs, - docsToRetry, - errors, - opTime, - electionId, - containsRetry, - retryAttemptsForDup); - if (!retryAttemptsForDup.empty()) { - BucketCatalog::get(opCtx).resetBucketOIDCounter(); - } + docsToRetry = _performUnorderedTimeseriesWrites( + opCtx, start, numDocs, docsToRetry, errors, opTime, electionId, containsRetry); } while (!docsToRetry.empty()); } @@ -1403,11 +1302,6 @@ public: curOp.getReadWriteType()); }); - // If an expected collection UUID is provided, always fail because the user-facing - // time-series namespace does not have a UUID. - checkCollectionUUIDMismatch( - opCtx, request().getNamespace(), nullptr, request().getCollectionUUID()); - uassert( ErrorCodes::OperationNotSupportedInTransaction, str::stream() << "Cannot insert into a time-series collection in a multi-document " @@ -1547,29 +1441,19 @@ public: invariant(!_commandObj.isEmpty()); + if (const auto& shardVersion = _commandObj.getField("shardVersion"); + !shardVersion.eoo()) { + bob->append(shardVersion); + } bob->append("find", _commandObj["update"].String()); extractQueryDetails(_updateOpObj, bob); bob->append("batchSize", 1); bob->append("singleBatch", true); - - if (const auto& shardVersion = _commandObj.getField("shardVersion"); - !shardVersion.eoo()) { - bob->append(shardVersion); - } - if (const auto& databaseVersion = _commandObj.getField("databaseVersion"); - !databaseVersion.eoo()) { - bob->append(databaseVersion); - } } write_ops::UpdateCommandReply typedRun(OperationContext* opCtx) final try { - // On debug builds, verify that the estimated size of the update command is at least as - // large as the size of the actual, serialized update command. This ensures that the - // logic which estimates the size of update commands is correct. - dassert(write_ops::verifySizeEstimate(request(), &unparsedRequest())); transactionChecks(opCtx, ns()); - write_ops::UpdateCommandReply updateReply; OperationSource source = OperationSource::kStandard; @@ -1676,7 +1560,6 @@ public: updateRequest.setLegacyRuntimeConstants(request().getLegacyRuntimeConstants().value_or( Variables::generateRuntimeConstants(opCtx))); updateRequest.setLetParameters(request().getLet()); - updateRequest.setBypassEmptyTsReplacement(request().getBypassEmptyTsReplacement()); updateRequest.setYieldPolicy(PlanYieldPolicy::YieldPolicy::YIELD_AUTO); updateRequest.setExplain(verbosity); @@ -1758,11 +1641,6 @@ public: } write_ops::DeleteCommandReply typedRun(OperationContext* opCtx) final try { - // On debug builds, verify that the estimated size of the deletes are at least as large - // as the actual, serialized size. This ensures that the logic that estimates the size - // of deletes for batch writes is correct. - dassert(write_ops::verifySizeEstimate(request(), &unparsedRequest())); - transactionChecks(opCtx, ns()); write_ops::DeleteCommandReply deleteReply; OperationSource source = OperationSource::kStandard; |