Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Oct 8, 2019
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d4b5e8d
PHPLIB-481: Add the ability to specify a pipeline to an update within…
alcaeus Oct 1, 2019
738022d
PHPLIB-481: Remove duplicated bulk write tests
alcaeus Oct 2, 2019
d44bd16
PHPLIB-481: Add functional test for pipeline-style bulk updates
alcaeus Oct 2, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions src/Operation/BulkWrite.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
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;

Expand Down Expand Up @@ -255,8 +256,8 @@ public function __construct($databaseName, $collectionName, array $operations, a
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])) {
Expand Down
92 changes: 30 additions & 62 deletions tests/Operation/BulkWriteFunctionalTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@
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;
Expand Down Expand Up @@ -226,67 +225,6 @@ public function testUnacknowledgedWriteConcernAccessesUpsertedIds(BulkWriteResul
$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', '<')) {
Expand Down Expand Up @@ -357,6 +295,36 @@ function (array $event) {
);
}

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.
*
Expand Down
8 changes: 4 additions & 4 deletions tests/Operation/BulkWriteTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -258,10 +258,10 @@ public function testUpdateManyUpdateArgumentTypeCheck($update)
]);
}

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]]],
]);
Expand Down Expand Up @@ -345,10 +345,10 @@ public function testUpdateOneUpdateArgumentTypeCheck($update)
]);
}

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]]],
]);
Expand Down
165 changes: 165 additions & 0 deletions tests/SpecTests/crud/updateWithPipelines.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
]
}
}
}
]
}