执行查询
目前支持插入、修改、保存（类似mysql的replace）查询
$use_connection_type 默认不传为自动判断，可传true/false,若传字符串(只支持a-z0-9的字符串)，则可以切换到另外一个连接，比如传other,则可以连接到$this->_connection_other_id所对应的ID的连接
Database_Driver_Mongo_Result Driver_Database_Driver_Mongo::query( array $options [, string $as_object = null , boolean $use_connection_type = null ] )
参数列表
参数 类型 描述 默认值 $optionsarray$options $as_objectstring是否返回对象 null $use_connection_typeboolean$use_master 是否使用主数据库，不设置则自动判断 null
Database_Driver_Mongo_Result public function query($options, $as_object = null, $use_connection_type = null)
{
if (IS_DEBUG)Core::debug()->log($options);
if (is_string($options))
{
# 设置连接类型
$this->_set_connection_type($use_connection_type);
// 必需数组
if (!is_array($as_object))$as_object = array();
return $this->connection()->execute($options,$as_object);
}
$type = strtoupper($options['type']);
$typeArr = array
(
'SELECT',
'SHOW', //显示表
'EXPLAIN', //分析
'DESCRIBE', //显示结结构
'INSERT',
'BATCHINSERT',
'REPLACE',
'SAVE',
'UPDATE',
'REMOVE',
);
$slaverType = array
(
'SELECT',
'SHOW',
'EXPLAIN'
);
if (in_array($type, $slaverType))
{
if (true===$use_connection_type)
{
$use_connection_type = 'master';
}
else if (is_string($use_connection_type))
{
if (!preg_match('#^[a-z0-9_]+$#i', $use_connection_type))$use_connection_type = 'master';
}
else
{
$use_connection_type = 'slaver';
}
}
else
{
$use_connection_type = 'master';
}
# 设置连接类型
$this->_set_connection_type($use_connection_type);
# 连接数据库
$connection = $this->connection();
if (!$options['table'])
{
throw new Exception('查询条件中缺少Collection');
}
$tablename = $this->config['table_prefix'] . $options['table'];
if(IS_DEBUG)
{
static $is_sql_debug = null;
if (null === $is_sql_debug) $is_sql_debug = (bool)Core::debug()->profiler('sql')->is_open();
if ($is_sql_debug)
{
$host = $this->_get_hostname_by_connection_hash($this->connection_id());
$benchmark = Core::debug()->profiler('sql')->start('Database', 'mongodb://'.($host['username']?$host['username'].'@':'') . $host['hostname'] . ($host['port'] && $host['port'] != '27017' ? ':' . $host['port'] : ''));
}
}
$explain = null;
try
{
switch ($type)
{
case 'SELECT':
if ($options['distinct'])
{
# 查询唯一值
$result = $connection->command(
array
(
'distinct' => $tablename,
'key' => $options['distinct'] ,
'query' => $options['where']
)
);
$last_query = 'db.'.$tablename.'.distinct('.$options['distinct'].', '.json_encode($options['where']).')';
if(IS_DEBUG && $is_sql_debug)
{
$count = count($result['values']);
}
if ($result && $result['ok']==1)
{
$rs = new Database_Driver_Mongo_Result(new ArrayIterator($result['values']), $options, $as_object ,$this->config);
}
else
{
throw new Exception($result['errmsg']);
}
}
elseif ($options['group_by'])
{
$have_dot = false;
$select = $options['select'];
# group by
$group_opt = array();
if (1===count($options['group_by']))
{
$k = current($options['group_by']);
$group_opt['_id'] = '$'.$k;
if (!isset($select[$k]))$select[$k] = 1;
}
else
{
$group_opt['_id'] = array();
foreach ($options['group_by'] as $item)
{
if (false!==strpos($item, '.'))
{
$have_dot = true;
$group_opt['_id'][str_replace('.', '->', $item)] = '$'.$item;
}
else
{
$group_opt['_id'][$item] = '$'.$item;
}
if (!isset($select[$item]))$select[$item] = 1;
}
}
$last_query = 'db.'.$tablename.'.aggregate(';
$ops = array();
if ($options['where'])
{
$last_query .= '{$match: '.json_encode($options['where']).'}, ';
$ops[] = array
(
'$match' => $options['where']
);
}
$group_opt['_count'] = array('$sum'=>1);
if ($select)
{
foreach ($select as $k=>$v)
{
if (1===$v)
{
if (false!==strpos($k,'.'))
{
$have_dot = true;
$group_opt[str_replace('.','->',$k)] = array('$first'=>'$'.$k);
}
else
{
$group_opt[$k] = array('$first'=>'$'.$k);
}
}
else
{
if (false!==strpos($v,'.'))
{
$have_dot = true;
$group_opt[str_replace('.','->',$v)] = array('$first'=>'$'.$k);
}
else
{
$group_opt[$v] = array('$first'=>'$'.$k);
}
}
}
}
// 处理高级查询条件
if ($options['select_adv'])foreach ($options['select_adv'] as $item)
{
if (!is_array($item))continue;
if (preg_match('#^(.*) AS (.*)$#i', $item[0] , $m))
{
$column = $m[1];
$alias = $m[2];
}
else
{
$column = $alias = $item[0];
}
$alias = str_replace('.','->',$alias);
if (false===$have_dot && false!==strpos($alias,'.'))
{
$have_dot = true;
}
switch ($item[1])
{
case 'max':
case 'min':
case 'avg':
case 'first':
case 'last':
$group_opt[$alias] = array
(
'$'.$item[1] => '$'.$column,
);
break;
case 'addToSet':
case 'concat':
$group_opt[$alias] = array
(
'$addToSet' => '$'.$column,
);
break;
case 'sum':
$group_opt[$alias] = array
(
'$sum' => isset($item[2])?$item[2]:'$'.$column,
);
break;
}
}
$ops[] = array
(
'$group' => $group_opt,
);
$last_query .= '{$group:'.json_encode($group_opt);
if (isset($options['sort']) && $options['sort'])
{
$ops[]['$sort'] = $options['sort'];
$last_query .= ',$sort:'.json_encode($options['sort']);
}
if (isset($options['skip']) && $options['skip']>0)
{
$ops[]['$skip'] = $options['skip'];
$last_query .= ',$skip:'.$options['skip'];
}
if (isset($options['limit']) && $options['limit']>0)
{
$ops[]['$limit'] = $options['limit'];
$last_query .= ',$limit:'.$options['limit'];
}
$last_query .= '}';
$last_query .= ')';
$result = $connection->selectCollection($tablename)->aggregate($ops);
// 兼容不同版本的aggregate返回
if ($result && ($result['ok']==1||!isset($result['errmsg'])))
{
if ($result['ok']==1 && is_array($result['result']))$result = $result['result'];
if ($have_dot)foreach ($result as &$item)
{
// 处理 _ID 字段
if (is_array($item['_id']))foreach ($item['_id'] as $k=>$v)
{
if (false!==strpos($k, '->'))
{
$item['_id'][str_replace('->', '.', $k)] = $v;
unset($item['_id'][$k]);
}
}
// 处理 select 的字段
foreach ($item as $k=>$v)
{
if (false!==strpos($k,'->'))
{
$item[str_replace('->','.',$k)] = $v;
unset($item[$k]);
}
}
}
if ($options['total_count'])
{
foreach ($result as &$item)
{
$item['total_count'] = $item['_count'];
}
}
$count = count($result);
$rs = new Database_Driver_Mongo_Result(new ArrayIterator($result), $options, $as_object ,$this->config);
}
else
{
throw new Exception($result['errmsg'].'.query:'.$last_query);
}
}
else
{
$last_query = 'db.'.$tablename.'.find(';
$last_query .= $options['where']?json_encode($options['where']):'{}';
$last_query .= $options['select']?','.json_encode($options['select']):'';
$last_query .= ')';
$result = $connection->selectCollection($tablename)->find($options['where'],(array)$options['select']);
if(IS_DEBUG && $is_sql_debug)
{
$explain = $result->explain();
$count = $result->count();
}
if ($options['total_count'])
{
$last_query .= '.count()';
$result = $result->count();
# 仅统计count
$rs = new Database_Driver_Mongo_Result(new ArrayIterator(array(array('total_row_count'=>$result))), $options, $as_object ,$this->config);
}
else
{
if ($options['sort'])
{
$last_query .= '.sort('.json_encode($options['sort']).')';
$result = $result->sort($options['sort']);
}
if ($options['skip'])
{
$last_query .= '.skip('.json_encode($options['skip']).')';
$result = $result->skip($options['skip']);
}
if ($options['limit'])
{
$last_query .= '.limit('.json_encode($options['limit']).')';
$result = $result->limit($options['limit']);
}
$rs = new Database_Driver_Mongo_Result($result, $options, $as_object ,$this->config);
}
}
break;
case 'UPDATE':
$result = $connection->selectCollection($tablename)->update($options['where'] , $options['data'] , $options['options']);
$count = $rs = $result['n'];
$last_query = 'db.'.$tablename.'.update('.json_encode($options['where']).','.json_encode($options['data']).')';
break;
case 'SAVE':
case 'INSERT':
case 'BATCHINSERT':
$fun = strtolower($type);
$result = $connection->selectCollection($tablename)->$fun($options['data'] , $options['options']);
if ($type=='BATCHINSERT')
{
$count = count($options['data']);
# 批量插入
$rs = array
(
'',
$count,
);
}
elseif (isset($result['data']['_id']) && $result['data']['_id'] instanceof MongoId)
{
$count = 1;
$rs = array
(
(string)$result['data']['_id'] ,
1 ,
);
}
else
{
$count = 0;
$rs = array
(
'',
0,
);
}
if ($type=='BATCHINSERT')
{
$last_query = '';
foreach ($options['data'] as $d)
{
$last_query .= 'db.'.$tablename.'.insert('.json_encode($d).');'."\n";
}
$last_query = trim($last_query);
}
else
{
$last_query = 'db.'.$tablename.'.'.$fun.'('.json_encode($options['data']).')';
}
break;
case 'REMOVE':
$result = $connection->selectCollection($tablename)->remove($options['where']);
$rs = $result['n'];
$last_query = 'db.'.$tablename.'.remove('.json_encode($options['where']).')';
break;
default:
throw new Exception('不支持的操作类型');
}
}
catch (Exception $e)
{
if(IS_DEBUG && isset($benchmark))
{
Core::debug()->profiler('sql')->stop();
}
throw $e;
}
$this->last_query = $last_query;
# 记录调试
if(IS_DEBUG)
{
Core::debug()->info($last_query,'MongoDB');
if (isset($benchmark))
{
if ($is_sql_debug)
{
$data = array();
$data[0]['db'] = $host['hostname'] . '/' . $this->config['connection']['database'] . '/';
$data[0]['cursor'] = '';
$data[0]['nscanned'] = '';
$data[0]['nscannedObjects'] = '';
$data[0]['n'] = '';
$data[0]['millis'] = '';
$data[0]['row'] = $count;
$data[0]['query'] = '';
$data[0]['nYields'] = '';
$data[0]['nChunkSkips'] = '';
$data[0]['isMultiKey'] = '';
$data[0]['indexOnly'] = '';
$data[0]['indexBounds'] = '';
if ($explain)
{
foreach ($explain as $k=>$v)
{
$data[0][$k] = $v;
}
}
$data[0]['query'] = $last_query;
}
else
{
$data = null;
}
Core::debug()->profiler('sql')->stop($data);
}
}
return $rs;
}