feat: add Client::insertPayload()

simPod · Jan 15, 2025 · c464ebc · c464ebc
1 parent 6f69f20
commit c464ebc
Show file tree

Hide file tree

Showing 12 changed files with 230 additions and 60 deletions.
diff --git a/composer.json b/composer.json
@@ -39,6 +39,7 @@
     "require-dev": {
         "cdn77/coding-standard": "^7.0",
         "infection/infection": "^0.29.0",
+        "kafkiansky/phpclick": "dev-master",
         "nyholm/psr7": "^1.2",
         "php-http/message-factory": "^1.1",
         "phpstan/extension-installer": "^1.1",

diff --git a/src/Client/ClickHouseClient.php b/src/Client/ClickHouseClient.php
@@ -5,6 +5,7 @@
 namespace SimPod\ClickHouseClient\Client;
 
 use Psr\Http\Client\ClientExceptionInterface;
+use Psr\Http\Message\StreamInterface;
 use SimPod\ClickHouseClient\Exception\CannotInsert;
 use SimPod\ClickHouseClient\Exception\ServerError;
 use SimPod\ClickHouseClient\Exception\UnsupportedParamType;
@@ -85,4 +86,20 @@ public function insert(string $table, array $values, array|null $columns = null,
      * @template O of Output
      */
     public function insertWithFormat(string $table, Format $inputFormat, string $data, array $settings = []): void;
+
+    /**
+     * @param array<string, float|int|string> $settings
+     * @param list<string> $columns
+     * @param Format<Output<mixed>> $inputFormat
+     *
+     * @throws ClientExceptionInterface
+     * @throws ServerError
+     */
+    public function insertPayload(
+        string $table,
+        Format $inputFormat,
+        StreamInterface $payload,
+        array $columns = [],
+        array $settings = [],
+    ): void;
 }
diff --git a/src/Client/Http/RequestFactory.php b/src/Client/Http/RequestFactory.php
@@ -49,11 +49,13 @@ public function __construct(
         $this->uri = $uri;
     }
 
-    /** @throws UnsupportedParamType */
-    public function prepareRequest(RequestOptions $requestOptions): RequestInterface
-    {
+    /** @param array<string, mixed> $additionalOptions */
+    public function initRequest(
+        RequestSettings $requestSettings,
+        array $additionalOptions = [],
+    ): RequestInterface {
         $query = http_build_query(
-            $requestOptions->settings,
+            $requestSettings->settings + $additionalOptions,
             '',
             '&',
             PHP_QUERY_RFC3986,
@@ -70,11 +72,20 @@ public function prepareRequest(RequestOptions $requestOptions): RequestInterface
             }
         }
 
-        $request = $this->requestFactory->createRequest('POST', $uri);
+        return $this->requestFactory->createRequest('POST', $uri);
+    }
 
-        preg_match_all('~\{([a-zA-Z\d]+):([a-zA-Z\d ]+(\(.+\))?)}~', $requestOptions->sql, $matches);
+    /** @throws UnsupportedParamType */
+    public function prepareSqlRequest(
+        string $sql,
+        RequestSettings $requestSettings,
+        RequestOptions $requestOptions,
+    ): RequestInterface {
+        $request = $this->initRequest($requestSettings);
+
+        preg_match_all('~\{([a-zA-Z\d]+):([a-zA-Z\d ]+(\(.+\))?)}~', $sql, $matches);
         if ($matches[0] === []) {
-            $body = $this->streamFactory->createStream($requestOptions->sql);
+            $body = $this->streamFactory->createStream($sql);
             try {
                 return $request->withBody($body);
             } catch (InvalidArgumentException) {
@@ -93,7 +104,7 @@ static function (array $acc, string|int $k) use ($matches) {
             [],
         );
 
-        $streamElements = [['name' => 'query', 'contents' => $requestOptions->sql]];
+        $streamElements = [['name' => 'query', 'contents' => $sql]];
         foreach ($requestOptions->params as $name => $value) {
             $type = $paramToType[$name] ?? null;
             if ($type === null) {

diff --git a/src/Client/Http/RequestOptions.php b/src/Client/Http/RequestOptions.php
@@ -6,20 +6,9 @@
 
 final class RequestOptions
 {
-    /** @var array<string, float|int|string> */
-    public array $settings;
-
-    /**
-     * @param array<string, mixed> $params
-     * @param array<string, float|int|string> $defaultSettings
-     * @param array<string, float|int|string> $querySettings
-     */
+    /** @param array<string, mixed> $params */
     public function __construct(
-        public string $sql,
         public array $params,
-        array $defaultSettings,
-        array $querySettings,
     ) {
-        $this->settings = $querySettings + $defaultSettings;
     }
 }
diff --git a/src/Client/Http/RequestSettings.php b/src/Client/Http/RequestSettings.php
@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SimPod\ClickHouseClient\Client\Http;
+
+final class RequestSettings
+{
+    /** @var array<string, float|int|string> */
+    public array $settings;
+
+    /**
+     * @param array<string, float|int|string> $defaultSettings
+     * @param array<string, float|int|string> $querySettings
+     */
+    public function __construct(
+        array $defaultSettings,
+        array $querySettings,
+    ) {
+        $this->settings = $querySettings + $defaultSettings;
+    }
+}
diff --git a/src/Client/PsrClickHouseAsyncClient.php b/src/Client/PsrClickHouseAsyncClient.php
@@ -12,6 +12,7 @@
 use Psr\Http\Message\ResponseInterface;
 use SimPod\ClickHouseClient\Client\Http\RequestFactory;
 use SimPod\ClickHouseClient\Client\Http\RequestOptions;
+use SimPod\ClickHouseClient\Client\Http\RequestSettings;
 use SimPod\ClickHouseClient\Exception\ServerError;
 use SimPod\ClickHouseClient\Format\Format;
 use SimPod\ClickHouseClient\Output\Output;
@@ -83,13 +84,15 @@ private function executeRequest(
         array $settings = [],
         callable|null $processResponse = null,
     ): PromiseInterface {
-        $request = $this->requestFactory->prepareRequest(
-            new RequestOptions(
-                $sql,
-                $params,
+        $request = $this->requestFactory->prepareSqlRequest(
+            $sql,
+            new RequestSettings(
                 $this->defaultSettings,
                 $settings,
             ),
+            new RequestOptions(
+                $params,
+            ),
         );
 
         return Create::promiseFor(

diff --git a/src/Client/PsrClickHouseClient.php b/src/Client/PsrClickHouseClient.php
@@ -5,11 +5,14 @@
 namespace SimPod\ClickHouseClient\Client;
 
 use DateTimeZone;
+use InvalidArgumentException;
 use Psr\Http\Client\ClientExceptionInterface;
 use Psr\Http\Client\ClientInterface;
 use Psr\Http\Message\ResponseInterface;
+use Psr\Http\Message\StreamInterface;
 use SimPod\ClickHouseClient\Client\Http\RequestFactory;
 use SimPod\ClickHouseClient\Client\Http\RequestOptions;
+use SimPod\ClickHouseClient\Client\Http\RequestSettings;
 use SimPod\ClickHouseClient\Exception\CannotInsert;
 use SimPod\ClickHouseClient\Exception\ServerError;
 use SimPod\ClickHouseClient\Exception\UnsupportedParamType;
@@ -198,6 +201,46 @@ public function insertWithFormat(string $table, Format $inputFormat, string $dat
         }
     }
 
+    public function insertPayload(
+        string $table,
+        Format $inputFormat,
+        StreamInterface $payload,
+        array $columns = [],
+        array $settings = [],
+    ): void {
+        $formatSql = $inputFormat::toSql();
+
+        $table = Escaper::quoteIdentifier($table);
+
+        $columnsSql = $columns === [] ? '' : sprintf('(%s)', implode(',', $columns));
+
+        $sql = <<<CLICKHOUSE
+        INSERT INTO $table $columnsSql $formatSql
+        CLICKHOUSE;
+
+        $request = $this->requestFactory->initRequest(
+            new RequestSettings(
+                $this->defaultSettings,
+                $settings,
+            ),
+            ['query' => $sql],
+        );
+
+        try {
+            $request = $request->withBody($payload);
+        } catch (InvalidArgumentException) {
+            absurd();
+        }
+
+        $response = $this->client->sendRequest($request);
+
+        if ($response->getStatusCode() !== 200) {
+            throw ServerError::fromResponse($response);
+        }
+
+        return;
+    }
+
     /**
      * @param array<string, mixed> $params
      * @param array<string, float|int|string> $settings
@@ -208,13 +251,15 @@ public function insertWithFormat(string $table, Format $inputFormat, string $dat
      */
     private function executeRequest(string $sql, array $params, array $settings): ResponseInterface
     {
-        $request = $this->requestFactory->prepareRequest(
-            new RequestOptions(
-                $sql,
-                $params,
+        $request = $this->requestFactory->prepareSqlRequest(
+            $sql,
+            new RequestSettings(
                 $this->defaultSettings,
                 $settings,
             ),
+            new RequestOptions(
+                $params,
+            ),
         );
 
         $response = $this->client->sendRequest($request);

diff --git a/src/Format/RowBinary.php b/src/Format/RowBinary.php
@@ -0,0 +1,28 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SimPod\ClickHouseClient\Format;
+
+use SimPod\ClickHouseClient\Output\Basic;
+use SimPod\ClickHouseClient\Output\Output;
+
+/**
+ * @template T
+ * @implements Format<Basic<T>>
+ */
+final class RowBinary implements Format
+{
+    public static function output(string $contents): Output
+    {
+        /** @var Basic<T> $output */
+        $output = new Basic($contents);
+
+        return $output;
+    }
+
+    public static function toSql(): string
+    {
+        return 'FORMAT RowBinary';
+    }
+}
diff --git a/tests/Client/Http/RequestFactoryTest.php b/tests/Client/Http/RequestFactoryTest.php
@@ -10,6 +10,7 @@
 use PHPUnit\Framework\Attributes\DataProvider;
 use SimPod\ClickHouseClient\Client\Http\RequestFactory;
 use SimPod\ClickHouseClient\Client\Http\RequestOptions;
+use SimPod\ClickHouseClient\Client\Http\RequestSettings;
 use SimPod\ClickHouseClient\Param\ParamValueConverterRegistry;
 use SimPod\ClickHouseClient\Tests\TestCaseBase;
 
@@ -28,12 +29,16 @@ public function testPrepareRequest(string $uri, string $expectedUri): void
             $uri,
         );
 
-        $request = $requestFactory->prepareRequest(new RequestOptions(
+        $request = $requestFactory->prepareSqlRequest(
             'SELECT 1',
-            [],
-            ['max_block_size' => 1],
-            ['database' => 'database'],
-        ));
+            new RequestSettings(
+                ['max_block_size' => 1],
+                ['database' => 'database'],
+            ),
+            new RequestOptions(
+                [],
+            ),
+        );
 
         self::assertSame('POST', $request->getMethod());
         self::assertSame(

diff --git a/tests/Client/Http/RequestOptionsTest.php b/tests/Client/Http/RequestOptionsTest.php
diff --git a/tests/Client/Http/RequestSettingsTest.php b/tests/Client/Http/RequestSettingsTest.php
@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SimPod\ClickHouseClient\Tests\Client\Http;
+
+use PHPUnit\Framework\Attributes\CoversClass;
+use SimPod\ClickHouseClient\Client\Http\RequestSettings;
+use SimPod\ClickHouseClient\Tests\TestCaseBase;
+
+#[CoversClass(RequestSettings::class)]
+final class RequestSettingsTest extends TestCaseBase
+{
+    public function testMergeSettings(): void
+    {
+        $requestSettings = new RequestSettings(
+            ['database' => 'foo', 'a' => 1],
+            ['database' => 'bar', 'b' => 2],
+        );
+
+        self::assertSame('bar', $requestSettings->settings['database']);
+        self::assertSame(1, $requestSettings->settings['a']);
+        self::assertSame(2, $requestSettings->settings['b']);
+    }
+}