smi2 / phpClickHouse

src/Client.php

<?php

declare(strict_types=1);

namespace ClickHouseDB;

use ClickHouseDB\Exception\QueryException;
use ClickHouseDB\Query\Degeneration;
use ClickHouseDB\Query\Degeneration\Bindings;
use ClickHouseDB\Query\Degeneration\Conditions;
use ClickHouseDB\Query\WhereInFile;
use ClickHouseDB\Query\WriteToFile;
use ClickHouseDB\Quote\FormatLine;
use ClickHouseDB\Transport\Http;
use ClickHouseDB\Transport\Stream;
use function array_flip;
use function array_keys;
use function array_rand;
use function array_values;
use function count;
use function date;
use function implode;
use function in_array;
use function is_array;
use function is_callable;
use function is_file;
use function is_readable;
use function is_string;
use function sprintf;
use function stripos;
use function strtotime;
use function trim;

/**
 * Class Client
 * @package ClickHouseDB
 */
class Client
{
    const SUPPORTED_FORMATS = ['TabSeparated', 'TabSeparatedWithNames', 'CSV', 'CSVWithNames', 'JSONEachRow'];

    /** @var Http */
    private $transport;

    /** @var string */
    private $connectUsername;

    /** @var string */
    private $connectPassword;

    /** @var string */
    private $connectHost;

    /** @var string */
    private $connectPort;

    /** @var bool */
    private $connectUserReadonly = false;

    /**
     * @param mixed[] $connectParams
     * @param mixed[] $settings
     */
    public function __construct(array $connectParams, array $settings = [])
    {
        if (! isset($connectParams['username'])) {
            throw  new \InvalidArgumentException('not set username');
        }

        if (! isset($connectParams['password'])) {
            throw  new \InvalidArgumentException('not set password');
        }

        if (! isset($connectParams['port'])) {
            throw  new \InvalidArgumentException('not set port');
        }

        if (! isset($connectParams['host'])) {
            throw  new \InvalidArgumentException('not set host');
        }

        $this->connectUsername = $connectParams['username'];
        $this->connectPassword = $connectParams['password'];
        $this->connectPort     = $connectParams['port'];
        $this->connectHost     = $connectParams['host'];

        // init transport class
        $this->transport = new Http(
            $this->connectHost,
            $this->connectPort,
            $this->connectUsername,
            $this->connectPassword
        );

        $this->transport->addQueryDegeneration(new Bindings());

        // apply settings to transport class
        $this->settings()->database('default');
        if (! empty($settings)) {
            $this->settings()->apply($settings);
        }

        if (isset($connectParams['readonly'])) {
            $this->setReadOnlyUser($connectParams['readonly']);
        }

        if (isset($connectParams['https'])) {
            $this->https($connectParams['https']);
        }

        $this->enableHttpCompression();
    }

    /**
     * if the user has only read in the config file
     */
    public function setReadOnlyUser(bool $flag)
    {
        $this->connectUserReadonly = $flag;
        $this->settings()->setReadOnlyUser($this->connectUserReadonly);
    }

    /**
     * Clear Degeneration processing request [template ]
     *
     * @return bool
     */
    public function cleanQueryDegeneration()
    {
        return $this->transport->cleanQueryDegeneration();
    }

    /**
     * add Degeneration processing
     *
     * @return bool
     */
    public function addQueryDegeneration(Degeneration $degeneration)
    {
        return $this->transport->addQueryDegeneration($degeneration);
    }

    /**
     * add Conditions in query
     *
     * @return bool
     */
    public function enableQueryConditions()
    {
        return $this->transport->addQueryDegeneration(new Conditions());
    }

    /**
     * Set connection host
     *
     * @param string $host
     */
    public function setHost($host)
    {
        $this->connectHost = $host;
        $this->transport()->setHost($host);
    }

    /**
     * @return Settings
     */
    public function setTimeout(float $timeout)
    {
        return $this->settings()->max_execution_time($timeout);
    }

    /**
     * @return mixed
     */
    public function getTimeout()
    {
        return $this->settings()->getTimeOut();
    }

    /**
     * ConnectTimeOut in seconds ( support 1.5 = 1500ms )
     */
    public function setConnectTimeOut(int $connectTimeOut)
    {
        $this->transport()->setConnectTimeOut($connectTimeOut);
    }

