Skip to content

Commit 076fd57

Browse files
lohanidamodarlohanidamodar
committed
feat(clickhouse): add bulk-insert builder for FORMAT envelope payloads
Routes ClickHouse's canonical `INSERT INTO <table> FORMAT <name>` bulk- ingest path through the typed builder. `Builder\ClickHouse::bulkInsert()` takes a `Format` enum and an iterable of associative rows and returns a `FormattedInsertStatement` whose `->query` is the envelope and whose new `->body` field is the serialized payload — callers ship both to the ClickHouse HTTP interface without hand-assembling either side. The `Format` enum supports `JSONEachRow` (encoded with JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE, one row per line, no trailing newline) and `TabSeparated` (escapes `\\`, `\t`, `\n`, `\r`; emits `\N` for null; booleans as 0/1). Empty iterables yield an empty body, which ClickHouse accepts as a zero-row ingest. Rows are materialized eagerly so the statement remains a plain readonly value object. `FormattedInsertStatement` gains an optional `?string $body` property (default null) that preserves back-compat for the existing `insertFormat()` + `insert()` envelope-only path. Callers who only want the envelope (e.g. when streaming the payload from elsewhere) keep using that path; callers who want a single typed call switch to `bulkInsert()`.
1 parent 77cdfbf commit 076fd57

5 files changed

Lines changed: 536 additions & 0 deletions

File tree

‎README.md‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1520,6 +1520,47 @@ $result->namedBindings;
15201520

15211521
Unregistered columns fall through to value-based inference: `int → Int64`, `float → Float64`, `bool → UInt8`, `null → Nullable(String)`, `DateTimeInterface → DateTime64(3)`, everything else → `String`. Register types via `withParamType($column, $type)` or `withParamTypes($map)` whenever the inference rule doesn't match the column's ClickHouse declaration. The positional `$bindings` array is still exposed on the resulting `Statement` for callers that prefer it.
15221522

1523+
**Bulk insert** — emit the canonical `INSERT INTO <table> FORMAT <name>` envelope together with the serialized row payload in a single typed call. The returned `FormattedInsertStatement` exposes `->query` (the envelope) and `->body` (the format-specific payload) so the caller can ship both to ClickHouse's HTTP interface without hand-assembling either side:
1524+
1525+
```php
1526+
use Utopia\Query\Builder\ClickHouse as Builder;
1527+
use Utopia\Query\Builder\ClickHouse\Format;
1528+
1529+
$statement = (new Builder())
1530+
->into('events')
1531+
->bulkInsert(Format::JSONEachRow, [
1532+
['id' => 1, 'event' => 'login', 'time' => '2024-01-01 00:00:00'],
1533+
['id' => 2, 'event' => 'logout', 'time' => '2024-01-01 00:00:05'],
1534+
]);
1535+
1536+
// $statement->query
1537+
// INSERT INTO `events` (`id`, `event`, `time`) FORMAT JSONEachRow
1538+
//
1539+
// $statement->body
1540+
// {"id":1,"event":"login","time":"2024-01-01 00:00:00"}
1541+
// {"id":2,"event":"logout","time":"2024-01-01 00:00:05"}
1542+
```
1543+
1544+
Ship the result over the HTTP interface by passing `$statement->query` as the `?query=` parameter and `$statement->body` as the POST body. Columns are derived from the first row's keys; pass an explicit third argument to pin the order or fill missing keys with `null`:
1545+
1546+
```php
1547+
$statement = (new Builder())
1548+
->into('events')
1549+
->bulkInsert(Format::JSONEachRow, $rows, ['id', 'event', 'time']);
1550+
```
1551+
1552+
The `Format` enum currently supports `Format::JSONEachRow` and `Format::TabSeparated`. JSONEachRow rows are encoded with `JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE` (slashes and non-ASCII are preserved verbatim); TabSeparated escapes `\\`, `\t`, `\n`, `\r` and emits `\N` for `null`. An empty row iterable produces an empty body, which ClickHouse accepts as a zero-row ingest. The iterable is consumed eagerly — pass a generator if you want to defer row construction, but the serialized body is materialized in full before the statement is returned.
1553+
1554+
For envelopes only (no body — e.g. when streaming the payload from elsewhere), the lower-level `insertFormat()` setter remains available and pairs with `insert()` as before:
1555+
1556+
```php
1557+
$statement = (new Builder())
1558+
->into('events')
1559+
->insertFormat('JSONEachRow', ['id', 'event', 'time'])
1560+
->insert();
1561+
// $statement->body is null; assemble the payload separately.
1562+
```
1563+
15231564
**UPDATE** — compiles to `ALTER TABLE ... UPDATE` with mandatory WHERE:
15241565

