Big 💥

simPod · Jan 19, 2020 · bef2f9f · bef2f9f
1 parent 9868436
commit bef2f9f
Show file tree

Hide file tree

Showing 32 changed files with 866 additions and 177 deletions.
diff --git a/README.md b/README.md
@@ -2,9 +2,9 @@
 
 [![Build Status](https://github.com/simPod/PhpClickHouseClient/workflows/Continuous%20Integration/badge.svg?branch=master)](https://github.com/simPod/PhpClickHouseClient/actions)
 [![Coverage Status](https://coveralls.io/repos/github/simPod/PhpClickHouseClient/badge.svg?branch=master)](https://coveralls.io/github/simPod/PhpClickHouseClient?branch=master)
-[![Downloads](https://poser.pugx.org/simpod/php-clickhouse-client/d/total.svg)](https://packagist.org/packages/simpod/php-clickhouse-client)
-[![Packagist](https://poser.pugx.org/simpod/php-clickhouse-client/v/stable.svg)](https://packagist.org/packages/simpod/php-clickhouse-client)
-[![Licence](https://poser.pugx.org/simpod/php-clickhouse-client/license.svg)](https://packagist.org/packages/simpod/php-clickhouse-client)
+[![Downloads](https://poser.pugx.org/simpod/clickhouse-client/d/total.svg)](https://packagist.org/packages/simpod/clickhouse-client)
+[![Packagist](https://poser.pugx.org/simpod/clickhouse-client/v/stable.svg)](https://packagist.org/packages/simpod/clickhouse-client)
+[![Licence](https://poser.pugx.org/simpod/clickhouse-client/license.svg)](https://packagist.org/packages/simpod/clickhouse-client)
 [![GitHub Issues](https://img.shields.io/github/issues/simPod/PhpClickHouseClient.svg?style=flat-square)](https://github.com/simPod/PhpClickHouseClient/issues)
 
 PHP Client that talks to ClickHouse via HTTP layer

diff --git a/composer.json b/composer.json
@@ -25,6 +25,7 @@
     },
     "require": {
         "php-64bit": "^7.3",
+        "guzzlehttp/promises": "^1.3",
         "php-ds/php-ds": "^1.2",
         "psr/http-client": "^1.0",
         "psr/http-factory": "^1.0",

diff --git a/src/ClickHouseClient.php b/src/ClickHouseClient.php
diff --git a/src/Client/ClickHouseAsyncClient.php b/src/Client/ClickHouseAsyncClient.php
@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SimPod\ClickHouseClient\Client;
+
+use GuzzleHttp\Promise\PromiseInterface;
+use SimPod\ClickHouseClient\Format\Format;
+use SimPod\ClickHouseClient\Output\Output;
+
+interface ClickHouseAsyncClient
+{
+    /**
+     * @see Output hack for IDe to preserve `use`
+     *
+     * @phpstan-template O of Output
+     * @phpstan-param    Format<O> $outputFormat
+     */
+    public function select(string $sql, Format $outputFormat) : PromiseInterface;
+}
diff --git a/src/Client/ClickHouseClient.php b/src/Client/ClickHouseClient.php
@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SimPod\ClickHouseClient\Client;
+
+use Ds\Set;
+use Ds\Vector;
+use SimPod\ClickHouseClient\Format\Format;
+use SimPod\ClickHouseClient\Output\Output;
+
+interface ClickHouseClient
+{
+    public function executeQuery(string $sql) : void;
+
+    /**
+     * @phpstan-template O of Output
+     * @phpstan-param    Format<O> $outputFormat
+     * @phpstan-return   O
+     */
+    public function select(string $sql, Format $outputFormat) : Output;
+
+    /**
+     * @param Vector<mixed>    $values
+     * @param Set<string>|null $columns
+     */
+    public function insert(string $table, Vector $values, ?Set $columns = null) : void;
+
+    /**
+     * @phpstan-template O of Output
+     * @phpstan-param    Format<O> $inputFormat
+     */
+    public function insertWithFormat(string $table, Format $inputFormat, string $data) : void;
+}
diff --git a/src/Client/HttpClickHouseAsyncClient.php b/src/Client/HttpClickHouseAsyncClient.php
@@ -0,0 +1,93 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SimPod\ClickHouseClient\Client;
+
+use GuzzleHttp\Promise\PromiseInterface;
+use Http\Client\HttpAsyncClient;
+use Psr\Http\Message\ResponseInterface;
+use SimPod\ClickHouseClient\Exception\ServerError;
+use SimPod\ClickHouseClient\Format\Format;
+use SimPod\ClickHouseClient\Http\RequestFactory;
+use SimPod\ClickHouseClient\Http\RequestOptions;
+use SimPod\ClickHouseClient\Output\Output;
+use function GuzzleHttp\Promise\promise_for;
+
+class HttpClickHouseAsyncClient implements ClickHouseAsyncClient
+{
+    /** @var HttpAsyncClient */
+    private $asyncClient;
+
+    /** @var RequestFactory */
+    private $requestFactory;
+
+    /** @var string */
+    private $endpoint;
+
+    /** @var array<string, int|string> */
+    private $defaultParameters;
+
+    /**
+     * @param array<string, int|string> $defaultParameters
+     */
+    public function __construct(
+        HttpAsyncClient $asyncClient,
+        RequestFactory $requestFactory,
+        string $endpoint,
+        array $defaultParameters = []
+    ) {
+        $this->asyncClient       = $asyncClient;
+        $this->requestFactory    = $requestFactory;
+        $this->endpoint          = $endpoint;
+        $this->defaultParameters = $defaultParameters;
+    }
+
+    public function select(string $sql, Format $outputFormat) : PromiseInterface
+    {
+        $formatClause = $outputFormat::toSql();
+
+        return $this->executeRequest(
+            <<<CLICKHOUSE
+$sql
+$formatClause
+CLICKHOUSE,
+            [],
+            static function (ResponseInterface $response) use ($outputFormat) : Output {
+                return $outputFormat::output((string) $response->getBody());
+            }
+        );
+    }
+
+    /**
+     * @param array<string, int|string>              $requestParameters
+     * @param callable(ResponseInterface):mixed|null $processResponse
+     */
+    private function executeRequest(string $sql, array $requestParameters = [], ?callable $processResponse = null) : PromiseInterface
+    {
+        $request = $this->requestFactory->prepareRequest(
+            $this->endpoint,
+            new RequestOptions(
+                $sql,
+                $this->defaultParameters,
+                $requestParameters
+            )
+        );
+
+        $promise = promise_for($this->asyncClient->sendAsyncRequest($request));
+
+        return $promise->then(
+            static function (ResponseInterface $response) use ($processResponse) {
+                if ($response->getStatusCode() !== 200) {
+                    throw ServerError::fromResponse($response);
+                }
+
+                if ($processResponse === null) {
+                    return $response;
+                }
+
+                return $processResponse($response);
+            }
+        );
+    }
+}
diff --git a/src/Client/HttpClickHouseClient.php b/src/Client/HttpClickHouseClient.php
@@ -0,0 +1,141 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SimPod\ClickHouseClient\Client;
+
+use Ds\Set;
+use Ds\Vector;
+use Psr\Http\Client\ClientInterface;
+use Psr\Http\Message\ResponseInterface;
+use SimPod\ClickHouseClient\Exception\CannotInsert;
+use SimPod\ClickHouseClient\Exception\ServerError;
+use SimPod\ClickHouseClient\Format\Format;
+use SimPod\ClickHouseClient\Http\RequestFactory;
+use SimPod\ClickHouseClient\Http\RequestOptions;
+use SimPod\ClickHouseClient\Output\Output;
+use function array_keys;
+use function implode;
+use function Safe\sprintf;
+
+class HttpClickHouseClient implements ClickHouseClient
+{
+    /** @var ClientInterface */
+    private $client;
+
+    /** @var RequestFactory */
+    private $requestFactory;
+
+    /** @var string */
+    private $endpoint;
+
+    /** @var array<string, int|string> */
+    private $defaultParameters;
+
+    /**
+     * @param array<string, int|string> $defaultParameters
+     */
+    public function __construct(
+        ClientInterface $client,
+        RequestFactory $requestFactory,
+        string $endpoint,
+        array $defaultParameters = []
+    ) {
+        $this->client            = $client;
+        $this->requestFactory    = $requestFactory;
+        $this->endpoint          = $endpoint;
+        $this->defaultParameters = $defaultParameters;
+    }
+
+    public function executeQuery(string $sql) : void
+    {
+        $response = $this->executeRequest($sql);
+
+        $contents = (string) $response->getBody();
+
+        return;
+    }
+
+    public function select(string $sql, Format $outputFormat) : Output
+    {
+        $formatClause = $outputFormat::toSql();
+
+        $response = $this->executeRequest(
+            <<<CLICKHOUSE
+$sql
+$formatClause
+CLICKHOUSE
+        );
+
+        return $outputFormat::output((string) $response->getBody());
+    }
+
+    public function insert(string $table, Vector $values, ?Set $columns = null) : void
+    {
+        if ($values->isEmpty()) {
+            throw CannotInsert::noValues();
+        }
+
+        if ($columns === null) {
+            /** @var array<mixed> $row */
+            $row         = $values->first();
+            $columnNames = array_keys($row);
+        } else {
+            $columnNames = $columns->toArray();
+        }
+
+        $columnsSql = implode(',', $columnNames);
+
+        $valuesSql = implode(
+            ',',
+            $values->map(
+                static function (array $map) : string {
+                    return sprintf(
+                        '(%s)',
+                        implode(',', $map)
+                    );
+                }
+            )->toArray()
+        );
+
+        $response = $this->executeRequest(
+            <<<CLICKHOUSE
+INSERT INTO $table
+($columnsSql)
+VALUES $valuesSql
+CLICKHOUSE
+        );
+    }
+
+    public function insertWithFormat(string $table, Format $inputFormat, string $data) : void
+    {
+        $formatSql = $inputFormat::toSql();
+
+        $this->executeRequest(
+            <<<CLICKHOUSE
+INSERT INTO $table $formatSql $data
+CLICKHOUSE
+        );
+    }
+
+    /**
+     * @param array<string, int|string> $requestParameters
+     */
+    private function executeRequest(string $sql, array $requestParameters = []) : ResponseInterface
+    {
+        $request  = $this->requestFactory->prepareRequest(
+            $this->endpoint,
+            new RequestOptions(
+                $sql,
+                $this->defaultParameters,
+                $requestParameters
+            )
+        );
+        $response = $this->client->sendRequest($request);
+        if ($response->getStatusCode() !== 200) {
+            throw ServerError::fromResponse($response);
+        }
+
+        return $response;
+    }
+}
diff --git a/src/Exception/CannotInsert.php b/src/Exception/CannotInsert.php
@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SimPod\ClickHouseClient\Exception;
+
+use Exception;
+
+final class CannotInsert extends Exception implements ClickHouseClientException
+{
+    public static function noValues() : self
+    {
+        return new self();
+    }
+}
diff --git a/src/Exception/ClickHouseClientException.php b/src/Exception/ClickHouseClientException.php
@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SimPod\ClickHouseClient\Exception;
+
+interface ClickHouseClientException
+{
+}
diff --git a/src/Exception/ServerException.php → src/Exception/ServerError.php b/src/Exception/ServerException.php → src/Exception/ServerError.php
@@ -7,7 +7,7 @@
 use Exception;
 use Psr\Http\Message\ResponseInterface;
 
-final class ServerException extends Exception
+final class ServerError extends Exception
 {
     public static function fromResponse(ResponseInterface $response) : self
     {

diff --git a/src/Exception/UnsupportedValueType.php b/src/Exception/UnsupportedValueType.php
@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SimPod\ClickHouseClient\Exception;
+
+use InvalidArgumentException;
+use function gettype;
+use function Safe\sprintf;
+
+final class UnsupportedValueType extends InvalidArgumentException implements ClickHouseClientException
+{
+    /**
+     * @param mixed $value
+     */
+    public static function value($value) : self
+    {
+        return new self(sprintf('Value of type "%s" is not supported as a parameter', gettype($value)));
+    }
+}