ApsaraDB for MongoDB:Capture data changes in real time using MongoDB change streams

Listen for additional DDL events

Prerequisites

MongoDB 6.0 or later (Upgrade guide).

Procedure

1. Use mongosh to connect to the database.

2. Run the watch command with showExpandedEvents: true:

   // mongo shell or mongosh v1.x
   cursor = db.getSiblingDB("test").watch([],
     {
       showExpandedEvents: true       // Enable listening for more DDL events
     }
   );
   cursor.next();

Verify results

1. In a new SQL window, run a change statement, such as db.createCollection("myCollection1").

2. Check the original SQL window for output related to the executed statement.

   mg xxx test> cursor = db.getSiblingDB("test").watch([],
   ... {
   ...     showExpandedEvents: true     // Enable more DDL event listening
   ... }
   ... );
   ... cursor.next();
   Warning: If there are no documents in the batch, next will block. Use tryNext if you want to check if there are any documents without waiting.
   {
     _id: {
       _data: '8268777614000000012B042C0100296E5A100434919C64814940F3A61F9D97A3F60D5C463C6Fxxx03C63726561746500466F70657274696F6E4465737372697074696F6E0046466964496E6465787800461E76002B04466B657990046E5F6964002B02003C6E616D65003C5F69645F000000004'
     },
     operationType: 'create',
     clusterTime: Timestamp({ t: 1752659476, i: 1 }),
     collectionUUID: UUID('34919c64-8149-40f3-a61f-9xxx'),
     wallTime: ISODate('2025-07-16T09:51:16.043Z'),
     ns: { db: 'test', coll: 'myCollection1' },
     operationDescription: { idIndex: { v: 2, key: { _id: 1 }, name: '_id_' } }
   }

Enable pre-image

A pre-image in MongoDB is a complete snapshot of a document before it is modified or deleted. It records the original values before a change occurs.

Prerequisites

MongoDB 6.0 or later (Upgrade guide).

Procedure

1. Enable pre-image at the database level:

   db.adminCommand({
     setClusterParameter: {
       changeStreamOptions: {
         preAndPostImages: { expireAfterSeconds: "off" } // "off" uses oplog retention period
       }
     }
   })

2. Enable pre-image at the collection level:

   Note

   To enable pre-image for all collections in a database, first enable it at the database level. Then enable it separately for each collection in that database.

   Modify an existing collection

   db.runCommand({
     collMod: "myCollection",
     changeStreamPreAndPostImages: { enabled: true }
   })

   Specify when creating a new collection

   db.createCollection("myCollection", { changeStreamPreAndPostImages: { enabled: true }})

3. Create a change stream listener (specify pre-image options):

   // Create a listener on the target collection
   cursor = db.getCollection("myCollection").watch([],
     {
       fullDocument: 'required', // or 'whenAvailable'
       fullDocumentBeforeChange: 'required' // or 'whenAvailable'
     }
   )
   cursor.next();

   • required: The server must return the pre/post-image. Otherwise, it returns an error.

   • whenAvailable: The server tries to return the image but does not guarantee it.

Verify results

1. Check whether the database-level setting is enabled:

   db.adminCommand( { getClusterParameter: "changeStreamOptions" } )

   If successful, the command returns output like the following:

   test> db.adminCommand( { getClusterParameter: "changeStreamOptions" } )
   {
     clusterParameters: [
       {
         _id: 'changeStreamOptions',
         clusterParameterTime: Timestamp({ t: 1752655937, i: 1 }),
         preAndPostImages: { expireAfterSeconds: Long('100') }
       }
     ],
     ok: 1,
     '$clusterTime': {
       clusterTime: Timestamp({ t: 1752656717, i: 1 }),
       signature: {
         hash: Binary.createFromBase64('xxx=', 0),
         keyId: Long('xxx')
       }
     },
     operationTime: Timestamp({ t: 1752656717, i: 1 })
   }