    /**
     * @return int
     */
    public function getConnectTimeOut()
    {
        return $this->transport()->getConnectTimeOut();
    }

    /**
     * @return Http
     */
    public function transport()
    {
        if (! $this->transport) {
            throw  new \InvalidArgumentException('Empty transport class');
        }

        return $this->transport;
    }

    /**
     * @return string
     */
    public function getConnectHost()
    {
        return $this->connectHost;
    }

    /**
     * @return string
     */
    public function getConnectPassword()
    {
        return $this->connectPassword;
    }

    /**
     * @return string
     */
    public function getConnectPort()
    {
        return $this->connectPort;
    }

    /**
     * @return string
     */
    public function getConnectUsername()
    {
        return $this->connectUsername;
    }

    /**
     * @return Http
     */
    public function getTransport()
    {
        return $this->transport;
    }

    /**
     * @return mixed
     */
    public function verbose()
    {
        return $this->transport()->verbose(true);
    }

    /**
     * @return Settings
     */
    public function settings()
    {
        return $this->transport()->settings();
    }

    /**
     * @param string|null $useSessionId
     * @return $this
     */
    public function useSession(string $useSessionId = null)
    {
        if (! $this->settings()->getSessionId()) {
            if (! $useSessionId) {
                $this->settings()->makeSessionId();
            } else {
                $this->settings()->session_id($useSessionId);
            }
        }
        return $this;
    }

    /**
     * @return mixed
     */
    public function getSession()
    {
        return $this->settings()->getSessionId();
    }

    /**
     * Query CREATE/DROP
     *
     * @param mixed[] $bindings
     * @return Statement
     */
    public function write(string $sql, array $bindings = [], bool $exception = true)
    {
        return $this->transport()->write($sql, $bindings, $exception);
    }

    /**
     * set db name
     * @return static
     */
    public function database(string $db)
    {
        $this->settings()->database($db);

        return $this;
    }

    /**
     * Write to system.query_log
     *
     * @return static
     */
    public function enableLogQueries(bool $flag = true)
    {
        $this->settings()->set('log_queries', (int) $flag);

        return $this;
    }

    /**
     * Compress the result if the HTTP client said that it understands data compressed with gzip or deflate
     *
     * @return static
     */
    public function enableHttpCompression(bool $flag = true)
    {
        $this->settings()->enableHttpCompression($flag);

        return $this;
    }

    /**
     * Enable / Disable HTTPS
     *
     * @return static
     */
    public function https(bool $flag = true)
    {
        $this->settings()->https($flag);

        return $this;
    }

    /**
     * Read extremes of the result columns. They can be output in JSON-formats.
     *
     * @return static
     */
    public function enableExtremes(bool $flag = true)
    {
        $this->settings()->set('extremes', (int) $flag);

        return $this;
    }

    /**
     * @param mixed[] $bindings
     * @return Statement
     */
    public function select(
        string $sql,
        array $bindings = [],
        WhereInFile $whereInFile = null,
        WriteToFile $writeToFile = null
    ) {
        return $this->transport()->select($sql, $bindings, $whereInFile, $writeToFile);
    }

    /**
     * @return bool
     */
    public function executeAsync()
    {
        return $this->transport()->executeAsync();
    }

    public function maxTimeExecutionAllAsync()
    {

    }

    /**
     * set progressFunction
     */
    public function progressFunction(callable $callback)
    {
        if (! is_callable($callback)) {
            throw new \InvalidArgumentException('Not is_callable progressFunction');
        }

        if (! $this->settings()->is('send_progress_in_http_headers')) {
            $this->settings()->set('send_progress_in_http_headers', 1);
        }
        if (! $this->settings()->is('http_headers_progress_interval_ms')) {
            $this->settings()->set('http_headers_progress_interval_ms', 100);
        }

        $this->transport()->setProgressFunction($callback);
    }

    /**
     * prepare select
     *
     * @param mixed[] $bindings
     * @return Statement
     */
    public function selectAsync(
        string $sql,
        array $bindings = [],
        WhereInFile $whereInFile = null,
        WriteToFile $writeToFile = null
    ) {
        return $this->transport()->selectAsync($sql, $bindings, $whereInFile, $writeToFile);
    }

    /**
     * SHOW PROCESSLIST
     *
     * @return array
     */
    public function showProcesslist()
    {
        return $this->select('SHOW PROCESSLIST')->rows();
    }

