1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
|
/**
* Copyright (C) 2018-present MongoDB, Inc.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the Server Side Public License, version 1,
* as published by MongoDB, Inc.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* Server Side Public License for more details.
*
* You should have received a copy of the Server Side Public License
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*
* As a special exception, the copyright holders give permission to link the
* code of portions of this program with the OpenSSL library under certain
* conditions as described in each individual source file and distribute
* linked combinations including the program with the OpenSSL library. You
* must comply with the Server Side Public License in all respects for
* all of the code used other than as permitted herein. If you modify file(s)
* with this exception, you may extend this exception to your version of the
* file(s), but you are not obligated to do so. If you do not wish to do so,
* delete this exception statement from your version. If you delete this
* exception statement from all source files in the program, then also delete
* it in the license file.
*/
#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kQuery
#include "mongo/platform/basic.h"
#include "mongo/db/auth/authorization_checks.h"
#include "mongo/db/auth/authorization_session.h"
#include "mongo/db/catalog/collection_uuid_mismatch.h"
#include "mongo/db/client.h"
#include "mongo/db/clientcursor.h"
#include "mongo/db/commands.h"
#include "mongo/db/commands/run_aggregate.h"
#include "mongo/db/commands/test_commands_enabled.h"
#include "mongo/db/cursor_manager.h"
#include "mongo/db/db_raii.h"
#include "mongo/db/exec/disk_use_options_gen.h"
#include "mongo/db/exec/working_set_common.h"
#include "mongo/db/fle_crud.h"
#include "mongo/db/matcher/extensions_callback_real.h"
#include "mongo/db/pipeline/aggregation_request_helper.h"
#include "mongo/db/pipeline/variables.h"
#include "mongo/db/query/collation/collator_factory_interface.h"
#include "mongo/db/query/cursor_response.h"
#include "mongo/db/query/explain.h"
#include "mongo/db/query/find.h"
#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/repl/replication_coordinator.h"
#include "mongo/db/service_context.h"
#include "mongo/db/stats/counters.h"
#include "mongo/db/stats/resource_consumption_metrics.h"
#include "mongo/db/stats/server_read_concern_metrics.h"
#include "mongo/db/storage/storage_engine.h"
#include "mongo/db/storage/storage_parameters_gen.h"
#include "mongo/db/transaction_participant.h"
#include "mongo/logv2/log.h"
#include "mongo/rpc/get_status_from_command_result.h"
#include "mongo/util/fail_point.h"
namespace mongo {
namespace {
MONGO_FAIL_POINT_DEFINE(allowExternalReadsForReverseOplogScanRule);
const auto kTermField = "term"_sd;
// Older client code before FCV 5.1 could still send 'ntoreturn' to mean a 'limit' or 'batchSize'.
// This helper translates an 'ntoreturn' to an appropriate combination of 'limit', 'singleBatch',
// and 'batchSize'. The translation rules are as follows: when 'ntoreturn' < 0 we have 'singleBatch'
// semantics so we set 'limit= -ntoreturn' and 'singleBatch=true' flag to 'true'. When 'ntoreturn'
// == 1 this is a limit-1 case so we set the 'limit' accordingly. Lastly when 'ntoreturn' > 1, it's
// interpreted as a 'batchSize'. When the cluster is fully upgraded to v5.1 'ntoreturn' should no
// longer be used on incoming requests.
//
// This helper can be removed when we no longer have to interoperate with 5.0 or older nodes.
// longer have to interoperate with 5.0 or older nodes.
std::unique_ptr<FindCommandRequest> translateNtoReturnToLimitOrBatchSize(
std::unique_ptr<FindCommandRequest> findCmd) {
const auto& fcvState = serverGlobalParams.featureCompatibility;
uassert(5746102,
"the ntoreturn find command parameter is not supported when FCV >= 5.1",
!(fcvState.isVersionInitialized() &&
fcvState.isGreaterThanOrEqualTo(
multiversion::FeatureCompatibilityVersion::kVersion_5_1) &&
findCmd->getNtoreturn()));
const auto ntoreturn = findCmd->getNtoreturn().get_value_or(0);
if (ntoreturn) {
if (ntoreturn < 0) {
uassert(5746100,
"bad ntoreturn value in query",
ntoreturn != std::numeric_limits<int>::min());
findCmd->setLimit(-ntoreturn);
findCmd->setSingleBatch(true);
} else if (ntoreturn == 1) {
findCmd->setLimit(1);
} else {
findCmd->setBatchSize(ntoreturn);
}
}
findCmd->setNtoreturn(boost::none);
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,
boost::optional<ExplainOptions::Verbosity> verbosity) {
std::unique_ptr<CollatorInterface> collator;
if (!findCommand.getCollation().isEmpty()) {
collator = uassertStatusOK(CollatorFactoryInterface::get(opCtx->getServiceContext())
->makeFromBSON(findCommand.getCollation()));
}
// 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
);
expCtx->tempDir = storageGlobalParams.dbpath + "/_tmp";
expCtx->startExpressionCounters();
return expCtx;
}
/**
* Fills out the CurOp for "opCtx" with information about this query.
*/
void beginQueryOp(OperationContext* opCtx, const NamespaceString& nss, const BSONObj& queryObj) {
auto curOp = CurOp::get(opCtx);
stdx::lock_guard<Client> lk(*opCtx->getClient());
curOp->setOpDescription_inlock(queryObj);
curOp->setNS_inlock(nss.ns());
}
/**
* A command for running .find() queries.
*/
class FindCmd final : public Command {
public:
FindCmd() : Command("find") {}
const std::set<std::string>& apiVersions() const {
return kApiVersions1;
}
std::unique_ptr<CommandInvocation> parse(OperationContext* opCtx,
const OpMsgRequest& opMsgRequest) override {
// TODO: Parse into a QueryRequest here.
return std::make_unique<Invocation>(this, opMsgRequest, opMsgRequest.getDatabase());
}
AllowedOnSecondary secondaryAllowed(ServiceContext* context) const override {
return AllowedOnSecondary::kOptIn;
}
bool maintenanceOk() const override {
return false;
}
bool adminOnly() const override {
return false;
}
std::string help() const override {
return "query for documents";
}
LogicalOp getLogicalOp() const override {
return LogicalOp::opQuery;
}
ReadWriteType getReadWriteType() const override {
return ReadWriteType::kRead;
}
std::size_t reserveBytesForReply() const override {
return FindCommon::kInitReplyBufferSize;
}
/**
* A find command does not increment the command counter, but rather increments the
* query counter.
*/
bool shouldAffectCommandCounter() const override {
return false;
}
bool shouldAffectReadConcernCounter() const override {
return true;
}
bool collectsResourceConsumptionMetrics() const override {
return true;
}
class Invocation final : public CommandInvocation {
public:
Invocation(const FindCmd* definition, const OpMsgRequest& request, StringData dbName)
: CommandInvocation(definition), _request(request), _dbName(dbName) {
invariant(_request.body.isOwned());
}
private:
bool supportsWriteConcern() const override {
return false;
}
ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
bool isImplicitDefault) const final {
return ReadConcernSupportResult::allSupportedAndDefaultPermitted();
}
bool supportsReadMirroring() const override {
return true;
}
bool canIgnorePrepareConflicts() const override {
return true;
}
bool allowsSpeculativeMajorityReads() const override {
// Find queries are only allowed to use speculative behavior if the 'allowsSpeculative'
// flag is passed. The find command will check for this flag internally and fail if
// necessary.
return true;
}
NamespaceString ns() const override {
// TODO get the ns from the parsed QueryRequest.
return NamespaceString(CommandHelpers::parseNsFromCommand(_dbName, _request.body));
}
void doCheckAuthorization(OperationContext* opCtx) const final {
AuthorizationSession* authSession = AuthorizationSession::get(opCtx->getClient());
uassert(ErrorCodes::Unauthorized,
"Unauthorized",
authSession->isAuthorizedToParseNamespaceElement(_request.body.firstElement()));
const auto hasTerm = _request.body.hasField(kTermField);
uassertStatusOK(auth::checkAuthForFind(
authSession,
CollectionCatalog::get(opCtx)->resolveNamespaceStringOrUUID(
opCtx, CommandHelpers::parseNsOrUUID(_dbName, _request.body)),
hasTerm));
}
void explain(OperationContext* opCtx,
ExplainOptions::Verbosity verbosity,
rpc::ReplyBuilderInterface* result) override {
// Acquire locks. The RAII object is optional, because in the case of a view, the locks
// need to be released.
boost::optional<AutoGetCollectionForReadCommandMaybeLockFree> ctx;
ctx.emplace(opCtx,
CommandHelpers::parseNsCollectionRequired(_dbName, _request.body),
AutoGetCollectionViewMode::kViewsPermitted);
const auto nss = ctx->getNss();
// Parse the command BSON to a FindCommandRequest.
auto findCommand = parseCmdObjectToFindCommandRequest(opCtx, nss, _request.body);
// Finish the parsing step by using the FindCommandRequest to create a CanonicalQuery.
const ExtensionsCallbackReal extensionsCallback(opCtx, &nss);
auto expCtx = makeExpressionContext(opCtx, *findCommand, verbosity);
const bool isExplain = true;
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.
if (ctx->getView() ||
(feature_flags::gfeatureFlagCommonQueryFramework.isEnabled(
serverGlobalParams.featureCompatibility) &&
internalQueryEnableCascadesOptimizer.load())) {
// Relinquish locks. The aggregation command will re-acquire them.
ctx.reset();
// Convert the find command into an aggregation using $match (and other stages, as
// necessary), if possible.
const auto& findCommand = cq->getFindCommandRequest();
auto viewAggregationCommand =
uassertStatusOK(query_request_helper::asAggregationCommand(findCommand));
auto viewAggCmd = OpMsgRequest::fromDBAndBody(_dbName, viewAggregationCommand).body;
// Create the agg request equivalent of the find operation, with the explain
// verbosity included.
auto aggRequest = aggregation_request_helper::parseFromBSON(
opCtx,
nss,
viewAggCmd,
verbosity,
APIParameters::get(opCtx).getAPIStrict().value_or(false));
try {
// An empty PrivilegeVector is acceptable because these privileges are only
// checked on getMore and explain will not open a cursor.
uassertStatusOK(runAggregate(
opCtx, nss, aggRequest, viewAggCmd, PrivilegeVector(), result));
} catch (DBException& error) {
if (error.code() == ErrorCodes::InvalidPipelineOperator) {
uasserted(ErrorCodes::InvalidPipelineOperator,
str::stream()
<< "Unsupported in view pipeline: " << error.what());
}
throw;
}
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 =
uassertStatusOK(getExecutorFind(opCtx,
&collection,
std::move(cq),
nullptr /* extractAndAttachPipelineStages */,
permitYield));
auto bodyBuilder = result->getBodyBuilder();
// Got the execution tree. Explain it.
Explain::explainStages(
exec.get(), collection, verbosity, BSONObj(), _request.body, &bodyBuilder);
}
/**
* Runs a query using the following steps:
* --Parsing.
* --Acquire locks.
* --Plan query, obtaining an executor that can run it.
* --Generate the first batch.
* --Save state for getMore, transferring ownership of the executor to a ClientCursor.
* --Generate response to send to the client.
*/
void run(OperationContext* opCtx, rpc::ReplyBuilderInterface* result) {
CommandHelpers::handleMarkKillOnClientDisconnect(opCtx);
// Although it is a command, a find command gets counted as a query.
globalOpCounters.gotQuery();
const BSONObj& cmdObj = _request.body;
// 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);
// Only allow speculative majority for internal commands that specify the correct flag.
uassert(ErrorCodes::ReadConcernMajorityNotEnabled,
"Majority read concern is not enabled.",
!(repl::ReadConcernArgs::get(opCtx).isSpeculativeMajority() &&
!findCommand->getAllowSpeculativeMajorityRead()));
uassert(ErrorCodes::InvalidOptions,
"When using the find command by UUID, the collectionUUID parameter cannot also "
"be specified",
!findCommand->getNamespaceOrUUID().uuid() || !findCommand->getCollectionUUID());
auto replCoord = repl::ReplicationCoordinator::get(opCtx);
const auto txnParticipant = TransactionParticipant::get(opCtx);
uassert(ErrorCodes::InvalidOptions,
"It is illegal to open a tailable cursor in a transaction",
!(opCtx->inMultiDocumentTransaction() && findCommand->getTailable()));
uassert(ErrorCodes::OperationNotSupportedInTransaction,
"The 'readOnce' option is not supported within a transaction.",
!txnParticipant || !opCtx->inMultiDocumentTransaction() ||
!findCommand->getReadOnce());
// Validate term before acquiring locks, if provided.
auto term = findCommand->getTerm();
if (term) {
// Note: updateTerm returns ok if term stayed the same.
uassertStatusOK(replCoord->updateTerm(opCtx, *term));
}
// The presence of a term in the request indicates that this is an internal replication
// oplog read request.
if (term && isOplogNss) {
// We do not want to take tickets for internal (replication) oplog reads. Stalling
// on ticket acquisition can cause complicated deadlocks. Primaries may depend on
// data reaching secondaries in order to proceed; and secondaries may get stalled
// replicating because of an inability to acquire a read ticket.
opCtx->lockState()->skipAcquireTicket();
}
// If this read represents a reverse oplog scan, we want to bypass oplog visibility
// rules in the case of secondaries. We normally only read from these nodes at batch
// boundaries, but in this specific case we should fetch all new entries, to be
// consistent with any catalog changes that might be observable before the batch is
// finalized. This special rule for reverse oplog scans is needed by replication
// initial sync, for the purposes of calculating the stopTimestamp correctly.
boost::optional<PinReadSourceBlock> pinReadSourceBlock;
if (isOplogNss) {
auto reverseScan = false;
auto cmdSort = findCommand->getSort();
if (!cmdSort.isEmpty()) {
BSONElement natural = cmdSort[query_request_helper::kNaturalSortField];
if (natural) {
reverseScan = natural.safeNumberInt() < 0;
}
}
auto isInternal = (opCtx->getClient()->session() &&
(opCtx->getClient()->session()->getTags() &
transport::Session::kInternalClient));
if (MONGO_unlikely(allowExternalReadsForReverseOplogScanRule.shouldFail())) {
isInternal = true;
}
if (reverseScan && isInternal) {
pinReadSourceBlock.emplace(opCtx->recoveryUnit());
}
}
// Acquire locks. If the query is on a view, we release our locks and convert the query
// request into an aggregation command.
boost::optional<AutoGetCollectionForReadCommandMaybeLockFree> ctx;
ctx.emplace(opCtx,
CommandHelpers::parseNsOrUUID(_dbName, _request.body),
AutoGetCollectionViewMode::kViewsPermitted);
const auto& nss = ctx->getNss();
uassert(ErrorCodes::NamespaceNotFound,
str::stream() << "UUID " << findCommand->getNamespaceOrUUID().uuid().get()
<< " specified in query request not found",
ctx || !findCommand->getNamespaceOrUUID().uuid());
checkCollectionUUIDMismatch(
opCtx, nss, ctx->getCollection(), findCommand->getCollectionUUID());
// Set the namespace if a collection was found, as opposed to nothing or a view.
if (ctx) {
query_request_helper::refreshNSS(ctx->getNss(), findCommand.get());
}
// Tailing a replicated capped clustered collection requires majority read concern.
const auto coll = ctx->getCollection().get();
if (coll) {
const bool isTailable = findCommand->getTailable();
const bool isMajorityReadConcern = repl::ReadConcernArgs::get(opCtx).getLevel() ==
repl::ReadConcernLevel::kMajorityReadConcern;
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 "
"read concern",
isMajorityReadConcern);
}
}
// Fill out curop information.
beginQueryOp(opCtx, nss, _request.body);
// 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.
if (ctx->getView() ||
(feature_flags::gfeatureFlagCommonQueryFramework.isEnabled(
serverGlobalParams.featureCompatibility) &&
internalQueryEnableCascadesOptimizer.load())) {
// Relinquish locks. The aggregation command will re-acquire them.
ctx.reset();
// Convert the find command into an aggregation using $match (and other stages, as
// necessary), if possible.
const auto& findCommand = cq->getFindCommandRequest();
auto viewAggregationCommand =
uassertStatusOK(query_request_helper::asAggregationCommand(findCommand));
BSONObj aggResult = CommandHelpers::runCommandDirectly(
opCtx, OpMsgRequest::fromDBAndBody(_dbName, std::move(viewAggregationCommand)));
auto status = getStatusFromCommandResult(aggResult);
if (status.code() == ErrorCodes::InvalidPipelineOperator) {
uasserted(ErrorCodes::InvalidPipelineOperator,
str::stream() << "Unsupported in view pipeline: " << status.reason());
}
uassertStatusOK(status);
result->getBodyBuilder().appendElements(aggResult);
return;
}
// Check whether we are allowed to read from this node after acquiring our locks.
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.
opCtx->recoveryUnit()->setReadOnce(true);
}
// Get the execution plan for the query.
bool permitYield = true;
auto exec =
uassertStatusOK(getExecutorFind(opCtx,
&collection,
std::move(cq),
nullptr /* extractAndAttachPipelineStages */,
permitYield));
// If the executor supports it, find operations will maintain the storage engine state
// across commands.
if (gYieldingSupportForSBE && !opCtx->inMultiDocumentTransaction() &&
repl::ReadConcernArgs::get(opCtx).getLevel() !=
repl::ReadConcernLevel::kSnapshotReadConcern) {
exec->enableSaveRecoveryUnitAcrossCommandsIfSupported();
}
{
stdx::lock_guard<Client> lk(*opCtx->getClient());
CurOp::get(opCtx)->setPlanSummary_inlock(exec->getPlanExplainer().getPlanSummary());
}
if (!collection) {
// No collection. Just fill out curop indicating that there were zero results and
// there is no ClientCursor id, and then return.
const long long numResults = 0;
const CursorId cursorId = 0;
endQueryOp(opCtx, collection, *exec, numResults, cursorId);
auto bodyBuilder = result->getBodyBuilder();
appendCursorResponseObject(
cursorId, nss.ns(), BSONArray(), boost::none, &bodyBuilder);
return;
}
FindCommon::waitInFindBeforeMakingBatch(opCtx, *exec->getCanonicalQuery());
const FindCommandRequest& originalFC =
exec->getCanonicalQuery()->getFindCommandRequest();
// Stream query results, adding them to a BSONArray as we go.
CursorResponseBuilder::Options options;
options.isInitialResponse = true;
if (!opCtx->inMultiDocumentTransaction()) {
options.atClusterTime = repl::ReadConcernArgs::get(opCtx).getArgsAtClusterTime();
}
CursorResponseBuilder firstBatch(result, options);
BSONObj obj;
PlanExecutor::ExecState state = PlanExecutor::ADVANCED;
std::uint64_t numResults = 0;
bool stashedResult = false;
ResourceConsumption::DocumentUnitCounter docUnitsReturned;
try {
while (!FindCommon::enoughForFirstBatch(originalFC, numResults) &&
PlanExecutor::ADVANCED == (state = exec->getNext(&obj, nullptr))) {
// If we can't fit this result inside the current batch, then we stash it for
// later.
if (!FindCommon::haveSpaceForNext(obj, numResults, firstBatch.bytesUsed())) {
exec->stashResult(obj);
stashedResult = true;
break;
}
// If this executor produces a postBatchResumeToken, add it to the response.
firstBatch.setPostBatchResumeToken(exec->getPostBatchResumeToken());
// Add result to output buffer.
firstBatch.append(obj);
numResults++;
docUnitsReturned.observeOne(obj.objsize());
}
} catch (DBException& exception) {
firstBatch.abandon();
auto&& explainer = exec->getPlanExplainer();
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),
"cmd"_attr = cmdObj);
exception.addContext("Executor error during find command");
throw;
}
// For empty batches, or in the case where the final result was added to the batch
// rather than being stashed, we update the PBRT to ensure that it is the most recent
// available.
if (!stashedResult) {
firstBatch.setPostBatchResumeToken(exec->getPostBatchResumeToken());
}
// Set up the cursor for getMore.
CursorId cursorId = 0;
if (shouldSaveCursor(opCtx, collection, state, exec.get())) {
const bool stashResourcesForGetMore =
exec->isSaveRecoveryUnitAcrossCommandsEnabled();
ClientCursorPin pinnedCursor = CursorManager::get(opCtx)->registerCursor(
opCtx,
{std::move(exec),
nss,
AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames(),
APIParameters::get(opCtx),
opCtx->getWriteConcern(),
repl::ReadConcernArgs::get(opCtx),
ReadPreferenceSetting::get(opCtx),
_request.body,
{Privilege(ResourcePattern::forExactNamespace(nss), ActionType::find)}});
cursorId = pinnedCursor.getCursor()->cursorid();
invariant(!exec);
PlanExecutor* cursorExec = pinnedCursor.getCursor()->getExecutor();
// State will be restored on getMore.
cursorExec->saveState();
cursorExec->detachFromOperationContext();
// We assume that cursors created through a DBDirectClient are always used from
// their original OperationContext, so we do not need to move time to and from the
// cursor.
if (!opCtx->getClient()->isInDirectClient()) {
pinnedCursor.getCursor()->setLeftoverMaxTimeMicros(
opCtx->getRemainingMaxTimeMicros());
}
pinnedCursor.getCursor()->setNReturnedSoFar(numResults);
pinnedCursor.getCursor()->incNBatches();
// Fill out curop based on the results.
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.
//
// 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()->getOperationStatistics();
}
} else {
endQueryOp(opCtx, collection, *exec, numResults, cursorId);
}
// Generate the response object to send to the client.
firstBatch.done(cursorId, nss.ns());
// Increment this metric once we have generated a response and we know it will return
// documents.
auto& metricsCollector = ResourceConsumption::MetricsCollector::get(opCtx);
metricsCollector.incrementDocUnitsReturned(docUnitsReturned);
query_request_helper::validateCursorResponse(result->getBodyBuilder().asTempObj());
}
void appendMirrorableRequest(BSONObjBuilder* bob) const override {
// Filter the keys that can be mirrored
static const auto kMirrorableKeys = [] {
BSONObjBuilder keyBob;
keyBob.append("find", 1);
keyBob.append("filter", 1);
keyBob.append("skip", 1);
keyBob.append("limit", 1);
keyBob.append("sort", 1);
keyBob.append("hint", 1);
keyBob.append("collation", 1);
keyBob.append("min", 1);
keyBob.append("max", 1);
keyBob.append("shardVersion", 1);
return keyBob.obj();
}();
_request.body.filterFieldsUndotted(bob, kMirrorableKeys, true);
// Tell the find to only return a single batch
bob->append("batchSize", 1);
bob->append("singleBatch", true);
}
private:
const OpMsgRequest _request;
const StringData _dbName;
};
} findCmd;
} // namespace
} // namespace mongo
|