15251566
```php

‎src/Query/Builder/ClickHouse.php‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
namespace Utopia\Query\Builder;
44

55
use Utopia\Query\Builder as BaseBuilder;
6+
use Utopia\Query\Builder\ClickHouse\Format;
67
use Utopia\Query\Builder\ClickHouse\FormattedInsertStatement;
78
use Utopia\Query\Builder\Feature\BitwiseAggregates;
89
use Utopia\Query\Builder\Feature\ClickHouse\ApproximateAggregates;
@@ -163,6 +164,72 @@ public function insertFormat(string $format, array $columns = []): static
163164
return $this;
164165
}
165166

167+
/**
168+
* Build a single statement that carries both the `INSERT INTO <table>
169+
* FORMAT <name>` envelope and the serialized row payload for a
170+
* ClickHouse bulk ingest. Returns a `FormattedInsertStatement` whose
171+
* `->query` is the envelope and whose `->body` is the formatted
172+
* payload to send as the HTTP request body.
173+
*
174+
* The target table must be set via `into()` first. Columns are derived
175+
* from the keys of the first row when `$columns` is omitted; pass
176+
* `$columns` explicitly to pin the order when row shapes vary or when
177+
* an empty iterable is passed. An empty iterable produces an empty
178+
* body — ClickHouse accepts this as a zero-row ingest.
179+
*
180+
* @param iterable<array<string, mixed>> $rows
181+
* @param list<string> $columns Optional explicit column ordering.
182+
*/
183+
public function bulkInsert(Format $format, iterable $rows, array $columns = []): FormattedInsertStatement
184+
{
185+
$this->bindings = [];
186+
$this->validateTable();
187+
188+
$materialized = [];
189+
foreach ($rows as $row) {
190+
/** @phpstan-ignore function.alreadyNarrowedType */
191+
if (!\is_array($row)) {
192+
throw new ValidationException('bulkInsert() rows must be associative arrays.');
193+
}
194+
$materialized[] = $row;
195+
}
196+
197+
if (empty($columns) && !empty($materialized)) {
198+
$columns = \array_keys($materialized[0]);
199+
}
200+
201+
foreach ($columns as $col) {
202+
if ($col === '') {
203+
throw new ValidationException('Column names for bulkInsert() must be non-empty strings.');
204+
}
205+
}
206+
207+
$wrappedColumns = empty($columns)
208+
? ''
209+
: ' (' . \implode(', ', \array_map(
210+
fn (string $col): string => $this->resolveAndWrap($col),
211+
$columns
212+
)) . ')';
213+
214+
$sql = 'INSERT INTO ' . $this->quote($this->table)
215+
. $wrappedColumns
216+
. ' FORMAT ' . $format->value;
217+
218+
$body = $format->serialize($materialized, empty($columns) ? null : $columns);
219+
220+
$this->insertFormat = $format->value;
221+
$this->insertFormatColumns = $columns;
222+
223+
return new FormattedInsertStatement(
224+
$sql,
225+
[],
226+
$columns,
227+
$format->value,
228+
$body,
229+
executor: $this->executor,
230+
);
231+
}
232+
166233
/**
167234
* @param array<string, string> $settings
168235
*/

‎src/Query/Builder/ClickHouse/Format.php‎

Lines changed: 117 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,117 @@
1+
<?php
2+
3+
namespace Utopia\Query\Builder\ClickHouse;
4+
5+
use Utopia\Query\Exception\ValidationException;
6+
7+
/**
8+
* ClickHouse bulk-ingest format identifiers.
9+
*
10+
* The values map 1:1 to the names ClickHouse accepts after the `FORMAT`
11+
* keyword in an `INSERT INTO <table> FORMAT <name>` envelope. Each case
12+
* knows how to serialize a row iterable into the request body that
13+
* ClickHouse expects for that format.
14+
*/
15+
enum Format: string
16+
{
17+
case JSONEachRow = 'JSONEachRow';
18+
case TabSeparated = 'TabSeparated';
19+
20+
/**
21+
* Serialize an iterable of associative rows into the body payload for
22+
* this format. Columns are derived from the first row; subsequent rows
23+
* use the same column ordering. An empty iterable yields an empty
24+
* string — ClickHouse accepts an empty body as a zero-row insert.
25+
*
26+
* @param iterable<array<string, mixed>> $rows
27+
* @param list<string>|null $columns Optional explicit column ordering. When null, derived from the first row.
28+
*/
29+
public function serialize(iterable $rows, ?array $columns = null): string
30+
{
31+
return match ($this) {
32+
self::JSONEachRow => $this->serializeJsonEachRow($rows, $columns),
33+
self::TabSeparated => $this->serializeTabSeparated($rows, $columns),
34+
};
35+
}
36+
37+
/**
38+
* @param iterable<array<string, mixed>> $rows
39+
* @param list<string>|null $columns
40+
*/
41+
private function serializeJsonEachRow(iterable $rows, ?array $columns): string
42+
{
43+
$lines = [];
44+
foreach ($rows as $row) {
45+
if ($columns !== null) {
46+
$ordered = [];
47+
foreach ($columns as $col) {
48+
$ordered[$col] = $row[$col] ?? null;
49+
}
50+
$row = $ordered;
51+
}
52+
53+
$lines[] = \json_encode(
54+
(object) $row,
55+
JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE,
56+
);
57+
}
58+
59+
return \implode("\n", $lines);
60+
}
61+
62+
/**
63+
* @param iterable<array<string, mixed>> $rows
64+
* @param list<string>|null $columns
65+
*/
66+
private function serializeTabSeparated(iterable $rows, ?array $columns): string
67+
{
68+
$lines = [];
69+
foreach ($rows as $row) {
70+
$values = [];
71+
72+
if ($columns === null) {
73+
foreach ($row as $value) {
74+
$values[] = $this->escapeTabSeparatedValue($value);
75+
}
76+
} else {
77+
foreach ($columns as $col) {
78+
$values[] = $this->escapeTabSeparatedValue($row[$col] ?? null);
79+
}
80+
}
81+
82+
$lines[] = \implode("\t", $values);
83+
}
84+
85+
return \implode("\n", $lines);
86+
}
87+
88+
private function escapeTabSeparatedValue(mixed $value): string
89+
{
90+
if ($value === null) {
91+
return '\\N';
92+
}
93+
94+
if (\is_bool($value)) {
95+
return $value ? '1' : '0';
96+
}
97+
98+
if (\is_int($value) || \is_float($value)) {
99+
return (string) $value;
100+
}
101+
102+
if (! \is_string($value)) {
103+
if (\is_object($value) && \method_exists($value, '__toString')) {
104+
$value = (string) $value;
105+
} else {
106+
throw new ValidationException('TabSeparated values must be scalar, null, or stringable. Received: ' . \get_debug_type($value));
107+
}
108+
}
109+
110+
return \strtr($value, [
111+
'\\' => '\\\\',
112+
"\t" => '\\t',
113+
"\n" => '\\n',
114+
"\r" => '\\r',
115+
]);
116+
}
117+
}

‎src/Query/Builder/ClickHouse/FormattedInsertStatement.php‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
* @param list<mixed> $bindings
1313
* @param list<string> $columns
1414
* @param string $format
15+
* @param ?string $body Serialized payload to ship as the HTTP request body alongside `$query`. Null when only the envelope query was produced (the caller assembles the body separately).
1516
* @param bool $readOnly
1617
* @param (Closure(Statement): (array<mixed>|int))|null $executor
1718
*/
@@ -20,6 +21,7 @@ public function __construct(
2021
array $bindings,
2122
public array $columns,
2223
public string $format,
24+
public ?string $body = null,
2325
bool $readOnly = false,
2426
?Closure $executor = null,
2527
) {
@@ -34,6 +36,7 @@ public function withExecutor(Closure $executor): self
3436
$this->bindings,
3537
$this->columns,
3638
$this->format,
39+
$this->body,
3740
$this->readOnly,
3841
$executor,
3942
);

0 commit comments

Comments
 (0)