xuemingqiang/mongo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SERVER-74716 Require splitEvent stage for splitEvent token

Browse Source
This commit is contained in:
Romans Kasperovics
2023-03-24 18:35:45 +00:00
committed by Evergreen Agent
parent f2d3e4874b
commit 0127ea7f63
16 changed files with 297 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

201
jstests/change_streams/split_large_event.js
View File

@@ -111,39 +111,6 @@ assert.commandWorked(testColl.update({_id: "bbb"}, {$set: {a: "y".repeat(kLargeS
assert(!fullEvent.splitEvent);
}
{
// Test that for events which are over the size limit, $changeStreamSplitLargeEvent is required.
// Additionally, test that 'changeStreams.largeEventsFailed' metric is counted correctly.
const oldChangeStreamsLargeEventsFailed = getChangeStreamMetricSum("largeEventsFailed");
const csCursor = testColl.watch([], {
batchSize: 0, // Ensure same behavior for replica sets and sharded clusters.
fullDocument: "required",
fullDocumentBeforeChange: "required",
resumeAfter: testStartToken
});
assert.throwsWithCode(() => assert.soon(() => csCursor.hasNext()),
ErrorCodes.BSONObjectTooLarge);
const newChangeStreamsLargeEventsFailed = getChangeStreamMetricSum("largeEventsFailed");
// We will hit this exception once on each shard that encounters a large change event document.
const numShardsWithLargeDocs =
Math.min(2, FixtureHelpers.numberOfShardsForCollection(testColl));
assert.eq(newChangeStreamsLargeEventsFailed,
oldChangeStreamsLargeEventsFailed + numShardsWithLargeDocs);
}
// Open a change stream with $changeStreamSplitLargeEvent and request both pre- and post-images.
csCursor =
testColl.watch([{$changeStreamSplitLargeEvent: {}}],
{
batchSize: 0, // Ensure same behavior for replica sets and sharded clusters.
fullDocument: "required",
fullDocumentBeforeChange: "required",
resumeAfter: testStartToken
});
/**
* Helper function to reconstruct the fragments of a split event into the original event. The
* fragments are expected to be the next 'expectedFragmentCount' events retrieved from the cursor.
@@ -176,54 +143,116 @@ function validateReconstructedEvent(event, expectedId) {
assert.eq(kLargeStringSize, event.updateDescription.updatedFields.a.length);
}
const oldChangeStreamsLargeEventsSplit = getChangeStreamMetricSum("largeEventSplit");
// We declare 'resumeTokens' array outside of the for-scope to collect and share resume tokens
// across several test-cases.
let resumeTokens = [];
const [reconstructedEvent, resumeTokens] = reconstructSplitEvent(csCursor, 3);
const fragmentCount = resumeTokens.length;
validateReconstructedEvent(reconstructedEvent, "aaa");
for (const postImageMode of ["required", "updateLookup"]) {
{
// Test that for events which are over the size limit, $changeStreamSplitLargeEvent is
// required. Additionally, test that 'changeStreams.largeEventsFailed' metric is counted
// correctly.
const [reconstructedEvent2, _] = reconstructSplitEvent(csCursor, 3);
validateReconstructedEvent(reconstructedEvent2, "bbb");
const oldChangeStreamsLargeEventsFailed = getChangeStreamMetricSum("largeEventsFailed");
const newChangeStreamsLargeEventsSplit = getChangeStreamMetricSum("largeEventSplit");
assert.eq(oldChangeStreamsLargeEventsSplit + 2, newChangeStreamsLargeEventsSplit);
{
// Test that we can filter on fields that sum to more than 16MB without throwing. Note that
// we construct this $match as an $or of the three large fields so that pipeline optimization
// cannot split this $match into multiple predicates and scatter them through the pipeline.
const csCursor = testColl.watch(
[
{
$match: {
$or: [
{"fullDocument": {$exists: true}},
{"fullDocumentBeforeChange": {$exists: true}},
{"updateDescription": {$exists: true}}
]
}
},
{$changeStreamSplitLargeEvent: {}}
],
{
fullDocument: "required",
const csCursor = testColl.watch([], {
batchSize: 0, // Ensure same behavior for replica sets and sharded clusters.
fullDocument: postImageMode,
fullDocumentBeforeChange: "required",
resumeAfter: testStartToken
});
assert.docEq(resumeTokens, reconstructSplitEvent(csCursor, 3)[1]);
}
assert.throwsWithCode(() => assert.soon(() => csCursor.hasNext()),
ErrorCodes.BSONObjectTooLarge);
{
// Resume the stream from the second-last fragment and test that we see only the last fragment.
const csCursor = testColl.watch([{$changeStreamSplitLargeEvent: {}}], {
fullDocument: "required",
fullDocumentBeforeChange: "required",
resumeAfter: resumeTokens[fragmentCount - 2]
});
assert.soon(() => csCursor.hasNext());
const resumedEvent = csCursor.next();
assert.eq(resumedEvent.updateDescription.updatedFields.a.length, kLargeStringSize);
assert.docEq({fragment: fragmentCount, of: fragmentCount}, resumedEvent.splitEvent);
const newChangeStreamsLargeEventsFailed = getChangeStreamMetricSum("largeEventsFailed");
// We will hit this exception once on each shard that encounters a large change event
// document.
const numShardsWithLargeDocs =
Math.min(2, FixtureHelpers.numberOfShardsForCollection(testColl));
assert.eq(newChangeStreamsLargeEventsFailed,
oldChangeStreamsLargeEventsFailed + numShardsWithLargeDocs);
}
{
// Test that oversized events are split into fragments and can be reassembled to form the
// original event, and that the largeEventSplit metric counter is correctly incremented.
const csCursor = testColl.watch(
[{$changeStreamSplitLargeEvent: {}}],
{
batchSize: 0, // Ensure same behavior for replica sets and sharded clusters.
fullDocument: postImageMode,
fullDocumentBeforeChange: "required",
resumeAfter: testStartToken
});
const oldChangeStreamsLargeEventsSplit = getChangeStreamMetricSum("largeEventSplit");
var reconstructedEvent;
[reconstructedEvent, resumeTokens] = reconstructSplitEvent(csCursor, 3);
validateReconstructedEvent(reconstructedEvent, "aaa");
const [reconstructedEvent2, _] = reconstructSplitEvent(csCursor, 3);
validateReconstructedEvent(reconstructedEvent2, "bbb");
const newChangeStreamsLargeEventsSplit = getChangeStreamMetricSum("largeEventSplit");
assert.eq(oldChangeStreamsLargeEventsSplit + 2, newChangeStreamsLargeEventsSplit);
}
{
// Test that we can filter on fields that sum to more than 16MB without throwing. Note that
// we construct this $match as an $or of the three large fields so that pipeline
// optimization cannot split this $match into multiple predicates and scatter them through
// the pipeline.
const csCursor = testColl.watch(
[
{
$match: {
$or: [
{"fullDocument": {$exists: true}},
{"fullDocumentBeforeChange": {$exists: true}},
{"updateDescription": {$exists: true}}
]
}
},
{$changeStreamSplitLargeEvent: {}}
],
{
fullDocument: postImageMode,
fullDocumentBeforeChange: "required",
resumeAfter: testStartToken
});
assert.docEq(resumeTokens, reconstructSplitEvent(csCursor, 3)[1]);
}
{
// Resume the stream from the second-last fragment and test that we see only the last
// fragment.
const csCursor = testColl.watch([{$changeStreamSplitLargeEvent: {}}], {
fullDocument: postImageMode,
fullDocumentBeforeChange: "required",
resumeAfter: resumeTokens[resumeTokens.length - 2]
});
assert.soon(() => csCursor.hasNext());
const resumedEvent = csCursor.next();
assert.eq(resumedEvent.updateDescription.updatedFields.a.length, kLargeStringSize);
assert.docEq({fragment: resumeTokens.length, of: resumeTokens.length},
resumedEvent.splitEvent);
}
{
// Test that projecting out one of the large fields in the resumed pipeline changes the
// split such that the resume point won't be generated, and we therefore throw an exception.
const csCursor = testColl.watch(
[{$project: {"fullDocument.a": 0}}, {$changeStreamSplitLargeEvent: {}}], {
batchSize: 0, // Ensure same behavior for replica sets and sharded clusters.
fullDocument: postImageMode,
fullDocumentBeforeChange: "required",
resumeAfter: resumeTokens[resumeTokens.length - 1]
});
assert.throwsWithCode(() => assert.soon(() => csCursor.hasNext()),
ErrorCodes.ChangeStreamFatalError);
}
}
{
@@ -232,7 +261,7 @@ assert.eq(oldChangeStreamsLargeEventsSplit + 2, newChangeStreamsLargeEventsSplit
assert.commandFailedWithCode(testDB.runCommand({
aggregate: testColl.getName(),
pipeline: [
{$changeStream: {resumeAfter: resumeTokens[fragmentCount - 2]}},
{$changeStream: {resumeAfter: resumeTokens[resumeTokens.length - 2]}},
{$_internalInhibitOptimization: {}},
{$changeStreamSplitLargeEvent: {}}
],
@@ -242,16 +271,20 @@ assert.eq(oldChangeStreamsLargeEventsSplit + 2, newChangeStreamsLargeEventsSplit
}
{
// Test that projecting out one of the large fields in the resumed pipeline changes the split
// such that the resume point won't be generated, and we therefore throw an exception.
const csCursor =
testColl.watch([{$project: {"fullDocument.a": 0}}, {$changeStreamSplitLargeEvent: {}}], {
batchSize: 0, // Ensure same behavior for replica sets and sharded clusters.
fullDocument: "required",
fullDocumentBeforeChange: "required",
resumeAfter: resumeTokens[fragmentCount - 1]
});
// Test that resuming from a split event token without requesting pre- and post- images fails,
// because the resulting event is too small to be split.
const csCursor = testColl.watch([{$changeStreamSplitLargeEvent: {}}], {
batchSize: 0, // Ensure same behavior for replica sets and sharded clusters.
resumeAfter: resumeTokens[resumeTokens.length - 1]
});
assert.throwsWithCode(() => assert.soon(() => csCursor.hasNext()),
ErrorCodes.ChangeStreamFatalError);
}
{
// Test that resuming from split event without the $changeStreamSplitLargeEvent stage fails.
assert.throwsWithCode(
() => testColl.watch([], {resumeAfter: resumeTokens[resumeTokens.length - 2]}),
ErrorCodes.ChangeStreamFatalError);
}
}());

13
src/mongo/db/pipeline/SConscript
View File

@@ -387,6 +387,7 @@ pipelineEnv.Library(
'$BUILD_DIR/third_party/shim_snappy',
'abt_utils',
'accumulator',
'change_stream_helpers',
'dependencies',
'document_path_support',
'document_sources_idl',
@@ -454,6 +455,7 @@ env.Library(
'$BUILD_DIR/mongo/db/server_feature_flags',
'$BUILD_DIR/mongo/db/update/update_driver',
'$BUILD_DIR/mongo/s/query/router_exec_stage',
'change_stream_helpers',
'change_stream_preimage',
],
)
@@ -526,6 +528,17 @@ env.Library(
],
)
env.Library(
target="change_stream_helpers",
source=[
"change_stream_helpers.cpp",
],
LIBDEPS=[
"document_sources_idl",
],
LIBDEPS_PRIVATE=[],
)
env.Library(
target="change_stream_test_helpers",
source=[

3
src/mongo/db/pipeline/change_stream_event_transform.cpp
View File

@@ -31,6 +31,7 @@
#include "mongo/db/pipeline/change_stream_document_diff_parser.h"
#include "mongo/db/pipeline/change_stream_filter_helpers.h"
#include "mongo/db/pipeline/change_stream_helpers.h"
#include "mongo/db/pipeline/change_stream_helpers_legacy.h"
#include "mongo/db/pipeline/change_stream_preimage_gen.h"
#include "mongo/db/pipeline/document_path_support.h"
@@ -46,7 +47,7 @@ namespace mongo {
namespace {
constexpr auto checkValueType = &DocumentSourceChangeStream::checkValueType;
constexpr auto checkValueTypeOrMissing = &DocumentSourceChangeStream::checkValueTypeOrMissing;
constexpr auto resolveResumeToken = &DocumentSourceChangeStream::resolveResumeTokenFromSpec;
constexpr auto resolveResumeToken = &change_stream::resolveResumeTokenFromSpec;
Document copyDocExceptFields(const Document& source, const std::set<StringData>& fieldNames) {
MutableDocument doc(source);

53
src/mongo/db/pipeline/change_stream_helpers.cpp Normal file
View File

@@ -0,0 +1,53 @@
/**
* Copyright (C) 2018-present MongoDB, Inc.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the Server Side Public License, version 1,
* as published by MongoDB, Inc.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* Server Side Public License for more details.
*
* You should have received a copy of the Server Side Public License
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*
* As a special exception, the copyright holders give permission to link the
* code of portions of this program with the OpenSSL library under certain
* conditions as described in each individual source file and distribute
* linked combinations including the program with the OpenSSL library. You
* must comply with the Server Side Public License in all respects for
* all of the code used other than as permitted herein. If you modify file(s)
* with this exception, you may extend this exception to your version of the
* file(s), but you are not obligated to do so. If you do not wish to do so,
* delete this exception statement from your version. If you delete this
* exception statement from all source files in the program, then also delete
* it in the license file.
*/
#include "mongo/db/pipeline/change_stream_helpers.h"
#include "mongo/db/pipeline/document_source_change_stream_gen.h"
namespace mongo {
namespace change_stream {
ResumeTokenData resolveResumeTokenFromSpec(const boost::intrusive_ptr<ExpressionContext>& expCtx,
const DocumentSourceChangeStreamSpec& spec) {
if (spec.getStartAfter()) {
return spec.getStartAfter()->getData();
} else if (spec.getResumeAfter()) {
return spec.getResumeAfter()->getData();
} else if (spec.getStartAtOperationTime()) {
return ResumeToken::makeHighWaterMarkToken(*spec.getStartAtOperationTime(),
expCtx->changeStreamTokenVersion)
.getData();
}
tasserted(5666901,
"Expected one of 'startAfter', 'resumeAfter' or 'startAtOperationTime' to be "
"populated in $changeStream spec");
}
} // namespace change_stream
} // namespace mongo

47
src/mongo/db/pipeline/change_stream_helpers.h Normal file
View File

@@ -0,0 +1,47 @@
/**
* 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/db/pipeline/document_source_change_stream_gen.h"
#include "mongo/db/pipeline/expression_context.h"
namespace mongo {
namespace change_stream {
/**
* Extracts the resume token from the given spec. If a 'startAtOperationTime' is specified,
* returns the equivalent high-watermark token. This method should only ever be called on a spec
* where one of 'resumeAfter', 'startAfter', or 'startAtOperationTime' is populated.
*/
ResumeTokenData resolveResumeTokenFromSpec(const boost::intrusive_ptr<ExpressionContext>& expCtx,
const DocumentSourceChangeStreamSpec& spec);
} // namespace change_stream
} // namespace mongo

22
src/mongo/db/pipeline/document_source_change_stream.cpp
View File

@@ -36,6 +36,7 @@
#include "mongo/db/pipeline/aggregate_command_gen.h"
#include "mongo/db/pipeline/change_stream_constants.h"
#include "mongo/db/pipeline/change_stream_filter_helpers.h"
#include "mongo/db/pipeline/change_stream_helpers.h"
#include "mongo/db/pipeline/change_stream_helpers_legacy.h"
#include "mongo/db/pipeline/document_path_support.h"
#include "mongo/db/pipeline/document_source_change_stream_add_post_image.h"
@@ -234,23 +235,6 @@ std::string DocumentSourceChangeStream::regexEscapeNsForChangeStream(StringData
return result;
}
ResumeTokenData DocumentSourceChangeStream::resolveResumeTokenFromSpec(
const boost::intrusive_ptr<ExpressionContext>& expCtx,
const DocumentSourceChangeStreamSpec& spec) {
if (spec.getStartAfter()) {
return spec.getStartAfter()->getData();
} else if (spec.getResumeAfter()) {
return spec.getResumeAfter()->getData();
} else if (spec.getStartAtOperationTime()) {
return ResumeToken::makeHighWaterMarkToken(*spec.getStartAtOperationTime(),
expCtx->changeStreamTokenVersion)
.getData();
}
tasserted(5666901,
"Expected one of 'startAfter', 'resumeAfter' or 'startAtOperationTime' to be "
"populated in $changeStream spec");
}
Timestamp DocumentSourceChangeStream::getStartTimeForNewStream(
const boost::intrusive_ptr<ExpressionContext>& expCtx) {
// If we do not have an explicit starting point, we should start from the latest majority
@@ -298,7 +282,7 @@ std::list<boost::intrusive_ptr<DocumentSource>> DocumentSourceChangeStream::_bui
std::list<boost::intrusive_ptr<DocumentSource>> stages;
// Obtain the resume token from the spec. This will be used when building the pipeline.
auto resumeToken = DocumentSourceChangeStream::resolveResumeTokenFromSpec(expCtx, spec);
auto resumeToken = change_stream::resolveResumeTokenFromSpec(expCtx, spec);
// Unfold the $changeStream into its constituent stages and add them to the pipeline.
stages.push_back(DocumentSourceChangeStreamOplogMatch::create(expCtx, spec));
@@ -408,7 +392,7 @@ void DocumentSourceChangeStream::assertIsLegalSpecification(
!spec.getResumeAfter() || !spec.getStartAfter());
auto resumeToken = (spec.getResumeAfter() || spec.getStartAfter())
? resolveResumeTokenFromSpec(expCtx, spec)
? change_stream::resolveResumeTokenFromSpec(expCtx, spec)
: boost::optional<ResumeTokenData>();
uassert(40674,

9
src/mongo/db/pipeline/document_source_change_stream.h
View File

@@ -294,15 +294,6 @@ public:
*/
static void checkValueTypeOrMissing(Value v, StringData fieldName, BSONType expectedType);
/**
* Extracts the resume token from the given spec. If a 'startAtOperationTime' is specified,
* returns the equivalent high-watermark token. This method should only ever be called on a spec
* where one of 'resumeAfter', 'startAfter', or 'startAtOperationTime' is populated.
*/
static ResumeTokenData resolveResumeTokenFromSpec(
const boost::intrusive_ptr<ExpressionContext>& expCtx,
const DocumentSourceChangeStreamSpec& spec);
/**
* For a change stream with no resume information supplied by the user, returns the clusterTime
* at which the new stream should begin scanning the oplog.

3
src/mongo/db/pipeline/document_source_change_stream_check_invalidate.cpp
View File

@@ -30,6 +30,7 @@
#include "mongo/platform/basic.h"
#include "mongo/db/pipeline/change_stream_helpers.h"
#include "mongo/db/pipeline/change_stream_start_after_invalidate_info.h"
#include "mongo/db/pipeline/document_source_change_stream.h"
#include "mongo/db/pipeline/document_source_change_stream_check_invalidate.h"
@@ -73,7 +74,7 @@ DocumentSourceChangeStreamCheckInvalidate::create(
const DocumentSourceChangeStreamSpec& spec) {
// If resuming from an "invalidate" using "startAfter", pass along the resume token data to
// DSCSCheckInvalidate to signify that another invalidate should not be generated.
auto resumeToken = DocumentSourceChangeStream::resolveResumeTokenFromSpec(expCtx, spec);
auto resumeToken = change_stream::resolveResumeTokenFromSpec(expCtx, spec);
return new DocumentSourceChangeStreamCheckInvalidate(
expCtx, boost::make_optional(resumeToken.fromInvalidate, std::move(resumeToken)));
}

3
src/mongo/db/pipeline/document_source_change_stream_check_resumability.cpp
View File

@@ -30,6 +30,7 @@
#include "mongo/platform/basic.h"
#include "mongo/db/curop.h"
#include "mongo/db/pipeline/change_stream_helpers.h"
#include "mongo/db/pipeline/document_source_change_stream_check_resumability.h"
#include "mongo/db/query/query_feature_flags_gen.h"
#include "mongo/db/repl/oplog_entry.h"
@@ -137,7 +138,7 @@ DocumentSourceChangeStreamCheckResumability::DocumentSourceChangeStreamCheckResu
intrusive_ptr<DocumentSourceChangeStreamCheckResumability>
DocumentSourceChangeStreamCheckResumability::create(const intrusive_ptr<ExpressionContext>& expCtx,
const DocumentSourceChangeStreamSpec& spec) {
auto resumeToken = DocumentSourceChangeStream::resolveResumeTokenFromSpec(expCtx, spec);
auto resumeToken = change_stream::resolveResumeTokenFromSpec(expCtx, spec);
return new DocumentSourceChangeStreamCheckResumability(expCtx, std::move(resumeToken));
}

3
src/mongo/db/pipeline/document_source_change_stream_ensure_resume_token_present.cpp
View File

@@ -31,6 +31,7 @@
#include "mongo/db/pipeline/document_source_change_stream_ensure_resume_token_present.h"
#include "mongo/db/pipeline/change_stream_helpers.h"
#include "mongo/db/pipeline/change_stream_start_after_invalidate_info.h"
#include "mongo/db/query/query_feature_flags_gen.h"
@@ -45,7 +46,7 @@ boost::intrusive_ptr<DocumentSourceChangeStreamEnsureResumeTokenPresent>
DocumentSourceChangeStreamEnsureResumeTokenPresent::create(
const boost::intrusive_ptr<ExpressionContext>& expCtx,
const DocumentSourceChangeStreamSpec& spec) {
auto resumeToken = DocumentSourceChangeStream::resolveResumeTokenFromSpec(expCtx, spec);
auto resumeToken = change_stream::resolveResumeTokenFromSpec(expCtx, spec);
tassert(5666902,
"Expected non-high-water-mark resume token",
!ResumeToken::isHighWaterMarkToken(resumeToken));

3
src/mongo/db/pipeline/document_source_change_stream_oplog_match.cpp
View File

@@ -31,6 +31,7 @@
#include "mongo/bson/bsonmisc.h"
#include "mongo/db/pipeline/change_stream_filter_helpers.h"
#include "mongo/db/pipeline/change_stream_helpers.h"
#include "mongo/db/pipeline/document_source_change_stream_unwind_transaction.h"
namespace mongo {
@@ -104,7 +105,7 @@ DocumentSourceChangeStreamOplogMatch::DocumentSourceChangeStreamOplogMatch(
boost::intrusive_ptr<DocumentSourceChangeStreamOplogMatch>
DocumentSourceChangeStreamOplogMatch::create(const boost::intrusive_ptr<ExpressionContext>& expCtx,
const DocumentSourceChangeStreamSpec& spec) {
auto resumeToken = DocumentSourceChangeStream::resolveResumeTokenFromSpec(expCtx, spec);
auto resumeToken = change_stream::resolveResumeTokenFromSpec(expCtx, spec);
return make_intrusive<DocumentSourceChangeStreamOplogMatch>(resumeToken.clusterTime, expCtx);
}

49
src/mongo/db/pipeline/document_source_change_stream_split_large_event.cpp
View File

@@ -29,6 +29,7 @@
#include "mongo/db/pipeline/document_source_change_stream_split_large_event.h"
#include "mongo/db/pipeline/change_stream_helpers.h"
#include "mongo/db/pipeline/change_stream_split_event_helpers.h"
#include "mongo/db/pipeline/document_source_change_stream_ensure_resume_token_present.h"
#include "mongo/db/pipeline/document_source_change_stream_handle_topology_change.h"
@@ -48,7 +49,7 @@ DocumentSourceChangeStreamSplitLargeEvent::create(
const DocumentSourceChangeStreamSpec& spec) {
// If resuming from a split event, pass along the resume token data to DSCSSplitEvent so that it
// can swallow fragments that precede the actual resume point.
auto resumeToken = DocumentSourceChangeStream::resolveResumeTokenFromSpec(expCtx, spec);
auto resumeToken = change_stream::resolveResumeTokenFromSpec(expCtx, spec);
auto resumeAfterSplit =
resumeToken.fragmentNum ? std::move(resumeToken) : boost::optional<ResumeTokenData>{};
return new DocumentSourceChangeStreamSplitLargeEvent(expCtx, std::move(resumeAfterSplit));
@@ -76,7 +77,7 @@ DocumentSourceChangeStreamSplitLargeEvent::DocumentSourceChangeStreamSplitLargeE
: DocumentSource(getSourceName(), expCtx), _resumeAfterSplit(std::move(resumeAfterSplit)) {
tassert(7182801,
"Expected a split event resume token, but found a non-split token",
!_resumeAfterSplit || resumeAfterSplit->fragmentNum);
!_resumeAfterSplit || _resumeAfterSplit->fragmentNum);
}
Value DocumentSourceChangeStreamSplitLargeEvent::serialize(SerializationOptions opts) const {
@@ -129,38 +130,29 @@ DocumentSource::GetNextResult DocumentSourceChangeStreamSplitLargeEvent::doGetNe
// TODO SERVER-74301: Consider 'this->pExpCtx->forPerShardCursor' here.
auto [eventDoc, eventBsonSize] = change_stream_split_event::processChangeEventBeforeSplit(
input.releaseDocument(), this->pExpCtx->needsMerge);
if (eventBsonSize <= kBSONObjMaxChangeEventSize) {
return std::move(eventDoc);
}
// If we are resuming from a split event, check whether this is it. If so, extract the fragment
// number from which we are resuming. Otherwise, we have already scanned past the resume point,
// which implies that it may be on another shard. Continue to split this event without skipping.
size_t skipFirstFragments = 0;
if (_resumeAfterSplit) {
using DSCSCR = DocumentSourceChangeStreamCheckResumability;
auto resumeStatus = DSCSCR::compareAgainstClientResumeToken(eventDoc, *_resumeAfterSplit);
tassert(7182805,
"Observed unexpected event before resume point",
resumeStatus != DSCSCR::ResumeStatus::kCheckNextDoc);
if (resumeStatus == DSCSCR::ResumeStatus::kNeedsSplit) {
skipFirstFragments = *_resumeAfterSplit->fragmentNum;
}
_resumeAfterSplit.reset();
size_t skipFragments = _handleResumeAfterSplit(eventDoc, eventBsonSize);
// Before proceeding, check whether the event is small enough to be returned as-is.
if (eventBsonSize <= kBSONObjMaxChangeEventSize) {
return std::move(eventDoc);
}
// Split the event into N appropriately-sized fragments. Make sure to leave some space for the
// postBatchResumeToken in the cursor response object.
size_t tokenSize = eventDoc.metadata().getSortKey().getDocument().toBson().objsize();
_splitEventQueue = change_stream_split_event::splitChangeEvent(
eventDoc, kBSONObjMaxChangeEventSize - tokenSize, skipFirstFragments);
eventDoc, kBSONObjMaxChangeEventSize - tokenSize, skipFragments);
// If the user is resuming from a split event but supplied a pipeline which produced a different
// split, we cannot reproduce the split point. Check if we're about to swallow all fragments.
uassert(ErrorCodes::ChangeStreamFatalError,
"Attempted to resume from a split event, but the resumed stream produced a different "
"split. Ensure that the pipeline used to resume is the same as the original",
!(skipFirstFragments > 0 && _splitEventQueue.empty()));
!(skipFragments > 0 && _splitEventQueue.empty()));
tassert(7182804,
"Unexpected empty fragment queue after splitting a change stream event",
!_splitEventQueue.empty());
@@ -178,6 +170,27 @@ Document DocumentSourceChangeStreamSplitLargeEvent::_popFromQueue() {
return nextFragment;
}
size_t DocumentSourceChangeStreamSplitLargeEvent::_handleResumeAfterSplit(const Document& eventDoc,
size_t eventBsonSize) {
if (!_resumeAfterSplit) {
return 0;
}
using DSCSCR = DocumentSourceChangeStreamCheckResumability;
auto resumeStatus = DSCSCR::compareAgainstClientResumeToken(eventDoc, *_resumeAfterSplit);
tassert(7182805,
"Observed unexpected event before resume point",
resumeStatus != DSCSCR::ResumeStatus::kCheckNextDoc);
uassert(ErrorCodes::ChangeStreamFatalError,
"Attempted to resume from a split event fragment, but the event in the resumed "
"stream was not large enough to be split",
resumeStatus != DSCSCR::ResumeStatus::kNeedsSplit ||
eventBsonSize > kBSONObjMaxChangeEventSize);
auto fragmentNum =
(resumeStatus == DSCSCR::ResumeStatus::kNeedsSplit ? *_resumeAfterSplit->fragmentNum : 0);
_resumeAfterSplit.reset();
return fragmentNum;
}
namespace {
// During pipeline optimization, the split stage must move ahead of these change stream stages.
static const std::set<StringData> kStagesToMoveAheadOf = {

6
src/mongo/db/pipeline/document_source_change_stream_split_large_event.h
View File

@@ -81,6 +81,12 @@ private:
Document _popFromQueue();
/**
* In case of resume after split, check whether 'eventDoc' is the split event. If so, extract
* and return the resume token's fragment number. Otherwise, return zero.
*/
size_t _handleResumeAfterSplit(const Document& eventDoc, size_t eventBsonSize);
boost::optional<ResumeTokenData> _resumeAfterSplit;
std::queue<Document> _splitEventQueue;
};

4
src/mongo/db/pipeline/document_source_change_stream_transform.cpp
View File

@@ -32,6 +32,7 @@
#include "mongo/db/pipeline/document_source_change_stream_transform.h"
#include "mongo/db/pipeline/change_stream_helpers.h"
#include "mongo/db/pipeline/expression.h"
#include "mongo/db/pipeline/lite_parsed_document_source.h"
#include "mongo/db/pipeline/resume_token.h"
@@ -80,8 +81,7 @@ DocumentSourceChangeStreamTransform::DocumentSourceChangeStreamTransform(
_isIndependentOfAnyCollection(expCtx->ns.isCollectionlessAggregateNS()) {
// Extract the resume token or high-water-mark from the spec.
auto tokenData =
DocumentSourceChangeStream::resolveResumeTokenFromSpec(expCtx, _changeStreamSpec);
auto tokenData = change_stream::resolveResumeTokenFromSpec(expCtx, _changeStreamSpec);
// Set the initialPostBatchResumeToken on the expression context.
expCtx->initialPostBatchResumeToken = ResumeToken(tokenData).toBSON();

15
src/mongo/db/pipeline/pipeline.cpp
View File

@@ -39,6 +39,7 @@
#include "mongo/db/jsobj.h"
#include "mongo/db/operation_context.h"
#include "mongo/db/pipeline/accumulator.h"
#include "mongo/db/pipeline/change_stream_helpers.h"
#include "mongo/db/pipeline/document_source.h"
#include "mongo/db/pipeline/document_source_match.h"
#include "mongo/db/pipeline/document_source_merge.h"
@@ -109,6 +110,8 @@ void validateTopLevelPipeline(const Pipeline& pipeline) {
// If the first stage is a $changeStream stage, then all stages in the pipeline must be
// either $changeStream stages or allowlisted as being able to run in a change stream.
const bool isChangeStream = firstStageConstraints.isChangeStreamStage();
// Record whether any of the stages in the pipeline is a $changeStreamSplitLargeEvent.
bool hasChangeStreamSplitLargeEventStage = false;
for (auto&& source : sources) {
uassert(ErrorCodes::IllegalOperation,
str::stream() << source->getSourceName()
@@ -119,7 +122,19 @@ void validateTopLevelPipeline(const Pipeline& pipeline) {
str::stream() << source->getSourceName()
<< " can only be used in a $changeStream pipeline",
!(source->constraints().requiresChangeStream() && !isChangeStream));
// Check whether this is a change stream split stage.
if ("$changeStreamSplitLargeEvent"_sd == source->getSourceName()) {
hasChangeStreamSplitLargeEventStage = true;
}
}
auto expCtx = pipeline.getContext();
auto spec = isChangeStream ? expCtx->changeStreamSpec : boost::none;
auto hasSplitEventResumeToken = spec &&
change_stream::resolveResumeTokenFromSpec(expCtx, *spec).fragmentNum.has_value();
uassert(ErrorCodes::ChangeStreamFatalError,
"To resume from a split event, the $changeStream pipeline must include a "
"$changeStreamSplitLargeEvent stage",
!(hasSplitEventResumeToken && !hasChangeStreamSplitLargeEventStage));
}
// Verify that usage of $searchMeta and $search is legal.

1
src/mongo/db/pipeline/resume_token.cpp
View File

@@ -37,7 +37,6 @@
#include "mongo/bson/bsonobjbuilder.h"
#include "mongo/db/exec/document_value/value_comparator.h"
#include "mongo/db/pipeline/change_stream_helpers_legacy.h"
#include "mongo/db/pipeline/document_source_change_stream_gen.h"
#include "mongo/db/storage/key_string.h"
#include "mongo/util/hex.h"
#include "mongo/util/optional_util.h"