diff options
Diffstat (limited to 'jstests/change_streams/collation.js')
| -rw-r--r-- | jstests/change_streams/collation.js | 662 |
1 files changed, 325 insertions, 337 deletions
diff --git a/jstests/change_streams/collation.js b/jstests/change_streams/collation.js index e99f6064b60..3d50b564711 100644 --- a/jstests/change_streams/collation.js +++ b/jstests/change_streams/collation.js @@ -3,344 +3,332 @@ * default collation, and uses the simple collation if none is provided. */ (function() { - "use strict"; - - load("jstests/libs/collection_drop_recreate.js"); // For assert[Drop|Create]Collection. - load("jstests/libs/change_stream_util.js"); // For 'ChangeStreamTest' and - // 'runCommandChangeStreamPassthroughAware'. - - let cst = new ChangeStreamTest(db); - - const caseInsensitive = {locale: "en_US", strength: 2}; - - let caseInsensitiveCollection = "change_stream_case_insensitive"; - assertDropCollection(db, caseInsensitiveCollection); - - // Test that you can open a change stream before the collection exists, and it will use the - // simple collation. Tag this stream as 'doNotModifyInPassthroughs', since only individual - // collections have the concept of a default collation. - const simpleCollationStream = cst.startWatchingChanges({ - pipeline: [ - {$changeStream: {}}, - { - $match: - {$or: [{"fullDocument._id": "INSERT_ONE"}, {"fullDocument._id": "INSERT_TWO"}]} - }, - {$project: {docId: "$fullDocument._id"}} - ], - collection: caseInsensitiveCollection, - doNotModifyInPassthroughs: true - }); - - // Create the collection with a non-default collation. The stream should continue to use the - // simple collation. - caseInsensitiveCollection = - assertCreateCollection(db, caseInsensitiveCollection, {collation: caseInsensitive}); - assert.commandWorked( - caseInsensitiveCollection.insert([{_id: "insert_one"}, {_id: "INSERT_TWO"}])); - cst.assertNextChangesEqual( - {cursor: simpleCollationStream, expectedChanges: [{docId: "INSERT_TWO"}]}); - - const caseInsensitivePipeline = [ +"use strict"; + +load("jstests/libs/collection_drop_recreate.js"); // For assert[Drop|Create]Collection. +load("jstests/libs/change_stream_util.js"); // For 'ChangeStreamTest' and + // 'runCommandChangeStreamPassthroughAware'. + +let cst = new ChangeStreamTest(db); + +const caseInsensitive = { + locale: "en_US", + strength: 2 +}; + +let caseInsensitiveCollection = "change_stream_case_insensitive"; +assertDropCollection(db, caseInsensitiveCollection); + +// Test that you can open a change stream before the collection exists, and it will use the +// simple collation. Tag this stream as 'doNotModifyInPassthroughs', since only individual +// collections have the concept of a default collation. +const simpleCollationStream = cst.startWatchingChanges({ + pipeline: [ + {$changeStream: {}}, + {$match: {$or: [{"fullDocument._id": "INSERT_ONE"}, {"fullDocument._id": "INSERT_TWO"}]}}, + {$project: {docId: "$fullDocument._id"}} + ], + collection: caseInsensitiveCollection, + doNotModifyInPassthroughs: true +}); + +// Create the collection with a non-default collation. The stream should continue to use the +// simple collation. +caseInsensitiveCollection = + assertCreateCollection(db, caseInsensitiveCollection, {collation: caseInsensitive}); +assert.commandWorked(caseInsensitiveCollection.insert([{_id: "insert_one"}, {_id: "INSERT_TWO"}])); +cst.assertNextChangesEqual( + {cursor: simpleCollationStream, expectedChanges: [{docId: "INSERT_TWO"}]}); + +const caseInsensitivePipeline = [ + {$changeStream: {}}, + {$match: {"fullDocument.text": "abc"}}, + {$project: {docId: "$documentKey._id"}} +]; + +// Test that $changeStream will not implicitly adopt the default collation of the collection on +// which it is run. Tag this stream as 'doNotModifyInPassthroughs'; whole-db and cluster-wide +// streams do not have default collations. +const didNotInheritCollationStream = cst.startWatchingChanges({ + pipeline: caseInsensitivePipeline, + collection: caseInsensitiveCollection, + doNotModifyInPassthroughs: true +}); +// Test that a collation can be explicitly specified for the $changeStream. This does not need +// to be tagged 'doNotModifyInPassthroughs', since whole-db and cluster-wide changeStreams will +// use an explicit collation if present. +let explicitCaseInsensitiveStream = cst.startWatchingChanges({ + pipeline: caseInsensitivePipeline, + collection: caseInsensitiveCollection, + aggregateOptions: {collation: caseInsensitive} +}); + +assert.writeOK(caseInsensitiveCollection.insert({_id: 0, text: "aBc"})); +assert.writeOK(caseInsensitiveCollection.insert({_id: 1, text: "abc"})); + +// 'didNotInheritCollationStream' should not have inherited the collection's case-insensitive +// default collation, and should only see the second insert. 'explicitCaseInsensitiveStream' +// should see both inserts. +cst.assertNextChangesEqual({cursor: didNotInheritCollationStream, expectedChanges: [{docId: 1}]}); +cst.assertNextChangesEqual( + {cursor: explicitCaseInsensitiveStream, expectedChanges: [{docId: 0}, {docId: 1}]}); + +// Test that the collation does not apply to the scan over the oplog. +const similarNameCollection = assertDropAndRecreateCollection( + db, "cHaNgE_sTrEaM_cAsE_iNsEnSiTiVe", {collation: {locale: "en_US"}}); + +// We must recreate the explicitCaseInsensitiveStream and set 'doNotModifyInPassthroughs'. Whole +// db and cluster-wide streams use the simple collation while scanning the oplog, but they don't +// filter the oplog by collection name. The subsequent $match stage which we inject into the +// pipeline to filter for a specific collection will obey the pipeline's case-insensitive +// collation, meaning that 'cHaNgE_sTrEaM_cAsE_iNsEnSiTiVe' will match +// 'change_stream_case_insensitive'. +explicitCaseInsensitiveStream = cst.startWatchingChanges({ + pipeline: caseInsensitivePipeline, + collection: caseInsensitiveCollection, + aggregateOptions: {collation: caseInsensitive}, + doNotModifyInPassthroughs: true +}); + +assert.writeOK(similarNameCollection.insert({_id: 0, text: "aBc"})); +assert.writeOK(caseInsensitiveCollection.insert({_id: 2, text: "ABC"})); + +// The case-insensitive stream should not see the first insert (to the other collection), only +// the second. We do not expect to see the insert in 'didNotInheritCollationStream'. +cst.assertNextChangesEqual({cursor: explicitCaseInsensitiveStream, expectedChanges: [{docId: 2}]}); + +// Test that creating a collection without a collation does not invalidate any change streams +// that were opened before the collection existed. +(function() { +let noCollationCollection = "change_stream_no_collation"; +assertDropCollection(db, noCollationCollection); + +const streamCreatedBeforeNoCollationCollection = cst.startWatchingChanges({ + pipeline: [{$changeStream: {}}, {$project: {docId: "$documentKey._id"}}], + collection: noCollationCollection +}); + +noCollationCollection = assertCreateCollection(db, noCollationCollection); +assert.writeOK(noCollationCollection.insert({_id: 0})); + +cst.assertNextChangesEqual( + {cursor: streamCreatedBeforeNoCollationCollection, expectedChanges: [{docId: 0}]}); +}()); + +// Test that creating a collection and explicitly specifying the simple collation does not +// invalidate any change streams that were opened before the collection existed. +(function() { +let simpleCollationCollection = "change_stream_simple_collation"; +assertDropCollection(db, simpleCollationCollection); + +const streamCreatedBeforeSimpleCollationCollection = cst.startWatchingChanges({ + pipeline: [{$changeStream: {}}, {$project: {docId: "$documentKey._id"}}], + collection: simpleCollationCollection +}); + +simpleCollationCollection = + assertCreateCollection(db, simpleCollationCollection, {collation: {locale: "simple"}}); +assert.writeOK(simpleCollationCollection.insert({_id: 0})); + +cst.assertNextChangesEqual( + {cursor: streamCreatedBeforeSimpleCollationCollection, expectedChanges: [{docId: 0}]}); +}()); + +// Test that creating a change stream with a non-default collation, then creating a collection +// with the same collation will not invalidate the change stream. +(function() { +let frenchCollection = "change_stream_french_collation"; +assertDropCollection(db, frenchCollection); + +const frenchChangeStream = cst.startWatchingChanges({ + pipeline: [{$changeStream: {}}, {$project: {docId: "$documentKey._id"}}], + aggregateOptions: {collation: {locale: "fr"}}, + collection: frenchCollection +}); + +frenchCollection = assertCreateCollection(db, frenchCollection, {collation: {locale: "fr"}}); +assert.writeOK(frenchCollection.insert({_id: 0})); + +cst.assertNextChangesEqual({cursor: frenchChangeStream, expectedChanges: [{docId: 0}]}); +}()); + +// Test that creating a change stream with a non-default collation, then creating a collection +// with *a different* collation will not invalidate the change stream. +(function() { +let germanCollection = "change_stream_german_collation"; +assertDropCollection(db, germanCollection); + +const englishCaseInsensitiveStream = cst.startWatchingChanges({ + pipeline: [ + {$changeStream: {}}, + {$match: {"fullDocument.text": "abc"}}, + {$project: {docId: "$documentKey._id"}} + ], + aggregateOptions: {collation: caseInsensitive}, + collection: germanCollection +}); + +germanCollection = assertCreateCollection(db, germanCollection, {collation: {locale: "de"}}); +assert.writeOK(germanCollection.insert({_id: 0, text: "aBc"})); + +cst.assertNextChangesEqual({cursor: englishCaseInsensitiveStream, expectedChanges: [{docId: 0}]}); +}()); + +// Test that creating a change stream with a non-default collation against a collection that has +// a non-simple default collation will use the collation specified on the operation. +(function() { +const caseInsensitiveCollection = assertDropAndRecreateCollection( + db, "change_stream_case_insensitive", {collation: caseInsensitive}); + +const englishCaseSensitiveStream = cst.startWatchingChanges({ + pipeline: [ {$changeStream: {}}, {$match: {"fullDocument.text": "abc"}}, {$project: {docId: "$documentKey._id"}} - ]; - - // Test that $changeStream will not implicitly adopt the default collation of the collection on - // which it is run. Tag this stream as 'doNotModifyInPassthroughs'; whole-db and cluster-wide - // streams do not have default collations. - const didNotInheritCollationStream = cst.startWatchingChanges({ - pipeline: caseInsensitivePipeline, - collection: caseInsensitiveCollection, - doNotModifyInPassthroughs: true - }); - // Test that a collation can be explicitly specified for the $changeStream. This does not need - // to be tagged 'doNotModifyInPassthroughs', since whole-db and cluster-wide changeStreams will - // use an explicit collation if present. - let explicitCaseInsensitiveStream = cst.startWatchingChanges({ - pipeline: caseInsensitivePipeline, - collection: caseInsensitiveCollection, - aggregateOptions: {collation: caseInsensitive} - }); - - assert.writeOK(caseInsensitiveCollection.insert({_id: 0, text: "aBc"})); - assert.writeOK(caseInsensitiveCollection.insert({_id: 1, text: "abc"})); - - // 'didNotInheritCollationStream' should not have inherited the collection's case-insensitive - // default collation, and should only see the second insert. 'explicitCaseInsensitiveStream' - // should see both inserts. - cst.assertNextChangesEqual( - {cursor: didNotInheritCollationStream, expectedChanges: [{docId: 1}]}); - cst.assertNextChangesEqual( - {cursor: explicitCaseInsensitiveStream, expectedChanges: [{docId: 0}, {docId: 1}]}); - - // Test that the collation does not apply to the scan over the oplog. - const similarNameCollection = assertDropAndRecreateCollection( - db, "cHaNgE_sTrEaM_cAsE_iNsEnSiTiVe", {collation: {locale: "en_US"}}); - - // We must recreate the explicitCaseInsensitiveStream and set 'doNotModifyInPassthroughs'. Whole - // db and cluster-wide streams use the simple collation while scanning the oplog, but they don't - // filter the oplog by collection name. The subsequent $match stage which we inject into the - // pipeline to filter for a specific collection will obey the pipeline's case-insensitive - // collation, meaning that 'cHaNgE_sTrEaM_cAsE_iNsEnSiTiVe' will match - // 'change_stream_case_insensitive'. - explicitCaseInsensitiveStream = cst.startWatchingChanges({ - pipeline: caseInsensitivePipeline, - collection: caseInsensitiveCollection, - aggregateOptions: {collation: caseInsensitive}, - doNotModifyInPassthroughs: true - }); - - assert.writeOK(similarNameCollection.insert({_id: 0, text: "aBc"})); - assert.writeOK(caseInsensitiveCollection.insert({_id: 2, text: "ABC"})); - - // The case-insensitive stream should not see the first insert (to the other collection), only - // the second. We do not expect to see the insert in 'didNotInheritCollationStream'. - cst.assertNextChangesEqual( - {cursor: explicitCaseInsensitiveStream, expectedChanges: [{docId: 2}]}); - - // Test that creating a collection without a collation does not invalidate any change streams - // that were opened before the collection existed. - (function() { - let noCollationCollection = "change_stream_no_collation"; - assertDropCollection(db, noCollationCollection); - - const streamCreatedBeforeNoCollationCollection = cst.startWatchingChanges({ - pipeline: [{$changeStream: {}}, {$project: {docId: "$documentKey._id"}}], - collection: noCollationCollection - }); - - noCollationCollection = assertCreateCollection(db, noCollationCollection); - assert.writeOK(noCollationCollection.insert({_id: 0})); - - cst.assertNextChangesEqual( - {cursor: streamCreatedBeforeNoCollationCollection, expectedChanges: [{docId: 0}]}); - }()); - - // Test that creating a collection and explicitly specifying the simple collation does not - // invalidate any change streams that were opened before the collection existed. - (function() { - let simpleCollationCollection = "change_stream_simple_collation"; - assertDropCollection(db, simpleCollationCollection); - - const streamCreatedBeforeSimpleCollationCollection = cst.startWatchingChanges({ - pipeline: [{$changeStream: {}}, {$project: {docId: "$documentKey._id"}}], - collection: simpleCollationCollection - }); - - simpleCollationCollection = - assertCreateCollection(db, simpleCollationCollection, {collation: {locale: "simple"}}); - assert.writeOK(simpleCollationCollection.insert({_id: 0})); - - cst.assertNextChangesEqual( - {cursor: streamCreatedBeforeSimpleCollationCollection, expectedChanges: [{docId: 0}]}); - }()); - - // Test that creating a change stream with a non-default collation, then creating a collection - // with the same collation will not invalidate the change stream. - (function() { - let frenchCollection = "change_stream_french_collation"; - assertDropCollection(db, frenchCollection); - - const frenchChangeStream = cst.startWatchingChanges({ - pipeline: [{$changeStream: {}}, {$project: {docId: "$documentKey._id"}}], - aggregateOptions: {collation: {locale: "fr"}}, - collection: frenchCollection - }); - - frenchCollection = - assertCreateCollection(db, frenchCollection, {collation: {locale: "fr"}}); - assert.writeOK(frenchCollection.insert({_id: 0})); - - cst.assertNextChangesEqual({cursor: frenchChangeStream, expectedChanges: [{docId: 0}]}); - }()); - - // Test that creating a change stream with a non-default collation, then creating a collection - // with *a different* collation will not invalidate the change stream. - (function() { - let germanCollection = "change_stream_german_collation"; - assertDropCollection(db, germanCollection); - - const englishCaseInsensitiveStream = cst.startWatchingChanges({ - pipeline: [ - {$changeStream: {}}, - {$match: {"fullDocument.text": "abc"}}, - {$project: {docId: "$documentKey._id"}} - ], - aggregateOptions: {collation: caseInsensitive}, - collection: germanCollection - }); - - germanCollection = - assertCreateCollection(db, germanCollection, {collation: {locale: "de"}}); - assert.writeOK(germanCollection.insert({_id: 0, text: "aBc"})); - - cst.assertNextChangesEqual( - {cursor: englishCaseInsensitiveStream, expectedChanges: [{docId: 0}]}); - }()); - - // Test that creating a change stream with a non-default collation against a collection that has - // a non-simple default collation will use the collation specified on the operation. - (function() { - const caseInsensitiveCollection = assertDropAndRecreateCollection( - db, "change_stream_case_insensitive", {collation: caseInsensitive}); - - const englishCaseSensitiveStream = cst.startWatchingChanges({ - pipeline: [ - {$changeStream: {}}, - {$match: {"fullDocument.text": "abc"}}, - {$project: {docId: "$documentKey._id"}} - ], - aggregateOptions: {collation: {locale: "en_US"}}, - collection: caseInsensitiveCollection - }); - - assert.writeOK(caseInsensitiveCollection.insert({_id: 0, text: "aBc"})); - assert.writeOK(caseInsensitiveCollection.insert({_id: 1, text: "abc"})); - - cst.assertNextChangesEqual( - {cursor: englishCaseSensitiveStream, expectedChanges: [{docId: 1}]}); - }()); - - // Test that collation is supported by the shell helper. Test that creating a change stream with - // a non-default collation against a collection that has a simple default collation will use the - // collation specified on the operation. - (function() { - const noCollationCollection = - assertDropAndRecreateCollection(db, "change_stream_no_collation"); - - const cursor = noCollationCollection.watch( - [{$match: {"fullDocument.text": "abc"}}, {$project: {docId: "$documentKey._id"}}], - {collation: caseInsensitive}); - assert(!cursor.hasNext()); - assert.writeOK(noCollationCollection.insert({_id: 0, text: "aBc"})); - assert.writeOK(noCollationCollection.insert({_id: 1, text: "abc"})); - assert.soon(() => cursor.hasNext()); - assertChangeStreamEventEq(cursor.next(), {docId: 0}); - assert.soon(() => cursor.hasNext()); - assertChangeStreamEventEq(cursor.next(), {docId: 1}); - assert(!cursor.hasNext()); - }()); - - // Test that we can resume a change stream on a collection that has been dropped without - // requiring the user to explicitly specify the collation. - (function() { - const collName = "change_stream_case_insensitive"; - let caseInsensitiveCollection = - assertDropAndRecreateCollection(db, collName, {collation: caseInsensitive}); - - let changeStream = caseInsensitiveCollection.watch([{$match: {"fullDocument.text": "abc"}}], - {collation: caseInsensitive}); - - assert.writeOK(caseInsensitiveCollection.insert({_id: 0, text: "abc"})); - - assert.soon(() => changeStream.hasNext()); - const next = changeStream.next(); - assert.docEq(next.documentKey, {_id: 0}); - const resumeToken = next._id; - - // Insert a second document to see after resuming. - assert.writeOK(caseInsensitiveCollection.insert({_id: "dropped_coll", text: "ABC"})); - - // Drop the collection to invalidate the stream. - assertDropCollection(db, collName); - - // Test that a $changeStream is allowed to resume on the dropped collection with an explicit - // collation, even if it doesn't match the original collection's default collation. - changeStream = caseInsensitiveCollection.watch( - [{$match: {"fullDocument.text": "ABC"}}], - {resumeAfter: resumeToken, collation: {locale: "simple"}}); - - assert.soon(() => changeStream.hasNext()); - assert.docEq(changeStream.next().documentKey, {_id: "dropped_coll"}); - - // Test that a pipeline without an explicit collation is allowed to resume the change stream - // after the collection has been dropped, and it will use the simple collation. Do not - // modify this in the passthrough suite(s) since only individual collections have the - // concept of a default collation. - const doNotModifyInPassthroughs = true; - const cmdRes = assert.commandWorked(runCommandChangeStreamPassthroughAware( - db, - { - aggregate: collName, - pipeline: [{$changeStream: {resumeAfter: resumeToken}}], - cursor: {}, - }, - doNotModifyInPassthroughs)); - - changeStream = new DBCommandCursor(db, cmdRes); - assert.soon(() => changeStream.hasNext()); - assert.docEq(changeStream.next().documentKey, {_id: "dropped_coll"}); - }()); - - // Test that the default collation of a new version of the collection is not applied when - // resuming a change stream from before a collection drop. - (function() { - const collName = "change_stream_case_insensitive"; - let caseInsensitiveCollection = - assertDropAndRecreateCollection(db, collName, {collation: caseInsensitive}); - - let changeStream = caseInsensitiveCollection.watch([{$match: {"fullDocument.text": "abc"}}], - {collation: caseInsensitive}); - - assert.writeOK(caseInsensitiveCollection.insert({_id: 0, text: "abc"})); - - assert.soon(() => changeStream.hasNext()); - const next = changeStream.next(); - assert.docEq(next.documentKey, {_id: 0}); - const resumeToken = next._id; - - // Insert a second document to see after resuming. - assert.writeOK(caseInsensitiveCollection.insert({_id: "dropped_coll", text: "ABC"})); - - // Recreate the collection with a different collation. - caseInsensitiveCollection = assertDropAndRecreateCollection( - db, caseInsensitiveCollection.getName(), {collation: {locale: "simple"}}); - assert.writeOK(caseInsensitiveCollection.insert({_id: "new collection", text: "abc"})); - - // Verify that the stream sees the insert before the drop and then is exhausted. We won't - // see the invalidate because the pipeline has a $match stage after the $changeStream. - assert.soon(() => changeStream.hasNext()); - assert.docEq(changeStream.next().fullDocument, {_id: "dropped_coll", text: "ABC"}); - // Only single-collection streams will be exhausted from the drop. Use 'next()' instead of - // 'isExhausted()' to force a getMore since the previous getMore may not include the - // collection drop, which is more likely with sharded collections on slow machines. - if (!isChangeStreamPassthrough()) { - assert.throws(() => changeStream.next()); - } - - // Test that a pipeline with an explicit collation is allowed to resume from before the - // collection is dropped and recreated. - changeStream = - caseInsensitiveCollection.watch([{$match: {"fullDocument.text": "ABC"}}], - {resumeAfter: resumeToken, collation: {locale: "fr"}}); - - assert.soon(() => changeStream.hasNext()); - assert.docEq(changeStream.next().documentKey, {_id: "dropped_coll"}); - // Only single-collection streams will be exhausted from the drop. Use 'next()' instead of - // 'isExhausted()' to force a getMore since the previous getMore may not include the - // collection drop, which is more likely with sharded collections on slow machines. - if (!isChangeStreamPassthrough()) { - assert.throws(() => changeStream.next()); - } - - // Test that a pipeline without an explicit collation is allowed to resume, even though the - // collection has been recreated with the same default collation as it had previously. Do - // not modify this command in the passthrough suite(s) since only individual collections - // have the concept of a default collation. - const doNotModifyInPassthroughs = true; - const cmdRes = assert.commandWorked(runCommandChangeStreamPassthroughAware( - db, - { - aggregate: collName, - pipeline: [{$changeStream: {resumeAfter: resumeToken}}], - cursor: {} - }, - doNotModifyInPassthroughs)); - - changeStream = new DBCommandCursor(db, cmdRes); - assert.soon(() => changeStream.hasNext()); - assert.docEq(changeStream.next().documentKey, {_id: "dropped_coll"}); - }()); + ], + aggregateOptions: {collation: {locale: "en_US"}}, + collection: caseInsensitiveCollection +}); + +assert.writeOK(caseInsensitiveCollection.insert({_id: 0, text: "aBc"})); +assert.writeOK(caseInsensitiveCollection.insert({_id: 1, text: "abc"})); + +cst.assertNextChangesEqual({cursor: englishCaseSensitiveStream, expectedChanges: [{docId: 1}]}); +}()); + +// Test that collation is supported by the shell helper. Test that creating a change stream with +// a non-default collation against a collection that has a simple default collation will use the +// collation specified on the operation. +(function() { +const noCollationCollection = assertDropAndRecreateCollection(db, "change_stream_no_collation"); + +const cursor = noCollationCollection.watch( + [{$match: {"fullDocument.text": "abc"}}, {$project: {docId: "$documentKey._id"}}], + {collation: caseInsensitive}); +assert(!cursor.hasNext()); +assert.writeOK(noCollationCollection.insert({_id: 0, text: "aBc"})); +assert.writeOK(noCollationCollection.insert({_id: 1, text: "abc"})); +assert.soon(() => cursor.hasNext()); +assertChangeStreamEventEq(cursor.next(), {docId: 0}); +assert.soon(() => cursor.hasNext()); +assertChangeStreamEventEq(cursor.next(), {docId: 1}); +assert(!cursor.hasNext()); +}()); + +// Test that we can resume a change stream on a collection that has been dropped without +// requiring the user to explicitly specify the collation. +(function() { +const collName = "change_stream_case_insensitive"; +let caseInsensitiveCollection = + assertDropAndRecreateCollection(db, collName, {collation: caseInsensitive}); + +let changeStream = caseInsensitiveCollection.watch([{$match: {"fullDocument.text": "abc"}}], + {collation: caseInsensitive}); + +assert.writeOK(caseInsensitiveCollection.insert({_id: 0, text: "abc"})); + +assert.soon(() => changeStream.hasNext()); +const next = changeStream.next(); +assert.docEq(next.documentKey, {_id: 0}); +const resumeToken = next._id; + +// Insert a second document to see after resuming. +assert.writeOK(caseInsensitiveCollection.insert({_id: "dropped_coll", text: "ABC"})); + +// Drop the collection to invalidate the stream. +assertDropCollection(db, collName); + +// Test that a $changeStream is allowed to resume on the dropped collection with an explicit +// collation, even if it doesn't match the original collection's default collation. +changeStream = + caseInsensitiveCollection.watch([{$match: {"fullDocument.text": "ABC"}}], + {resumeAfter: resumeToken, collation: {locale: "simple"}}); + +assert.soon(() => changeStream.hasNext()); +assert.docEq(changeStream.next().documentKey, {_id: "dropped_coll"}); + +// Test that a pipeline without an explicit collation is allowed to resume the change stream +// after the collection has been dropped, and it will use the simple collation. Do not +// modify this in the passthrough suite(s) since only individual collections have the +// concept of a default collation. +const doNotModifyInPassthroughs = true; +const cmdRes = assert.commandWorked(runCommandChangeStreamPassthroughAware( + db, + { + aggregate: collName, + pipeline: [{$changeStream: {resumeAfter: resumeToken}}], + cursor: {}, + }, + doNotModifyInPassthroughs)); + +changeStream = new DBCommandCursor(db, cmdRes); +assert.soon(() => changeStream.hasNext()); +assert.docEq(changeStream.next().documentKey, {_id: "dropped_coll"}); +}()); + +// Test that the default collation of a new version of the collection is not applied when +// resuming a change stream from before a collection drop. +(function() { +const collName = "change_stream_case_insensitive"; +let caseInsensitiveCollection = + assertDropAndRecreateCollection(db, collName, {collation: caseInsensitive}); + +let changeStream = caseInsensitiveCollection.watch([{$match: {"fullDocument.text": "abc"}}], + {collation: caseInsensitive}); + +assert.writeOK(caseInsensitiveCollection.insert({_id: 0, text: "abc"})); + +assert.soon(() => changeStream.hasNext()); +const next = changeStream.next(); +assert.docEq(next.documentKey, {_id: 0}); +const resumeToken = next._id; + +// Insert a second document to see after resuming. +assert.writeOK(caseInsensitiveCollection.insert({_id: "dropped_coll", text: "ABC"})); + +// Recreate the collection with a different collation. +caseInsensitiveCollection = assertDropAndRecreateCollection( + db, caseInsensitiveCollection.getName(), {collation: {locale: "simple"}}); +assert.writeOK(caseInsensitiveCollection.insert({_id: "new collection", text: "abc"})); + +// Verify that the stream sees the insert before the drop and then is exhausted. We won't +// see the invalidate because the pipeline has a $match stage after the $changeStream. +assert.soon(() => changeStream.hasNext()); +assert.docEq(changeStream.next().fullDocument, {_id: "dropped_coll", text: "ABC"}); +// Only single-collection streams will be exhausted from the drop. Use 'next()' instead of +// 'isExhausted()' to force a getMore since the previous getMore may not include the +// collection drop, which is more likely with sharded collections on slow machines. +if (!isChangeStreamPassthrough()) { + assert.throws(() => changeStream.next()); +} + +// Test that a pipeline with an explicit collation is allowed to resume from before the +// collection is dropped and recreated. +changeStream = + caseInsensitiveCollection.watch([{$match: {"fullDocument.text": "ABC"}}], + {resumeAfter: resumeToken, collation: {locale: "fr"}}); + +assert.soon(() => changeStream.hasNext()); +assert.docEq(changeStream.next().documentKey, {_id: "dropped_coll"}); +// Only single-collection streams will be exhausted from the drop. Use 'next()' instead of +// 'isExhausted()' to force a getMore since the previous getMore may not include the +// collection drop, which is more likely with sharded collections on slow machines. +if (!isChangeStreamPassthrough()) { + assert.throws(() => changeStream.next()); +} + +// Test that a pipeline without an explicit collation is allowed to resume, even though the +// collection has been recreated with the same default collation as it had previously. Do +// not modify this command in the passthrough suite(s) since only individual collections +// have the concept of a default collation. +const doNotModifyInPassthroughs = true; +const cmdRes = assert.commandWorked(runCommandChangeStreamPassthroughAware( + db, + {aggregate: collName, pipeline: [{$changeStream: {resumeAfter: resumeToken}}], cursor: {}}, + doNotModifyInPassthroughs)); + +changeStream = new DBCommandCursor(db, cmdRes); +assert.soon(() => changeStream.hasNext()); +assert.docEq(changeStream.next().documentKey, {_id: "dropped_coll"}); +}()); })(); |