diff options
Diffstat (limited to 'src/mongo/db/pipeline/plan_executor_pipeline.cpp')
| -rw-r--r-- | src/mongo/db/pipeline/plan_executor_pipeline.cpp | 32 |
1 files changed, 8 insertions, 24 deletions
diff --git a/src/mongo/db/pipeline/plan_executor_pipeline.cpp b/src/mongo/db/pipeline/plan_executor_pipeline.cpp index 7e39571b589..0b26a7db813 100644 --- a/src/mongo/db/pipeline/plan_executor_pipeline.cpp +++ b/src/mongo/db/pipeline/plan_executor_pipeline.cpp @@ -37,14 +37,8 @@ #include "mongo/db/pipeline/plan_explainer_pipeline.h" #include "mongo/db/pipeline/resume_token.h" #include "mongo/db/repl/speculative_majority_read_info.h" -#include "mongo/util/duration.h" namespace mongo { -namespace { -Counter64 changeStreamsLargeEventsFailedCounter; -ServerStatusMetricField<Counter64> dChangeStreamsLargeEventsFailedCounter( - "changeStreams.largeEventsFailed", &changeStreamsLargeEventsFailedCounter); -} // namespace PlanExecutorPipeline::PlanExecutorPipeline(boost::intrusive_ptr<ExpressionContext> expCtx, std::unique_ptr<Pipeline, PipelineDeleter> pipeline, @@ -84,7 +78,9 @@ PlanExecutor::ExecState PlanExecutorPipeline::getNext(BSONObj* objOut, RecordId* Document docOut; auto execState = getNextDocument(&docOut, nullptr); if (execState == PlanExecutor::ADVANCED) { - *objOut = _trySerializeToBson(docOut); + // Include metadata if the output will be consumed by a merging node. + *objOut = _expCtx->needsMerge || _expCtx->forPerShardCursor ? docOut.toBsonWithMetaData() + : docOut.toBson(); } return execState; } @@ -144,19 +140,6 @@ boost::optional<Document> PlanExecutorPipeline::_tryGetNext() try { return Document::fromBsonWithMetaData(extraInfo->getStartAfterInvalidateEvent()); } -BSONObj PlanExecutorPipeline::_trySerializeToBson(const Document& doc) try { - // Include metadata if the output will be consumed by a merging node. - return _expCtx->needsMerge || _expCtx->forPerShardCursor ? doc.toBsonWithMetaData() - : doc.toBson(); -} catch (const ExceptionFor<ErrorCodes::BSONObjectTooLarge>&) { - // If in a change stream pipeline, increment change stream large event failed error - // count metric. - if (ResumableScanType::kChangeStream == _resumableScanType) { - changeStreamsLargeEventsFailedCounter.increment(); - } - throw; -} - void PlanExecutorPipeline::_updateResumableScanState(const boost::optional<Document>& document) { switch (_resumableScanType) { case ResumableScanType::kChangeStream: @@ -202,8 +185,9 @@ void PlanExecutorPipeline::_performChangeStreamsAccounting(const boost::optional void PlanExecutorPipeline::_validateChangeStreamsResumeToken(const Document& event) const { // Confirm that the document _id field matches the original resume token in the sort key field. + auto eventBSON = event.toBson(); auto resumeToken = event.metadata().getSortKey(); - auto idField = event.getField("_id"); + auto idField = eventBSON.getObjectField("_id"); invariant(!resumeToken.missing()); uassert(ErrorCodes::ChangeStreamFatalError, str::stream() << "Encountered an event whose _id field, which contains the resume " @@ -212,9 +196,9 @@ void PlanExecutorPipeline::_validateChangeStreamsResumeToken(const Document& eve "transformations that retain the unmodified _id field are allowed. " "Expected: " << BSON("_id" << resumeToken) << " but found: " - << (idField.missing() ? BSONObj() : BSON("_id" << idField)), - resumeToken.getType() == BSONType::Object && - ValueComparator::kInstance.evaluate(idField == resumeToken)); + << (eventBSON["_id"] ? BSON("_id" << eventBSON["_id"]) : BSONObj()), + (resumeToken.getType() == BSONType::Object) && + idField.binaryEqual(resumeToken.getDocument().toBson())); } void PlanExecutorPipeline::_performResumableOplogScanAccounting() { |