    /**
     * show databases
     *
     * @return array
     */
    public function showDatabases()
    {
        return $this->select('show databases')->rows();
    }

    /**
     * statement = SHOW CREATE TABLE
     *
     * @return mixed
     */
    public function showCreateTable(string $table)
    {
        return $this->select('SHOW CREATE TABLE ' . $table)->fetchOne('statement');
    }

    /**
     * SHOW TABLES
     *
     * @return mixed[]
     */
    public function showTables()
    {
        return $this->select('SHOW TABLES')->rowsAsTree('name');
    }

    /**
     * Get the number of simultaneous/Pending requests
     *
     * @return int
     */
    public function getCountPendingQueue()
    {
        return $this->transport()->getCountPendingQueue();
    }

    /**
     * @param mixed[][] $values
     * @param string[]  $columns
     * @return Statement
     * @throws Exception\TransportException
     */
    public function insert(string $table, array $values, array $columns = []) : Statement
    {
        if (empty($values)) {
            throw QueryException::cannotInsertEmptyValues();
        }

        if (stripos($table, '`') === false && stripos($table, '.') === false) {
            $table = '`' . $table . '`'; //quote table name for dot names
        }
        $sql = 'INSERT INTO ' . $table;

        if (count($columns) !== 0) {
            $sql .= ' (`' . implode('`,`', $columns) . '`) ';
        }

        $sql .= ' VALUES ';

        foreach ($values as $row) {
            $sql .= ' (' . FormatLine::Insert($row) . '), ';
        }
        $sql = trim($sql, ', ');

        return $this->transport()->write($sql);
    }

    /**
     *       * Prepares the values to insert from the associative array.
     *       * There may be one or more lines inserted, but then the keys inside the array list must match (including in the sequence)
     *       *
     *       * @param mixed[] $values - array column_name => value (if we insert one row) or array list column_name => value if we insert many lines
     *       * @return mixed[][] - list of arrays - 0 => fields, 1 => list of value arrays for insertion
     *       */
    public function prepareInsertAssocBulk(array $values)
    {
        if (isset($values[0]) && is_array($values[0])) { //случай, когда много строк вставляется
            $preparedFields = array_keys($values[0]);
            $preparedValues = [];
            foreach ($values as $idx => $row) {
                $_fields = array_keys($row);
                if ($_fields !== $preparedFields) {
                    throw new QueryException(
                        sprintf(
                            'Fields not match: %s and %s on element %s',
                            implode(',', $_fields),
                            implode(',', $preparedFields),
                            $idx
                        )
                    );
                }
                $preparedValues[] = array_values($row);
            }
        } else {
            $preparedFields = array_keys($values);
            $preparedValues = [array_values($values)];
        }

        return [$preparedFields, $preparedValues];
    }

    /**
     * Inserts one or more rows from an associative array.
     * If there is a discrepancy between the keys of the value arrays (or their order) - throws an exception.
     *
     * @param mixed[] $values - array column_name => value (if we insert one row) or array list column_name => value if we insert many lines
     * @return Statement
     */
    public function insertAssocBulk(string $tableName, array $values)
    {
        list($columns, $vals) = $this->prepareInsertAssocBulk($values);

        return $this->insert($tableName, $vals, $columns);
    }

    /**
     * insert TabSeparated files
     *
     * @param string|string[] $fileNames
     * @param string[]        $columns
     * @return mixed
     */
    public function insertBatchTSVFiles(string $tableName, $fileNames, array $columns = [])
    {
        return $this->insertBatchFiles($tableName, $fileNames, $columns, 'TabSeparated');
    }

