Merge pull request #684 from alcaeus/fix-bulkwrite-pipeline

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

@@ -8,7 +8,6 @@ use MongoDB\Collection;
 use MongoDB\Driver\BulkWrite as Bulk;
 use MongoDB\Driver\WriteConcern;
 use MongoDB\Exception\BadMethodCallException;
-use MongoDB\Exception\InvalidArgumentException;
 use MongoDB\Model\BSONDocument;
 use MongoDB\Operation\BulkWrite;
 use MongoDB\Tests\CommandObserver;
@@ -226,67 +225,6 @@ class BulkWriteFunctionalTest extends FunctionalTestCase
         $result->getUpsertedIds();
     }
 
-    public function testUnknownOperation()
-    {
-        $this->expectException(InvalidArgumentException::class);
-        $this->expectExceptionMessage('Unknown operation type "foo" in $operations[0]');
-        new BulkWrite($this->getDatabaseName(), $this->getCollectionName(), [
-            ['foo' => [['_id' => 1]]],
-        ]);
-    }
-
-    /**
-     * @dataProvider provideOpsWithMissingArguments
-     */
-    public function testMissingArguments(array $ops)
-    {
-        $this->expectException(InvalidArgumentException::class);
-        $this->expectExceptionMessageRegExp('/Missing (first|second) argument for \$operations\[\d+\]\["\w+\"]/');
-        new BulkWrite($this->getDatabaseName(), $this->getCollectionName(), $ops);
-    }
-
-    public function provideOpsWithMissingArguments()
-    {
-        return [
-            [[['insertOne' => []]]],
-            [[['updateOne' => []]]],
-            [[['updateOne' => [['_id' => 1]]]]],
-            [[['updateMany' => []]]],
-            [[['updateMany' => [['_id' => 1]]]]],
-            [[['replaceOne' => []]]],
-            [[['replaceOne' => [['_id' => 1]]]]],
-            [[['deleteOne' => []]]],
-            [[['deleteMany' => []]]],
-        ];
-    }
-
-    public function testUpdateOneRequiresUpdateOperators()
-    {
-        $this->expectException(InvalidArgumentException::class);
-        $this->expectExceptionMessage('First key in $operations[0]["updateOne"][1] is not an update operator');
-        new BulkWrite($this->getDatabaseName(), $this->getCollectionName(), [
-            ['updateOne' => [['_id' => 1], ['x' => 1]]],
-        ]);
-    }
-
-    public function testUpdateManyRequiresUpdateOperators()
-    {
-        $this->expectException(InvalidArgumentException::class);
-        $this->expectExceptionMessage('First key in $operations[0]["updateMany"][1] is not an update operator');
-        new BulkWrite($this->getDatabaseName(), $this->getCollectionName(), [
-            ['updateMany' => [['_id' => ['$gt' => 1]], ['x' => 1]]],
-        ]);
-    }
-
-    public function testReplaceOneRequiresReplacementDocument()
-    {
-        $this->expectException(InvalidArgumentException::class);
-        $this->expectExceptionMessage('First key in $operations[0]["replaceOne"][1] is an update operator');
-        new BulkWrite($this->getDatabaseName(), $this->getCollectionName(), [
-            ['replaceOne' => [['_id' => 1], ['$inc' => ['x' => 1]]]],
-        ]);
-    }
-
     public function testSessionOption()
     {
         if (version_compare($this->getServerVersion(), '3.6.0', '<')) {
@@ -357,6 +295,36 @@ class BulkWriteFunctionalTest extends FunctionalTestCase
         );
     }
 
+    public function testBulkWriteWithPipelineUpdates()
+    {
+        if (version_compare($this->getServerVersion(), '4.2.0', '<')) {
+            $this->markTestSkipped('Pipeline-style updates are not supported');
+        }
+
+        $this->createFixtures(4);
+
+        $ops = [
+            ['updateOne' => [['_id' => 2], [['$addFields' => ['y' => 2]]]]],
+            ['updateMany' => [['_id' => ['$gt' => 2]], [['$addFields' => ['y' => '$_id']]]]],
+        ];
+
+        $operation = new BulkWrite($this->getDatabaseName(), $this->getCollectionName(), $ops);
+        $result = $operation->execute($this->getPrimaryServer());
+
+        $this->assertInstanceOf(BulkWriteResult::class, $result);
+        $this->assertSame(3, $result->getMatchedCount());
+        $this->assertSame(3, $result->getModifiedCount());
+
+        $expected = [
+            ['_id' => 1, 'x' => 11],
+            ['_id' => 2, 'x' => 22, 'y' => 2],
+            ['_id' => 3, 'x' => 33, 'y' => 3],
+            ['_id' => 4, 'x' => 44, 'y' => 4],
+        ];
+
+        $this->assertSameDocuments($expected, $this->collection->find());
+    }
+
     /**
      * Create data fixtures.
      *

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
+            }
+          ]
+        }
+      }
     }
   ]
 }