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