PHPLIB-846: Improved change stream event visibility for C2C Replication (#949)

Spec tests synced with mongodb/specifications@8da1a89

Co-authored-by: Jeremy Mikola <jmikola@gmail.com>

Author: aleksandr-rudo

docs/includes/apiargs-MongoDBClient-method-watch-option.yaml

@@ -49,6 +49,12 @@ source:
   file: apiargs-common-option.yaml
   ref: session
 ---
+source:
+  file: apiargs-method-watch-option.yaml
+  ref: showExpandedEvents
+post: |
+  .. versionadded:: 1.13
+---
 source:
   file: apiargs-method-watch-option.yaml
   ref: startAfter

docs/includes/apiargs-MongoDBCollection-method-watch-option.yaml

@@ -49,6 +49,12 @@ source:
   file: apiargs-common-option.yaml
   ref: session
 ---
+source:
+  file: apiargs-method-watch-option.yaml
+  ref: showExpandedEvents
+post: |
+  .. versionadded:: 1.13
+---
 source:
   file: apiargs-method-watch-option.yaml
   ref: startAfter

docs/includes/apiargs-MongoDBDatabase-method-watch-option.yaml

@@ -49,6 +49,12 @@ source:
   file: apiargs-common-option.yaml
   ref: session
 ---
+source:
+  file: apiargs-method-watch-option.yaml
+  ref: showExpandedEvents
+post: |
+  .. versionadded:: 1.13
+---
 source:
   file: apiargs-method-watch-option.yaml
   ref: startAfter

docs/includes/apiargs-method-watch-option.yaml

@@ -97,6 +97,31 @@ operation: ~
 optional: true
 ---
 arg_name: option
+name: showExpandedEvents
+type: boolean
+description: |
+  If true, instructs the server to include additional DDL events in the change
+  stream. The additional events that may be included are:
+
+  - ``createIndexes``
+  - ``dropIndexes``
+  - ``modify``
+  - ``create``
+  - ``shardCollection``
+  - ``reshardCollection`` (server 6.1+)
+  - ``refineCollectionShardKey`` (server 6.1+)
+
+  This is not supported for server versions prior to 6.0 and will result in an
+  exception at execution time if used.
+
+  .. note::
+
+     This is an option of the ``$changeStream`` pipeline stage.
+interface: phpmethod
+operation: ~
+optional: true
+---
+arg_name: option
 name: startAfter
 type: array|object
 description: |

src/Operation/Watch.php

@@ -38,6 +38,7 @@
 use function array_unshift;
 use function count;
 use function is_array;
+use function is_bool;
 use function is_object;
 use function is_string;
 use function MongoDB\Driver\Monitoring\addSubscriber;
@@ -161,6 +162,11 @@ class Watch implements Executable, /* @internal */ CommandSubscriber
      *
      *  * session (MongoDB\Driver\Session): Client session.
      *
+     *  * showExpandedEvents (boolean): Enables the server to send the expanded
+     *    list of change stream events.
+     *
+     *    This option is not supported for server versions < 6.0.
+     *
      *  * startAfter (document): Specifies the logical starting point for the
      *    new change stream. Unlike "resumeAfter", this option can be used with
      *    a resume token from an "invalidate" event.
@@ -229,6 +235,10 @@ public function __construct(Manager $manager, $databaseName, $collectionName, ar
             throw InvalidArgumentException::invalidType('"startAtOperationTime" option', $options['startAtOperationTime'], TimestampInterface::class);
         }
 
+        if (isset($options['showExpandedEvents']) && ! is_bool($options['showExpandedEvents'])) {
+            throw InvalidArgumentException::invalidType('"showExpandedEvents" option', $options['showExpandedEvents'], 'bool');
+        }
+
         /* In the absence of an explicit session, create one to ensure that the
          * initial aggregation and any resume attempts can use the same session
          * ("implicit from the user's perspective" per PHPLIB-342). Since this
@@ -244,7 +254,7 @@ public function __construct(Manager $manager, $databaseName, $collectionName, ar
         }
 
         $this->aggregateOptions = array_intersect_key($options, ['batchSize' => 1, 'collation' => 1, 'comment' => 1, 'maxAwaitTimeMS' => 1, 'readConcern' => 1, 'readPreference' => 1, 'session' => 1, 'typeMap' => 1]);
-        $this->changeStreamOptions = array_intersect_key($options, ['fullDocument' => 1, 'fullDocumentBeforeChange' => 1, 'resumeAfter' => 1, 'startAfter' => 1, 'startAtOperationTime' => 1]);
+        $this->changeStreamOptions = array_intersect_key($options, ['fullDocument' => 1, 'fullDocumentBeforeChange' => 1, 'resumeAfter' => 1, 'showExpandedEvents' => 1, 'startAfter' => 1, 'startAtOperationTime' => 1]);
 
         // Null database name implies a cluster-wide change stream
         if ($databaseName === null) {

tests/UnifiedSpecTests/Operation.php

@@ -491,7 +491,11 @@ function (IndexInfo $info) {
                 assertArrayHasKey('to', $args);
                 assertIsString($args['to']);
 
-                return $collection->rename($args['to']);
+                return $collection->rename(
+                    $args['to'],
+                    null, /* $toDatabaseName */
+                    array_diff_key($args, ['to' => 1])
+                );
 
             default:
                 Assert::fail('Unsupported collection operation: ' . $this->name);

tests/UnifiedSpecTests/Util.php

