Client::prepareInsertAssocBulk() - Code Metrics - Inspection of "Refactor partitions()" - smi2/phpClickHouse - Measure and Improve Code Quality continuously with Scrutinizer

Passed
Pull Request — master (#90)

by Šimon
created 2018-09-11 08:40 UTC
Client::prepareInsertAssocBulk() A

↳ Parent: Client
Complexity

Conditions	5
Paths	3
Size

Total Lines	17
Code Lines	12
Duplication

Lines	0
Ratio	0 %
Code Coverage

Tests	12
CRAP Score	5
Importance

Changes
Metric	Value
eloc	12
dl	0
loc	17
c	0
b	0
f	0
ccs	12
cts	12
cp	1
rs	9.5555
cc	5
nc	3
nop	1
crap	5
<?php


namespace ClickHouseDB;

use ClickHouseDB\Exception\QueryException;
use ClickHouseDB\Query\Degeneration\Bindings;
use ClickHouseDB\Query\WhereInFile;
use ClickHouseDB\Query\WriteToFile;
use ClickHouseDB\Quote\FormatLine;
use ClickHouseDB\Transport\Http;
use ClickHouseDB\Transport\Stream;
use function sprintf;

class Client
{
    /**

     * @var Http
     */
    private $_transport = null;

    /**

     * @var string
     */
    private $_connect_username = '';

    /**

     * @var string
     */
    private $_connect_password = '';

    /**

     * @var string
     */
    private $_connect_host = '';

    /**

     * @var string
     */
    private $_connect_port = '';

    /**

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

     * @var array

     */
    private $_support_format = ['TabSeparated', 'TabSeparatedWithNames', 'CSV', 'CSVWithNames', 'JSONEachRow'];

    /**
     * Client constructor.

     * @param array $connect_params

     * @param array $settings

     */
    public function __construct($connect_params, $settings = [])

    {
        if (!isset($connect_params['username'])) {

            throw  new \InvalidArgumentException('not set username');
        }

        if (!isset($connect_params['password'])) {

            throw  new \InvalidArgumentException('not set password');
        }

        if (!isset($connect_params['port'])) {

            throw  new \InvalidArgumentException('not set port');
        }

        if (!isset($connect_params['host'])) {

            throw  new \InvalidArgumentException('not set host');
        }

        if (isset($connect_params['settings']) && is_array($connect_params['settings'])) {

            if (empty($settings)) {
                $settings = $connect_params['settings'];
            }
        }

        $this->_connect_username    = $connect_params['username'];
$a = "a";
$ab = "ab";
$abc = "abc";
        $this->_connect_password    = $connect_params['password'];
$a = "a";
$ab = "ab";
$abc = "abc";
        $this->_connect_port        = $connect_params['port'];
$a = "a";
$ab = "ab";
$abc = "abc";
        $this->_connect_host        = $connect_params['host'];
$a = "a";
$ab = "ab";
$abc = "abc";



        // init transport class
        $this->_transport = new Http(
            $this->_connect_host,
            $this->_connect_port,
            $this->_connect_username,
            $this->_connect_password
        );



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

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

            $this->settings()->apply($settings);
        }



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

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

        $this->enableHttpCompression();



    }


    /**
     * if the user has only read in the config file
     *
     * @param bool $flag
     */
    public function setReadOnlyUser($flag)

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

    {
        return $this->_transport->cleanQueryDegeneration();
    }

    /**
     * add Degeneration processing
     *
     * @param Query\Degeneration $degeneration

     * @return bool
     */
    public function addQueryDegeneration(Query\Degeneration $degeneration)

    {
        return $this->_transport->addQueryDegeneration($degeneration);
    }

    /**
     * add Conditions in query
     *
     * @return bool
     */
    public function enableQueryConditions()

    {
        return $this->_transport->addQueryDegeneration(new \ClickHouseDB\Query\Degeneration\Conditions());

    }
    /**
     * Set connection host
     *
     * @param string|array $host

     */
    public function setHost($host)

    {


        if (is_array($host))

        {
            $host = array_rand(array_flip($host));

        }

        $this->_connect_host = $host;
        $this->transport()->setHost($host);
    }

    /**
     * Таймаут
     *
     * @param int $timeout
     * @return Settings
     */
    public function setTimeout($timeout)

    {
        return $this->settings()->max_execution_time($timeout);
    }

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

    /**
     * ConnectTimeOut in seconds ( support 1.5 = 1500ms )
     *
     * @param int $connectTimeOut
     */
    public function setConnectTimeOut($connectTimeOut)

    {
        $this->transport()->setConnectTimeOut($connectTimeOut);
    }

    /**
     * get ConnectTimeOut
     *
     * @return int
     */
    public function getConnectTimeOut()

    {
        return $this->transport()->getConnectTimeOut();
    }


    /**
     * transport
     *
     * @return Http
     */
    public function transport()

    {
        if (!$this->_transport) {

            throw  new \InvalidArgumentException('Empty transport class');
        }
        return $this->_transport;
    }

    /**
     * @return string
     */
    public function getConnectHost()

    {
        return $this->_connect_host;
    }

    /**
     * @return string
     */
    public function getConnectPassword()

    {
        return $this->_connect_password;
    }

    /**
     * @return string
     */
    public function getConnectPort()

    {
        return $this->_connect_port;
    }

    /**
     * @return string
     */
    public function getConnectUsername()

    {
        return $this->_connect_username;
    }

    /**
     * transport
     *
     * @return Http
     */
    public function getTransport()

    {
        return $this->_transport;
    }


    /**
     * Режим отладки CURL
     *
     * @return mixed
     */
    public function verbose()
    {
        return $this->transport()->verbose(true);
    }

    /**
     * @return Settings
     */
    public function settings()

    {
        return $this->transport()->settings();
    }

    /**
     * @return $this
     */
    public function useSession($useSessionId = false)

    {
        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 string $sql
     * @param array $bindings

     * @param bool $exception

     * @return Statement
     * @throws Exception\TransportException
     */
    public function write($sql, $bindings = [], $exception = true)

    {
        return $this->transport()->write($sql, $bindings, $exception);
    }

    /**
     * set db name
     * @param string $db
     * @return $this
     */
    public function database($db)

    {
        $this->settings()->database($db);
        return $this;
    }

    /**
     * Write to system.query_log
     *
     * @param bool $flag
     * @return $this
     */
    public function enableLogQueries($flag = true)

    {
        $this->settings()->set('log_queries', intval($flag));

        return $this;
    }

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

    {
        $this->settings()->enableHttpCompression($flag);
        return $this;
    }

    /**
     * Enable / Disable HTTPS
     *
     * @param bool $flag
     * @return $this
     */
    public function https($flag = true)

    {
        $this->settings()->https($flag);
        return $this;
    }

    /**
     * Read extremes of the result columns. They can be output in JSON-formats.
     *
     * @param bool $flag
     * @return $this
     */
    public function enableExtremes($flag = true)

    {
        $this->settings()->set('extremes', intval($flag));

        return $this;
    }

    /**
     * SELECT
     *
     * @param string $sql

     * @param array $bindings

     * @param null|WhereInFile $whereInFile
     * @param null|WriteToFile $writeToFile
     * @return Statement
     * @throws Exception\TransportException
     * @throws \Exception
     */
    public function select($sql, $bindings = [], $whereInFile = null, $writeToFile = null)

    {
        return $this->transport()->select($sql, $bindings, $whereInFile, $writeToFile);
    }

    /**
     * execute run
     *
     * @return bool
     * @throws Exception\TransportException
     */
    public function executeAsync()

    {
        return $this->transport()->executeAsync();
    }

    /**
     * set progressFunction
     *
     * @param callable $callback
     */
    public function progressFunction($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 string $sql

     * @param array $bindings

     * @param null|WhereInFile $whereInFile
     * @param null|WriteToFile $writeToFile
     * @return Statement
     * @throws Exception\TransportException
     * @throws \Exception
     */
    public function selectAsync($sql, $bindings = [], $whereInFile = null, $writeToFile = null)

    {
        return $this->transport()->selectAsync($sql, $bindings, $whereInFile, $writeToFile);
    }

    /**
     * SHOW PROCESSLIST
     *
     * @return array

     * @throws Exception\TransportException
     * @throws \Exception
     */
    public function showProcesslist()

    {
        return $this->select('SHOW PROCESSLIST')->rows();
    }

    /**
     * show databases
     *
     * @return array

     * @throws Exception\TransportException
     * @throws \Exception
     */
    public function showDatabases()

    {
        return $this->select('show databases')->rows();
    }

    /**
     * statement = SHOW CREATE TABLE
     *
     * @param string $table
     * @return mixed
     * @throws Exception\TransportException
     * @throws \Exception
     */
    public function showCreateTable($table)

    {
        return ($this->select('SHOW CREATE TABLE ' . $table)->fetchOne('statement'));

    }

    /**
     * SHOW TABLES
     *
     * @return array

     * @throws Exception\TransportException
     * @throws \Exception
     */
    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();
    }

    /**
     * Insert Array
     *
     * @param string $table
     * @param array $values

     * @param array $columns

     * @return Statement
     * @throws Exception\TransportException
     */
    public function insert($table, $values, $columns = [])

    {
        $sql = 'INSERT INTO `' . $table . "`";
<?php

$singleQuoted = 'Value';
$doubleQuoted = "\tSingle is $singleQuoted";

print $doubleQuoted;

        if (0 !== count($columns)) {

            $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 array $values - array column_name => value (if we insert one row) or array list column_name => value if we insert many lines
      * @return array - 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("Fields not match: " . implode(',', $_fields) . " and " . implode(',', $preparedFields) . " on element $idx");
<?php

$singleQuoted = 'Value';
$doubleQuoted = "\tSingle is $singleQuoted";

print $doubleQuoted;
                }
                $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 string $table - table name

     * @param array $values - array column_name => value (if we insert one row) or array list column_name => value if we insert many lines

     * @return Statement
     * @throws QueryException
     * @throws Exception\TransportException
     */
    public function insertAssocBulk($table, array $values)

    {
        list($columns, $vals) = $this->prepareInsertAssocBulk($values);
        return $this->insert($table, $vals, $columns);
    }

    /**
     * insert TabSeparated files
     *
     * @param string $table_name

     * @param string|array $file_names

     * @param array $columns_array

     * @return mixed
     * @throws Exception\TransportException
     */
    public function insertBatchTSVFiles($table_name, $file_names, $columns_array=[])

    {
        return $this->insertBatchFiles($table_name, $file_names, $columns_array, 'TabSeparated');
    }

    /**
     * insert Batch Files
     *
     * @param string $table_name

     * @param string|array $file_names

     * @param array $columns_array

     * @param string $format ['TabSeparated','TabSeparatedWithNames','CSV','CSVWithNames']

     * @return array

     * @throws Exception\TransportException
     */
    public function insertBatchFiles($table_name, $file_names, $columns_array=[], $format = "CSV")
<?php

$singleQuoted = 'Value';
$doubleQuoted = "\tSingle is $singleQuoted";

print $doubleQuoted;
    {
        if (is_string($file_names))

        {
            $file_names = [$file_names];
        }
        if ($this->getCountPendingQueue() > 0) {
            throw new QueryException('Queue must be empty, before insertBatch, need executeAsync');
        }

        if (!in_array($format, $this->_support_format))

        {
            throw new QueryException('Format not support in insertBatchFiles');
        }

        $result = [];

        foreach ($file_names 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_array))
            {
                $sql = 'INSERT INTO ' . $table_name . ' FORMAT ' . $format;


            } else

            {
                $sql = 'INSERT INTO ' . $table_name . ' ( ' . implode(',', $columns_array) . ' ) FORMAT ' . $format;



            }
            $result[$fileName] = $this->transport()->writeAsyncCSV($sql, $fileName);
        }

        // exec
        $this->executeAsync();

        // fetch resutl
        foreach ($file_names as $fileName) {
            if ($result[$fileName]->isError()) {

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

        return $result;
    }

    /**
     * insert Batch Stream
     *
     * @param string $table_name
     * @param array $columns_array

     * @param string $format ['TabSeparated','TabSeparatedWithNames','CSV','CSVWithNames']

     * @return Transport\CurlerRequest
     */
    public function insertBatchStream($table_name, $columns_array=[], $format = "CSV")
<?php

$singleQuoted = 'Value';
$doubleQuoted = "\tSingle is $singleQuoted";

print $doubleQuoted;
    {
        if ($this->getCountPendingQueue() > 0) {
            throw new QueryException('Queue must be empty, before insertBatch, need executeAsync');
        }

        if (!in_array($format, $this->_support_format))

        {
            throw new QueryException('Format not support in insertBatchFiles');
        }

        if (empty($columns_array))
        {
            $sql = 'INSERT INTO ' . $table_name . ' FORMAT ' . $format;


        } else

        {
            $sql = 'INSERT INTO ' . $table_name . ' ( ' . implode(',', $columns_array) . ' ) FORMAT ' . $format;



        }

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


    /**
     * stream Write
     *
     * @param Stream $stream

     * @param string $sql
     * @param array $bind

     * @return Statement
     * @throws Exception\TransportException
     */
    public function streamWrite(Stream $stream,$sql,$bind=[])

    {
        if ($this->getCountPendingQueue() > 0) {
            throw new QueryException('Queue must be empty, before streamWrite');
        }
        return $this->transport()->streamWrite($stream,$sql,$bind);
    }


    /**
     * stream Read
     *
     * @param Stream $streamRead

     * @param string $sql
     * @param array $bind

     * @return Statement
     * @throws Exception\TransportException
     */
    public function streamRead(Stream $streamRead,$sql,$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
     * @throws Exception\TransportException
     * @throws \Exception
     */
    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
     *
     * @param string $tableName
     * @return mixed
     * @throws Exception\TransportException
     * @throws \Exception
     */
    public function tableSize($tableName)

    {
        $tables = $this->tablesSize();

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

        return null;
    }

    /**
     * Ping server
     *
     * @return bool
     * @throws Exception\TransportException
     */
    public function ping()

    {
        return $this->transport()->ping();
    }

    /**
     * Tables sizes
     *
     * @param bool $flatList
     * @return array

     * @throws Exception\TransportException
     * @throws \Exception
     */
    public function tablesSize($flatList = false)

    {
        $z = $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 $z->rows();
        }



        return $z->rowsAsTree('table');



    }



    /**
     * isExists
     *
     * @param string $database
     * @param string $table
     * @return array

     * @throws Exception\TransportException
     * @throws \Exception
     */
    public function isExists($database, $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
     * @param string $dataBaseTableName database_name.table_name
     * @param string $partition_id
     * @return Statement
     * @throws Exception\TransportException
     */
    public function dropPartition($dataBaseTableName, $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
     * @param string $tableName
     * @return array

     * @throws Exception\TransportException
     * @throws \Exception
     */
    public function truncateTable($tableName)

    {
        $partions = $this->partitions($tableName);
        $out = [];
$a = "a";
$ab = "ab";
$abc = "abc";
        foreach ($partions as $part_key=>$part)

        {
            $part_id = $part['partition'];
$a = "a";
$ab = "ab";
$abc = "abc";
            $out[$part_id] = $this->dropPartition($tableName, $part_id);
            $out[$part_id] = /** @scrutinizer ignore-deprecated */ $this->dropPartition($tableName, $part_id);
        }
        return $out;
    }

    /**
     * Returns the server's uptime in seconds.
     *
     * @return array

     * @throws Exception\TransportException
     */
    public function getServerUptime()

    {
        return $this->select('SELECT uptime() as uptime')->fetchOne('uptime');
    }

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


    /**
     * Read system.settings table
     *
     * @param string $like
     * @return array

     * @throws Exception\TransportException
     */
    public function getServerSystemSettings($like='')

    {
        $l=[];
$a = "a";
$ab = "ab";
$abc = "abc";
        $list=$this->select('SELECT * FROM system.settings'.($like ? ' WHERE name LIKE :like':'' ),['like'=>'%'.$like.'%'])->rows();
$a = "a";
$ab = "ab";
$abc = "abc";
        foreach ($list as $row) {
            if (isset($row['name'])) {$n=$row['name']; unset($row['name']) ; $l[$n]=$row;}
// Bad
$a = 5; $b = 6; $c = 7;

// Good
$a = 5;
$b = 6;
$c = 7;
        }
        return $l;
    }



    /**
     * dropOldPartitions by day_ago
     * @deprecated
     *
     * @param string $table_name
     * @param int $days_ago

     * @param int $count_partitons_per_one

     * @return array

     * @throws Exception\TransportException
     * @throws \Exception
     */
    public function dropOldPartitions($table_name, $days_ago, $count_partitons_per_one = 100)

    {
        $days_ago = strtotime(date('Y-m-d 00:00:00', strtotime('-' . $days_ago . ' day')));


        $drop = [];
$a = "a";
$ab = "ab";
$abc = "abc";
        $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);
            $result[$partition_id] = /** @scrutinizer ignore-deprecated */ $this->dropPartition($table_name, $partition_id);
        }

        return $result;
    }
}

smi2 / phpClickHouse

Pull Request — master (#90)

Client::prepareInsertAssocBulk() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like
