PHPLIB-418: Add the ability to specify a pipeline to an update command

Author: alcaeus

docs/includes/apiargs-MongoDBCollection-common-param.yaml

@@ -24,7 +24,9 @@ type: array|object
 description: |
   Specifies the field and value combinations to update and any relevant update
   operators. ``$update`` uses MongoDB's :method:`update operators
-  </reference/operator/update>`.
+  </reference/operator/update>`. Starting with MongoDB 4.2, an `aggregation
+  pipeline <https://docs.mongodb.com/master/reference/command/update/#update-with-an-aggregation-pipeline>`_
+  can be passed as this parameter.
 interface: phpmethod
 operation: ~
 optional: false

src/Operation/FindAndModify.php

@@ -31,6 +31,7 @@
 use function is_integer;
 use function is_object;
 use function MongoDB\create_field_path_type_map;
+use function MongoDB\is_pipeline;
 use function MongoDB\server_supports_feature;
 
 /**
@@ -255,12 +256,18 @@ private function createCommandDocument(Server $server)
             $cmd['upsert'] = $this->options['upsert'];
         }
 
-        foreach (['collation', 'fields', 'query', 'sort', 'update'] as $option) {
+        foreach (['collation', 'fields', 'query', 'sort'] as $option) {
             if (isset($this->options[$option])) {
                 $cmd[$option] = (object) $this->options[$option];
             }
         }
 
+        if (isset($this->options['update'])) {
+            $cmd['update'] = is_pipeline($this->options['update'])
+                ? $this->options['update']
+                : (object) $this->options['update'];
+        }
+
         if (isset($this->options['arrayFilters'])) {
             $cmd['arrayFilters'] = $this->options['arrayFilters'];
         }

src/Operation/FindOneAndUpdate.php

@@ -25,6 +25,7 @@
 use function is_integer;
 use function is_object;
 use function MongoDB\is_first_key_operator;
+use function MongoDB\is_pipeline;
 
 /**
  * Operation for updating a document with the findAndModify command.
@@ -105,8 +106,8 @@ public function __construct($databaseName, $collectionName, $filter, $update, ar
             throw InvalidArgumentException::invalidType('$update', $update, 'array or object');
         }
 
-        if (! is_first_key_operator($update)) {
-            throw new InvalidArgumentException('First key in $update argument is not an update operator');
+        if (! is_first_key_operator($update) && ! is_pipeline($update)) {
+            throw new InvalidArgumentException('Expected an update document with operator as first key or a pipeline');
         }
 
         $options += [

src/Operation/ReplaceOne.php

@@ -25,6 +25,7 @@
 use function is_array;
 use function is_object;
 use function MongoDB\is_first_key_operator;
+use function MongoDB\is_pipeline;
 
 /**
  * Operation for replacing a single document with the update command.
@@ -79,6 +80,10 @@ public function __construct($databaseName, $collectionName, $filter, $replacemen
             throw new InvalidArgumentException('First key in $replacement argument is an update operator');
         }
 
+        if (is_pipeline($replacement)) {
+            throw new InvalidArgumentException('$replacement argument is a pipeline');
+        }
+
         $this->update = new Update(
             $databaseName,
             $collectionName,

src/Operation/Update.php

@@ -29,6 +29,7 @@
 use function is_bool;
 use function is_object;
 use function MongoDB\is_first_key_operator;
+use function MongoDB\is_pipeline;
 use function MongoDB\server_supports_feature;
 
 /**
@@ -126,7 +127,7 @@ public function __construct($databaseName, $collectionName, $filter, $update, ar
             throw InvalidArgumentException::invalidType('"multi" option', $options['multi'], 'boolean');
         }
 
-        if ($options['multi'] && ! is_first_key_operator($update)) {
+        if ($options['multi'] && ! is_first_key_operator($update) && ! is_pipeline($update)) {
             throw new InvalidArgumentException('"multi" option cannot be true if $update is a replacement document');
         }
 

src/Operation/UpdateMany.php

@@ -25,6 +25,7 @@
 use function is_array;
 use function is_object;
 use function MongoDB\is_first_key_operator;
+use function MongoDB\is_pipeline;
 
 /**
  * Operation for updating multiple documents with the update command.
@@ -81,8 +82,8 @@ public function __construct($databaseName, $collectionName, $filter, $update, ar
             throw InvalidArgumentException::invalidType('$update', $update, 'array or object');
         }
 
-        if (! is_first_key_operator($update)) {
-            throw new InvalidArgumentException('First key in $update argument is not an update operator');
+        if (! is_first_key_operator($update) && ! is_pipeline($update)) {
+            throw new InvalidArgumentException('Expected an update document with operator as first key or a pipeline');
         }
 
         $this->update = new Update(

src/Operation/UpdateOne.php

@@ -25,6 +25,7 @@
 use function is_array;
 use function is_object;
 use function MongoDB\is_first_key_operator;
+use function MongoDB\is_pipeline;
 
 /**
  * Operation for updating a single document with the update command.
@@ -81,8 +82,8 @@ public function __construct($databaseName, $collectionName, $filter, $update, ar
             throw InvalidArgumentException::invalidType('$update', $update, 'array or object');
         }
 
-        if (! is_first_key_operator($update)) {
-            throw new InvalidArgumentException('First key in $update argument is not an update operator');
+        if (! is_first_key_operator($update) && ! is_pipeline($update)) {
+            throw new InvalidArgumentException('Expected an update document with operator as first key or a pipeline');
         }
 
         $this->update = new Update(

src/functions.php

@@ -119,6 +119,47 @@ function is_first_key_operator($document)
     return isset($firstKey[0]) && $firstKey[0] === '$';
 }
 
+/**
+ * Returns whether an update specification is a valid aggregation pipeline.
+ *
+ * @internal
+ * @param mixed $pipeline
+ * @return boolean
+ */
+function is_pipeline($pipeline)
+{
+    if (! is_array($pipeline)) {
+        return false;
+    }
+
+    if ($pipeline === []) {
+        return false;
+    }
+
+    $expectedKey = 0;
+
+    foreach ($pipeline as $key => $stage) {
+        if (! is_array($stage) && ! is_object($stage)) {
+            return false;
+        }
+
+        if ($expectedKey !== $key) {
+            return false;
+        }
+
+        $expectedKey++;
+        $stage = (array) $stage;
+        reset($stage);
+        $key = key($stage);
+
+        if (! isset($key[0]) || $key[0] !== '$') {
+            return false;
+        }
+    }
+
+    return true;
+}
+
 /**
  * Returns whether we are currently in a transaction.
  *

tests/FunctionsTest.php

@@ -10,6 +10,7 @@
 use function MongoDB\generate_index_name;
 use function MongoDB\is_first_key_operator;
 use function MongoDB\is_mapreduce_output_inline;
+use function MongoDB\is_pipeline;
 
 /**
  * Unit tests for utility functions.
@@ -224,4 +225,33 @@ public function provideTypeMapValues()
             ],
         ];
     }
+
+    /**
+     * @dataProvider providePipelines
+     */
+    public function testIsPipeline($expected, $pipeline)
+    {
+        $this->assertSame($expected, is_pipeline($pipeline));
+    }
+
+    public function providePipelines()
+    {
+        return [
+            'Not an array' => [false, (object) []],
+            'Empty array' => [false, []],
+            'Non-sequential indexes in array' => [false, [1 => ['$group' => []]]],
+            'Update document instead of pipeline' => [false, ['$set' => ['foo' => 'bar']]],
+            'Invalid pipeline stage' => [false, [['group' => []]]],
+            'Update with DbRef' => [false, ['x' => ['$ref' => 'foo', '$id' => 'bar']]],
+            'Valid pipeline' => [
+                true,
+                [
+                    ['$match' => ['foo' => 'bar']],
+                    ['$group' => ['_id' => 1]],
+                ],
+            ],
+            'False positive with DbRef in numeric field' => [true, ['0' => ['$ref' => 'foo', '$id' => 'bar']]],
+            'DbRef in numeric field as object' => [false, (object) ['0' => ['$ref' => 'foo', '$id' => 'bar']]],
+        ];
+    }
 }

tests/Operation/FindOneAndUpdateTest.php

@@ -25,10 +25,10 @@ public function testConstructorUpdateArgumentTypeCheck($update)
         new FindOneAndUpdate($this->getDatabaseName(), $this->getCollectionName(), [], $update);
     }
 
-    public function testConstructorUpdateArgumentRequiresOperators()
+    public function testConstructorUpdateArgumentRequiresOperatorsOrPipeline()
     {
         $this->expectException(InvalidArgumentException::class);
-        $this->expectExceptionMessage('First key in $update argument is not an update operator');
+        $this->expectExceptionMessage('Expected an update document with operator as first key or a pipeline');
         new FindOneAndUpdate($this->getDatabaseName(), $this->getCollectionName(), [], []);
     }