PHPLIB-53: Create mapReduce command helper

src/MapReduceResult.php

+<?php
+/*
+ * Copyright 2017 MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace MongoDB;
+
+use IteratorAggregate;
+use stdClass;
+
+/**
+ * Result class for mapReduce command results.
+ *
+ * This class allows for iteration of mapReduce results irrespective of the
+ * output method (e.g. inline, collection) via the IteratorAggregate interface.
+ * It also provides access to command statistics.
+ *
+ * @api
+ * @see \MongoDB\Collection::mapReduce()
+ * @see https://docs.mongodb.com/manual/reference/command/mapReduce/
+ */
+class MapReduceResult implements IteratorAggregate
+{
+    private $getIterator;
+    private $executionTimeMS;
+    private $counts;
+    private $timing;
+
+    /**
+     * Constructor.
+     *
+     * @internal
+     * @param callable $getIterator Callback that returns a Traversable for mapReduce results
+     * @param stdClass $result      Result document from the mapReduce command
+     */
+    public function __construct(callable $getIterator, stdClass $result)
+    {
+        $this->getIterator = $getIterator;
+        $this->executionTimeMS = (integer) $result->timeMillis;
+        $this->counts = (array) $result->counts;
+        $this->timing = isset($result->timing) ? (array) $result->timing : [];
+    }
+
+    /**
+     * Returns various count statistics from the mapReduce command.
+     *
+     * @return array
+     */
+    public function getCounts()
+    {
+        return $this->counts;
+    }
+
+    /**
+     * Return the command execution time in milliseconds.
+     *
+     * @return integer
+     */
+    public function getExecutionTimeMS()
+    {
+        return $this->executionTimeMS;
+    }
+
+    /**
+     * Return the mapReduce results as a Traversable.
+     *
+     * @see http://php.net/iteratoraggregate.getiterator
+     * @return Traversable
+     */
+    public function getIterator()
+    {
+        return call_user_func($this->getIterator);
+    }
+
+    /**
+     * Returns various timing statistics from the mapReduce command.
+     *
+     * Note: timing statistics are only available if the mapReduce command's
+     * "verbose" option was true; otherwise, an empty array will be returned.
+     *
+     * @return array
+     */
+    public function getTiming()
+    {
+        return $this->timing;
+    }
+}

src/Operation/MapReduce.php

