Executes the compiled query

@param string $type
@param string $table
@param string $sql
@param array|null $vals
@param boolean $reset
@return PDOStatement
335367
protected function _run(string $type, string $table, string $sql=NULL, array $vals=NULL, bool $reset=TRUE): PDOStatement
	{
		if ($sql === NULL)
		{
			$sql = $this->_compile($type, $table);
		}

		if ($vals === NULL)
		{
			$vals = array_merge($this->state->getValues(), $this->state->getWhereValues());
		}

		$startTime = microtime(TRUE);

		$res = empty($vals)
			? $this->driver->query($sql)
			: $this->driver->prepareExecute($sql, $vals);

		$endTime = microtime(TRUE);
		$totalTime = number_format($endTime - $startTime, 5);

		// Add this query to the list of executed queries
		$this->_appendQuery($vals, $sql, (int) $totalTime);

		// Reset class state for next query
		if ($reset)
		{
			$this->resetQuery();
		}

		return $res;
	}
875159
protected function _execute($type, $sql, $bind = [])
    {
        $context = $this->_context;

        $context->sql = $sql;
        $context->bind = $bind;

        $context->affected_rows = 0;

        $this->eventsManager->fireEvent('db:before' . ucfirst($type), $this);

        if ($context->connection) {
            $connection = $context->connection;
        } else {
            $connection = $this->poolManager->pop($this, $this->_timeout);
        }

        try {
            $start_time = microtime(true);
            $context->affected_rows = $connection->execute($sql, $bind);
            $elapsed = round(microtime(true) - $start_time, 3);
        } finally {
            if (!$context->connection) {
                $this->poolManager->push($this, $connection);
            }
        }

        $count = $context->affected_rows;
        $event_data = compact('count', 'sql', 'bind', 'elapsed');
        if (is_int($context->affected_rows)) {
            $this->eventsManager->fireEvent('db:after' . ucfirst($type), $this, $event_data);
        }

        $this->logger->info($event_data, 'db.' . $type);
        return $count;
    }
408334
public function query($type = 'object')
    {
        $sql = 'select '.$this->columns.' from '.$this->getTables($this->tables).' '.$this->where.' '.$this->order.' '.$this->group;

        static::$sql = static::$register[] = $sql;
        //
        if ($data = Database::read($sql)) {
            return self::fetch($data, $type);
        } elseif (Database::execerr()) {
            throw new QueryException();
        }
    }
1177916
private function makeSql(int $type, string $sql, int $count): string
    {
        switch ($type) {
            case self::TYPE_FUNCTION_QUERY:
                $limit = $count !== self::UNLIMIT ? sprintf('limit %s', $count) : '';
                $result = vsprintf('select * from %s %s', [$sql, $limit,]);
                break;
            case self::TYPE_FUNCTION_CURSOR:
                $fetch = $count === self::UNLIMIT ? 'all' : 'FORWARD ' . $count;
                $result = vsprintf('select %s; fetch %s from _result;', [$sql, $fetch,]);
                break;
            case self::TYPE_SIMPLE_QUERY:
            default:
                $result = $sql;
        }

        return $result;
    }
59621
public function buildQuery( $table_name, $type, $params = null )
    {
        if ( $type == "find" )
        {
            $sub_sql = isset($params["sub_sql"]) ? $params["sub_sql"] : null;
        
            $sql = "SELECT * FROM $table_name";
            
            if ( isset($sub_sql) )
            {
                $sql .= " " . $sub_sql;
            }
            
            $sql .= ";";
        }
        else if ( $type == "update" )
        {
            $columns = isset($params["columns"]) ? $params["columns"] : null;
            
            $sql = "UPDATE $table_name SET ";
            
            foreach ( $columns as $column )
            {
                $sql .= $column["name"] . "=:" . $column["name"] . ",";
            }
            
            $sql = rtrim( $sql, ", " );
            $sql .= " WHERE ";
            
            foreach ( $columns as $column )
            {
                $sql .= $column["name"] . "=:snap_" . $column["name"] . " AND ";
            }
            
            $sql = rtrim( $sql, " AND " );
            $sql .= ";";
        }
        else if ( $type == "insert" )
        {
            $columns = isset($params["columns"]) ? $params["columns"] : null;
            
            $sql = "INSERT INTO $table_name (";
            
            foreach ( $columns as $column )
            {
                $sql .= $column["name"] . ", ";;
            }
            
            $sql = rtrim( $sql, ", " );
            
            $sql .= ") VALUES (";
            
            foreach ( $columns as $column )
            {
                $sql .= ":" . $column["name"] . ", ";
            }
            
            $sql = rtrim( $sql, ", " );
            $sql .= ");";
        }
        
        return $this->prepare( $sql );
    }
