| 1 | <?php |
||||
| 2 | |||||
| 3 | declare(strict_types=1); |
||||
| 4 | |||||
| 5 | namespace ClickHouseDB; |
||||
| 6 | |||||
| 7 | use ClickHouseDB\Exception\QueryException; |
||||
| 8 | use ClickHouseDB\Exception\TransportException; |
||||
| 9 | use ClickHouseDB\Query\Degeneration; |
||||
| 10 | use ClickHouseDB\Query\Degeneration\Bindings; |
||||
| 11 | use ClickHouseDB\Query\Degeneration\Conditions; |
||||
| 12 | use ClickHouseDB\Query\WhereInFile; |
||||
| 13 | use ClickHouseDB\Query\WriteToFile; |
||||
| 14 | use ClickHouseDB\Quote\FormatLine; |
||||
| 15 | use ClickHouseDB\Transport\Http; |
||||
| 16 | use ClickHouseDB\Transport\Stream; |
||||
| 17 | use function array_flip; |
||||
| 18 | use function array_keys; |
||||
| 19 | use function array_rand; |
||||
| 20 | use function array_values; |
||||
| 21 | use function count; |
||||
| 22 | use function date; |
||||
| 23 | use function implode; |
||||
| 24 | use function in_array; |
||||
| 25 | use function is_array; |
||||
| 26 | use function is_callable; |
||||
| 27 | use function is_file; |
||||
| 28 | use function is_readable; |
||||
| 29 | use function is_string; |
||||
| 30 | use function sprintf; |
||||
| 31 | use function stripos; |
||||
| 32 | use function strtotime; |
||||
| 33 | use function trim; |
||||
| 34 | |||||
| 35 | /** |
||||
| 36 | * Class Client |
||||
|
0 ignored issues
–
show
introduced
by
Loading history...
|
|||||
| 37 | * @package ClickHouseDB |
||||
|
0 ignored issues
–
show
|
|||||
| 38 | */ |
||||
| 39 | class Client |
||||
| 40 | { |
||||
| 41 | const SUPPORTED_FORMATS = ['TabSeparated', 'TabSeparatedWithNames', 'CSV', 'CSVWithNames', 'JSONEachRow','CSVWithNamesAndTypes','TSVWithNamesAndTypes']; |
||||
| 42 | |||||
| 43 | /** @var Http */ |
||||
| 44 | private $transport; |
||||
| 45 | |||||
| 46 | /** @var string */ |
||||
| 47 | private $connectUsername; |
||||
| 48 | |||||
| 49 | /** @var string */ |
||||
| 50 | private $connectPassword; |
||||
| 51 | |||||
| 52 | /** @var string */ |
||||
| 53 | private $connectHost; |
||||
| 54 | |||||
| 55 | /** @var int */ |
||||
| 56 | private $connectPort; |
||||
| 57 | |||||
| 58 | /** @var int */ |
||||
| 59 | private $authMethod; |
||||
| 60 | |||||
| 61 | /** @var bool */ |
||||
| 62 | private $connectUserReadonly = false; |
||||
| 63 | |||||
| 64 | /** |
||||
| 65 | * @param mixed[] $connectParams |
||||
| 66 | * @param mixed[] $settings |
||||
| 67 | 68 | */ |
|||
| 68 | public function __construct(array $connectParams, array $settings = []) |
||||
| 69 | 68 | { |
|||
| 70 | if (!isset($connectParams['username'])) { |
||||
| 71 | throw new \InvalidArgumentException('not set username'); |
||||
| 72 | } |
||||
| 73 | 68 | ||||
| 74 | if (!isset($connectParams['password'])) { |
||||
| 75 | throw new \InvalidArgumentException('not set password'); |
||||
| 76 | } |
||||
| 77 | 68 | ||||
| 78 | if (!isset($connectParams['port'])) { |
||||
| 79 | throw new \InvalidArgumentException('not set port'); |
||||
| 80 | } |
||||
| 81 | 68 | ||||
| 82 | if (!isset($connectParams['host'])) { |
||||
| 83 | throw new \InvalidArgumentException('not set host'); |
||||
| 84 | } |
||||
| 85 | 68 | ||||
| 86 | if (array_key_exists('auth_method', $connectParams)) { |
||||
| 87 | if (false === in_array($connectParams['auth_method'], Http::AUTH_METHODS_LIST)) { |
||||
| 88 | $errorMessage = sprintf( |
||||
| 89 | 'Invalid value for "auth_method" param. Should be one of [%s].', |
||||
| 90 | json_encode(Http::AUTH_METHODS_LIST) |
||||
| 91 | ); |
||||
| 92 | throw new \InvalidArgumentException($errorMessage); |
||||
| 93 | } |
||||
| 94 | |||||
| 95 | $this->authMethod = $connectParams['auth_method']; |
||||
| 96 | } |
||||
| 97 | 68 | ||||
| 98 | 68 | $this->connectUsername = $connectParams['username']; |
|||
| 99 | 68 | $this->connectPassword = $connectParams['password']; |
|||
| 100 | 68 | $this->connectPort = intval($connectParams['port']); |
|||
| 101 | $this->connectHost = $connectParams['host']; |
||||
| 102 | |||||
| 103 | 68 | // init transport class |
|||
| 104 | 68 | $this->transport = new Http( |
|||
| 105 | 68 | $this->connectHost, |
|||
| 106 | 68 | $this->connectPort, |
|||
| 107 | 68 | $this->connectUsername, |
|||
| 108 | 68 | $this->connectPassword, |
|||
| 109 | $this->authMethod |
||||
| 110 | ); |
||||
| 111 | 68 | ||||
| 112 | $this->transport->addQueryDegeneration(new Bindings()); |
||||
| 113 | |||||
| 114 | 68 | // apply settings to transport class |
|||
| 115 | 68 | $this->settings()->database('default'); |
|||
| 116 | 1 | if (!empty($settings)) { |
|||
| 117 | $this->settings()->apply($settings); |
||||
| 118 | } |
||||
| 119 | 68 | ||||
| 120 | if (isset($connectParams['readonly'])) { |
||||
| 121 | $this->setReadOnlyUser($connectParams['readonly']); |
||||
| 122 | } |
||||
| 123 | 68 | ||||
| 124 | if (isset($connectParams['https'])) { |
||||
| 125 | $this->https($connectParams['https']); |
||||
| 126 | } |
||||
| 127 | 68 | ||||
| 128 | if (isset($connectParams['sslCA'])) { |
||||
| 129 | $this->transport->setSslCa($connectParams['sslCA']); |
||||
| 130 | } |
||||
| 131 | 68 | } |
|||
| 132 | 68 | ||||
| 133 | /** |
||||
| 134 | * if the user has only read in the config file |
||||
| 135 | */ |
||||
| 136 | public function setReadOnlyUser(bool $flag) |
||||
| 137 | { |
||||
| 138 | $this->connectUserReadonly = $flag; |
||||
| 139 | $this->settings()->setReadOnlyUser($this->connectUserReadonly); |
||||
| 140 | } |
||||
| 141 | |||||
| 142 | /** |
||||
| 143 | * Clear Degeneration processing request [template ] |
||||
| 144 | * |
||||
| 145 | * @return bool |
||||
| 146 | */ |
||||
| 147 | public function cleanQueryDegeneration() |
||||
|
0 ignored issues
–
show
|
|||||
| 148 | 1 | { |
|||
| 149 | return $this->transport->cleanQueryDegeneration(); |
||||
| 150 | 1 | } |
|||
| 151 | |||||
| 152 | /** |
||||
| 153 | * add Degeneration processing |
||||
| 154 | * |
||||
| 155 | * @return bool |
||||
| 156 | */ |
||||
| 157 | public function addQueryDegeneration(Degeneration $degeneration) |
||||
| 158 | { |
||||
| 159 | return $this->transport->addQueryDegeneration($degeneration); |
||||
| 160 | } |
||||
| 161 | |||||
| 162 | /** |
||||
| 163 | * add Conditions in query |
||||
| 164 | * |
||||
| 165 | * @return bool |
||||
|
0 ignored issues
–
show
|
|||||
| 166 | */ |
||||
| 167 | public function enableQueryConditions(): bool |
||||
| 168 | 3 | { |
|||
| 169 | return $this->transport->addQueryDegeneration(new Conditions()); |
||||
| 170 | 3 | } |
|||
| 171 | |||||
| 172 | /** |
||||
| 173 | * Set connection host |
||||
| 174 | * |
||||
| 175 | * @param string $host |
||||
| 176 | */ |
||||
| 177 | public function setHost($host): void |
||||
|
0 ignored issues
–
show
|
|||||
| 178 | { |
||||
| 179 | $this->connectHost = $host; |
||||
| 180 | $this->transport()->setHost($host); |
||||
| 181 | } |
||||
| 182 | |||||
| 183 | /** |
||||
| 184 | * max_execution_time , in int value (seconds) |
||||
| 185 | */ |
||||
| 186 | public function setTimeout($timeout): Settings |
||||
|
0 ignored issues
–
show
|
|||||
| 187 | 2 | { |
|||
| 188 | return $this->settings()->max_execution_time(intval($timeout)); |
||||
| 189 | 2 | } |
|||
| 190 | |||||
| 191 | /** |
||||
| 192 | * @return int |
||||
| 193 | */ |
||||
| 194 | public function getTimeout(): int |
||||
| 195 | 1 | { |
|||
| 196 | return $this->settings()->getTimeOut(); |
||||
| 197 | 1 | } |
|||
| 198 | |||||
| 199 | /** |
||||
| 200 | * ConnectTimeOut in seconds ( support 1.5 = 1500ms ) |
||||
| 201 | */ |
||||
| 202 | public function setConnectTimeOut(float $connectTimeOut): void |
||||
| 203 | 2 | { |
|||
| 204 | $this->transport()->setConnectTimeOut($connectTimeOut); |
||||
| 205 | 2 | } |
|||
| 206 | 2 | ||||
| 207 | /** |
||||
| 208 | * @return float |
||||
| 209 | */ |
||||
| 210 | public function getConnectTimeOut(): float |
||||
| 211 | 1 | { |
|||
| 212 | return $this->transport()->getConnectTimeOut(); |
||||
| 213 | 1 | } |
|||
| 214 | |||||
| 215 | /** |
||||
| 216 | * @return Http |
||||
| 217 | */ |
||||
| 218 | public function transport(): Http |
||||
| 219 | 68 | { |
|||
| 220 | if (!$this->transport) { |
||||
| 221 | 68 | throw new \InvalidArgumentException('Empty transport class'); |
|||
|
0 ignored issues
–
show
|
|||||
| 222 | } |
||||
| 223 | |||||
| 224 | return $this->transport; |
||||
| 225 | 68 | } |
|||
| 226 | |||||
| 227 | /** |
||||
| 228 | * @return string |
||||
|
0 ignored issues
–
show
|
|||||
| 229 | */ |
||||
| 230 | public function getConnectHost(): string |
||||
| 231 | { |
||||
| 232 | return $this->connectHost; |
||||
| 233 | } |
||||
| 234 | |||||
| 235 | /** |
||||
| 236 | * @return string |
||||
|
0 ignored issues
–
show
|
|||||
| 237 | */ |
||||
| 238 | public function getConnectPassword(): string |
||||
| 239 | { |
||||
| 240 | return $this->connectPassword; |
||||
| 241 | } |
||||
| 242 | |||||
| 243 | /** |
||||
| 244 | * @return string |
||||
|
0 ignored issues
–
show
|
|||||
| 245 | */ |
||||
| 246 | public function getConnectPort(): string |
||||
| 247 | { |
||||
| 248 | return strval($this->connectPort); |
||||
| 249 | } |
||||
| 250 | |||||
| 251 | /** |
||||
| 252 | * @return string |
||||
| 253 | */ |
||||
| 254 | public function getConnectUsername() |
||||
|
0 ignored issues
–
show
|
|||||
| 255 | { |
||||
| 256 | return $this->connectUsername; |
||||
| 257 | } |
||||
| 258 | |||||
| 259 | /** |
||||
| 260 | * @return int |
||||
| 261 | */ |
||||
| 262 | public function getAuthMethod(): int |
||||
| 263 | { |
||||
| 264 | return $this->authMethod; |
||||
| 265 | } |
||||
| 266 | |||||
| 267 | /** |
||||
| 268 | * @return Http |
||||
| 269 | */ |
||||
| 270 | public function getTransport() |
||||
|
0 ignored issues
–
show
|
|||||
| 271 | { |
||||
| 272 | return $this->transport; |
||||
| 273 | } |
||||
| 274 | |||||
| 275 | /** |
||||
| 276 | * @return bool |
||||
|
0 ignored issues
–
show
|
|||||
| 277 | */ |
||||
| 278 | public function verbose(bool $flag = true):bool |
||||
|
0 ignored issues
–
show
The parameter
$flag is not used and could be removed.
(
Ignorable by Annotation
)
If this is a false-positive, you can also ignore this issue in your code via the
This check looks for parameters that have been defined for a function or method, but which are not used in the method body. Loading history...
|
|||||
| 279 | { |
||||
| 280 | return $this->transport()->verbose(true); |
||||
| 281 | } |
||||
| 282 | |||||
| 283 | /** |
||||
| 284 | * @return Settings |
||||
| 285 | */ |
||||
| 286 | public function settings() |
||||
|
0 ignored issues
–
show
|
|||||
| 287 | 68 | { |
|||
| 288 | return $this->transport()->settings(); |
||||
| 289 | 68 | } |
|||
| 290 | |||||
| 291 | /** |
||||
| 292 | * @param string|null $useSessionId |
||||
|
0 ignored issues
–
show
|
|||||
| 293 | * @return $this |
||||
| 294 | */ |
||||
| 295 | public function useSession(string $useSessionId = null) |
||||
|
0 ignored issues
–
show
|
|||||
| 296 | 2 | { |
|||
| 297 | if (!$this->settings()->getSessionId()) { |
||||
| 298 | 2 | if (!$useSessionId) { |
|||
| 299 | 2 | $this->settings()->makeSessionId(); |
|||
| 300 | 2 | } else { |
|||
| 301 | $this->settings()->session_id($useSessionId); |
||||
| 302 | } |
||||
| 303 | } |
||||
| 304 | return $this; |
||||
|
0 ignored issues
–
show
|
|||||
| 305 | 2 | } |
|||
| 306 | |||||
| 307 | /** |
||||
| 308 | * @return mixed |
||||
| 309 | */ |
||||
| 310 | public function getSession() |
||||
| 311 | 2 | { |
|||
| 312 | return $this->settings()->getSessionId(); |
||||
| 313 | 2 | } |
|||
| 314 | |||||
| 315 | /** |
||||
| 316 | * Query CREATE/DROP |
||||
| 317 | * |
||||
| 318 | * @param mixed[] $bindings |
||||
| 319 | * @return Statement |
||||
| 320 | */ |
||||
| 321 | public function write(string $sql, array $bindings = [], bool $exception = true) |
||||
| 322 | 27 | { |
|||
| 323 | return $this->transport()->write($sql, $bindings, $exception); |
||||
| 324 | 27 | } |
|||
| 325 | |||||
| 326 | /** |
||||
| 327 | * set db name |
||||
| 328 | * @return static |
||||
| 329 | */ |
||||
| 330 | public function database(string $db) |
||||
| 331 | 68 | { |
|||
| 332 | $this->settings()->database($db); |
||||
| 333 | 68 | ||||
| 334 | return $this; |
||||
| 335 | 68 | } |
|||
| 336 | |||||
| 337 | /** |
||||
| 338 | * Write to system.query_log |
||||
| 339 | * |
||||
| 340 | * @return static |
||||
| 341 | */ |
||||
| 342 | public function enableLogQueries(bool $flag = true) |
||||
| 343 | { |
||||
| 344 | $this->settings()->set('log_queries', (int)$flag); |
||||
| 345 | |||||
| 346 | return $this; |
||||
| 347 | } |
||||
| 348 | |||||
| 349 | /** |
||||
| 350 | * Compress the result if the HTTP client said that it understands data compressed with gzip or deflate |
||||
| 351 | * |
||||
| 352 | * @return static |
||||
| 353 | */ |
||||
| 354 | public function enableHttpCompression(bool $flag = true) |
||||
| 355 | 68 | { |
|||
| 356 | $this->settings()->enableHttpCompression($flag); |
||||
| 357 | 68 | ||||
| 358 | return $this; |
||||
| 359 | 68 | } |
|||
| 360 | |||||
| 361 | /** |
||||
| 362 | * Enable / Disable HTTPS |
||||
| 363 | * |
||||
| 364 | * @return static |
||||
| 365 | */ |
||||
| 366 | public function https(bool $flag = true) |
||||
| 367 | 1 | { |
|||
| 368 | $this->settings()->https($flag); |
||||
| 369 | 1 | ||||
| 370 | return $this; |
||||
| 371 | 1 | } |
|||
| 372 | |||||
| 373 | /** |
||||
| 374 | * Read extremes of the result columns. They can be output in JSON-formats. |
||||
| 375 | * |
||||
| 376 | * @return static |
||||
| 377 | */ |
||||
| 378 | public function enableExtremes(bool $flag = true) |
||||
| 379 | 2 | { |
|||
| 380 | $this->settings()->set('extremes', (int)$flag); |
||||
| 381 | 2 | ||||
| 382 | return $this; |
||||
| 383 | 2 | } |
|||
| 384 | |||||
| 385 | /** |
||||
| 386 | * @param mixed[] $bindings |
||||
| 387 | * @return Statement |
||||
| 388 | */ |
||||
| 389 | public function select( |
||||
| 390 | 32 | string $sql, |
|||
| 391 | array $bindings = [], |
||||
| 392 | WhereInFile $whereInFile = null, |
||||
| 393 | WriteToFile $writeToFile = null |
||||
| 394 | ) |
||||
| 395 | { |
||||
| 396 | return $this->transport()->select($sql, $bindings, $whereInFile, $writeToFile); |
||||
| 397 | 32 | } |
|||
| 398 | |||||
| 399 | /** |
||||
| 400 | * @return bool |
||||
| 401 | */ |
||||
| 402 | public function executeAsync() |
||||
|
0 ignored issues
–
show
|
|||||
| 403 | 10 | { |
|||
| 404 | return $this->transport()->executeAsync(); |
||||
| 405 | 10 | } |
|||
| 406 | |||||
| 407 | public function maxTimeExecutionAllAsync() |
||||
|
0 ignored issues
–
show
|
|||||
| 408 | { |
||||
| 409 | |||||
| 410 | } |
||||
|
0 ignored issues
–
show
|
|||||
| 411 | |||||
| 412 | /** |
||||
| 413 | * set progressFunction |
||||
| 414 | */ |
||||
| 415 | public function progressFunction(callable $callback) |
||||
| 416 | 1 | { |
|||
| 417 | if (!is_callable($callback)) { |
||||
| 418 | 1 | throw new \InvalidArgumentException('Not is_callable progressFunction'); |
|||
| 419 | } |
||||
| 420 | |||||
| 421 | if (!$this->settings()->is('send_progress_in_http_headers')) { |
||||
| 422 | 1 | $this->settings()->set('send_progress_in_http_headers', 1); |
|||
| 423 | 1 | } |
|||
| 424 | if (!$this->settings()->is('http_headers_progress_interval_ms')) { |
||||
| 425 | 1 | $this->settings()->set('http_headers_progress_interval_ms', 100); |
|||
| 426 | 1 | } |
|||
| 427 | |||||
| 428 | $this->transport()->setProgressFunction($callback); |
||||
| 429 | 1 | } |
|||
| 430 | 1 | ||||
| 431 | /** |
||||
| 432 | * prepare select |
||||
| 433 | * |
||||
| 434 | * @param mixed[] $bindings |
||||
| 435 | * @return Statement |
||||
| 436 | */ |
||||
| 437 | public function selectAsync( |
||||
| 438 | 7 | string $sql, |
|||
| 439 | array $bindings = [], |
||||
| 440 | WhereInFile $whereInFile = null, |
||||
| 441 | WriteToFile $writeToFile = null |
||||
| 442 | ) |
||||
| 443 | { |
||||
| 444 | return $this->transport()->selectAsync($sql, $bindings, $whereInFile, $writeToFile); |
||||
| 445 | 7 | } |
|||
| 446 | |||||
| 447 | /** |
||||
| 448 | * SHOW PROCESSLIST |
||||
| 449 | * |
||||
| 450 | * @return array |
||||
|
0 ignored issues
–
show
|
|||||
| 451 | */ |
||||
| 452 | public function showProcesslist() |
||||
|
0 ignored issues
–
show
|
|||||
| 453 | { |
||||
| 454 | return $this->select('SHOW PROCESSLIST')->rows(); |
||||
| 455 | } |
||||
| 456 | |||||
| 457 | /** |
||||
| 458 | * show databases |
||||
| 459 | * |
||||
| 460 | * @return array |
||||
|
0 ignored issues
–
show
|
|||||
| 461 | */ |
||||
| 462 | public function showDatabases() |
||||
|
0 ignored issues
–
show
|
|||||
| 463 | { |
||||
| 464 | return $this->select('show databases')->rows(); |
||||
| 465 | } |
||||
| 466 | |||||
| 467 | /** |
||||
| 468 | * statement = SHOW CREATE TABLE |
||||
| 469 | * |
||||
| 470 | * @return mixed |
||||
| 471 | */ |
||||
| 472 | public function showCreateTable(string $table) |
||||
| 473 | { |
||||
| 474 | return $this->select('SHOW CREATE TABLE ' . $table)->fetchOne('statement'); |
||||
| 475 | } |
||||
| 476 | |||||
| 477 | /** |
||||
| 478 | * SHOW TABLES |
||||
| 479 | * |
||||
| 480 | * @return mixed[] |
||||
| 481 | */ |
||||
| 482 | public function showTables() |
||||
|
0 ignored issues
–
show
|
|||||
| 483 | 1 | { |
|||
| 484 | return $this->select('SHOW TABLES')->rowsAsTree('name'); |
||||
| 485 | 1 | } |
|||
| 486 | |||||
| 487 | /** |
||||
| 488 | * Get the number of simultaneous/Pending requests |
||||
| 489 | * |
||||
| 490 | * @return int |
||||
| 491 | */ |
||||
| 492 | public function getCountPendingQueue() |
||||
|
0 ignored issues
–
show
|
|||||
| 493 | 12 | { |
|||
| 494 | return $this->transport()->getCountPendingQueue(); |
||||
| 495 | 12 | } |
|||
| 496 | |||||
| 497 | /** |
||||
| 498 | * @param mixed[][] $values |
||||
| 499 | * @param string[] $columns |
||||
| 500 | * @return Statement |
||||
|
0 ignored issues
–
show
|
|||||
| 501 | * @throws Exception\TransportException |
||||
| 502 | */ |
||||
| 503 | public function insert(string $table, array $values, array $columns = []): Statement |
||||
| 504 | 10 | { |
|||
| 505 | if (empty($values)) { |
||||
| 506 | 10 | throw QueryException::cannotInsertEmptyValues(); |
|||
| 507 | 1 | } |
|||
| 508 | |||||
| 509 | if (stripos($table, '`') === false && stripos($table, '.') === false) { |
||||
| 510 | 9 | $table = '`' . $table . '`'; //quote table name for dot names |
|||
| 511 | 6 | } |
|||
| 512 | $sql = 'INSERT INTO ' . $table; |
||||
| 513 | 9 | ||||
| 514 | if (count($columns) !== 0) { |
||||
| 515 | 9 | $sql .= ' (`' . implode('`,`', $columns) . '`) '; |
|||
| 516 | 8 | } |
|||
| 517 | |||||
| 518 | $sql .= ' VALUES '; |
||||
| 519 | 9 | ||||
| 520 | foreach ($values as $row) { |
||||
|
0 ignored issues
–
show
|
|||||
| 521 | 9 | $sql .= ' (' . FormatLine::Insert($row) . '), '; |
|||
| 522 | 9 | } |
|||
| 523 | $sql = trim($sql, ', '); |
||||
| 524 | 9 | ||||
| 525 | return $this->transport()->write($sql); |
||||
| 526 | 9 | } |
|||
| 527 | |||||
| 528 | /** |
||||
| 529 | * Prepares the values to insert from the associative array. |
||||
| 530 | * There may be one or more lines inserted, but then the keys inside the array list must match (including in the sequence) |
||||
| 531 | * |
||||
| 532 | * @param mixed[] $values - array column_name => value (if we insert one row) or array list column_name => value if we insert many lines |
||||
| 533 | * @return mixed[][] - list of arrays - 0 => fields, 1 => list of value arrays for insertion |
||||
| 534 | **/ |
||||
| 535 | public function prepareInsertAssocBulk(array $values) |
||||
| 536 | 3 | { |
|||
| 537 | if (isset($values[0]) && is_array($values[0])) { //случай, когда много строк вставляется |
||||
| 538 | 3 | $preparedFields = array_keys($values[0]); |
|||
| 539 | 2 | $preparedValues = []; |
|||
| 540 | 2 | foreach ($values as $idx => $row) { |
|||
| 541 | 2 | $_fields = array_keys($row); |
|||
| 542 | 2 | if ($_fields !== $preparedFields) { |
|||
| 543 | 2 | throw new QueryException( |
|||
| 544 | 1 | sprintf( |
|||
| 545 | 1 | 'Fields not match: %s and %s on element %s', |
|||
| 546 | 1 | implode(',', $_fields), |
|||
| 547 | 1 | implode(',', $preparedFields), |
|||
| 548 | 1 | $idx |
|||
| 549 | 1 | ) |
|||
| 550 | ); |
||||
| 551 | } |
||||
| 552 | $preparedValues[] = array_values($row); |
||||
| 553 | 2 | } |
|||
| 554 | } else { |
||||
| 555 | $preparedFields = array_keys($values); |
||||
| 556 | 1 | $preparedValues = [array_values($values)]; |
|||
| 557 | 1 | } |
|||
| 558 | |||||
| 559 | return [$preparedFields, $preparedValues]; |
||||
| 560 | 2 | } |
|||
| 561 | |||||
| 562 | /** |
||||
| 563 | * Inserts one or more rows from an associative array. |
||||
| 564 | * If there is a discrepancy between the keys of the value arrays (or their order) - throws an exception. |
||||
| 565 | * |
||||
| 566 | * @param string $tableName - name table |
||||
|
0 ignored issues
–
show
|
|||||
| 567 | * @param mixed[] $values - array column_name => value (if we insert one row) or array list column_name => value if we insert many lines |
||||
| 568 | * @return Statement |
||||
| 569 | */ |
||||
| 570 | public function insertAssocBulk(string $tableName, array $values) |
||||
| 571 | { |
||||
| 572 | list($columns, $vals) = $this->prepareInsertAssocBulk($values); |
||||
| 573 | |||||
| 574 | return $this->insert($tableName, $vals, $columns); |
||||
| 575 | } |
||||
| 576 | |||||
| 577 | /** |
||||
| 578 | * insert TabSeparated files |
||||
| 579 | * |
||||
| 580 | * @param string|string[] $fileNames |
||||
| 581 | * @param string[] $columns |
||||
| 582 | * @return mixed |
||||
| 583 | */ |
||||
| 584 | public function insertBatchTSVFiles(string $tableName, $fileNames, array $columns = []) |
||||
| 585 | 1 | { |
|||
| 586 | return $this->insertBatchFiles($tableName, $fileNames, $columns, 'TabSeparated'); |
||||
| 587 | 1 | } |
|||
| 588 | |||||
| 589 | /** |
||||
| 590 | * insert Batch Files |
||||
| 591 | * |
||||
| 592 | * @param string|string[] $fileNames |
||||
| 593 | * @param string[] $columns |
||||
|
0 ignored issues
–
show
|
|||||
| 594 | * @param string $format ['TabSeparated','TabSeparatedWithNames','CSV','CSVWithNames'] |
||||
| 595 | * @return Statement[] |
||||
| 596 | * @throws Exception\TransportException |
||||
| 597 | */ |
||||
| 598 | public function insertBatchFiles(string $tableName, $fileNames, array $columns = [], string $format = 'CSV') |
||||
| 599 | 8 | { |
|||
| 600 | if (is_string($fileNames)) { |
||||
| 601 | 8 | $fileNames = [$fileNames]; |
|||
| 602 | } |
||||
| 603 | if ($this->getCountPendingQueue() > 0) { |
||||
| 604 | 8 | throw new QueryException('Queue must be empty, before insertBatch, need executeAsync'); |
|||
| 605 | } |
||||
| 606 | |||||
| 607 | if (!in_array($format, self::SUPPORTED_FORMATS, true)) { |
||||
| 608 | 8 | throw new QueryException('Format not support in insertBatchFiles'); |
|||
| 609 | } |
||||
| 610 | |||||
| 611 | $result = []; |
||||
| 612 | 8 | ||||
| 613 | foreach ($fileNames as $fileName) { |
||||
| 614 | 8 | if (!is_file($fileName) || !is_readable($fileName)) { |
|||
| 615 | 8 | throw new QueryException('Cant read file: ' . $fileName . ' ' . (is_file($fileName) ? '' : ' is not file')); |
|||
| 616 | } |
||||
| 617 | |||||
| 618 | if (empty($columns)) { |
||||
| 619 | 8 | $sql = 'INSERT INTO ' . $tableName . ' FORMAT ' . $format; |
|||
| 620 | } else { |
||||
| 621 | $sql = 'INSERT INTO ' . $tableName . ' ( ' . implode(',', $columns) . ' ) FORMAT ' . $format; |
||||
| 622 | 8 | } |
|||
| 623 | $result[$fileName] = $this->transport()->writeAsyncCSV($sql, $fileName); |
||||
| 624 | 8 | } |
|||
| 625 | |||||
| 626 | // exec |
||||
| 627 | $this->executeAsync(); |
||||
| 628 | 8 | ||||
| 629 | // fetch resutl |
||||
| 630 | foreach ($fileNames as $fileName) { |
||||
| 631 | 8 | if (!$result[$fileName]->isError()) { |
|||
| 632 | 8 | continue; |
|||
| 633 | 6 | } |
|||
| 634 | |||||
| 635 | $result[$fileName]->error(); |
||||
| 636 | 2 | } |
|||
| 637 | |||||
| 638 | return $result; |
||||
| 639 | 6 | } |
|||
| 640 | |||||
| 641 | /** |
||||
| 642 | * insert Batch Stream |
||||
| 643 | * |
||||
| 644 | * @param string[] $columns |
||||
| 645 | * @param string $format ['TabSeparated','TabSeparatedWithNames','CSV','CSVWithNames'] |
||||
| 646 | * @return Transport\CurlerRequest |
||||
| 647 | */ |
||||
| 648 | public function insertBatchStream(string $tableName, array $columns = [], string $format = 'CSV') |
||||
| 649 | 2 | { |
|||
| 650 | if ($this->getCountPendingQueue() > 0) { |
||||
| 651 | 2 | throw new QueryException('Queue must be empty, before insertBatch, need executeAsync'); |
|||
| 652 | } |
||||
| 653 | |||||
| 654 | if (!in_array($format, self::SUPPORTED_FORMATS, true)) { |
||||
| 655 | 2 | throw new QueryException('Format not support in insertBatchFiles'); |
|||
| 656 | } |
||||
| 657 | |||||
| 658 | if (empty($columns)) { |
||||
| 659 | 2 | $sql = 'INSERT INTO ' . $tableName . ' FORMAT ' . $format; |
|||
| 660 | } else { |
||||
| 661 | $sql = 'INSERT INTO ' . $tableName . ' ( ' . implode(',', $columns) . ' ) FORMAT ' . $format; |
||||
| 662 | 2 | } |
|||
| 663 | |||||
| 664 | return $this->transport()->writeStreamData($sql); |
||||
| 665 | 2 | } |
|||
| 666 | |||||
| 667 | /** |
||||
| 668 | * stream Write |
||||
| 669 | * |
||||
| 670 | * @param string[] $bind |
||||
| 671 | * @return Statement |
||||
| 672 | * @throws Exception\TransportException |
||||
| 673 | */ |
||||
| 674 | public function streamWrite(Stream $stream, string $sql, array $bind = []) |
||||
| 675 | 1 | { |
|||
| 676 | if ($this->getCountPendingQueue() > 0) { |
||||
| 677 | 1 | throw new QueryException('Queue must be empty, before streamWrite'); |
|||
| 678 | } |
||||
| 679 | |||||
| 680 | return $this->transport()->streamWrite($stream, $sql, $bind); |
||||
| 681 | 1 | } |
|||
| 682 | |||||
| 683 | /** |
||||
| 684 | * stream Read |
||||
| 685 | * |
||||
| 686 | * @param string[] $bind |
||||
| 687 | * @return Statement |
||||
| 688 | */ |
||||
| 689 | public function streamRead(Stream $streamRead, string $sql, array $bind = []) |
||||
| 690 | 1 | { |
|||
| 691 | if ($this->getCountPendingQueue() > 0) { |
||||
| 692 | 1 | throw new QueryException('Queue must be empty, before streamRead'); |
|||
| 693 | } |
||||
| 694 | |||||
| 695 | return $this->transport()->streamRead($streamRead, $sql, $bind); |
||||
| 696 | 1 | } |
|||
| 697 | |||||
| 698 | /** |
||||
| 699 | * Size of database |
||||
| 700 | * |
||||
| 701 | * @return mixed|null |
||||
|
0 ignored issues
–
show
|
|||||
| 702 | * @throws \Exception |
||||
|
0 ignored issues
–
show
|
|||||
| 703 | */ |
||||
| 704 | public function databaseSize() |
||||
| 705 | { |
||||
| 706 | $b = $this->settings()->getDatabase(); |
||||
| 707 | |||||
| 708 | return $this->select( |
||||
| 709 | ' |
||||
| 710 | SELECT database,formatReadableSize(sum(bytes)) as size |
||||
| 711 | FROM system.parts |
||||
| 712 | WHERE active AND database=:database |
||||
| 713 | GROUP BY database |
||||
| 714 | ', |
||||
| 715 | ['database' => $b] |
||||
| 716 | )->fetchOne(); |
||||
| 717 | } |
||||
| 718 | |||||
| 719 | /** |
||||
| 720 | * Size of tables |
||||
| 721 | * |
||||
| 722 | * @return mixed |
||||
|
0 ignored issues
–
show
|
|||||
| 723 | * @throws \Exception |
||||
|
0 ignored issues
–
show
|
|||||
| 724 | */ |
||||
| 725 | public function tableSize(string $tableName) |
||||
| 726 | 1 | { |
|||
| 727 | $tables = $this->tablesSize(); |
||||
| 728 | 1 | ||||
| 729 | if (isset($tables[$tableName])) { |
||||
| 730 | 1 | return $tables[$tableName]; |
|||
| 731 | 1 | } |
|||
| 732 | |||||
| 733 | return null; |
||||
| 734 | } |
||||
| 735 | |||||
| 736 | /** |
||||
| 737 | * Ping server |
||||
| 738 | * |
||||
| 739 | * @param bool $throwException |
||||
|
0 ignored issues
–
show
|
|||||
| 740 | * @return bool |
||||
| 741 | * @throws TransportException |
||||
| 742 | 38 | */ |
|||
| 743 | public function ping(bool $throwException=false): bool |
||||
|
0 ignored issues
–
show
|
|||||
| 744 | 38 | { |
|||
| 745 | $result=$this->transport()->ping(); |
||||
|
0 ignored issues
–
show
|
|||||
| 746 | if ($throwException && !$result) throw new TransportException('Can`t ping server'); |
||||
|
0 ignored issues
–
show
|
|||||
| 747 | return $result; |
||||
|
0 ignored issues
–
show
|
|||||
| 748 | } |
||||
| 749 | |||||
| 750 | /** |
||||
| 751 | * Tables sizes |
||||
| 752 | * |
||||
| 753 | * @param bool $flatList |
||||
|
0 ignored issues
–
show
|
|||||
| 754 | 1 | * @return mixed[][] |
|||
| 755 | * @throws \Exception |
||||
|
0 ignored issues
–
show
|
|||||
| 756 | 1 | */ |
|||
| 757 | public function tablesSize($flatList = false) |
||||
| 758 | { |
||||
| 759 | $result = $this->select(' |
||||
| 760 | SELECT name as table,database, |
||||
| 761 | max(sizebytes) as sizebytes, |
||||
| 762 | max(size) as size, |
||||
| 763 | min(min_date) as min_date, |
||||
| 764 | max(max_date) as max_date |
||||
| 765 | FROM system.tables |
||||
| 766 | ANY LEFT JOIN |
||||
| 767 | ( |
||||
| 768 | SELECT table,database, |
||||
| 769 | formatReadableSize(sum(bytes)) as size, |
||||
| 770 | sum(bytes) as sizebytes, |
||||
| 771 | min(min_date) as min_date, |
||||
| 772 | max(max_date) as max_date |
||||
| 773 | FROM system.parts |
||||
| 774 | WHERE active AND database=:database |
||||
| 775 | GROUP BY table,database |
||||
| 776 | ) as s USING ( table,database ) |
||||
| 777 | 1 | WHERE database=:database |
|||
| 778 | GROUP BY table,database |
||||
| 779 | 1 | ', |
|||
| 780 | ['database' => $this->settings()->getDatabase()]); |
||||
| 781 | |||||
| 782 | if ($flatList) { |
||||
| 783 | 1 | return $result->rows(); |
|||
| 784 | } |
||||
| 785 | |||||
| 786 | return $result->rowsAsTree('table'); |
||||
| 787 | } |
||||
| 788 | |||||
| 789 | /** |
||||
| 790 | * isExists |
||||
| 791 | * |
||||
| 792 | * @return array |
||||
|
0 ignored issues
–
show
|
|||||
| 793 | * @throws \Exception |
||||
|
0 ignored issues
–
show
|
|||||
| 794 | */ |
||||
| 795 | public function isExists(string $database, string $table) |
||||
| 796 | { |
||||
| 797 | return $this->select( |
||||
| 798 | ' |
||||
| 799 | SELECT * |
||||
| 800 | FROM system.tables |
||||
| 801 | WHERE name=\'' . $table . '\' AND database=\'' . $database . '\'' |
||||
| 802 | )->rowsAsTree('name'); |
||||
| 803 | } |
||||
| 804 | |||||
| 805 | /** |
||||
| 806 | * List of partitions |
||||
| 807 | * |
||||
| 808 | * @return mixed[][] |
||||
|
0 ignored issues
–
show
|
|||||
| 809 | * @throws \Exception |
||||
|
0 ignored issues
–
show
|
|||||
| 810 | */ |
||||
| 811 | public function partitions(string $table, int $limit = null, bool $active = null) |
||||
| 812 | { |
||||
| 813 | $database = $this->settings()->getDatabase(); |
||||
| 814 | $whereActiveClause = $active === null ? '' : sprintf(' AND active = %s', (int)$active); |
||||
| 815 | $limitClause = $limit !== null ? ' LIMIT ' . $limit : ''; |
||||
| 816 | |||||
| 817 | return $this->select(<<<CLICKHOUSE |
||||
| 818 | SELECT * |
||||
| 819 | FROM system.parts |
||||
| 820 | WHERE table={tbl:String} AND database = {db:String} |
||||
| 821 | $whereActiveClause |
||||
| 822 | ORDER BY max_date $limitClause |
||||
| 823 | CLICKHOUSE, |
||||
| 824 | [ |
||||
| 825 | 'db'=>$database, |
||||
|
0 ignored issues
–
show
|
|||||
| 826 | 'tbl'=>$table |
||||
|
0 ignored issues
–
show
|
|||||
| 827 | ] |
||||
| 828 | )->rowsAsTree('name'); |
||||
| 829 | } |
||||
| 830 | |||||
| 831 | /** |
||||
| 832 | * dropPartition |
||||
| 833 | * @return Statement |
||||
|
0 ignored issues
–
show
|
|||||
| 834 | * @deprecated |
||||
| 835 | */ |
||||
| 836 | public function dropPartition(string $dataBaseTableName, string $partition_id) |
||||
| 837 | { |
||||
|
0 ignored issues
–
show
|
|||||
| 838 | |||||
| 839 | $partition_id = trim($partition_id, '\''); |
||||
| 840 | $this->settings()->set('replication_alter_partitions_sync', 2); |
||||
| 841 | $state = $this->write('ALTER TABLE {dataBaseTableName} DROP PARTITION :partion_id', |
||||
| 842 | [ |
||||
| 843 | 'dataBaseTableName' => $dataBaseTableName, |
||||
| 844 | 'partion_id' => $partition_id, |
||||
| 845 | ]); |
||||
| 846 | |||||
| 847 | return $state; |
||||
| 848 | } |
||||
| 849 | |||||
| 850 | /** |
||||
| 851 | * Truncate ( drop all partitions ) |
||||
| 852 | * @return array |
||||
|
0 ignored issues
–
show
|
|||||
| 853 | * @throws \Exception |
||||
|
0 ignored issues
–
show
|
|||||
| 854 | * @deprecated |
||||
| 855 | */ |
||||
| 856 | public function truncateTable(string $tableName) |
||||
| 857 | { |
||||
| 858 | $partions = $this->partitions($tableName); |
||||
| 859 | $out = []; |
||||
| 860 | foreach ($partions as $part_key => $part) { |
||||
| 861 | $part_id = $part['partition']; |
||||
| 862 | $out[$part_id] = $this->dropPartition($tableName, $part_id); |
||||
| 863 | } |
||||
| 864 | |||||
| 865 | return $out; |
||||
| 866 | } |
||||
| 867 | 1 | ||||
| 868 | /** |
||||
| 869 | 1 | * Returns the server's uptime in seconds. |
|||
| 870 | * |
||||
| 871 | * @return int |
||||
| 872 | * @throws Exception\TransportException |
||||
| 873 | * @throws \Exception |
||||
|
0 ignored issues
–
show
|
|||||
| 874 | */ |
||||
| 875 | 1 | public function getServerUptime() |
|||
|
0 ignored issues
–
show
|
|||||
| 876 | { |
||||
| 877 | 1 | return $this->select('SELECT uptime() as uptime')->fetchOne('uptime'); |
|||
| 878 | } |
||||
| 879 | |||||
| 880 | /** |
||||
| 881 | * Returns string with the server version. |
||||
| 882 | */ |
||||
| 883 | public function getServerVersion(): string |
||||
| 884 | { |
||||
| 885 | return (string)$this->select('SELECT version() as version')->fetchOne('version'); |
||||
| 886 | 1 | } |
|||
| 887 | |||||
| 888 | 1 | /** |
|||
| 889 | 1 | * Read system.settings table |
|||
| 890 | 1 | * |
|||
| 891 | 1 | * @return mixed[][] |
|||
| 892 | 1 | * @throws \Exception |
|||
|
0 ignored issues
–
show
|
|||||
| 893 | 1 | */ |
|||
| 894 | 1 | public function getServerSystemSettings(string $like = '') |
|||
| 895 | 1 | { |
|||
| 896 | $l = []; |
||||
| 897 | $list = $this->select('SELECT * FROM system.settings' . ($like ? ' WHERE name LIKE :like' : ''), |
||||
| 898 | ['like' => '%' . $like . '%'])->rows(); |
||||
| 899 | 1 | foreach ($list as $row) { |
|||
| 900 | if (isset($row['name'])) { |
||||
| 901 | $n = $row['name']; |
||||
| 902 | unset($row['name']); |
||||
| 903 | $l[$n] = $row; |
||||
| 904 | } |
||||
| 905 | } |
||||
| 906 | |||||
| 907 | return $l; |
||||
| 908 | } |
||||
|
0 ignored issues
–
show
|
|||||
| 909 | |||||
| 910 | } |
||||
| 911 |