@@ -55,13 +55,13 @@ final class Util
             'loop' => ['operations', 'storeErrorsAsEntity', 'storeFailuresAsEntity', 'storeSuccessesAsEntity', 'storeIterationsAsEntity'],
         ],
         Client::class => [
-            'createChangeStream' => ['pipeline', 'session', 'fullDocument', 'resumeAfter', 'startAfter', 'startAtOperationTime', 'batchSize', 'collation', 'maxAwaitTimeMS'],
+            'createChangeStream' => ['pipeline', 'session', 'fullDocument', 'resumeAfter', 'startAfter', 'startAtOperationTime', 'batchSize', 'collation', 'maxAwaitTimeMS', 'showExpandedEvents'],
             'listDatabaseNames' => ['authorizedDatabases', 'filter', 'maxTimeMS', 'session'],
             'listDatabases' => ['authorizedDatabases', 'filter', 'maxTimeMS', 'session'],
         ],
         Database::class => [
             'aggregate' => ['pipeline', 'session', 'useCursor', 'allowDiskUse', 'batchSize', 'bypassDocumentValidation', 'collation', 'comment', 'explain', 'hint', 'let', 'maxAwaitTimeMS', 'maxTimeMS'],
-            'createChangeStream' => ['pipeline', 'session', 'fullDocument', 'resumeAfter', 'startAfter', 'startAtOperationTime', 'batchSize', 'collation', 'maxAwaitTimeMS'],
+            'createChangeStream' => ['pipeline', 'session', 'fullDocument', 'resumeAfter', 'startAfter', 'startAtOperationTime', 'batchSize', 'collation', 'maxAwaitTimeMS', 'showExpandedEvents'],
             'createCollection' => ['collection', 'session', 'autoIndexId', 'capped', 'changeStreamPreAndPostImages', 'clusteredIndex', 'collation', 'expireAfterSeconds', 'flags', 'indexOptionDefaults', 'max', 'maxTimeMS', 'pipeline', 'size', 'storageEngine', 'timeseries', 'validationAction', 'validationLevel', 'validator', 'viewOn'],
             'dropCollection' => ['collection', 'session'],
             'listCollectionNames' => ['authorizedCollections', 'filter', 'maxTimeMS', 'session'],
@@ -73,7 +73,7 @@ final class Util
         Collection::class => [
             'aggregate' => ['pipeline', 'session', 'useCursor', 'allowDiskUse', 'batchSize', 'bypassDocumentValidation', 'collation', 'comment', 'explain', 'hint', 'let', 'maxAwaitTimeMS', 'maxTimeMS'],
             'bulkWrite' => ['let', 'requests', 'session', 'ordered', 'bypassDocumentValidation', 'comment'],
-            'createChangeStream' => ['pipeline', 'session', 'fullDocument', 'fullDocumentBeforeChange', 'resumeAfter', 'startAfter', 'startAtOperationTime', 'batchSize', 'collation', 'maxAwaitTimeMS', 'comment'],
+            'createChangeStream' => ['pipeline', 'session', 'fullDocument', 'fullDocumentBeforeChange', 'resumeAfter', 'startAfter', 'startAtOperationTime', 'batchSize', 'collation', 'maxAwaitTimeMS', 'comment', 'showExpandedEvents'],
             'createFindCursor' => ['filter', 'session', 'allowDiskUse', 'allowPartialResults', 'batchSize', 'collation', 'comment', 'cursorType', 'hint', 'limit', 'max', 'maxAwaitTimeMS', 'maxScan', 'maxTimeMS', 'min', 'modifiers', 'noCursorTimeout', 'oplogReplay', 'projection', 'returnKey', 'showRecordId', 'skip', 'snapshot', 'sort'],
             'createIndex' => ['keys', 'commitQuorum', 'maxTimeMS', 'name', 'session', 'comment'],
             'dropIndex' => ['name', 'session', 'maxTimeMS', 'comment'],
@@ -88,7 +88,7 @@ final class Util
             'find' => ['let', 'filter', 'session', 'allowDiskUse', 'allowPartialResults', 'batchSize', 'collation', 'comment', 'cursorType', 'hint', 'limit', 'max', 'maxAwaitTimeMS', 'maxScan', 'maxTimeMS', 'min', 'modifiers', 'noCursorTimeout', 'oplogReplay', 'projection', 'returnKey', 'showRecordId', 'skip', 'snapshot', 'sort'],
             'findOne' => ['let', 'filter', 'session', 'allowDiskUse', 'allowPartialResults', 'batchSize', 'collation', 'comment', 'cursorType', 'hint', 'max', 'maxAwaitTimeMS', 'maxScan', 'maxTimeMS', 'min', 'modifiers', 'noCursorTimeout', 'oplogReplay', 'projection', 'returnKey', 'showRecordId', 'skip', 'snapshot', 'sort'],
             'findOneAndReplace' => ['let', 'returnDocument', 'filter', 'replacement', 'session', 'projection', 'returnDocument', 'upsert', 'arrayFilters', 'bypassDocumentValidation', 'collation', 'hint', 'maxTimeMS', 'new', 'remove', 'sort', 'comment'],
-            'rename' => ['to', 'comment'],
+            'rename' => ['to', 'comment', 'dropTarget'],
             'replaceOne' => ['let', 'filter', 'replacement', 'session', 'upsert', 'arrayFilters', 'bypassDocumentValidation', 'collation', 'hint', 'comment'],
             'findOneAndUpdate' => ['let', 'returnDocument', 'filter', 'update', 'session', 'upsert', 'projection', 'remove', 'arrayFilters', 'bypassDocumentValidation', 'collation', 'hint', 'maxTimeMS', 'sort', 'comment'],
             'updateMany' => ['let', 'filter', 'update', 'session', 'upsert', 'arrayFilters', 'bypassDocumentValidation', 'collation', 'hint', 'comment'],