Merge branch 'v1.7'

* v1.7:
  PHPLIB-558: Don't inherit read preference in Database::command
  PHPLIB-561: Retry replSetStepDown in connection failover test
  PHPLIB-559: Make aggregate command explainable

Author: alcaeus

docs/reference/method/MongoDBCollection-explain.txt

@@ -50,6 +50,7 @@ Explainable Commands
 
 Explainable commands include, but are not limited to:
 
+   - :phpclass:`MongoDB\\Operation\\Aggregate`
    - :phpclass:`MongoDB\\Operation\\Count`
    - :phpclass:`MongoDB\\Operation\\DeleteMany`
    - :phpclass:`MongoDB\\Operation\\DeleteOne`

src/Database.php

@@ -249,10 +249,6 @@ public function aggregate(array $pipeline, array $options = [])
      */
     public function command($command, array $options = [])
     {
-        if (! isset($options['readPreference']) && ! is_in_transaction($options)) {
-            $options['readPreference'] = $this->readPreference;
-        }
-
         if (! isset($options['typeMap'])) {
             $options['typeMap'] = $this->typeMap;
         }

src/Operation/Aggregate.php

@@ -48,7 +48,7 @@
  * @see \MongoDB\Collection::aggregate()
  * @see http://docs.mongodb.org/manual/reference/command/aggregate/
  */
-class Aggregate implements Executable
+class Aggregate implements Executable, Explainable
 {
     /** @var integer */
     private static $wireVersionForCollation = 5;
@@ -285,9 +285,12 @@ public function execute(Server $server)
         }
 
         $hasExplain = ! empty($this->options['explain']);
-        $hasWriteStage = is_last_pipeline_operator_write($this->pipeline);
+        $hasWriteStage = $this->hasWriteStage();
 
-        $command = $this->createCommand($server, $hasWriteStage);
+        $command = new Command(
+            $this->createCommandDocument($server, $hasWriteStage),
+            $this->createCommandOptions()
+        );
         $options = $this->createOptions($hasWriteStage, $hasExplain);
 
         $cursor = $hasWriteStage && ! $hasExplain
@@ -315,20 +318,17 @@ public function execute(Server $server)
         return new ArrayIterator($result->result);
     }
 
-    /**
-     * Create the aggregate command.
-     *
-     * @param Server  $server
-     * @param boolean $hasWriteStage
-     * @return Command
-     */
-    private function createCommand(Server $server, $hasWriteStage)
+    public function getCommandDocument(Server $server)
+    {
+        return $this->createCommandDocument($server, $this->hasWriteStage());
+    }
+
+    private function createCommandDocument(Server $server, bool $hasWriteStage) : array
     {
         $cmd = [
             'aggregate' => $this->collectionName ?? 1,
             'pipeline' => $this->pipeline,
         ];
-        $cmdOptions = [];
 
         $cmd['allowDiskUse'] = $this->options['allowDiskUse'];
 
@@ -352,10 +352,6 @@ private function createCommand(Server $server, $hasWriteStage)
             $cmd['hint'] = is_array($this->options['hint']) ? (object) $this->options['hint'] : $this->options['hint'];
         }
 
