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
{
    private const DEFAULT_USERNAME  = 'default';
    private const DEFAULT_PASSWORD  = '';
    private const DEFAULT_PORT      = 8123;
    private const DEFAULT_HOST      = '127.0.0.1';
    private const DEFAULT_DATABASE  = 'default';
    private const SUPPORTED_FORMATS = ['TabSeparated', 'TabSeparatedWithNames', 'CSV', 'CSVWithNames', 'JSONEachRow'];

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

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

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

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

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

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

    /**
     * @param mixed[] $connectParams
     * @param mixed[] $settings
     */
    public function __construct(array $connectParams, array $settings = [])
    {
        $this->username = $connectParams['username'] ?? self::DEFAULT_USERNAME;
        $this->password = $connectParams['password'] ?? self::DEFAULT_PASSWORD;
        $this->port     = $connectParams['port'] ?? self::DEFAULT_PORT;
        $this->host     = $connectParams['host'] ?? self::DEFAULT_HOST;
        $this->database = $connectParams['database'] ?? self::DEFAULT_DATABASE;

        $this->transport = new Http(
            $this->host,
            $this->port,
            $this->username,
            $this->password,
            $this->database
        );

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

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

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

        $this->setHttpCompression(true);
    }

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

    /**
     * Degeneration processing
     */
    public function addQueryDegeneration(Degeneration $degeneration) : void
    {
        $this->transport->addQueryDegeneration($degeneration);
    }

    /**
     * Add Conditions Degeneration to query processing
     */
    public function enableQueryConditions() : void
    {
        $this->transport->addQueryDegeneration(new Conditions());
    }

    public function setTimeout(float $seconds) : void
    {
        $this->transport->setTimeout($seconds);
    }

    public function setConnectTimeout(float $seconds) : void
    {
        $this->transport->setConnectTimeout($seconds);
    }

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

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

    public function getPassword() : string
    {
        return $this->password;
    }

    public function getPort() : int
    {
        return $this->port;

The expression return $this->port returns the type string which is incompatible with the type-hinted return integer.

    }

    public function getUsername() : string
    {
        return $this->username;
    }

    public function setDatabase(string $database) : self
    {
        $this->database = $database;
        $this->transport->setDatabase($database);

        return $this;
    }

    public function getTransport() : Http
    {
        return $this->transport;
    }

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

    public function getSettings() : Settings
    {
        return $this->transport->getSettings();
    }

    /**
     * @return static
     */
    public function useSession(bool $useSessionId = false)
    {
        if (! $this->getSettings()->getSessionId()) {
            if (! $useSessionId) {
                $this->getSettings()->makeSessionId();
            } else {
                $this->getSettings()->session_id($useSessionId);
            }
        }

        return $this;
    }

    /**
     * @return mixed
     */
    public function getSession()
    {
        return $this->getSettings()->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);
    }

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

        return $this;
    }

    /**
     * Compress the result if the HTTP client said that it understands data compressed with gzip or deflate
     */
    public function setHttpCompression(bool $enable) : self
    {
        $this->getSettings()->setHttpCompression($enable);

        return $this;
    }

    /**
     * Enable / Disable HTTPS
     *
     * @return static
     */
    public function setHttps(bool $flag)
    {
        $this->transport->setHttps($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->getSettings()->set('extremes', (int) $flag);

        return $this;
    }

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

    /**
     * @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);
    }

    /**
     * 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);
    }

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

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

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

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

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

    /**
     * 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);
    }

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

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

The method getDatabase() does not exist on ClickHouseDB\Transport\Http.

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

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

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

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

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

        return null;
    }

    /**
     * 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->database]);

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

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

    /**
     * isExists
     *
     * @return array
     */
    public function tableExists(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->database;
        $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
     * @return Statement
     */
    public function dropPartition(string $dataBaseTableName, string $partition_id)
    {

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

        return $state;
    }

    /**
     * dropOldPartitions by day_ago
     * @deprecated
     *
     * @return array
     * @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;
    }

    /**
     * Truncate ( drop all partitions )
     * @deprecated
     * @return array
     */
    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;
    }
}