PHPLIB-481: Add the ability to specify a pipeline to an update within bulk write

src/Operation/BulkWrite.php

@@ -33,6 +33,7 @@ use function is_bool;
 use function is_object;
 use function key;
 use function MongoDB\is_first_key_operator;
+use function MongoDB\is_pipeline;
 use function MongoDB\server_supports_feature;
 use function sprintf;
 
@@ -255,8 +256,8 @@ class BulkWrite implements Executable
                         throw InvalidArgumentException::invalidType(sprintf('$operations[%d]["%s"][1]', $i, $type), $args[1], 'array or object');
                     }
 
-                    if (! is_first_key_operator($args[1])) {
-                        throw new InvalidArgumentException(sprintf('First key in $operations[%d]["%s"][1] is not an update operator', $i, $type));
+                    if (! is_first_key_operator($args[1]) && ! is_pipeline($args[1])) {
+                        throw new InvalidArgumentException(sprintf('First key in $operations[%d]["%s"][1] is neither an update operator nor a pipeline', $i, $type));
                     }
 
                     if (! isset($args[2])) {

tests/Operation/BulkWriteFunctionalTest.php

@@ -263,7 +263,7 @@ class BulkWriteFunctionalTest extends FunctionalTestCase
     public function testUpdateOneRequiresUpdateOperators()
     {
         $this->expectException(InvalidArgumentException::class);
-        $this->expectExceptionMessage('First key in $operations[0]["updateOne"][1] is not an update operator');
+        $this->expectExceptionMessage('First key in $operations[0]["updateOne"][1] is neither an update operator nor a pipeline');
         new BulkWrite($this->getDatabaseName(), $this->getCollectionName(), [
             ['updateOne' => [['_id' => 1], ['x' => 1]]],
         ]);
@@ -272,7 +272,7 @@ class BulkWriteFunctionalTest extends FunctionalTestCase
     public function testUpdateManyRequiresUpdateOperators()
     {
         $this->expectException(InvalidArgumentException::class);
-        $this->expectExceptionMessage('First key in $operations[0]["updateMany"][1] is not an update operator');
+        $this->expectExceptionMessage('First key in $operations[0]["updateMany"][1] is neither an update operator nor a pipeline');
         new BulkWrite($this->getDatabaseName(), $this->getCollectionName(), [
             ['updateMany' => [['_id' => ['$gt' => 1]], ['x' => 1]]],
         ]);

tests/Operation/BulkWriteTest.php

@@ -258,10 +258,10 @@ class BulkWriteTest extends TestCase
         ]);
     }
 
-    public function testUpdateManyUpdateArgumentRequiresOperators()
+    public function testUpdateManyUpdateArgumentRequiresOperatorsOrPipeline()
     {
         $this->expectException(InvalidArgumentException::class);
-        $this->expectExceptionMessage('First key in $operations[0]["updateMany"][1] is not an update operator');
+        $this->expectExceptionMessage('First key in $operations[0]["updateMany"][1] is neither an update operator nor a pipeline');
         new BulkWrite($this->getDatabaseName(), $this->getCollectionName(), [
             [BulkWrite::UPDATE_MANY => [['_id' => ['$gt' => 1]], ['x' => 1]]],
         ]);
@@ -345,10 +345,10 @@ class BulkWriteTest extends TestCase
         ]);
     }
 
-    public function testUpdateOneUpdateArgumentRequiresOperators()
+    public function testUpdateOneUpdateArgumentRequiresOperatorsOrPipeline()
     {
         $this->expectException(InvalidArgumentException::class);
-        $this->expectExceptionMessage('First key in $operations[0]["updateOne"][1] is not an update operator');
+        $this->expectExceptionMessage('First key in $operations[0]["updateOne"][1] is neither an update operator nor a pipeline');
         new BulkWrite($this->getDatabaseName(), $this->getCollectionName(), [
             [BulkWrite::UPDATE_ONE => [['_id' => 1], ['x' => 1]]],
         ]);

tests/SpecTests/crud/updateWithPipelines.json

@@ -238,6 +238,171 @@
           ]
         }
       }
+    },
+    {
+      "description": "UpdateOne in bulk write using pipelines",
+      "operations": [
+        {
+          "name": "bulkWrite",
+          "arguments": {
+            "requests": [
+              {
+                "name": "updateOne",
+                "arguments": {
+                  "filter": {
+                    "_id": 1
+                  },
+                  "update": [
+                    {
+                      "$replaceRoot": {
+                        "newRoot": "$t"
+                      }
+                    },
+                    {
+                      "$addFields": {
+                        "foo": 1
+                      }
+                    }
+                  ]
+                }
+              }
+            ]
+          },
+          "result": {
+            "matchedCount": 1,
+            "modifiedCount": 1,
+            "upsertedCount": 0
+          }
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "update": "test",
+              "updates": [
+                {
+                  "q": {
+                    "_id": 1
+                  },
+                  "u": [
+                    {
+                      "$replaceRoot": {
+                        "newRoot": "$t"
+                      }
+                    },
+                    {
+                      "$addFields": {
+                        "foo": 1
+                      }
+                    }
+                  ]
+                }
+              ]
+            },
+            "command_name": "update",
+            "database_name": "crud-tests"
+          }
+        }
+      ],
+      "outcome": {
+        "collection": {
+          "data": [
+            {
+              "_id": 1,
+              "u": {
+                "v": 1
+              },
+              "foo": 1
+            },
+            {
+              "_id": 2,
+              "x": 2,
+              "y": 1
+            }
+          ]
+        }
+      }
+    },
+    {
+      "description": "UpdateMany in bulk write using pipelines",
+      "operations": [
+        {
+          "name": "bulkWrite",
+          "arguments": {
+            "requests": [
+              {
+                "name": "updateMany",
+                "arguments": {
+                  "filter": {},
+                  "update": [
+                    {
+                      "$project": {
+                        "x": 1
+                      }
+                    },
+                    {
+                      "$addFields": {
+                        "foo": 1
+                      }
+                    }
+                  ]
+                }
+              }
+            ]
+          },
+          "result": {
+            "matchedCount": 2,
+            "modifiedCount": 2,
+            "upsertedCount": 0
+          }
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "update": "test",
+              "updates": [
+                {
+                  "q": {},
+                  "u": [
+                    {
+                      "$project": {
+                        "x": 1
+                      }
+                    },
+                    {
+                      "$addFields": {
+                        "foo": 1
+                      }
+                    }
+                  ],
+                  "multi": true
+                }
+              ]
+            },
+            "command_name": "update",
+            "database_name": "crud-tests"
+          }
+        }
+      ],
+      "outcome": {
+        "collection": {
+          "data": [
+            {
+              "_id": 1,
+              "x": 1,
+              "foo": 1
+            },
+            {
+              "_id": 2,
+              "x": 2,
+              "foo": 1
+            }
+          ]
+        }
+      }
     }
   ]
 }