    /**
     * insert Batch Files
     *
     * @param string|string[] $fileNames
     * @param string[]        $columns
     * @param string          $format ['TabSeparated','TabSeparatedWithNames','CSV','CSVWithNames']
     * @return Statement[]
     * @throws Exception\TransportException
     */
    public function insertBatchFiles(string $tableName, $fileNames, array $columns = [], string $format = 'CSV')
    {
        if (is_string($fileNames)) {
            $fileNames = [$fileNames];
        }
        if ($this->getCountPendingQueue() > 0) {
            throw new QueryException('Queue must be empty, before insertBatch, need executeAsync');
        }

        if (! in_array($format, self::SUPPORTED_FORMATS, true)) {
            throw new QueryException('Format not support in insertBatchFiles');
        }

        $result = [];

        foreach ($fileNames as $fileName) {
            if (! is_file($fileName) || ! is_readable($fileName)) {
                throw  new QueryException('Cant read file: ' . $fileName . ' ' . (is_file($fileName) ? '' : ' is not file'));
            }

            if (empty($columns)) {
                $sql = 'INSERT INTO ' . $tableName . ' FORMAT ' . $format;
            } else {
                $sql = 'INSERT INTO ' . $tableName . ' ( ' . implode(',', $columns) . ' ) FORMAT ' . $format;
            }
            $result[$fileName] = $this->transport()->writeAsyncCSV($sql, $fileName);
        }

        // exec
        $this->executeAsync();

        // fetch resutl
        foreach ($fileNames as $fileName) {
            if (! $result[$fileName]->isError()) {
                continue;
            }

            $result[$fileName]->error();
        }

        return $result;
    }

    /**
     * insert Batch Stream
     *
     * @param string[] $columns
     * @param string   $format ['TabSeparated','TabSeparatedWithNames','CSV','CSVWithNames']
     * @return Transport\CurlerRequest
     */
    public function insertBatchStream(string $tableName, array $columns = [], string $format = 'CSV')
    {
        if ($this->getCountPendingQueue() > 0) {
            throw new QueryException('Queue must be empty, before insertBatch, need executeAsync');
        }

        if (! in_array($format, self::SUPPORTED_FORMATS, true)) {
            throw new QueryException('Format not support in insertBatchFiles');
        }

        if (empty($columns)) {
            $sql = 'INSERT INTO ' . $tableName . ' FORMAT ' . $format;
        } else {
            $sql = 'INSERT INTO ' . $tableName . ' ( ' . implode(',', $columns) . ' ) FORMAT ' . $format;
        }

        return $this->transport()->writeStreamData($sql);
    }

    /**
     * stream Write
     *
     * @param string[] $bind
     * @return Statement
     * @throws Exception\TransportException
     */
    public function streamWrite(Stream $stream, string $sql, array $bind = [])
    {
        if ($this->getCountPendingQueue() > 0) {
            throw new QueryException('Queue must be empty, before streamWrite');
        }

        return $this->transport()->streamWrite($stream, $sql, $bind);
    }

    /**
     * stream Read
     *
     * @param string[] $bind
     * @return Statement
     */
    public function streamRead(Stream $streamRead, string $sql, array $bind = [])
    {
        if ($this->getCountPendingQueue() > 0) {
            throw new QueryException('Queue must be empty, before streamWrite');
        }

        return $this->transport()->streamRead($streamRead, $sql, $bind);
    }

    /**
     * Size of database
     *
     * @return mixed|null
     */
    public function databaseSize()
    {
        $b = $this->settings()->getDatabase();

        return $this->select(
            '
            SELECT database,formatReadableSize(sum(bytes)) as size
            FROM system.parts
            WHERE active AND database=:database
            GROUP BY database
            ',
            ['database' => $b]
        )->fetchOne();
    }

    /**
     * Size of tables
     *
     * @return mixed
     */
    public function tableSize(string $tableName)
    {
        $tables = $this->tablesSize();

        if (isset($tables[$tableName])) {
            return $tables[$tableName];
        }

        return null;
    }

    /**
     * Ping server
     *
     * @return bool
     */
    public function ping()
    {
        return $this->transport()->ping();
    }

    /**
     * Tables sizes
     *
     * @param bool $flatList
     * @return mixed[][]
     */
    public function tablesSize($flatList = false)
    {
        $result = $this->select('
        SELECT name as table,database,
            max(sizebytes) as sizebytes,
            max(size) as size,
            min(min_date) as min_date,
            max(max_date) as max_date
            FROM system.tables
            ANY LEFT JOIN 
            (
            SELECT table,database,
                        formatReadableSize(sum(bytes)) as size,
                        sum(bytes) as sizebytes,
                        min(min_date) as min_date,
                        max(max_date) as max_date
                        FROM system.parts 
                        WHERE active AND database=:database
                        GROUP BY table,database
            ) USING ( table,database )
            WHERE database=:database
            GROUP BY table,database
        ',
            ['database' => $this->settings()->getDatabase()]);

        if ($flatList) {
            return $result->rows();
        }

        return $result->rowsAsTree('table');
    }

