PHPLIB-1419 Encode Agg builder objects in Collection methods

psalm-baseline.xml

@@ -191,13 +191,20 @@
   <file src="src/Client.php">
     <MixedArgument>
       <code><![CDATA[$driverOptions['driver'] ?? []]]></code>
+      <code><![CDATA[$pipeline]]></code>
     </MixedArgument>
     <MixedAssignment>
       <code><![CDATA[$mergedDriver['platform']]]></code>
     </MixedAssignment>
     <MixedPropertyTypeCoercion>
       <code><![CDATA[$driverOptions['builderEncoder'] ?? new BuilderEncoder()]]></code>
     </MixedPropertyTypeCoercion>
+    <NamedArgumentNotAllowed>
+      <code><![CDATA[$pipeline]]></code>
+    </NamedArgumentNotAllowed>
+    <PossiblyInvalidArgument>
+      <code><![CDATA[$pipeline]]></code>
+    </PossiblyInvalidArgument>
   </file>
   <file src="src/Codec/EncodeIfSupported.php">
     <ArgumentTypeCoercion>
@@ -220,9 +227,22 @@
     </MixedArgumentTypeCoercion>
   </file>
   <file src="src/Collection.php">
+    <MixedArgument>
+      <code><![CDATA[$pipeline]]></code>
+      <code><![CDATA[$pipeline]]></code>
+    </MixedArgument>
     <MixedPropertyTypeCoercion>
       <code><![CDATA[$options['builderEncoder'] ?? new BuilderEncoder()]]></code>
     </MixedPropertyTypeCoercion>
+    <NamedArgumentNotAllowed>
+      <code><![CDATA[$pipeline]]></code>
+      <code><![CDATA[$pipeline]]></code>
+    </NamedArgumentNotAllowed>
+    <PossiblyInvalidArgument>
+      <code><![CDATA[$pipeline]]></code>
+      <code><![CDATA[$pipeline]]></code>
+      <code><![CDATA[$pipeline]]></code>
+    </PossiblyInvalidArgument>
   </file>
   <file src="src/Command/ListCollections.php">
     <MixedAssignment>
@@ -237,9 +257,22 @@
     </MixedAssignment>
   </file>
   <file src="src/Database.php">
+    <MixedArgument>
+      <code><![CDATA[$pipeline]]></code>
+      <code><![CDATA[$pipeline]]></code>
+    </MixedArgument>
     <MixedPropertyTypeCoercion>
       <code><![CDATA[$options['builderEncoder'] ?? new BuilderEncoder()]]></code>
     </MixedPropertyTypeCoercion>
+    <NamedArgumentNotAllowed>
+      <code><![CDATA[$pipeline]]></code>
+      <code><![CDATA[$pipeline]]></code>
+    </NamedArgumentNotAllowed>
+    <PossiblyInvalidArgument>
+      <code><![CDATA[$pipeline]]></code>
+      <code><![CDATA[$pipeline]]></code>
+      <code><![CDATA[$pipeline]]></code>
+    </PossiblyInvalidArgument>
   </file>
   <file src="src/GridFS/Bucket.php">
     <MixedArgument>
@@ -857,6 +890,7 @@
     <MixedAssignment>
       <code><![CDATA[$element[$key]]]></code>
       <code><![CDATA[$stage]]></code>
+      <code><![CDATA[$stage]]></code>
       <code><![CDATA[$type]]></code>
       <code><![CDATA[$typeMap['fieldPaths'][$fieldPath . '.' . $existingFieldPath]]]></code>
       <code><![CDATA[$typeMap['fieldPaths'][$fieldPath]]]></code>

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;
         }

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 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,12 @@
      */
     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);
 

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 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 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;
         }

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,27 @@
     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 || ! array_is_list($pipeline)) {
+        return false;
+    }
+
+    foreach ($pipeline as $stage) {
+        if (is_object($stage) && $stage instanceof StageInterface) {
+            return true;
+        }
+    }
+
+    return false;
+}
+
 /**
  * Returns whether we are currently in a transaction.
  *

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);
+    }
 }

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,22 @@ 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);
+        $this->collection->insertOne(['x' => 3]);
+        $changeStream->next();
+        $this->assertTrue($changeStream->valid());
+        $this->assertEquals('insert', $changeStream->current()->operationType);
     }
 }

tests/Database/BuilderDatabaseFunctionalTest.php

@@ -0,0 +1,63 @@
+<?php
+
+namespace MongoDB\Tests\Database;
+
+use MongoDB\Builder\Expression;
+use MongoDB\Builder\Pipeline;
+use MongoDB\Builder\Query;
+use MongoDB\Builder\Stage;
+
+use function iterator_to_array;
+
+class BuilderDatabaseFunctionalTest extends FunctionalTestCase
+{
+    public function tearDown(): void
+    {
+        $this->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);
+    }
+}