691560
function Execute($sql)
	{
		$result = null;
		
		if ($this->ConnectionSetting->IsReadOnlySlave) {
			
			// this is a read-only slave connection attempting a write operation.  we
			// will only proceed if the connection specifies a "master" delegate connection
			if (!$this->_masterAdapter) {
				
				if ($this->ConnectionSetting->MasterConnectionDelegate) {
					
					$this->Observe("DataAdapter ($this->_label) (DataAdapter.Execute) Delegating write operation from Slave to Master Connection", OBSERVE_INFO);
				
					$this->_masterAdapter = new DataAdapter($this->ConnectionSetting->MasterConnectionDelegate);
					
					foreach ($this->_observers as $observer) {
						$this->_masterAdapter->AttachObserver($observer);
					}
				}
				else {
					throw new Exception('DB Write operation was attempted on a read-only slave connection');
				}
			}
			
			// we have a master connection initialized and ready to use
			$result = $this->_masterAdapter->Execute($sql);

		}
		else {
			$this->RequireConnection(true);
			$this->Observe("DataAdapter ($this->_label) (DataAdapter.Execute) " . $sql, OBSERVE_QUERY);
			$result = -1;
			
			try
			{
				$result = $this->_driver->Execute($this->_dbconn, $sql);
				$this->_num_retries = 0;
			}
			catch (Exception $ex)
			{
				// retry one time a communication error occurs
				if ($this->_num_retries == 0 && DataAdapter::$RETRY_ON_COMMUNICATION_ERROR && $this->IsCommunicationError($ex))
				{
					$this->_num_retries++;
					$this->Observe("DataAdapter ($this->_label) Communication error.  Retry attempt " . $this->_num_retries, OBSERVE_WARN);
					sleep(2); // slight delay to prevent throttling
					return $this->Execute($sql);
				}
				
				$msg = "DataAdapter ($this->_label)" . ' Error Executing SQL: ' . $ex->getMessage() . ' (retry attempts: '.$this->_num_retries.')';
				
				$this->Observe($msg,OBSERVE_FATAL);
				throw new Exception($msg,$ex->getCode());
			}
			}
		return $result;
	}
62831
public static function dbQ2ArrayAll($sql, $atype = "assoc") {

        try {
            $res = self::query($sql);
            if (is_object($res)) {
                $type = self::getPDOFetchType($atype);
                return $res->fetchAll($type);
            }
        } catch (PDOException $e) {
            die(self::dbError($sql));
        }
    }
995166
private function executeStatement($type = 'execute')
    {
        $options = [
            'consistency' => \Cassandra::CONSISTENCY_LOCAL_QUORUM,
        ];

        if ($this->getConsistency()) {
            $options['consistency'] = $this->getConsistency();
        }

        if ($this->getRetryPolicy()) {
            $options['retryPolicy'] = $this->getRetryPolicy();
        }

        if (is_array($this->getArguments())) {
            $options['arguments'] = $this->getArguments();
        }

        $casOptions = new \Cassandra\ExecutionOptions($options);
        $session = $this->getSession();

        // execute
        /** @var \Cassandra\Rows $result */
        $result = $session->$type($this->getStatement(), $casOptions);

        // deal with args
        if (array_key_exists('arguments', $options)) {
            $this->previousArgs($options['arguments']);
        }

        return $result;
    }
