refactor(clickhouse): use Builder bulkInsert for addBatch, drop manual JSONEachRow body assembly

addBatch() now hands its built rows directly to the typed
bulkInsert(Format::JSONEachRow, ...) entry point on the ClickHouse builder
and ships the eager $statement->body to the HTTP layer, replacing the
hand-rolled array_map(json_encode, ...) + implode("\n", ...) assembly.
The runtime instanceof guard on FormattedInsertStatement is gone too -
bulkInsert() returns the typed statement by signature.

insert() now takes the serialized body string instead of an array of
pre-encoded rows; the only caller is addBatch().

createDailyMaterializedView() picks up the createMaterializedView()
argument-order change in the new query branch (name, body, targetTable,
ifNotExists). The snapshot test for the MV path is updated to match.

New snapshots:
  - testAddBatchEmitsBulkInsertQueryAndBody asserts the envelope query
    and the serialized JSONEachRow body for a two-row fixture.
  - testNestedColumnDotQuoting validates that columns containing a dot
    (ClickHouse nested-array convention) remain single-backtick-wrapped
    atomic identifiers, exercising the QuotesIdentifiers::quoteLiteral()
    fix shipped in utopia-php/query#13.

Author: lohanidamodar

src/Usage/Adapter/ClickHouse.php

@@ -8,7 +8,7 @@
 use InvalidArgumentException;
 use Utopia\Fetch\Client;
 use Utopia\Query\Builder\ClickHouse as ClickHouseBuilder;
-use Utopia\Query\Builder\ClickHouse\FormattedInsertStatement;
+use Utopia\Query\Builder\ClickHouse\Format;
 use Utopia\Query\Method;
 use Utopia\Query\Query;
 use Utopia\Query\Schema\ClickHouse as ClickHouseSchema;
@@ -27,7 +27,7 @@
  * the utopia-php/query 0.3.x builder / schema layer:
  *
  * - DDL (setup, daily table, materialized view) to Utopia\Query\Schema\ClickHouse
- * - INSERT (addBatch) to Utopia\Query\Builder\ClickHouse::insertFormat
+ * - INSERT (addBatch) to Utopia\Query\Builder\ClickHouse::bulkInsert
  * - SELECT (find, count, sum, daily, totals, time series) to Utopia\Query\Builder\ClickHouse
  *   with `useNamedBindings()` for `{paramN:Type}` placeholders
  * - DELETE (purge) to Utopia\Query\Builder\ClickHouse::delete
@@ -719,19 +719,20 @@ function (Exception $e, int $attempt) use ($sql): Exception {
     }
 
     /**
-     * Stream a list of JSON rows into a `INSERT ... FORMAT JSONEachRow` statement.
+     * Ship a serialized `INSERT ... FORMAT <name>` body to ClickHouse's HTTP
+     * interface. The body must already be formatted to match the FORMAT
+     * envelope declared in $sql (the builder's `bulkInsert()` emits both).
      *
-     * @param array<string> $data Pre-encoded JSON rows
      * @throws Exception
      */