2. Check the collection configuration:

   db.getCollectionInfos({name: "myCollection"}) // or db.runCommand({listCollections: 1})

   Expected output: Find a field similar to "options" : { "changeStreamPreAndPostImages" : { "enabled" : true } } in the returned document.

   test> db.getCollectionInfos({name: "myCollection"})
   [
     {
       name: 'myCollection',
       type: 'collection',
       options: { changeStreamPreAndPostImages: { enabled: true } },
       info: {
         readOnly: false,
         uuid: UUID('55ed1b7c-7575-4ba3-8afa-xxx')
       },
       idIndex: { v: 2, key: { _id: 1 }, name: '_id_' }
     }
   ]

3. In another mongosh window, update a document in myCollection.

4. Observe the event returned by cursor. It should include the fullDocumentBeforeChange field (the document before the change).

   ... cursor.next();
   {
     _id: {
       _data: '8268786B15000000012B042C0100296E5A100455ED1B7C75754BA38AFA0C5B56A360BC463C6F7065726174696F6E54797065003C7570646174650046646F63756D656E744B65790046xxx697CDDF6EE279xxx'
     },
     operationType: 'update',
     clusterTime: Timestamp({ t: 1752722197, i: 1 }),
     wallTime: ISODate('2025-07-17T03:16:37.626Z'),
     fullDocument: {
       _id: ObjectId('6878697cddfxxx'),
       name: 'test',
       count: 111
     },
     ns: { db: 'test', coll: 'myCollection' },
     documentKey: { _id: ObjectId('6878697cddfxxx') },
     updateDescription: {
       updatedFields: { count: 111 },
       removedFields: [],
       truncatedArrays: []
     },
     fullDocumentBeforeChange: { _id: ObjectId('6878697cddf6xxx'), name: 'test' }
   }

For more information, see Change Streams with Document Pre- and Post-Images.

Enable post-image

A post-image in MongoDB is a complete snapshot of a document after a change occurs. It records the full document content after the change.

Prerequisites

MongoDB 3.6 or later (Upgrade guide).

Procedure

When running the watch command, set fullDocument: 'updateLookup'.

cursor = db.getSiblingDB("test").myCollection.watch([], 
  {
    fullDocument: 'updateLookup'
  }
);
cursor.next();
              

Verify results

1. In another mongosh window, insert or update a document in myCollection.

2. Observe the event returned by cursor. It should include the fullDocument field (the document after the change).

   xxx [primary] test> cursor = db.getSiblingDB("test").myCollection.watch([],
   ... {
   ...     fullDocument: 'updateLookup'
   ... }
   ... );
   ... cursor.next();
   [...
   {
     _id: {
       _data: 'xxx100296E5A10040D6C3FBC28484F08240555576066945463C6F7065726174696F6E54797065003C757064617465500046646F63756D656E744B6579005F6964004B657390046645F696400646881ADE741E7D4B638ED7CA1000004'
     },
     operationType: 'update',
     clusterTime: Timestamp({ t: 1753329166, i: 1 }),
     wallTime: ISODate('2025-07-24T03:52:46.966Z'),
     fullDocument: {
       _id: ObjectId('xxx8ed7ca1'),
       name: 'test1',
       age: 12,
       count: 2222
     },
     ns: { db: 'test', coll: 'myCollection' },
     documentKey: { _id: ObjectId('xxx7ca1') },
     updateDescription: {
       updatedFields: { count: 2222 },
       removedFields: [],
       truncatedArrays: []
     }
   }]

Handle very large change events (>16 MB)

Prerequisites

MongoDB 7.0 or later (Upgrade guide).

Procedure

Include the $changeStreamSplitLargeEvent stage in the pipeline of the watch() command:

myChangeStreamCursor = db.myCollection.watch(
  [ { $changeStreamSplitLargeEvent: {} } ], // Add split stage
  {
    fullDocument: "required",
    fullDocumentBeforeChange: "required"
  }
)

Verify results

• Perform an operation that generates a change event larger than 16 MB (for example, updating a document containing a very large array).

• Observe that the returned event stream is split into multiple consecutive fragment events, ending with a final fragment event.

