改进Mongo驱动支持3.5+版本聚合查询

增加多聚合查询支持修正事务问题
top-think · Jan 30, 2018 · 2af4656 · 2af4656
1 parent a672147
commit 2af4656
Show file tree

Hide file tree

Showing 3 changed files with 96 additions and 1 deletion.
diff --git a/src/db/Mongo.php b/src/db/Mongo.php
@@ -182,7 +182,7 @@ public function aggregate($aggregate, $field, $force = false)
         $this->parseOptions();
 
         $result = $this->cmd('aggregate', [strtolower($aggregate), $field]);
-        $value  = isset($result[0]['result'][0]['aggregate']) ? $result[0]['result'][0]['aggregate'] : 0;
+        $value  = isset($result[0]['aggregate']) ? $result[0]['aggregate'] : 0;
 
         if ($force) {
             $value += 0;
@@ -191,6 +191,31 @@ public function aggregate($aggregate, $field, $force = false)
         return $value;
     }
 
+    /**
+     * 多聚合操作
+     *
+     * @param array $aggregate 聚合指令, 可以聚合多个参数, 如 ['sum' => 'field1', 'avg' => 'field2']
+     * @param array $groupBy 类似mysql里面的group字段, 可以传入多个字段, 如 ['field_a', 'field_b', 'field_c']
+     * @return array 查询结果
+     */
+    public function multiAggregate($aggregate, $groupBy)
+    {
+        $this->parseOptions();
+
+        $result = $this->cmd('multiAggregate', [$aggregate, $groupBy]);
+
+        foreach ($result as &$row) {
+            if (isset($row['_id']) && !empty($row['_id'])) {
+                foreach ($row['_id'] as $k => $v) {
+                    $row[$k] = $v;
+                }
+                unset($row['_id']);
+            }
+        }
+
+        return $result;
+    }
+
     /**
      * 字段值(延迟)增长
      * @access public

diff --git a/src/db/builder/Mongo.php b/src/db/builder/Mongo.php
@@ -535,6 +535,7 @@ public function aggregate(Query $query, $extra)
             'aggregate'    => $options['table'],
             'allowDiskUse' => true,
             'pipeline'     => $pipeline,
+            'cursor'       => new \stdClass,
         ];
 
         foreach (['explain', 'collation', 'bypassDocumentValidation', 'readConcern'] as $option) {
@@ -550,6 +551,47 @@ public function aggregate(Query $query, $extra)
         return $command;
     }
 
+    /**
+     * 多聚合查询命令, 可以对多个字段进行 group by 操作
+     *
+     * @param array $options 参数
+     * @param array $extra 指令和字段
+     * @return Command
+     */
+    public function multiAggregate(Query $query, $extra)
+    {
+        $options = $query->getOptions();
+
+        list($aggregate, $groupBy) = $extra;
+        $groups                    = ['_id' => []];
+        foreach ($groupBy as $field) {
+            $groups['_id'][$field] = '$' . $field;
+        }
+
+        foreach ($aggregate as $fun => $field) {
+            $groups[$field . '_' . $fun] = ['$' . $fun => '$' . $field];
+        }
+        $pipeline = [
+            ['$match' => (object) $this->parseWhere($query, $options['where'])],
+            ['$group' => $groups],
+        ];
+        $cmd = [
+            'aggregate'    => $options['table'],
+            'allowDiskUse' => true,
+            'pipeline'     => $pipeline,
+            'cursor'       => new \stdClass,
+        ];
+
+        foreach (['explain', 'collation', 'bypassDocumentValidation', 'readConcern'] as $option) {
+            if (isset($options[$option])) {
+                $cmd[$option] = $options[$option];
+            }
+        }
+        $command = new Command($cmd);
+        $this->log('group', $cmd);
+        return $command;
+    }
+
     /**
      * 生成distinct命令
      * @access public

diff --git a/src/db/connector/Mongo.php b/src/db/connector/Mongo.php
@@ -279,6 +279,34 @@ public function parseSqlTable($sql)
         return $sql;
     }
 
+    /**
+     * 启动事务
+     * @access public
+     * @return void
+     * @throws \PDOException
+     * @throws \Exception
+     */
+    public function startTrans()
+    {}
+
+    /**
+     * 用于非自动提交状态下面的查询提交
+     * @access public
+     * @return void
+     * @throws PDOException
+     */
+    public function commit()
+    {}
+
+    /**
+     * 事务回滚
+     * @access public
+     * @return void
+     * @throws PDOException
+     */
+    public function rollback()
+    {}
+
     /**
      * 执行查询
      * @access public