From d3caa38451c8cfcb09041f57cd7305c3f0754aeb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=A9r=C3=B4me=20Tamarelle?= Date: Tue, 10 Sep 2024 11:05:09 +0200 Subject: [PATCH 1/3] PHPLIB-1419 Encode Agg builder objects in Collection methods Ignore static analysis issues PHPLIB-1419 Fix BC break --- psalm-baseline.xml | 5 +++ src/Collection.php | 12 +++++++ src/functions.php | 33 +++++++++++++++++ .../BuilderCollectionFunctionalTest.php | 36 +++++++++++++++++-- tests/FunctionsTest.php | 19 ++++++++++ 5 files changed, 103 insertions(+), 2 deletions(-) diff --git a/psalm-baseline.xml b/psalm-baseline.xml index f51206d0d..1d8ee73f0 100644 --- a/psalm-baseline.xml +++ b/psalm-baseline.xml @@ -223,6 +223,11 @@ + + + + + diff --git a/src/Collection.php b/src/Collection.php index 7769b9a5b..7fff8cd87 100644 --- a/src/Collection.php +++ b/src/Collection.php @@ -23,6 +23,7 @@ use MongoDB\BSON\JavascriptInterface; use MongoDB\BSON\PackedArray; use MongoDB\Builder\BuilderEncoder; +use MongoDB\Builder\Pipeline; use MongoDB\Codec\DocumentCodec; use MongoDB\Codec\Encoder; use MongoDB\Driver\CursorInterface; @@ -223,6 +224,12 @@ public function __toString() */ public function aggregate(array $pipeline, array $options = []) { + if (is_builder_pipeline($pipeline)) { + $pipeline = new Pipeline(...$pipeline); + } + + $pipeline = $this->builderEncoder->encodeIfSupported($pipeline); + $hasWriteStage = is_last_pipeline_operator_write($pipeline); $options = $this->inheritReadPreference($options); @@ -1098,6 +1105,11 @@ public function updateSearchIndex(string $name, array|object $definition, array */ public function watch(array $pipeline = [], array $options = []) { + if (is_builder_pipeline($pipeline)) { + $pipeline = new Pipeline(...$pipeline); + } + + $pipeline = $this->builderEncoder->encodeIfSupported($pipeline); $options = $this->inheritReadOptions($options); $options = $this->inheritCodecOrTypeMap($options); diff --git a/src/functions.php b/src/functions.php index ca30fdbc0..2d1c61c45 100644 --- a/src/functions.php +++ b/src/functions.php @@ -21,6 +21,7 @@ use MongoDB\BSON\Document; use MongoDB\BSON\PackedArray; use MongoDB\BSON\Serializable; +use MongoDB\Builder\Type\StageInterface; use MongoDB\Driver\Exception\RuntimeException as DriverRuntimeException; use MongoDB\Driver\Manager; use MongoDB\Driver\ReadPreference; @@ -327,6 +328,38 @@ function is_pipeline(array|object $pipeline, bool $allowEmpty = false): bool return true; } +/** + * Returns whether the argument is a list that contains at least one + * {@see StageInterface} object. + * + * @internal + */ +function is_builder_pipeline(array $pipeline): bool +{ + if (! $pipeline) { + return false; + } + + if (! array_is_list($pipeline)) { + return false; + } + + $result = false; + foreach ($pipeline as $stage) { + if (! is_array($stage) && ! is_object($stage)) { + return false; + } + + if ($stage instanceof StageInterface) { + $result = true; + } elseif (! is_first_key_operator($stage)) { + return false; + } + } + + return $result; +} + /** * Returns whether we are currently in a transaction. * diff --git a/tests/Collection/BuilderCollectionFunctionalTest.php b/tests/Collection/BuilderCollectionFunctionalTest.php index 15a89cda5..7fedb96e9 100644 --- a/tests/Collection/BuilderCollectionFunctionalTest.php +++ b/tests/Collection/BuilderCollectionFunctionalTest.php @@ -2,10 +2,13 @@ namespace MongoDB\Tests\Collection; +use MongoDB\Builder\Expression; use MongoDB\Builder\Pipeline; use MongoDB\Builder\Query; use MongoDB\Builder\Stage; +use function iterator_to_array; + class BuilderCollectionFunctionalTest extends FunctionalTestCase { public function setUp(): void @@ -17,7 +20,18 @@ public function setUp(): void public function testAggregate(): void { - $this->markTestSkipped('Not supported yet'); + $this->collection->insertMany([['x' => 10], ['x' => 10], ['x' => 10]]); + $pipeline = new Pipeline( + Stage::bucketAuto( + groupBy: Expression::intFieldPath('x'), + buckets: 2, + ), + ); + // Extract the list of stages for arg type restriction + $pipeline = iterator_to_array($pipeline); + + $results = $this->collection->aggregate($pipeline)->toArray(); + $this->assertCount(2, $results); } public function testBulkWriteDeleteMany(): void @@ -245,6 +259,24 @@ public function testUpdateManyWithPipeline(): void public function testWatch(): void { - $this->markTestSkipped('Not supported yet'); + $this->skipIfChangeStreamIsNotSupported(); + + if ($this->isShardedCluster()) { + $this->markTestSkipped('Test does not apply on sharded clusters: need more than a single getMore call on the change stream.'); + } + + $pipeline = new Pipeline( + Stage::match(operationType: Query::eq('insert')), + ); + // Extract the list of stages for arg type restriction + $pipeline = iterator_to_array($pipeline); + + $changeStream = $this->collection->watch($pipeline); + $changeStream->rewind(); + $this->assertNull($changeStream->current()); + $this->collection->insertOne(['x' => 3]); + $changeStream->next(); + $this->assertTrue($changeStream->valid()); + $this->assertEquals('insert', $changeStream->current()->operationType); } } diff --git a/tests/FunctionsTest.php b/tests/FunctionsTest.php index 7098cb24a..48039290a 100644 --- a/tests/FunctionsTest.php +++ b/tests/FunctionsTest.php @@ -4,6 +4,8 @@ use MongoDB\BSON\Document; use MongoDB\BSON\PackedArray; +use MongoDB\Builder\Stage\LimitStage; +use MongoDB\Builder\Stage\MatchStage; use MongoDB\Driver\WriteConcern; use MongoDB\Model\BSONArray; use MongoDB\Model\BSONDocument; @@ -12,6 +14,7 @@ use function MongoDB\apply_type_map_to_document; use function MongoDB\create_field_path_type_map; use function MongoDB\document_to_array; +use function MongoDB\is_builder_pipeline; use function MongoDB\is_first_key_operator; use function MongoDB\is_last_pipeline_operator_write; use function MongoDB\is_mapreduce_output_inline; @@ -311,6 +314,22 @@ public function providePipelines(): array ]; } + /** @dataProvider provideStagePipelines */ + public function testIsBuilderPipeline($expected, $pipeline): void + { + $this->assertSame($expected, is_builder_pipeline($pipeline)); + } + + public function provideStagePipelines(): iterable + { + yield 'empty array' => [false, []]; + yield 'array of arrays' => [false, [['$match' => ['x' => 1]]]]; + yield 'map of stages' => [false, [1 => new MatchStage([])]]; + yield 'stages' => [true, [new MatchStage([]), new LimitStage(1)]]; + yield 'stages and operators' => [true, [new MatchStage([]), ['$limit' => 1]]]; + yield 'stages and invalid' => [false, [new MatchStage([]), ['foo' => 'bar']]]; + } + /** @dataProvider provideWriteConcerns */ public function testIsWriteConcernAcknowledged($expected, WriteConcern $writeConcern): void { From f3b4961341b62ef8aeecbec54a35a19f8205a637 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=A9r=C3=B4me=20Tamarelle?= Date: Wed, 11 Sep 2024 12:11:29 +0200 Subject: [PATCH 2/3] PHPLIB-1419 Support builder Pipeline in Client and Database watch --- psalm-baseline.xml | 28 +++++++++ src/Client.php | 7 +++ src/Collection.php | 1 + src/Database.php | 13 ++++ tests/ClientFunctionalTest.php | 25 ++++++++ .../BuilderCollectionFunctionalTest.php | 2 - .../BuilderDatabaseFunctionalTest.php | 63 +++++++++++++++++++ 7 files changed, 137 insertions(+), 2 deletions(-) create mode 100644 tests/Database/BuilderDatabaseFunctionalTest.php diff --git a/psalm-baseline.xml b/psalm-baseline.xml index 1d8ee73f0..9d7e47185 100644 --- a/psalm-baseline.xml +++ b/psalm-baseline.xml @@ -191,6 +191,7 @@ + @@ -198,6 +199,12 @@ + + + + + + @@ -220,9 +227,17 @@ + + + + + + + + @@ -242,9 +257,22 @@ + + + + + + + + + + + + + diff --git a/src/Client.php b/src/Client.php index 5b02aa44c..2e161707b 100644 --- a/src/Client.php +++ b/src/Client.php @@ -22,6 +22,7 @@ use MongoDB\BSON\Document; use MongoDB\BSON\PackedArray; use MongoDB\Builder\BuilderEncoder; +use MongoDB\Builder\Pipeline; use MongoDB\Codec\Encoder; use MongoDB\Driver\ClientEncryption; use MongoDB\Driver\Exception\InvalidArgumentException as DriverInvalidArgumentException; @@ -391,6 +392,12 @@ public function startSession(array $options = []) */ public function watch(array $pipeline = [], array $options = []) { + if (is_builder_pipeline($pipeline)) { + $pipeline = new Pipeline(...$pipeline); + } + + $pipeline = $this->builderEncoder->encodeIfSupported($pipeline); + if (! isset($options['readPreference']) && ! is_in_transaction($options)) { $options['readPreference'] = $this->readPreference; } diff --git a/src/Collection.php b/src/Collection.php index 7fff8cd87..8ed460c4e 100644 --- a/src/Collection.php +++ b/src/Collection.php @@ -1110,6 +1110,7 @@ public function watch(array $pipeline = [], array $options = []) } $pipeline = $this->builderEncoder->encodeIfSupported($pipeline); + $options = $this->inheritReadOptions($options); $options = $this->inheritCodecOrTypeMap($options); diff --git a/src/Database.php b/src/Database.php index 8b7742b70..125012c81 100644 --- a/src/Database.php +++ b/src/Database.php @@ -21,6 +21,7 @@ use MongoDB\BSON\Document; use MongoDB\BSON\PackedArray; use MongoDB\Builder\BuilderEncoder; +use MongoDB\Builder\Pipeline; use MongoDB\Codec\Encoder; use MongoDB\Driver\ClientEncryption; use MongoDB\Driver\Cursor; @@ -202,6 +203,12 @@ public function __toString() */ public function aggregate(array $pipeline, array $options = []) { + if (is_builder_pipeline($pipeline)) { + $pipeline = new Pipeline(...$pipeline); + } + + $pipeline = $this->builderEncoder->encodeIfSupported($pipeline); + $hasWriteStage = is_last_pipeline_operator_write($pipeline); if (! isset($options['readPreference']) && ! is_in_transaction($options)) { @@ -611,6 +618,12 @@ public function selectGridFSBucket(array $options = []) */ public function watch(array $pipeline = [], array $options = []) { + if (is_builder_pipeline($pipeline)) { + $pipeline = new Pipeline(...$pipeline); + } + + $pipeline = $this->builderEncoder->encodeIfSupported($pipeline); + if (! isset($options['readPreference']) && ! is_in_transaction($options)) { $options['readPreference'] = $this->readPreference; } diff --git a/tests/ClientFunctionalTest.php b/tests/ClientFunctionalTest.php index 3c04ecd67..ca0ea2722 100644 --- a/tests/ClientFunctionalTest.php +++ b/tests/ClientFunctionalTest.php @@ -2,6 +2,9 @@ namespace MongoDB\Tests; +use MongoDB\Builder\Pipeline; +use MongoDB\Builder\Query; +use MongoDB\Builder\Stage; use MongoDB\Client; use MongoDB\Driver\BulkWrite; use MongoDB\Driver\Command; @@ -13,6 +16,7 @@ use function call_user_func; use function is_callable; +use function iterator_to_array; use function sprintf; /** @@ -137,4 +141,25 @@ public function testAddAndRemoveSubscriber(): void $client->getManager()->executeCommand('admin', new Command(['ping' => 1])); } + + public function testWatchWithBuilderPipeline(): void + { + $this->skipIfChangeStreamIsNotSupported(); + + if ($this->isShardedCluster()) { + $this->markTestSkipped('Test does not apply on sharded clusters: need more than a single getMore call on the change stream.'); + } + + $pipeline = new Pipeline( + Stage::match(operationType: Query::eq('insert')), + ); + // Extract the list of stages for arg type restriction + $pipeline = iterator_to_array($pipeline); + + $changeStream = $this->client->watch($pipeline); + $this->client->selectCollection($this->getDatabaseName(), $this->getCollectionName())->insertOne(['x' => 3]); + $changeStream->next(); + $this->assertTrue($changeStream->valid()); + $this->assertEquals('insert', $changeStream->current()->operationType); + } } diff --git a/tests/Collection/BuilderCollectionFunctionalTest.php b/tests/Collection/BuilderCollectionFunctionalTest.php index 7fedb96e9..2ace15109 100644 --- a/tests/Collection/BuilderCollectionFunctionalTest.php +++ b/tests/Collection/BuilderCollectionFunctionalTest.php @@ -272,8 +272,6 @@ public function testWatch(): void $pipeline = iterator_to_array($pipeline); $changeStream = $this->collection->watch($pipeline); - $changeStream->rewind(); - $this->assertNull($changeStream->current()); $this->collection->insertOne(['x' => 3]); $changeStream->next(); $this->assertTrue($changeStream->valid()); diff --git a/tests/Database/BuilderDatabaseFunctionalTest.php b/tests/Database/BuilderDatabaseFunctionalTest.php new file mode 100644 index 000000000..9b89d87e1 --- /dev/null +++ b/tests/Database/BuilderDatabaseFunctionalTest.php @@ -0,0 +1,63 @@ +dropCollection($this->getDatabaseName(), $this->getCollectionName()); + + parent::tearDown(); + } + + public function testAggregate(): void + { + $this->skipIfServerVersion('<', '6.0.0', '$documents stage is not supported'); + + $pipeline = new Pipeline( + Stage::documents([ + ['x' => 1], + ['x' => 2], + ['x' => 3], + ]), + Stage::bucketAuto( + groupBy: Expression::intFieldPath('x'), + buckets: 2, + ), + ); + // Extract the list of stages for arg type restriction + $pipeline = iterator_to_array($pipeline); + + $results = $this->database->aggregate($pipeline)->toArray(); + $this->assertCount(2, $results); + } + + public function testWatch(): void + { + $this->skipIfChangeStreamIsNotSupported(); + + if ($this->isShardedCluster()) { + $this->markTestSkipped('Test does not apply on sharded clusters: need more than a single getMore call on the change stream.'); + } + + $pipeline = new Pipeline( + Stage::match(operationType: Query::eq('insert')), + ); + // Extract the list of stages for arg type restriction + $pipeline = iterator_to_array($pipeline); + + $changeStream = $this->database->watch($pipeline); + $this->database->selectCollection($this->getCollectionName())->insertOne(['x' => 3]); + $changeStream->next(); + $this->assertTrue($changeStream->valid()); + $this->assertEquals('insert', $changeStream->current()->operationType); + } +} From f9539b673c308bb2d9de9f48401dee99a1d0070e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=A9r=C3=B4me=20Tamarelle?= Date: Wed, 11 Sep 2024 18:20:51 +0200 Subject: [PATCH 3/3] Simplify is_builder_pipeline --- psalm-baseline.xml | 1 + src/functions.php | 19 ++++--------------- tests/FunctionsTest.php | 1 - 3 files changed, 5 insertions(+), 16 deletions(-) diff --git a/psalm-baseline.xml b/psalm-baseline.xml index 9d7e47185..3a55977f8 100644 --- a/psalm-baseline.xml +++ b/psalm-baseline.xml @@ -890,6 +890,7 @@ + diff --git a/src/functions.php b/src/functions.php index 2d1c61c45..a445467ba 100644 --- a/src/functions.php +++ b/src/functions.php @@ -336,28 +336,17 @@ function is_pipeline(array|object $pipeline, bool $allowEmpty = false): bool */ function is_builder_pipeline(array $pipeline): bool { - if (! $pipeline) { + if (! $pipeline || ! array_is_list($pipeline)) { return false; } - if (! array_is_list($pipeline)) { - return false; - } - - $result = false; foreach ($pipeline as $stage) { - if (! is_array($stage) && ! is_object($stage)) { - return false; - } - - if ($stage instanceof StageInterface) { - $result = true; - } elseif (! is_first_key_operator($stage)) { - return false; + if (is_object($stage) && $stage instanceof StageInterface) { + return true; } } - return $result; + return false; } /** diff --git a/tests/FunctionsTest.php b/tests/FunctionsTest.php index 48039290a..a673ef719 100644 --- a/tests/FunctionsTest.php +++ b/tests/FunctionsTest.php @@ -327,7 +327,6 @@ public function provideStagePipelines(): iterable yield 'map of stages' => [false, [1 => new MatchStage([])]]; yield 'stages' => [true, [new MatchStage([]), new LimitStage(1)]]; yield 'stages and operators' => [true, [new MatchStage([]), ['$limit' => 1]]]; - yield 'stages and invalid' => [false, [new MatchStage([]), ['foo' => 'bar']]]; } /** @dataProvider provideWriteConcerns */