protected function doAggregate(array $pipeline, array $options = array())
    {
        if (isset($options['cursor']) && ($options['cursor'] || is_array($options['cursor']))) {
            return $this->doAggregateCursor($pipeline, $options);
        }

        unset($options['cursor']);

        list($commandOptions, $clientOptions) = isset($options['socketTimeoutMS']) || isset($options['timeout'])
            ? $this->splitCommandAndClientOptions($options)
            : array($options, array());

        $command = array();
        $command['aggregate'] = $this->riakCollection->getName();
        $command['pipeline'] = $pipeline;
        $command = array_merge($command, $commandOptions);

        $database = $this->database;
        $result = $this->retry(function() use ($database, $command, $clientOptions) {
            return $database->command($command, $clientOptions);
        });

        if (empty($result['ok'])) {
            throw new ResultException($result);
        }

        /* If the pipeline ends with an $out operator, return a cursor on that
         * collection so a table scan may be performed.
         */
        if (isset($pipeline[count($pipeline) - 1]['$out'])) {
            $outputCollection = $pipeline[count($pipeline) - 1]['$out'];

            return $database->selectCollection($outputCollection)->find();
        }

        $arrayIterator = new ArrayIterator(isset($result['result']) ? $result['result'] : array());
        $arrayIterator->setCommandResult($result);

        return $arrayIterator;
    }