    /**
     * isExists
     *
     * @return array
     */
    public function isExists(string $database, string $table)
    {
        return $this->select(
            '
            SELECT *
            FROM system.tables 
            WHERE name=\'' . $table . '\' AND database=\'' . $database . '\''
        )->rowsAsTree('name');
    }

    /**
     * List of partitions
     *
     * @return mixed[][]
     */
    public function partitions(string $table, int $limit = null, bool $active = null)
    {
        $database          = $this->settings()->getDatabase();
        $whereActiveClause = $active === null ? '' : sprintf(' AND active = %s', (int) $active);
        $limitClause       = $limit !== null ? ' LIMIT ' . $limit : '';

        return $this->select(<<<CLICKHOUSE
SELECT *
FROM system.parts 
WHERE like(table,'%$table%') AND database='$database'$whereActiveClause
ORDER BY max_date $limitClause
CLICKHOUSE
        )->rowsAsTree('name');
    }

    /**
     * dropPartition
     * @deprecated

Expected 1 lines between description and annotations, found 0.

Incorrect annotations group.

     * @return Statement
     */
    public function dropPartition(string $dataBaseTableName, string $partition_id)
    {

        $partition_id = trim($partition_id, '\'');
        $this->settings()->set('replication_alter_partitions_sync', 2);
        $state = $this->write('ALTER TABLE {dataBaseTableName} DROP PARTITION :partion_id',
            [
                'dataBaseTableName' => $dataBaseTableName,
                'partion_id'        => $partition_id,
            ]);

        return $state;
    }

    /**
     * Truncate ( drop all partitions )
     * @deprecated

Expected 1 lines between description and annotations, found 0.

Incorrect annotations group.

     * @return array

@return annotation of method \ClickHouseDB\Client::truncateTable() does not specify type hint for items of its traversable return value.

     */
    public function truncateTable(string $tableName)
    {
        $partions = $this->partitions($tableName);
        $out      = [];
        foreach ($partions as $part_key => $part) {
            $part_id       = $part['partition'];
            $out[$part_id] = $this->dropPartition($tableName, $part_id);
        }

        return $out;
    }

    /**
     * Returns the server's uptime in seconds.
     *
     * @return int
     * @throws Exception\TransportException
     */
    public function getServerUptime()
    {
        return $this->select('SELECT uptime() as uptime')->fetchOne('uptime');
    }

    /**
     * Returns string with the server version.
     */
    public function getServerVersion() : string
    {
        return (string) $this->select('SELECT version() as version')->fetchOne('version');
    }

    /**
     * Read system.settings table
     *
     * @return mixed[][]
     */
    public function getServerSystemSettings(string $like = '')
    {
        $l    = [];
        $list = $this->select('SELECT * FROM system.settings' . ($like ? ' WHERE name LIKE :like' : ''),
            ['like' => '%' . $like . '%'])->rows();
        foreach ($list as $row) {
            if (isset($row['name'])) {
                $n = $row['name'];
                unset($row['name']);
                $l[$n] = $row;
            }
        }

        return $l;
    }

    /**
     * dropOldPartitions by day_ago
     * @deprecated

Expected 1 lines between description and annotations, found 0.

     *
     * @return array

Incorrect annotations group.

@return annotation of method \ClickHouseDB\Client::dropOldPartitions() does not specify type hint for items of its traversable return value.

     * @throws Exception\TransportException
     * @throws \Exception
     */
    public function dropOldPartitions(string $table_name, int $days_ago, int $count_partitons_per_one = 100)
    {
        $days_ago = strtotime(date('Y-m-d 00:00:00', strtotime('-' . $days_ago . ' day')));

        $drop           = [];
        $list_patitions = $this->partitions($table_name, $count_partitons_per_one);

        foreach ($list_patitions as $partion_id => $partition) {
            if (stripos($partition['engine'], 'mergetree') === false) {
                continue;
            }

            // $min_date = strtotime($partition['min_date']);
            $max_date = strtotime($partition['max_date']);

            if ($max_date < $days_ago) {
                $drop[] = $partition['partition'];
            }
        }

        $result = [];
        foreach ($drop as $partition_id) {
            $result[$partition_id] = $this->dropPartition($table_name, $partition_id);
        }

        return $result;
    }
}