-        if (isset($this->options['maxAwaitTimeMS'])) {
-            $cmdOptions['maxAwaitTimeMS'] = $this->options['maxAwaitTimeMS'];
-        }
-
         if ($this->options['useCursor']) {
             /* Ignore batchSize if pipeline includes an $out or $merge stage, as
              * no documents will be returned and sending a batchSize of zero
@@ -365,7 +361,18 @@ private function createCommand(Server $server, $hasWriteStage)
                 : new stdClass();
         }
 
-        return new Command($cmd, $cmdOptions);
+        return $cmd;
+    }
+
+    private function createCommandOptions() : array
+    {
+        $cmdOptions = [];
+
+        if (isset($this->options['maxAwaitTimeMS'])) {
+            $cmdOptions['maxAwaitTimeMS'] = $this->options['maxAwaitTimeMS'];
+        }
+
+        return $cmdOptions;
     }
 
     /**
@@ -399,4 +406,9 @@ private function createOptions($hasWriteStage, $hasExplain)
 
         return $options;
     }
+
+    private function hasWriteStage() : bool
+    {
+        return is_last_pipeline_operator_write($this->pipeline);
+    }
 }

src/Operation/Explain.php

@@ -41,6 +41,9 @@ class Explain implements Executable
     const VERBOSITY_EXEC_STATS = 'executionStats';
     const VERBOSITY_QUERY = 'queryPlanner';
 
+    /** @var integer */
+    private static $wireVersionForAggregate = 7;
+
     /** @var integer */
     private static $wireVersionForDistinct = 4;
 
@@ -108,6 +111,10 @@ public function execute(Server $server)
             throw UnsupportedException::explainNotSupported();
         }
 
+        if ($this->explainable instanceof Aggregate && ! server_supports_feature($server, self::$wireVersionForAggregate)) {
+            throw UnsupportedException::explainNotSupported();
+        }
+
         $cmd = ['explain' => $this->explainable->getCommandDocument($server)];
 
         if (isset($this->options['verbosity'])) {

src/Operation/Explainable.php

@@ -20,8 +20,8 @@
 use MongoDB\Driver\Server;
 
 /**
- * Explainable interface for explainable operations (count, distinct, find,
- * findAndModify, delete, and update).
+ * Explainable interface for explainable operations (aggregate, count, distinct,
+ * find, findAndModify, delete, and update).
  *
  * @internal
  */

tests/Database/DatabaseFunctionalTest.php

@@ -7,6 +7,7 @@
 use MongoDB\Driver\Cursor;
 use MongoDB\Driver\ReadConcern;
 use MongoDB\Driver\ReadPreference;
+use MongoDB\Driver\Server;
 use MongoDB\Driver\WriteConcern;
 use MongoDB\Exception\InvalidArgumentException;
 use MongoDB\Operation\CreateIndexes;
@@ -100,6 +101,22 @@ public function testCommand()
         $this->assertTrue($commandResult->ismaster);
     }
 
+    public function testCommandDoesNotInheritReadPreference()
+    {
+        if ($this->isReplicaSet()) {
+            $this->markTestSkipped('Test only applies to replica sets');
+        }
+
+        $this->database = new Database($this->manager, $this->getDatabaseName(), ['readPreference' => new ReadPreference(ReadPreference::RP_SECONDARY)]);
+
+        $command = ['ping' => 1];
+
+        $cursor = $this->database->command($command);
+
+        $this->assertInstanceOf(Cursor::class, $cursor);
+        $this->assertTrue($cursor->getServer()->isPrimary());
+    }
+
     public function testCommandAppliesTypeMapToCursor()
     {
         $command = ['isMaster' => 1];

tests/FunctionalTestCase.php

@@ -301,6 +301,11 @@ protected function getServerStorageEngine(ReadPreference $readPreference = null)
         throw new UnexpectedValueException('Could not determine server storage engine');
     }
 