Reduce pre-image storage overhead

By default, pre-images expire along with the oplog. Set a shorter expiration time to save space:

db.adminCommand({
  setClusterParameter: {
    changeStreamOptions: {
      preAndPostImages: { expireAfterSeconds: 100 } // Unit: seconds
    }
  }
})

1. Why do change stream slow logs always show COLLSCAN?

This behavior is normal and expected. No optimization is needed. A change stream cursor ultimately reads from local.oplog.rs (the only place that records all instance modifications). This collection has no indexes, so it always performs a COLLSCAN. This cannot be avoided and offers no room for optimization.

If you use COLLSCAN as a keyword to filter slow logs while searching for queries to optimize, also filter out the $changeStream keyword.

Only investigate change stream–related slow logs if you encounter performance issues (for example, increasing latency in downstream consumption).

2. Why, on a sharded instance, do I see change stream cursors on shards other than the primary shard when listening to an unsharded collection?

This prepares for possible shardCollection operations. You can convert an unsharded collection with an active listener into a sharded collection at any time, distributing its data across all shards (similar to movePrimary). Creating change stream cursors on other shards in advance handles this scenario. Typically, these cursors on other shards return no change events and incur minimal performance overhead.

Similarly, to handle potential addShard/removeShard operations, when you set up a listener on a sharded instance, mongos also creates the corresponding cursors on the Config Server.

3. Why does the slow log for a change stream cursor appear on the primary node even though I specified readPreference:secondary when creating the cursor?

A cursor becomes "pinned" to a specific node after creation and does not automatically migrate when node roles change. Once a change stream cursor is created, it remains fixed on a specific mongod node (primary or secondary) and continues consuming via getMore. After events such as primary/secondary switchover, instance resizing, or migration, a cursor originally on a secondary might end up on the primary.

If you do not want this load to persist on the primary node, clean up the relevant change stream cursors using killCursors. Your downstream consumer logic can then restore the cursor to a secondary node using the resumeToken and readPreference. This operation does not cause event loss or interruption.

db.runCommand(
   {
     killCursors: <collection>,
     cursors: [ <cursor id1>, ... ], comment: <any>
   }
)
db.getSiblingDB("<testDB>").runCommand( { killCursors: "<testColl>", cursors: [NumberLong("2452840976689696187") ] } ) 

4. Why do I see many slow logs about change streams on mongos that last 1000 ms?

2020-08-26T04:34:45.045+0000 I COMMAND [conn21283] command altconfig-b2b-perf.oplog command: getMore \{ getMore: 3513599116181216748, collection: "oplog", $db: "altconfig-b2b-perf", $clusterTime: { clusterTime: Timestamp(1598416483, 1), signature: { hash: BinData(0, EC3841EB1FB7A34F897688BB5983E32E2ADF6763), keyId: 6855385090000683010 } }, lsid: \{ id: UUID("72864d11-c0f8-48bb-823e-de5f18a9c409") } } originatingCommand: \{ find: "oplog", filter: { timestamp: { $gte: new Date(1597895390009) } }, tailable: true, awaitData: true, $db: "altconfig-b2b-perf", $clusterTime: \{ clusterTime: Timestamp(1597895487, 1), signature: { hash: BinData(0, AAAE33D5688F935C70469CBDB8EB1F6882749A40), keyId: 6855385090000683010 } }, lsid: \{ id: UUID("72864d11-c0f8-48bb-823e-de5f18a9c409") } } nShards:1 cursorid:3513599116181216748 numYields:0 nreturned:0 reslen:237 protocol:op_msg 1000ms

This is normal behavior in MongoDB versions earlier than 6.0. It indicates a wait timeout, not a performance bottleneck. In sharded cluster architectures, mongos creates change stream cursors on shards with tailable:true, awaitData:true, and maxTimeMS:1000 to return as many change events as possible within 1000 ms.

In major versions before 6.0, mongos logs these 1000 ms slow logs, which can mislead users. MongoDB has optimized this behavior. For details, see SERVER-50559.