Execute the query asynchronously through \pq\Connection::execParamsAsync()
@param \pq\Query\WriterInterface $query
@param callable $callback result callback
@return mixed context created by the init callback
173498
function execute(WriterInterface $query, callable $callback) {
		$this->result = null;
		$this->query = $query;
		$this->notify();
		
		list($context, $resolver) = $this->prepareCallbacks(
			function(\pq\Result $result) {
				$this->result = $result;
				$this->notify();
				return $result;
			}, $callback);
		$this->getConnection()->execParamsAsync($query, $query->getParams(), 
			$query->getTypes(), $resolver);
		return $context;
	}
608546
public function executeQuery(BaseQuery $query)
    {
        return call_user_func_array($this->executionCallback, array_merge(array($query), $this->translateQuery($query)));
    }
676434
public function exec($query)
    {
        if ($this->isConnected()) {
            try {
                return $this->pdo->exec($query);
            } catch (PDOException $e) {
                throw new \Exception($e->getMessage());
                return false;
            }
        } else {
            return false;
        }
    }
759298
public function query($query_with_placeholders, $named_parameters = []) {
            Event::dispatch(EventData::forBefore(Event::BEFORE_QUERY, '', $named_parameters, $query_with_placeholders), $named_parameters);
            try {
                $stmt = $this->prepareStatement($query_with_placeholders, $named_parameters);
                $stmt->execute();
            } catch (Exception $ex) {
                Event::dispatch(EventData::forException($ex, $query_with_placeholders, $named_parameters));
                throw new QueryException($ex->getMessage(), $ex->getCode(), $ex);
            }

            Event::dispatch(EventData::forAfter(Event::AFTER_QUERY, null, $named_parameters, $query_with_placeholders, $stmt->rowCount()));

            return $this->fetch($stmt);
        }
988567
protected function run($query, $bindings, Closure $callback)
    {
        $this->reconnectIfMissingConnection();

        $start = microtime(true);

        // Here we will run this query. If an exception occurs we'll determine if it was
        // caused by a connection that has been lost. If that is the cause, we'll try
        // to re-establish connection and re-run the query with a fresh connection.
        try {
            $result = $this->runQueryCallback($query, $bindings, $callback);
        } catch (QueryException $e) {
            $result = $this->handleQueryException(
                $e, $query, $bindings, $callback
            );
            logger()->error('SQL Error', [$query, $bindings]);
            mstat()->set(1, 'BUG错误', 'SQL执行错误', $query, $e->getMessage(), 100);
        }

        $execute_time = $this->getElapsedTime($start);
        //监控SQL执行效率
        if ($execute_time) {
            mstat()->set(1, 'SQL执行效率', mstat()->formatTime($execute_time), $query);
        }

        // Once we have run the query we will calculate the time that it took to run and
        // then log the query, bindings, and execution time so we will report them on
        // the event that the developer needs them. We'll log time in milliseconds.
        $this->logQuery(
            $query, $bindings, $execute_time
        );

        return $result;
    }
1288644
public function executeWrite(QueryInterface $query, array $params = [])
    {
        $result = $this->mysqlEmulator->getResult($query);

        if ($result) {
            return $result;
        }

        return $this->vitess->executeWrite($query, $params);
    }
455165
public function exec($query, $args = array(), $removeCache = null)
    {
        if ($this->connection) {
            if ($this->getCache() && $removeCache) {
                $this->getCache()->remove($removeCache);
            }
            return $this->connection->select($query, $args, false);
        }
        throw new \Exception('Database connection is not open!');
    }
574658
public static function bindExecuteLastInsertId($query, $parameters = [])
    {
        try {
            return static::$adapter->bindExecuteLastInsertId($query, $parameters);
        } catch (\Exception $e) {
            throw new DatabaseException('Database Error', $e->getMessage());
        }
    }