229983
public function exec($sql, $errorLevel = E_USER_ERROR)
    {
        $this->beforeQuery($sql);

        // Directly exec this query
        $result = $this->pdoConnection->exec($sql);

        // Check for errors
        if ($result !== false) {
            return $this->rowCount = $result;
        }

        $this->databaseError($this->getLastError(), $errorLevel, $sql);
        return null;
    }
633700
public function insertBatch(string $table, $data=[]): ?PDOStatement
	{
		// Get the generated values and sql string
		[$sql, $data] = $this->driver->insertBatch($table, $data);

		return $sql !== NULL
			? $this->_run('', $table, $sql, $data)
			: NULL;
	}
922750
protected function _compileType(string $type='', string $table=''): string
	{
		$setArrayKeys = $this->state->getSetArrayKeys();
		switch($type)
		{
			case 'insert':
				$paramCount = count($setArrayKeys);
				$params = array_fill(0, $paramCount, '?');
				$sql = "INSERT INTO {$table} ("
					. implode(',', $setArrayKeys)
					. ")\nVALUES (".implode(',', $params).')';
				break;

			case 'update':
				$setString = $this->state->getSetString();
				$sql = "UPDATE {$table}\nSET {$setString}";
				break;

			case 'replace':
				// @TODO implement
				$sql = '';
				break;

			case 'delete':
				$sql = "DELETE FROM {$table}";
				break;

			// Get queries
			default:
				$fromString = $this->state->getFromString();
				$selectString = $this->state->getSelectString();

				$sql = "SELECT * \nFROM {$fromString}";

				// Set the select string
				if ( ! empty($selectString))
				{
					// Replace the star with the selected fields
					$sql = str_replace('*', $selectString, $sql);
				}
				break;
		}

		return $sql;
	}
1006770
public function join($table, $type = NULL)
	{
		$this->_join[] = $this->_last_join = new \Database_Query_Builder_Join($table, $type);

		return $this;
	}
707647
public function updateBatch(string $table, $data, $where): ?PDOStatement
	{
		// Get the generated values and sql string
		list($sql, $data) = $this->driver->updateBatch($table, $data, $where);

		return $sql !== NULL
			? $this->_run('', $table, $sql, $data)
			: NULL;
	}
360295
public function prepare($sql, $link = null, $resetParams = true)
    {
        $resetParams && $this->reset();
        is_null($link) && $link = $this->currentQueryIsMaster ? $this->wlink : $this->rlink;

        $sqlParams = [];
        foreach ($this->bindParams as $key => $val) {
            $sqlParams[] = ':param' . $key;
        }

        $this->currentSql = $sql;
        $sql = vsprintf($sql, $sqlParams);

        $stmt = $link->prepare($sql);//pdo默认情况prepare出错不抛出异常只返回Pdo::errorInfo
        if ($stmt === false) {
            $error = $link->errorInfo();
            if (in_array($error[1], [2006, 2013])) {
                $link = $this->connectDb($this->currentQueryIsMaster ? 'wlink' : 'rlink', true);
                $stmt = $link->prepare($sql);
                if ($stmt === false) {
                    $error = $link->errorInfo();
                } else {
                    return $stmt;
                }
            }
            throw new \InvalidArgumentException(
                'Pdo Prepare Sql error! ,【Sql: ' . $this->buildDebugSql() . '】,【Code: ' . $link->errorCode() . '】, 【ErrorInfo!: 
                ' . $error[2] . '】 '
            );
        }
        return $stmt;
    }
1298721
public function startQuery($sql, array $params = null, array $types = null)
    {
        $this->logger->addDebug($sql);
        $this->startTime = microtime(true);
    }
482903
public function countAllResults(string $table='', bool $reset = TRUE): int
	{
		// Set the table
		if ( ! empty($table))
		{
			$this->from($table);
		}

		$result = $this->_run('get', $table, NULL, NULL, $reset);
		$rows = $result->fetchAll();

		return (int) count($rows);
	}
1364737
function sql($sql = null, $params = NULL) {
        $params = $this->autoarr($params);
        $sql = str_replace('table.',$this->alias.'.',$sql);
        return $this->execute($sql, $params);
        return $this;
    }