+<?php
+/*
+ * Copyright 2015-2017 MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace MongoDB\Operation;
+
+use MongoDB\BSON\Javascript;
+use MongoDB\Driver\Command;
+use MongoDB\Driver\ReadConcern;
+use MongoDB\Driver\ReadPreference;
+use MongoDB\Driver\Server;
+use MongoDB\Driver\WriteConcern;
+use MongoDB\Driver\Exception\RuntimeException as DriverRuntimeException;
+use MongoDB\Exception\InvalidArgumentException;
+use MongoDB\Exception\UnexpectedValueException;
+use MongoDB\Exception\UnsupportedException;
+use MongoDB\Model\TypeMapArrayIterator;
+use MongoDB\MapReduceResult;
+use ArrayIterator;
+use stdClass;
+
+/**
+ * Operation for the mapReduce command.
+ *
+ * @api
+ * @see \MongoDB\Collection::mapReduce()
+ * @see https://docs.mongodb.com/manual/reference/command/mapReduce/
+ */
+class MapReduce implements Executable
+{
+    private static $wireVersionForCollation = 5;
+    private static $wireVersionForDocumentLevelValidation = 4;
+    private static $wireVersionForReadConcern = 4;
+    private static $wireVersionForWriteConcern = 4;
+
+    private $databaseName;
+    private $collectionName;
+    private $map;
+    private $reduce;
+    private $out;
+    private $options;
+
+    /**
+     * Constructs a mapReduce command.
+     *
+     * Required arguments:
+     *
+     *  * map (MongoDB\BSON\Javascript): A JavaScript function that associates
+     *    or "maps" a value with a key and emits the key and value pair.
+     *
+     *  * reduce (MongoDB\BSON\Javascript): A JavaScript function that "reduces"
+     *    to a single object all the values associated with a particular key.
+     *
+     *  * out (string|document): Specifies where to output the result of the
+     *    map-reduce operation. You can either output to a collection or return
+     *    the result inline. On a primary member of a replica set you can output
+     *    either to a collection or inline, but on a secondary, only inline
+     *    output is possible.
+     *
+     * Supported options:
+     *
+     *  * bypassDocumentValidation (boolean): If true, allows the write to
+     *    circumvent document level validation. This only applies when results
+     *    are output to a collection.
+     *
+     *    For servers < 3.2, this option is ignored as document level validation
+     *    is not available.
+     *
+     *  * collation (document): Collation specification.
+     *
+     *    This is not supported for server versions < 3.4 and will result in an
+     *    exception at execution time if used.
+     *
+     *  * finalize (MongoDB\BSON\Javascript): Follows the reduce method and
+     *    modifies the output.
+     *
+     *  * jsMode (boolean): Specifies whether to convert intermediate data into
+     *    BSON format between the execution of the map and reduce functions. The
+     *    default is false.
+     *
+     *  * limit (integer): Specifies a maximum number of documents for the input
+     *    into the map function.
+     *
+     *  * maxTimeMS (integer): The maximum amount of time to allow the query to
+     *    run.
+     *
+     *  * query (document): Specifies the selection criteria using query
+     *    operators for determining the documents input to the map function.
+     *
+     *  * readConcern (MongoDB\Driver\ReadConcern): Read concern. This is not
+     *    supported when results are returned inline.
+     *
+     *    This is not supported for server versions < 3.2 and will result in an
+     *    exception at execution time if used.
+     *
+     *  * readPreference (MongoDB\Driver\ReadPreference): Read preference.
+     *
+     *  * scope (document): Specifies global variables that are accessible in
+     *    the map, reduce and finalize functions.
+     *
+     *  * sort (document): Sorts the input documents. This option is useful for
+     *    optimization. For example, specify the sort key to be the same as the
+     *    emit key so that there are fewer reduce operations. The sort key must
+     *    be in an existing index for this collection.
+     *
+     *  * typeMap (array): Type map for BSON deserialization. This will be
+     *    applied to the returned Cursor (it is not sent to the server).
+     *
+     *  * verbose (boolean): Specifies whether to include the timing information
+     *    in the result information. The default is true.
+     *
+     *  * writeConcern (MongoDB\Driver\WriteConcern): Write concern. This only
+     *    applies when results are output to a collection.
+     *
+     *    This is not supported for server versions < 3.4 and will result in an
+     *    exception at execution time if used.
+     *
+     * @param string              $databaseName   Database name
+     * @param string              $collectionName Collection name
+     * @param Javascript          $map            Map function
+     * @param Javascript          $reduce         Reduce function
+     * @param string|array|object $out            Output specification
+     * @param array               $options        Command options
+     * @throws InvalidArgumentException for parameter/option parsing errors
+     */
+    public function __construct($databaseName, $collectionName, Javascript $map, Javascript $reduce, $out, array $options = [])
+    {
+        if ( ! is_string($out) && ! is_array($out) && ! is_object($out)) {
+            throw InvalidArgumentException::invalidType('$out', $out, 'string or array or object');
+        }
+
+        $options += [
+            'jsMode' => false,
+            'verbose' => true,
+        ];
+
+        if (isset($options['bypassDocumentValidation']) && ! is_bool($options['bypassDocumentValidation'])) {
+            throw InvalidArgumentException::invalidType('"bypassDocumentValidation" option', $options['bypassDocumentValidation'], 'boolean');
+        }
+
+        if (isset($options['collation']) && ! is_array($options['collation']) && ! is_object($options['collation'])) {
+            throw InvalidArgumentException::invalidType('"collation" option', $options['collation'], 'array or object');
+        }
+
+        if (isset($options['finalize']) && ! $options['finalize'] instanceof Javascript) {
+            throw InvalidArgumentException::invalidType('"finalize" option', $options['finalize'], 'MongoDB\Driver\Javascript');
+        }
+
+        if (isset($options['jsMode']) && ! is_bool($options['jsMode'])) {
+            throw InvalidArgumentException::invalidType('"jsMode" option', $options['jsMode'], 'boolean');
+        }
+
+        if (isset($options['limit']) && ! is_integer($options['limit'])) {
+            throw InvalidArgumentException::invalidType('"limit" option', $options['limit'], 'integer');
+        }
+
+        if (isset($options['maxTimeMS']) && ! is_integer($options['maxTimeMS'])) {
+            throw InvalidArgumentException::invalidType('"maxTimeMS" option', $options['maxTimeMS'], 'integer');
+        }
+
+        if (isset($options['query']) && ! is_array($options['query']) && ! is_object($options['query'])) {
+            throw InvalidArgumentException::invalidType('"query" option', $options['query'], 'array or object');
+        }
+
+        if (isset($options['readConcern']) && ! $options['readConcern'] instanceof ReadConcern) {
+            throw InvalidArgumentException::invalidType('"readConcern" option', $options['readConcern'], 'MongoDB\Driver\ReadConcern');
+        }
+
+        if (isset($options['readPreference']) && ! $options['readPreference'] instanceof ReadPreference) {
+            throw InvalidArgumentException::invalidType('"readPreference" option', $options['readPreference'], 'MongoDB\Driver\ReadPreference');
+        }
+
+        if (isset($options['scope']) && ! is_array($options['scope']) && ! is_object($options['scope'])) {
+            throw InvalidArgumentException::invalidType('"scope" option', $options['scope'], 'array or object');
+        }
+
+        if (isset($options['sort']) && ! is_array($options['sort']) && ! is_object($options['sort'])) {
+            throw InvalidArgumentException::invalidType('"sort" option', $options['sort'], 'array or object');
+        }
+
+        if (isset($options['typeMap']) && ! is_array($options['typeMap'])) {
+            throw InvalidArgumentException::invalidType('"typeMap" option', $options['typeMap'], 'array');
+        }
+
+        if (isset($options['verbose']) && ! is_bool($options['verbose'])) {
+            throw InvalidArgumentException::invalidType('"verbose" option', $options['verbose'], 'boolean');
+        }
+
+        if (isset($options['writeConcern']) && ! $options['writeConcern'] instanceof WriteConcern) {
+            throw InvalidArgumentException::invalidType('"writeConcern" option', $options['writeConcern'], 'MongoDB\Driver\WriteConcern');
+        }
+
+        $this->databaseName = (string) $databaseName;
+        $this->collectionName = (string) $collectionName;
+        $this->map = $map;
+        $this->reduce = $reduce;
+        $this->out = $out;
+        $this->options = $options;
+    }
+
+    /**
+     * Execute the operation.
+     *
+     * @see Executable::execute()
+     * @param Server $server
+     * @return MapReduceResult
+     * @throws UnexpectedValueException if the command response was malformed
+     * @throws UnsupportedException if collation, read concern, or write concern is used and unsupported
+     * @throws DriverRuntimeException for other driver errors (e.g. connection errors)
+     */
+    public function execute(Server $server)
+    {
+        if (isset($this->options['collation']) && ! \MongoDB\server_supports_feature($server, self::$wireVersionForCollation)) {
+            throw UnsupportedException::collationNotSupported();
+        }
+
+        if (isset($this->options['readConcern']) && ! \MongoDB\server_supports_feature($server, self::$wireVersionForReadConcern)) {
+            throw UnsupportedException::readConcernNotSupported();
+        }
+
+        if (isset($this->options['writeConcern']) && ! \MongoDB\server_supports_feature($server, self::$wireVersionForWriteConcern)) {
+            throw UnsupportedException::writeConcernNotSupported();
+        }
+
+        $readPreference = isset($this->options['readPreference']) ? $this->options['readPreference'] : null;
+        $cursor = $server->executeCommand($this->databaseName, $this->createCommand($server), $readPreference);
+        $result = current($cursor->toArray());
+
+        $getIterator = $this->createGetIteratorCallable($result, $server);
+
+        return new MapReduceResult($getIterator, $result);
+    }
+
+    /**
+     * Create the mapReduce command.
+     *
+     * @param Server $server
+     * @return Command
+     */
+    private function createCommand(Server $server)
+    {
+        $cmd = [
+            'mapReduce' => $this->collectionName,
+            'map' => $this->map,
+            'reduce' => $this->reduce,
+            'out' => $this->out,
+        ];
+
+        foreach (['finalize', 'jsMode', 'limit', 'maxTimeMS', 'readConcern', 'verbose', 'writeConcern'] as $option) {
+            if (isset($this->options[$option])) {
+                $cmd[$option] = $this->options[$option];
+            }
+        }
+
+        foreach (['collation', 'query', 'scope', 'sort'] as $option) {
+            if (isset($this->options[$option])) {
+                $cmd[$option] = (object) $this->options[$option];
+            }
+        }
+
+        if (isset($this->options['bypassDocumentValidation']) && \MongoDB\server_supports_feature($server, self::$wireVersionForDocumentLevelValidation)) {
+            $cmd['bypassDocumentValidation'] = $this->options['bypassDocumentValidation'];
+        }
+
+        return new Command($cmd);
+    }
+
+    /**
+     * Creates a callable for MapReduceResult::getIterator().
+     *
+     * @param stdClass $result
+     * @param Server   $server
+     * @return callable
+     * @throws UnexpectedValueException if the command response was malformed
+     */
+    private function createGetIteratorCallable(stdClass $result, Server $server)
+    {
+        // Inline results can be wrapped with an ArrayIterator
+        if (isset($result->results) && is_array($result->results)) {
+            $results = $result->results;
+
+            return function() use ($results) {
+                if (isset($this->options['typeMap'])) {
+                    return new TypeMapArrayIterator($results, $this->options['typeMap']);
+                }
+
+                return new ArrayIterator($results);
+            };
+        }
+
+        if (isset($result->result) && (is_string($result->result) || is_object($result->result))) {
+            $options = isset($this->options['typeMap']) ? ['typeMap' => $this->options['typeMap']] : [];
+
+            $find = is_string($result->result)
+                ? new Find($this->databaseName, $result->result, [], $options)
+                : new Find($result->result->db, $result->result->collection, [], $options);
+
+            return function() use ($find, $server) {
+                return $find->execute($server);
+            };
+        }
+
+        throw new UnexpectedValueException('mapReduce command did not return inline results or an output collection');
+    }
+}