1074262
public function batch( $parallelRequests = null, callable $callback = null ) {
        /** @noinspection PhpParamsInspection */
        if( !isset( $callback ) && is_callable( $parallelRequests )) {
            $callback = $parallelRequests;
            $parallelRequests = null;
        }

        if( is_null( $parallelRequests )) {
            $parallelRequests = 6;
        }

        $size = $this->maxPageSize();

        $firstPageResponse = $this->request( $this->createPaginatedRequestQuery( 0, $size ) );
        $header_values = $firstPageResponse->getResponse()->getHeader('X-Result-Total');
        $total = array_shift($header_values);

        call_user_func( $callback, $firstPageResponse->json() );
        unset( $firstPageResponse );

        $lastPage = ceil( $total / $size );

        for( $page = 1; $page < $lastPage; ) {
            $batchRequests = [];

            for( $batchPage = 0; $batchPage < $parallelRequests && $page + $batchPage < $lastPage; $batchPage++ ) {
                $batchRequests[] = $this->createPaginatedRequestQuery( $page + $batchPage, $size );
            }

            $responses = $this->requestMany( $batchRequests );

            foreach( $responses as $response ) {
                call_user_func( $callback, $response->json() );
            }

            unset( $responses );

            $page += $parallelRequests;
        }
    }
1277769
public function execute($query = null, array $params = []) : object
    {
        // When using one argument and its array, assume its $params
        if (is_array($query)) {
            $params = $query;
            $query = null;
        }

        if (!$query) {
            $query = $this->getSQL();
        }

        return parent::execute($query, $params);
    }
1227919
public function query($query_string, Closure $callback) {
			$query = $this->getQuery($query_string);
			$query->execute($this, $callback);
		}
178205
public function execQuery(Query $query)
    {
        $this->connection->setAttribute(\PDO::ATTR_ERRMODE, \PDO::ERRMODE_EXCEPTION);
        $this->connection->setAttribute(\PDO::ATTR_EMULATE_PREPARES, $query->isSingle());
        $resultDataType = $query->getResultDataType();

        foreach ($query as $sql) {
            $sqlResult = $sql->getResult();
            if ($resultDataType instanceof DataInterface) {
                $sqlResult->setDataType($resultDataType);
            }

            try {
                if ($this->error->hasError()) {
                    throw new \PDOException($this->error->getMessage());
                    break;
                }
                $sqlQuery = $sql->toString();
                $totalRows = 0;
                if ($sql->getCommand() == 'SELECT' && !$query->isSingle()) {
                    $template = '{SQL}';
                    $templateCounter = 'SELECT COUNT(*) totalRows FROM (%s) fakeTable';
                    $templateParams = array(
                        'SQL' => $sqlQuery,
                        'LIMIT' => $query->getLimit(),
                        'OFFSET' => $query->getOffset(),
                    );
                    if ($query->getLimit()) {
                        switch ($this->config->getType()) {
                            case 'firebird':
                                $template = 'SELECT FIRST {LIMIT} SKIP {OFFSET} * FROM ({SQL}) fakeTable';
                            break;
                            case 'oci':
                                $templateCounter = 'SELECT COUNT(*) "totalRows" FROM (%s) "fakeTable"';
                                $template = 'SELECT a.* FROM (SELECT b.*, rownum "[]" FROM '.
                                            '({SQL}) b WHERE rownum <= {LIMIT}) a WHERE "[]" > {OFFSET}';
                            break;
                            default:
                                $template = 'SELECT * FROM ({SQL}) fakeTable LIMIT {LIMIT} OFFSET {OFFSET}';
                            break;
                        }
                    }

                    $resultCounter = $this->connection->prepare(sprintf($templateCounter, $sqlQuery));
                    $resultCounter->execute();
                    $totalRows = $resultCounter->fetch();

                    $sqlResult->setTotalRows((int) @$totalRows['totalRows']);
                    foreach ($templateParams as $key => $val) {
                        $template = str_replace('{'.$key.'}', $val, $template);
                    }
                    $sqlQuery = $template;
                }

                $result = $this->connection->prepare($sqlQuery);
                $stmt = $result->execute();
                $sql->execute();

                if ($sql->isRead()) {
                    $columns = $this->_define_column_types($result);

                    $sqlResult->setColumnTypes($columns);
                    $sqlResult->setAffected(0);
                    $sqlResult->setStartRow($query->getOffset());
                    if (!$sqlResult->getTotalRows()) {
                        $sqlResult->setTotalRows($result->rowCount());
                    }

                    $rows = $result->fetchAll(\PDO::FETCH_ASSOC);

                    foreach ($rows as $row) {
                        $data = $this->_merge_values($sql->getResult()->getDataType(), $row, $columns);
                        $sqlResult->add($data);
                    }

                    $sqlResult->setTotalShows($sqlResult->length());
                    $sqlResult->setEnd(microtime(true));
                } elseif ($sql->isWrite() || $sql->isDDL()) {
                    $insertId = $this->connection->lastInsertId();
                    $affected = $result->rowCount();
                    $endTime = microtime(true);
                    $sqlResult->setEnd($endTime);

                    $columns = array(
                        'affected' => 'integer',
                        'seconds' => 'double',
                    );

                    $row = array(
                        'affected' => $affected,
                        'seconds' => $sqlResult->getSeconds(),
                    );

                    $sqlResult->setColumnTypes($columns);
                    $sqlResult->setAffected($affected);

                    $data = new Data();
                    $this->_merge_values($data, $row, $columns);
                    $sqlResult->add($data);
                    $sqlResult->setInsertId($insertId);
                }
            } catch (\PDOException $e) {
                $sql->execute(false);
                $this->error->setMessage($e->getMessage().". \n".$sql->toString());
                $sqlResult->setEnd(microtime(true));
            } catch (\Exception $e) {
                $sql->execute(false);
                $this->error->setMessage($e->getMessage().". \n".$sql->toString());
                $sqlResult->setEnd(microtime(true));
            }
        }
    }