-    private function insert(string $sql, array $data): void
+    private function insert(string $sql, string $body): void
     {
-        if (empty($data)) {
+        if ($body === '') {
             return;
         }
 
         $this->executeWithRetry(
-            function (int $attempt) use ($sql, $data): void {
+            function (int $attempt) use ($sql, $body): void {
                 $startTime = microtime(true);
                 $scheme = $this->secure ? 'https' : 'http';
 
@@ -759,9 +760,9 @@ function (int $attempt) use ($sql, $data): void {
                     $this->requestCount++;
                 }
 
-                $body = implode("\n", $data);
+                $rowCount = \substr_count($body, "\n") + 1;
 
-                $params = ['rows' => count($data), 'bytes' => strlen($body)];
+                $params = ['rows' => $rowCount, 'bytes' => strlen($body)];
 
                 try {
                     $response = $this->client->fetch(
@@ -776,7 +777,6 @@ function (int $attempt) use ($sql, $data): void {
                         $bodyStr = $response->getBody();
                         $bodyStr = is_string($bodyStr) ? $bodyStr : '';
                         $duration = microtime(true) - $startTime;
-                        $rowCount = count($data);
                         $baseError = "ClickHouse insert failed with HTTP {$httpCode}: {$bodyStr}";
                         $errorMsg = $this->buildErrorMessage($baseError, null, $sql . " ({$rowCount} rows)");
                         $this->logQuery($sql, $params, $duration, false, $errorMsg, $attempt);
@@ -793,15 +793,15 @@ function (int $attempt) use ($sql, $data): void {
             function (Exception $e, ?int $httpCode): bool {
                 return false;
             },
-            function (Exception $e, int $attempt) use ($sql, $data): Exception {
+            function (Exception $e, int $attempt) use ($sql, $body): Exception {
                 $cleanMessage = preg_replace('/\|HTTP_CODE:\d+$/', '', $e->getMessage());
                 $cleanMessage = is_string($cleanMessage) ? $cleanMessage : $e->getMessage();
 
                 if (strpos($cleanMessage, '[Operation:') !== false) {
                     return new Exception($cleanMessage, 0, $e);
                 }
 
-                $rowCount = count($data);
+                $rowCount = \substr_count($body, "\n") + 1;
                 $baseError = "ClickHouse insert execution failed after " . ($attempt + 1) . " attempt(s): {$cleanMessage}";
                 $errorMsg = $this->buildErrorMessage($baseError, null, $sql . " ({$rowCount} rows)");
                 return new Exception($errorMsg, 0, $e);
@@ -1139,8 +1139,8 @@ private function createDailyMaterializedView(): void
 
         $statement = $this->newSchema()->createMaterializedView(
             $this->getEventsDailyMvName(),
-            $this->getEventsDailyTableName(),
             $body,
+            $this->getEventsDailyTableName(),
             true,
         );
 
@@ -1330,17 +1330,6 @@ public function addBatch(array $metrics, string $type, int $batchSize = self::IN
         $tableName = $this->getTableForType($type);
         $columns = $this->getInsertColumns($type);
 
-        $statement = $this->newBuilder($type)
-            ->into($tableName)
-            ->insertFormat('JSONEachRow', $columns)
-            ->insert();
-
-        if (!$statement instanceof FormattedInsertStatement) {
-            throw new Exception('Expected FormattedInsertStatement from builder insertFormat()');
-        }
-
-        $sql = $this->qualifyDdl($statement->query, $tableName);
-
         foreach (\array_chunk($metrics, $batchSize) as $metricsBatch) {
             $rows = [];
 
@@ -1392,14 +1381,16 @@ public function addBatch(array $metrics, string $type, int $batchSize = self::IN
                     $row['tenant'] = $tenant;
                 }
 
-                $encoded = json_encode($row);
-                if ($encoded === false) {
-                    throw new Exception("Failed to JSON encode metric row: " . json_last_error_msg());
-                }
-                $rows[] = $encoded;
+                $rows[] = $row;
             }
 
-            $this->insert($sql, $rows);
+            $statement = $this->newBuilder($type)
+                ->into($tableName)
+                ->bulkInsert(Format::JSONEachRow, $rows, $columns);
+
+            $sql = $this->qualifyDdl($statement->query, $tableName);
+
+            $this->insert($sql, $statement->body ?? '');
         }
 
         return true;

tests/Usage/Adapter/ClickHouseSqlSnapshotTest.php

@@ -4,6 +4,7 @@
 
 use PHPUnit\Framework\TestCase;
 use Utopia\Query\Builder\ClickHouse as ClickHouseBuilder;
+use Utopia\Query\Builder\ClickHouse\Format;
 use Utopia\Query\Builder\ClickHouse\FormattedInsertStatement;
 use Utopia\Query\Query;
 use Utopia\Query\Schema\ClickHouse as ClickHouseSchema;
@@ -111,8 +112,8 @@ public function testDailyMaterializedViewDdl(): void
 
         $statement = (new ClickHouseSchema())->createMaterializedView(
             'usage_events_daily_mv',
-            'usage_events_daily',
             $body,
+            'usage_events_daily',
             true,
         );
 
@@ -197,14 +198,32 @@ public function testFindAggregatedWithGroupByTimeBucket(): void
     }
 
     /**
-     * Snapshot for `addBatch()` INSERT … FORMAT JSONEachRow.
+     * Snapshot for `addBatch()`: `bulkInsert(Format::JSONEachRow, ...)`
+     * emits both the INSERT envelope and the serialized JSONEachRow body
+     * in one typed call.
      */
-    public function testAddBatchInsertFormat(): void
+    public function testAddBatchEmitsBulkInsertQueryAndBody(): void
     {
+        $rows = [
+            [
+                'id' => 'row1',
+                'metric' => 'requests',
+                'value' => 7,
+                'time' => '2026-03-01 00:00:00',
+                'tags' => ['region' => 'us-east'],
+            ],
+            [
+                'id' => 'row2',
+                'metric' => 'requests',
+                'value' => 3,
+                'time' => '2026-03-01 00:01:00',
+                'tags' => ['region' => 'eu-west'],
+            ],
+        ];
+
         $statement = (new ClickHouseBuilder())
             ->into('events')
-            ->insertFormat('JSONEachRow', ['id', 'metric', 'value', 'time', 'tags'])
-            ->insert();
+            ->bulkInsert(Format::JSONEachRow, $rows, ['id', 'metric', 'value', 'time', 'tags']);
 
         $this->assertInstanceOf(FormattedInsertStatement::class, $statement);
         $this->assertSame(
@@ -214,6 +233,37 @@ public function testAddBatchInsertFormat(): void
         $this->assertSame([], $statement->bindings);
         $this->assertSame('JSONEachRow', $statement->format);
         $this->assertSame(['id', 'metric', 'value', 'time', 'tags'], $statement->columns);
+        $this->assertSame(
+            '{"id":"row1","metric":"requests","value":7,"time":"2026-03-01 00:00:00","tags":{"region":"us-east"}}' . "\n"
+            . '{"id":"row2","metric":"requests","value":3,"time":"2026-03-01 00:01:00","tags":{"region":"eu-west"}}',
+            $statement->body,
+        );
+        $this->assertStringEndsNotWith("\n", (string) $statement->body);
+    }
+
+    /**
+     * Snapshot for atomic identifier quoting: column names containing a dot
+     * (ClickHouse nested-array convention, e.g. `meta.key Array(String)`)
+     * must remain single-backtick-wrapped atomic identifiers, not split
+     * across `.` into `` `meta`.`key` ``.
+     */
+    public function testNestedColumnDotQuoting(): void
+    {
+        $table = (new ClickHouseSchema())->table('events_with_meta');
+        $table->string('id')->primary();
+        $table->string('metric');
+        $table->datetime('time', 3)->nullable();
+        $table->addColumn('meta.key', ColumnType::String);
+        $table->addColumn('meta.value', ColumnType::String);
+        $table->engine(Engine::MergeTree)
+            ->orderBy(['metric', 'time', 'id']);
+
+        $statement = $table->createIfNotExists();
+
+        $this->assertStringContainsString('`meta.key` String', $statement->query);
+        $this->assertStringContainsString('`meta.value` String', $statement->query);
+        $this->assertStringNotContainsString('`meta`.`key`', $statement->query);
+        $this->assertStringNotContainsString('`meta`.`value`', $statement->query);
     }
 
     /**