tests/Operation/MapReduceFunctionalTest.php

+<?php
+
+namespace MongoDB\Tests\Operation;
+
+use MongoDB\BSON\Javascript;
+use MongoDB\Driver\BulkWrite;
+use MongoDB\Operation\Find;
+use MongoDB\Operation\MapReduce;
+
+class MapReduceFunctionalTest extends FunctionalTestCase
+{
+    public function testResult()
+    {
+        $this->createFixtures(3);
+
+        $map = new Javascript('function() { emit(this.x, this.y); }');
+        $reduce = new Javascript('function(key, values) { return Array.sum(values); }');
+        $out = ['inline' => 1];
+
+        $operation = new MapReduce($this->getDatabaseName(), $this->getCollectionName(), $map, $reduce, $out);
+        $result = $operation->execute($this->getPrimaryServer());
+
+        $this->assertInstanceOf('MongoDB\MapReduceResult', $result);
+        $this->assertGreaterThanOrEqual(0, $result->getExecutionTimeMS());
+        $this->assertNotEmpty($result->getCounts());
+        $this->assertNotEmpty($result->getTiming());
+    }
+
+    public function testResultDoesNotIncludeTimingWithoutVerboseOption()
+    {
+        $this->createFixtures(3);
+
+        $map = new Javascript('function() { emit(this.x, this.y); }');
+        $reduce = new Javascript('function(key, values) { return Array.sum(values); }');
+        $out = ['inline' => 1];
+
+        $operation = new MapReduce($this->getDatabaseName(), $this->getCollectionName(), $map, $reduce, $out, ['verbose' => false]);
+        $result = $operation->execute($this->getPrimaryServer());
+
+        $this->assertInstanceOf('MongoDB\MapReduceResult', $result);
+        $this->assertGreaterThanOrEqual(0, $result->getExecutionTimeMS());
+        $this->assertNotEmpty($result->getCounts());
+        $this->assertEmpty($result->getTiming());
+    }
+
+    /**
+     * @dataProvider provideTypeMapOptionsAndExpectedDocuments
+     */
+    public function testTypeMapOptionWithInlineResults(array $typeMap = null, array $expectedDocuments)
+    {
+        $this->createFixtures(3);
+
+        $map = new Javascript('function() { emit(this.x, this.y); }');
+        $reduce = new Javascript('function(key, values) { return Array.sum(values); }');
+        $out = ['inline' => 1];
+
+        $operation = new MapReduce($this->getDatabaseName(), $this->getCollectionName(), $map, $reduce, $out, ['typeMap' => $typeMap]);
+        $results = iterator_to_array($operation->execute($this->getPrimaryServer()));
+
+        $this->assertEquals($expectedDocuments, $results);
+    }
+
+    public function provideTypeMapOptionsAndExpectedDocuments()
+    {
+        return [
+            [
+                null,
+                [
+                    (object) ['_id' => 1, 'value' => 3],
+                    (object) ['_id' => 2, 'value' => 6],
+                    (object) ['_id' => 3, 'value' => 9],
+                ],
+            ],
+            [
+                ['root' => 'array'],
+                [
+                    ['_id' => 1, 'value' => 3],
+                    ['_id' => 2, 'value' => 6],
+                    ['_id' => 3, 'value' => 9],
+                ],
+            ],
+            [
+                ['root' => 'object'],
+                [
+                    (object) ['_id' => 1, 'value' => 3],
+                    (object) ['_id' => 2, 'value' => 6],
+                    (object) ['_id' => 3, 'value' => 9],
+                ],
+            ],
+        ];
+    }
+
+    /**
+     * @dataProvider provideTypeMapOptionsAndExpectedDocuments
+     */
+    public function testTypeMapOptionWithOutputCollection(array $typeMap = null, array $expectedDocuments)
+    {
+        $this->createFixtures(3);
+
+        $map = new Javascript('function() { emit(this.x, this.y); }');
+        $reduce = new Javascript('function(key, values) { return Array.sum(values); }');
+        $out = $this->getCollectionName() . '.output';
+
+        $operation = new MapReduce($this->getDatabaseName(), $this->getCollectionName(), $map, $reduce, $out, ['typeMap' => $typeMap]);
+        $results = iterator_to_array($operation->execute($this->getPrimaryServer()));
+
+        $this->assertEquals($expectedDocuments, $results);
+
+        $operation = new Find($this->getDatabaseName(), $out, [], ['typeMap' => $typeMap]);
+        $cursor = $operation->execute($this->getPrimaryServer());
+
+        $this->assertEquals($expectedDocuments, iterator_to_array($cursor));
+    }
+
+    /**
+     * Create data fixtures.
+     *
+     * @param integer $n
+     */
+    private function createFixtures($n)
+    {
+        $bulkWrite = new BulkWrite(['ordered' => true]);
+
+        for ($i = 1; $i <= $n; $i++) {
+            $bulkWrite->insert(['x' => $i, 'y' => $i]);
+            $bulkWrite->insert(['x' => $i, 'y' => $i * 2]);
+        }
+
+        $result = $this->manager->executeBulkWrite($this->getNamespace(), $bulkWrite);
+
+        $this->assertEquals($n * 2, $result->getInsertedCount());
+    }
+}