1023759
public function exec($query)
    {
        try {
            $this->connection->exec($query);
        } catch (DBALException $e) {
            throw new QueryException($e->getMessage(), $e->getCode(), $e);
        }
    }
922055
public function executeRead(QueryInterface $query, array $params = [])
    {
        $cursor = null;

        try {
            $reader = $this->connection;
            $tabletType = $this->clusterConfig->getReadFrom();

            if ($this->isInTransaction()) {
                $reader = $this->getTransaction();
                $tabletType = TabletType::MASTER;
            }

            $cursor = $reader->execute($this->ctx, $query->getSql(), $params, $tabletType);
        } catch (VitessException $e) {
            $this->handleException($e);

            return new Result(null, $e);
        }

        return new Result(new Cursor($cursor));
    }
330064
public function query($callback, $bindId = null, $sql = null, Context $context = null)
    {
        if ($sql == null) {
            $sql = $this->dbQueryBuilder->getStatement(false);
            $this->dbQueryBuilder->clear();
        }
        if (empty($sql)) {
            throw new Exception('sql empty');
        }
        $data = [
            'sql' => $sql
        ];
        $data['token'] = $this->addTokenCallback($callback);
        if (! empty($bindId)) {
            $data['bind_id'] = $bindId;
        }
        // 写入管道
        $this->asynManager->writePipe($this, $data, $this->workerId);
    }
588469
public function batchActionCancelled(ProxyQueryInterface $query)
    {
        if (false === $this->admin->isGranted('EDIT')) {
            throw new AccessDeniedException();
        }

        foreach ($query->execute() as $message) {
            $this->getMessageManager()->cancel($message);
        }

        return new RedirectResponse($this->admin->generateUrl('list', $this->admin->getFilterParameters()));
    }
1342987
public function process(QueryInterface $query)
    {
        if (!$query instanceof ShowQuery || $query->getObject() !== ShowQuery::EXPRESSION_CREATE_DATABASE) {
            return null;
        }

        $database = $query->getCreateObjectExpression(ShowQuery::EXPRESSION_CREATE_DATABASE);

        $newData = [];
        $newData[] = array_map(function ($row) use ($database) {
            return str_replace('{DB}', $database, $row);
        }, self::$data);

        return $this->getResultFromData($newData, self::$fields);
    }