+    protected function isReplicaSet()
+    {
+        return $this->getPrimaryServer()->getType() !== Server::TYPE_RS_PRIMARY;
+    }
+
     protected function isShardedCluster()
     {
         if ($this->getPrimaryServer()->getType() == Server::TYPE_MONGOS) {

tests/Operation/ExplainFunctionalTest.php

@@ -3,6 +3,7 @@
 namespace MongoDB\Tests\Operation;
 
 use MongoDB\Driver\BulkWrite;
+use MongoDB\Operation\Aggregate;
 use MongoDB\Operation\Count;
 use MongoDB\Operation\CreateCollection;
 use MongoDB\Operation\Delete;
@@ -387,6 +388,43 @@ public function testUpdateOne($verbosity, $executionStatsExpected, $allPlansExec
         $this->assertExplainResult($result, $executionStatsExpected, $allPlansExecutionExpected);
     }
 
+    public function testAggregate()
+    {
+        if (version_compare($this->getServerVersion(), '4.0.0', '<')) {
+            $this->markTestSkipped('Explaining aggregate command requires server version >= 4.0');
+        }
+
+        $this->createFixtures(3);
+
+        $pipeline = [['$group' => ['_id' => null]]];
+        $operation = new Aggregate($this->getDatabaseName(), $this->getCollectionName(), $pipeline);
+
+        $explainOperation = new Explain($this->getDatabaseName(), $operation, ['verbosity' => Explain::VERBOSITY_QUERY, 'typeMap' => ['root' => 'array', 'document' => 'array']]);
+        $result = $explainOperation->execute($this->getPrimaryServer());
+
+        $this->assertExplainResult($result, false, false, true);
+    }
+
+    /**
+     * @dataProvider provideVerbosityInformation
+     */
+    public function testAggregateOptimizedToQuery($verbosity, $executionStatsExpected, $allPlansExecutionExpected)
+    {
+        if (version_compare($this->getServerVersion(), '4.2.0', '<')) {
+            $this->markTestSkipped('MongoDB < 4.2 does not optimize simple aggregation pipelines');
+        }
+
+        $this->createFixtures(3);
+
+        $pipeline = [['$match' => ['_id' => ['$ne' => 2]]]];
+        $operation = new Aggregate($this->getDatabaseName(), $this->getCollectionName(), $pipeline);
+
+        $explainOperation = new Explain($this->getDatabaseName(), $operation, ['verbosity' => $verbosity, 'typeMap' => ['root' => 'array', 'document' => 'array']]);
+        $result = $explainOperation->execute($this->getPrimaryServer());
+
+        $this->assertExplainResult($result, $executionStatsExpected, $allPlansExecutionExpected);
+    }
+
     public function provideVerbosityInformation()
     {
         return [
@@ -396,9 +434,14 @@ public function provideVerbosityInformation()
         ];
     }
 
-    private function assertExplainResult($result, $executionStatsExpected, $allPlansExecutionExpected)
+    private function assertExplainResult($result, $executionStatsExpected, $allPlansExecutionExpected, $stagesExpected = false)
     {
-        $this->assertArrayHasKey('queryPlanner', $result);
+        if ($stagesExpected) {
+            $this->assertArrayHasKey('stages', $result);
+        } else {
+            $this->assertArrayHasKey('queryPlanner', $result);
+        }
+
         if ($executionStatsExpected) {
             $this->assertArrayHasKey('executionStats', $result);
             if ($allPlansExecutionExpected) {

tests/SpecTests/PrimaryStepDownSpecTest.php

@@ -16,6 +16,7 @@
 use Symfony\Bridge\PhpUnit\SetUpTearDownTrait;
 use UnexpectedValueException;
 use function current;
+use function sprintf;
 
 /**
  * @see https://github.com/mongodb/specifications/tree/master/source/connections-survive-step-down/tests
@@ -215,7 +216,20 @@ public function testGetMoreIteration()
 
         // Send a {replSetStepDown: 5, force: true} command to the current primary and verify that the command succeeded
         $primary = $this->client->getManager()->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
-        $primary->executeCommand('admin', new Command(['replSetStepDown' => 5, 'force' => true]));
+
+        $success = false;
+        $attempts = 0;
+        do {
+            try {
+                $attempts++;
+                $primary->executeCommand('admin', new Command(['replSetStepDown' => 5, 'force' => true]));
+                $success = true;
+            } catch (DriverException $e) {
+                if ($attempts == 10) {
+                    $this->fail(sprintf('Could not successfully execute replSetStepDown within %d attempts', $attempts));
+                }
+            }
+        } while (! $success);
 
         // Retrieve the next batch of results from the cursor obtained in the find operation, and verify that this operation succeeded.
         $events = [];