tests/Operation/MapReduceTest.php

+<?php
+
+namespace MongoDB\Tests\Operation;
+
+use MongoDB\BSON\Javascript;
+use MongoDB\BSON\ObjectID;
+use MongoDB\Operation\MapReduce;
+use stdClass;
+
+class MapReduceTest extends TestCase
+{
+    /**
+     * @expectedException MongoDB\Exception\InvalidArgumentException
+     * @dataProvider provideInvalidOutValues
+     */
+    public function testConstructorOutArgumentTypeCheck($out)
+    {
+        $map = new Javascript('function() { emit(this.x, this.y); }');
+        $reduce = new Javascript('function(key, values) { return Array.sum(values); }');
+
+        new MapReduce($this->getDatabaseName(), $this->getCollectionName(), $map, $reduce, $out);
+    }
+
+    public function provideInvalidOutValues()
+    {
+        return $this->wrapValuesForDataProvider([123, 3.14, true]);
+    }
+
+    /**
+     * @expectedException MongoDB\Exception\InvalidArgumentException
+     * @dataProvider provideInvalidConstructorOptions
+     */
+    public function testConstructorOptionTypeChecks(array $options)
+    {
+        $map = new Javascript('function() { emit(this.x, this.y); }');
+        $reduce = new Javascript('function(key, values) { return Array.sum(values); }');
+        $out = ['inline' => 1];
+
+        new MapReduce($this->getDatabaseName(), $this->getCollectionName(), $map, $reduce, $out, $options);
+    }
+
+    public function provideInvalidConstructorOptions()
+    {
+        $options = [];
+
+        foreach ($this->getInvalidBooleanValues() as $value) {
+            $options[][] = ['bypassDocumentValidation' => $value];
+        }
+
+        foreach ($this->getInvalidDocumentValues() as $value) {
+            $options[][] = ['collation' => $value];
+        }
+
+        foreach ($this->getInvalidJavascriptValues() as $value) {
+            $options[][] = ['finalize' => $value];
+        }
+
+        foreach ($this->getInvalidBooleanValues() as $value) {
+            $options[][] = ['jsMode' => $value];
+        }
+
+        foreach ($this->getInvalidIntegerValues() as $value) {
+            $options[][] = ['limit' => $value];
+        }
+
+        foreach ($this->getInvalidIntegerValues() as $value) {
+            $options[][] = ['maxTimeMS' => $value];
+        }
+
+        foreach ($this->getInvalidDocumentValues() as $value) {
+            $options[][] = ['query' => $value];
+        }
+
+        foreach ($this->getInvalidReadConcernValues() as $value) {
+            $options[][] = ['readConcern' => $value];
+        }
+
+        foreach ($this->getInvalidReadPreferenceValues() as $value) {
+            $options[][] = ['readPreference' => $value];
+        }
+
+        foreach ($this->getInvalidDocumentValues() as $value) {
+            $options[][] = ['scope' => $value];
+        }
+
+        foreach ($this->getInvalidDocumentValues() as $value) {
+            $options[][] = ['sort' => $value];
+        }
+
+        foreach ($this->getInvalidArrayValues() as $value) {
+            $options[][] = ['typeMap' => $value];
+        }
+
+        foreach ($this->getInvalidBooleanValues() as $value) {
+            $options[][] = ['verbose' => $value];
+        }
+
+        foreach ($this->getInvalidWriteConcernValues() as $value) {
+            $options[][] = ['writeConcern' => $value];
+        }
+
+        return $options;
+    }
+
+    private function getInvalidJavascriptValues()
+    {
+        return [123, 3.14, 'foo', true, [], new stdClass, new ObjectID];
+    }
+}