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
|
/**
* 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_session.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/working_set_common.h"
#include "mongo/db/matcher/extensions_callback_real.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/repl/replication_coordinator.h"
#include "mongo/db/service_context.h"
#include "mongo/db/stats/counters.h"
#include "mongo/db/stats/server_read_concern_metrics.h"
#include "mongo/db/storage/storage_engine.h"
#include "mongo/db/transaction_participant.h"
#include "mongo/logv2/log.h"
#include "mongo/rpc/get_status_from_command_result.h"
namespace mongo {
namespace {
const auto kTermField = "term"_sd;
// Parses the command object to a QueryRequest. If the client request did not specify any runtime
// constants, make them available to the query here.
std::unique_ptr<QueryRequest> parseCmdObjectToQueryRequest(OperationContext* opCtx,
NamespaceString nss,
BSONObj cmdObj,
bool isExplain) {
auto qr = uassertStatusOK(
QueryRequest::makeFromFindCommand(std::move(nss), std::move(cmdObj), isExplain));
if (!qr->getRuntimeConstants()) {
qr->setRuntimeConstants(Variables::generateRuntimeConstants(opCtx));
}
return qr;
}
boost::intrusive_ptr<ExpressionContext> makeExpressionContext(
OperationContext* opCtx,
const QueryRequest& queryRequest,
boost::optional<ExplainOptions::Verbosity> verbosity) {
std::unique_ptr<CollatorInterface> collator;
if (!queryRequest.getCollation().isEmpty()) {
collator = uassertStatusOK(CollatorFactoryInterface::get(opCtx->getServiceContext())
->makeFromBSON(queryRequest.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
queryRequest.allowDiskUse(),
false, // bypassDocumentValidation
false, // isMapReduceCommand
queryRequest.nss(),
queryRequest.getRuntimeConstants(),
std::move(collator),
nullptr, // mongoProcessInterface
StringMap<ExpressionContext::ResolvedNamespace>{},
boost::none, // uuid
queryRequest.getLetParameters(), // let
CurOp::get(opCtx)->dbProfileLevel() > 0 // mayDbProfile
);
expCtx->tempDir = storageGlobalParams.dbpath + "/_tmp";
return expCtx;
}
/**
* A command for running .find() queries.
*/
class FindCmd final : public Command {
public:
FindCmd() : Command("find") {}
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;
}
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) 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(authSession->checkAuthForFind(
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<AutoGetCollectionForReadCommand> ctx;
ctx.emplace(opCtx,
CommandHelpers::parseNsCollectionRequired(_dbName, _request.body),
AutoGetCollection::ViewMode::kViewsPermitted);
const auto nss = ctx->getNss();
// Parse the command BSON to a QueryRequest.
const bool isExplain = true;
auto qr = parseCmdObjectToQueryRequest(opCtx, nss, _request.body, isExplain);
// Finish the parsing step by using the QueryRequest to create a CanonicalQuery.
const ExtensionsCallbackReal extensionsCallback(opCtx, &nss);
auto expCtx = makeExpressionContext(opCtx, *qr, verbosity);
auto cq = uassertStatusOK(
CanonicalQuery::canonicalize(opCtx,
std::move(qr),
std::move(expCtx),
extensionsCallback,
MatchExpressionParser::kAllowAllSpecialFeatures));
if (ctx->getView()) {
// 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& qr = cq->getQueryRequest();
auto viewAggregationCommand = uassertStatusOK(qr.asAggregationCommand());
// Create the agg request equivalent of the find operation, with the explain
// verbosity included.
auto aggRequest = uassertStatusOK(
AggregationRequest::parseFromBSON(nss, viewAggregationCommand, verbosity));
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, viewAggregationCommand, 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.
Collection* const collection = ctx->getCollection();
// Get the execution plan for the query.
bool permitYield = true;
auto exec =
uassertStatusOK(getExecutorFind(opCtx, collection, std::move(cq), permitYield));
auto bodyBuilder = result->getBodyBuilder();
// Got the execution tree. Explain it.
Explain::explainStages(exec.get(), collection, verbosity, BSONObj(), &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();
// Parse the command BSON to a QueryRequest. Pass in the parsedNss in case _request.body
// does not have a UUID.
auto parsedNss =
NamespaceString{CommandHelpers::parseNsFromCommand(_dbName, _request.body)};
const bool isExplain = false;
auto qr =
parseCmdObjectToQueryRequest(opCtx, std::move(parsedNss), _request.body, isExplain);
// 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() &&
!qr->allowSpeculativeMajorityRead()));
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() && qr->isTailable()));
uassert(ErrorCodes::OperationNotSupportedInTransaction,
"The 'readOnce' option is not supported within a transaction.",
!txnParticipant || !opCtx->inMultiDocumentTransaction() || !qr->isReadOnce());
uassert(ErrorCodes::InvalidOptions,
"The '$_internalReadAtClusterTime' option is only supported when testing"
" commands are enabled",
!qr->getReadAtClusterTime() || getTestCommandsEnabled());
uassert(
ErrorCodes::OperationNotSupportedInTransaction,
"The '$_internalReadAtClusterTime' option is not supported within a transaction.",
!txnParticipant || !opCtx->inMultiDocumentTransaction() ||
!qr->getReadAtClusterTime());
uassert(ErrorCodes::InvalidOptions,
"The '$_internalReadAtClusterTime' option is only supported when replication is"
" enabled",
!qr->getReadAtClusterTime() || replCoord->isReplEnabled());
auto* storageEngine = opCtx->getServiceContext()->getStorageEngine();
uassert(ErrorCodes::InvalidOptions,
"The '$_internalReadAtClusterTime' option is only supported by storage engines"
" that support document-level concurrency",
!qr->getReadAtClusterTime() || storageEngine->supportsDocLocking());
// Validate term before acquiring locks, if provided.
auto term = qr->getReplicationTerm();
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 && parsedNss == NamespaceString::kRsOplogNamespace) {
// 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();
}
// We call RecoveryUnit::setTimestampReadSource() before acquiring a lock on the
// collection via AutoGetCollectionForRead in order to ensure the comparison to the
// collection's minimum visible snapshot is accurate.
if (auto targetClusterTime = qr->getReadAtClusterTime()) {
uassert(ErrorCodes::InvalidOptions,
str::stream() << "$_internalReadAtClusterTime value must not be a null"
" timestamp.",
!targetClusterTime->isNull());
// We aren't holding the global lock in intent mode, so it is possible after
// comparing 'targetClusterTime' to 'lastAppliedOpTime' for the last applied opTime
// to go backwards or for the term to change due to replication rollback. This isn't
// an actual concern because the testing infrastructure won't use the
// $_internalReadAtClusterTime option in any test suite where rollback is expected
// to occur.
auto lastAppliedOpTime = replCoord->getMyLastAppliedOpTime();
uassert(ErrorCodes::InvalidOptions,
str::stream() << "$_internalReadAtClusterTime value must not be greater"
" than the last applied opTime. Requested clusterTime: "
<< targetClusterTime->toString()
<< "; last applied opTime: " << lastAppliedOpTime.toString(),
lastAppliedOpTime.getTimestamp() >= targetClusterTime);
// We aren't holding the global lock in intent mode, so it is possible for the
// global storage engine to have been destructed already as a result of the server
// shutting down. This isn't an actual concern because the testing infrastructure
// won't use the $_internalReadAtClusterTime option in any test suite where clean
// shutdown is expected to occur concurrently with tests running.
auto allDurableTime = storageEngine->getAllDurableTimestamp();
invariant(!allDurableTime.isNull());
uassert(ErrorCodes::InvalidOptions,
str::stream() << "$_internalReadAtClusterTime value must not be greater"
" than the all_durable timestamp. Requested"
" clusterTime: "
<< targetClusterTime->toString()
<< "; all_durable timestamp: " << allDurableTime.toString(),
allDurableTime >= targetClusterTime);
// The $_internalReadAtClusterTime option causes any storage-layer cursors created
// during plan execution to read from a consistent snapshot of data at the supplied
// clusterTime, even across yields.
opCtx->recoveryUnit()->setTimestampReadSource(RecoveryUnit::ReadSource::kProvided,
targetClusterTime);
// The $_internalReadAtClusterTime option also causes any storage-layer cursors
// created during plan execution to block on prepared transactions. Since the find
// command ignores prepare conflicts by default, change the behavior.
opCtx->recoveryUnit()->setPrepareConflictBehavior(
PrepareConflictBehavior::kEnforce);
}
// Acquire locks. If the query is on a view, we release our locks and convert the query
// request into an aggregation command.
boost::optional<AutoGetCollectionForReadCommand> ctx;
ctx.emplace(opCtx,
CommandHelpers::parseNsOrUUID(_dbName, _request.body),
AutoGetCollection::ViewMode::kViewsPermitted);
const auto& nss = ctx->getNss();
qr->refreshNSS(opCtx);
// Check whether we are allowed to read from this node after acquiring our locks.
uassertStatusOK(replCoord->checkCanServeReadsFor(
opCtx, nss, ReadPreferenceSetting::get(opCtx).canRunOnSecondary()));
// Fill out curop information.
//
// We pass negative values for 'ntoreturn' and 'ntoskip' to indicate that these values
// should be omitted from the log line. Limit and skip information is already present in
// the find command parameters, so these fields are redundant.
const int ntoreturn = -1;
const int ntoskip = -1;
beginQueryOp(opCtx, nss, _request.body, ntoreturn, ntoskip);
// Finish the parsing step by using the QueryRequest to create a CanonicalQuery.
const ExtensionsCallbackReal extensionsCallback(opCtx, &nss);
auto expCtx = makeExpressionContext(opCtx, *qr, boost::none /* verbosity */);
auto cq = uassertStatusOK(
CanonicalQuery::canonicalize(opCtx,
std::move(qr),
std::move(expCtx),
extensionsCallback,
MatchExpressionParser::kAllowAllSpecialFeatures));
if (ctx->getView()) {
// 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& qr = cq->getQueryRequest();
auto viewAggregationCommand = uassertStatusOK(qr.asAggregationCommand());
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;
}
Collection* const collection = ctx->getCollection();
if (cq->getQueryRequest().isReadOnce()) {
// 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), permitYield));
{
stdx::lock_guard<Client> lk(*opCtx->getClient());
CurOp::get(opCtx)->setPlanSummary_inlock(Explain::getPlanSummary(exec.get()));
}
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(), &bodyBuilder);
return;
}
FindCommon::waitInFindBeforeMakingBatch(opCtx, *exec->getCanonicalQuery());
const QueryRequest& originalQR = exec->getCanonicalQuery()->getQueryRequest();
// Stream query results, adding them to a BSONArray as we go.
CursorResponseBuilder::Options options;
options.isInitialResponse = true;
options.atClusterTime = repl::ReadConcernArgs::get(opCtx).getArgsAtClusterTime();
CursorResponseBuilder firstBatch(result, options);
Document doc;
PlanExecutor::ExecState state = PlanExecutor::ADVANCED;
std::uint64_t numResults = 0;
try {
while (!FindCommon::enoughForFirstBatch(originalQR, numResults) &&
PlanExecutor::ADVANCED == (state = exec->getNext(&doc, nullptr))) {
// If we can't fit this result inside the current batch, then we stash it for
// later.
BSONObj obj = doc.toBson();
if (!FindCommon::haveSpaceForNext(obj, numResults, firstBatch.bytesUsed())) {
exec->enqueue(obj);
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++;
}
} catch (DBException& exception) {
firstBatch.abandon();
LOGV2_WARNING(23798,
"Plan executor error during find command: {error}, "
"stats: {stats}",
"Plan executor error during find command",
"error"_attr = exception.toStatus(),
"stats"_attr = redact(Explain::getWinningPlanStats(exec.get())));
exception.addContext("Executor error during find command");
throw;
}
// Set up the cursor for getMore.
CursorId cursorId = 0;
if (shouldSaveCursor(opCtx, collection, state, exec.get())) {
ClientCursorPin pinnedCursor = CursorManager::get(opCtx)->registerCursor(
opCtx,
{std::move(exec),
nss,
AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames(),
opCtx->getWriteConcern(),
repl::ReadConcernArgs::get(opCtx),
_request.body,
{Privilege(ResourcePattern::forExactNamespace(nss), ActionType::find)},
expCtx->needsMerge});
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);
} else {
endQueryOp(opCtx, collection, *exec, numResults, cursorId);
}
// Generate the response object to send to the client.
firstBatch.done(cursorId, nss.ns());
}
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);
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
|