EV-ClickHouse
view release on metacpan or search on metacpan
# METHODS
## query
$ch->query($sql, sub { my ($rows, $err) = @_ });
$ch->query($sql, \%settings, sub { my ($rows, $err) = @_ });
Executes a SQL statement. The callback receives:
- `($arrayref_of_arrayrefs)` for select with at least one row
- `(undef)` for DDL/DML on success and for select with zero rows
(both protocols). When in doubt, treat `undef` and `[]` equivalently
with `my @rows = @{$rows // []};`.
- `(undef, $error_message)` on error (server exception or
connection error)
The optional `\%settings` hashref passes per-query ClickHouse settings
(`max_execution_time`, `max_threads`, `async_insert`, etc.), overriding
connection-level defaults.
The following keys are intercepted by the client and not sent verbatim
to the server:
- `params => \%hash`
Parameterized values for `{name:Type}` placeholders in the SQL. Encoding
and quoting is the server's job, so values do not need escaping:
$ch->query(
"select * from t where id = {id:UInt64} and name = {n:String}",
{ params => { id => 42, n => "O'Brien" } },
sub { ... },
);
Works on both protocols (HTTP uses URL-encoded `param_*` query string;
native uses dedicated wire fields).
- `query_id => $string`
Set the protocol-level query identifier. Retrievable later via
["last\_query\_id"](#last_query_id).
- `raw => 1`
HTTP only. The callback receives the raw response body as a scalar string
instead of parsed rows. Use with an explicit `format` clause:
$ch->query("select * from t format CSV", { raw => 1 }, sub {
my ($body, $err) = @_;
});
Croaks if used with the native protocol.
- `query_timeout => $seconds`
Per-query timeout, overriding the connection-level `query_timeout`.
- `on_data => sub { my ($rows) = @_; ... }`
Native protocol only. A code ref called for each data block as it arrives,
for streaming large result sets. Rows are delivered incrementally and
**not** accumulated, so the final callback receives `(undef)` rather than
all rows. The final callback always fires on completion or error, even if
no data block was emitted (empty result, server-side error before the
first block).
$ch->query("select * from big_table",
{ on_data => sub { my ($rows) = @_; process_batch($rows) } },
sub { my (undef, $err) = @_; warn $err if $err },
);
- `external => \%tables`
Native protocol only. Ships one or more in-memory data blocks that the
query can reference as tables, JOIN against, or filter with `IN` -
without creating a server-side temporary table. Each entry maps a table
name to `{ structure => [...], data => [...] }`:
$ch->query(
"select u.id, u.name from users u where u.id in _wanted",
{ external => {
_wanted => {
structure => [ id => 'UInt64' ],
data => [ [7], [42], [911] ],
},
} },
sub { my ($rows, $err) = @_; ... },
);
`structure` is a flat list of `name => type` pairs (ClickHouse type
names, e.g. `UInt64`, `String`, `Float64`); `data` is an arrayref of
row arrayrefs, encoded with the same type machinery as ["insert"](#insert). An
empty `data` arrayref is a valid zero-row table. Several external tables
may be supplied at once. Croaks on the HTTP protocol or on a malformed
spec (odd structure list, non-arrayref row, or a column type that
cannot be encoded).
**Native protocol type notes:** values come back as typed Perl scalars.
By default `Date`/`DateTime` are integers (days since epoch / Unix
timestamps); enable `decode_datetime` for strings. `Enum` values are
numeric codes; `decode_enum` returns labels. `Decimal` values are
unscaled integers; `decode_decimal` scales them to floats.
`SimpleAggregateFunction` is transparently decoded as its inner type.
`Nested` columns become arrays of tuples. `LowCardinality` works
correctly across multi-block results with shared dictionaries.
## insert
$ch->insert($table, $data, sub { my (undef, $err) = @_ });
$ch->insert($table, $data, \%settings, sub { my (undef, $err) = @_ });
`$data` may be either:
- A pre-formatted TabSeparated string (tabs separate columns,
newlines separate rows, with the standard ClickHouse escapes).
- An arrayref of arrayrefs (rows of column values).
When using arrayrefs, no TSV escaping is needed: `undef` maps to null
and strings may contain tabs and newlines freely.
Nested arrayrefs (Array/Tuple columns) and hashrefs (Map columns) are
common transient failure that warrants automatic retry: timeouts,
network errors, memory pressure, replica catch-up, keeper exceptions,
etc. Authoritative-looking source list curated against ClickHouse's
`src/Common/ErrorCodes.cpp`; expect the set to grow conservatively.
$ch->query($sql, sub {
my ($r, $err) = @_;
if ($err && EV::ClickHouse->is_retryable_error($ch->last_error_code)) {
schedule_retry($sql);
}
});
## server\_supports
$ch->server_supports($feature_name)
Returns true if the live native server's protocol revision is high
enough to support the given feature. Feature names map to documented
protocol-revision thresholds so user code can branch cleanly on
capability instead of hard-coding revision numbers. Supported names:
block_info 51903 block_info packet in DATA blocks
server_display_name 54372 ServerHello carries display name
version_patch 54401 ServerHello carries patch version
progress_writes 54420 Progress packets include write counters
server_timezone 54423 Server timezone string in ServerHello
addendum 54458 Native ClientHello addendum block
HTTP connections have no protocol revision (`server_revision` is `0`),
so `server_supports` returns false on HTTP for any feature. Unknown
feature names also return false. Use `server_revision` directly if you
need the raw integer.
## for\_table
$ch->for_table('events', sub {
my ($info, $err) = @_;
die $err if $err;
for my $col (@{ $info->{columns} }) {
printf "%-20s %s\n", $col->{name}, $col->{type};
}
});
Schema introspection: issues `describe table $name` and delivers
`{ columns => [{name=>..., type=>...}, ...] }` to the
callback. Useful for generic insert pipelines that need column types
without hard-coding them. `$name` may be `table` or `db.table`;
non-identifier characters are rejected up-front.
## iterate
my $it = $ch->iterate("select number from numbers(1_000_000)");
while (my $batch = $it->next($timeout)) {
process($_) for @$batch;
}
die $it->error if $it->error;
**Native protocol only** - relies on the per-block `on_data` hook and
will croak if invoked on an HTTP connection.
Synchronous-feeling pull iterator over a streaming select. Internally
wraps the native `on_data` per-block callback and drives the EV loop
from inside `->next` until the next block arrives, the query
completes, or the optional timeout (seconds) expires. Useful for
procedural ETL / export code that doesn't fit a callback shape.
`->error`, `->is_done`, and `->cancel` are also
available on the returned iterator object.
## on\_log
$ch->on_log(sub {
my ($entry) = @_;
# $entry: { event_time, host_name, query_id, thread_id,
# priority, source, text }
printf "[CH %s] %s\n", $entry->{priority}, $entry->{text};
});
Native protocol only. Fires once per row inside any `SERVER_LOG`
packet the server emits. Useful for surfacing
`send_logs_level => 'information'` server-side trace events to
the application's own log stream without polling `system.text_log`.
The row hash keys mirror the server-side log block schema; missing
keys (older revisions) come through as `undef`.
## on\_query\_start
$ch->on_query_start(sub {
my ($query_id) = @_;
log_metric_start($query_id);
});
Optional connection-level hook that fires the moment a query is
dispatched to the wire (after the query\_id has been resolved, before
the first send byte). Symmetric with ["on\_query\_complete"](#on_query_complete); useful for
deriving accurate "query in flight" durations without depending on
the per-query callback closure. Keepalive PINGs are suppressed, the
same as for `on_query_complete`. Also accepted as a constructor
argument.
## on\_query\_complete
$ch->on_query_complete(sub {
my ($query_id, $rows, $bytes, $error_code, $duration_s, $err) = @_;
log_metric(...);
});
Optional connection-level hook that fires after every query (success
or error). Arguments: query\_id (or undef), profile\_rows, profile\_bytes,
last\_error\_code, wall-clock duration in seconds, error message (or
undef). Useful for statsd/Prometheus-style instrumentation. Also
accepted as a constructor argument.
A per-query override may be passed in the `\%settings` hashref of
["query"](#query) or ["insert"](#insert). When set, it **replaces** (does not augment)
the connection-level handler for that single call, so per-query
instrumentation doesn't double-count against global metrics:
$ch->query(
$sql,
{ on_query_complete => sub {
my ($qid, $rows, $bytes, $code, $dur, $err) = @_;
record_slow_query($qid, $dur);
} },
$cb,
);
## insert\_streamer
my $s = $ch->insert_streamer('events',
batch_size => 5_000,
settings => { query_id => 'ingest-1' }, # optional
on_batch_error => sub { warn "batch err: $_[0]" }, # per-failure
);
while (my $row = next_event()) {
$s->push_row($row);
}
$s->finish(sub {
my (undef, $err) = @_;
die "ingest failed: $err" if $err;
});
Buffered streaming insert for ETL workloads. Rows are buffered until
`batch_size` is reached, then dispatched as a single `insert()`.
Dispatches are serialised; push\_row keeps buffering while a batch is
in flight (the native protocol cannot pipeline INSERTs). `finish`
flushes the remaining buffer and fires its callback once all batches
complete; if any batch failed the first error is delivered as
`$err`. The streamer also offers `buffered_count` and `in_flight`
accessors for backpressure logic.
`$streamer->reset` discards any rows still in the local buffer
and clears the sticky error so the streamer can be reused after a
permanent error (e.g. a schema fix). Does **not** touch the underlying
`$ch` - any batch already on the wire still completes normally. Any
callback registered via `finish` or `await_drain` that has not yet
fired is invoked with a `'streamer reset'` error rather than being
silently dropped.
`high_water` + `on_high_water` trigger a one-shot notification when
the buffered row count crosses the watermark, intended as a hint to
slow the producer. Set `high_water` below `batch_size`; if
`high_water > batch_size`, the buffer drains via `batch_size`
flushes before the watermark is reached and `on_high_water` never
fires. When `high_water == batch_size` the watermark and the flush
threshold coincide: `on_high_water` fires once per batch, right after
the flush dispatches (so `in_flight` is already 1). The notification
re-arms only after the buffer drops below `high_water`.
`$streamer->await_drain($cb)` registers a callback that fires
once the buffer drops to `low_water` (default `high_water / 2`; 0
when `high_water` isn't set). Pairs with `on_high_water` to close
the backpressure loop:
my $s = $ch->insert_streamer($table,
batch_size => 5_000,
high_water => 10_000,
low_water => 3_000,
on_high_water => sub { $producer->pause },
);
$s->await_drain(sub { $producer->resume });
Fires synchronously if the buffer is already at/below `low_water`.
Re-arms each call: register again from inside the callback if you
want to keep watching. The callback receives one argument: `undef`
on a normal drain, or an error string if the streamer was `reset`
before the buffer drained (so `my ($err) = @_` distinguishes them).
**Named-row mode:** pass `columns => [@col_names]` at construction
to accept hashref rows instead of positional arrayrefs. The streamer
reorders each pushed hash into the declared column order, so producer
code does not have to know where each column lives in the table.
my $s = $ch->insert_streamer('events',
columns => [qw(ts user_id action payload)],
batch_size => 5_000,
);
$s->push_row({ user_id => 7, action => 'click', ts => time });
$s->push_row([ 1735, 7, 'view', '...' ]); # arrayref still works
Hash keys missing from a row become `undef`; extra keys are ignored.
Mixing arrayref and hashref pushes is allowed.
- Insert silently dropped (counts don't match)
Likely `insert_deduplication_token` dedupe; either you're reusing a token
across distinct batches, or the table is `ReplicatedMergeTree` with the
default dedupe window. See `eg/idempotent_insert.pl`.
- Hangs on connect when host is a hostname
Without [EV::cares](https://metacpan.org/pod/EV%3A%3Acares), DNS resolution falls back to blocking
`getaddrinfo`. Install [EV::cares](https://metacpan.org/pod/EV%3A%3Acares) for non-blocking lookup; otherwise
use an IP literal or a local caching resolver (nscd / systemd-resolved).
- `connect_timeout` doesn't fire
It does across TCP connect, TLS handshake, and native ServerHello. If
the timer doesn't fire, the underlying issue is usually a synchronous
DNS stall (see above) which happens before `start_connect` arms the
timer; install [EV::cares](https://metacpan.org/pod/EV%3A%3Acares) to push DNS off the loop.
- Per-query `query_timeout` is ignored
Set it inside the `\%settings` hashref, not as a top-level argument:
`$ch->query($sql, { query_timeout => 5 }, $cb)`.
- Which host am I currently pointed at after failover?
`$ch->current_host` and `$ch->current_port` reflect the
live target after a multi-host rotation. Use `on_failover =>
sub { ... }` to get notified at the moment of each rotation.
- How do I retry only on transient errors?
`EV::ClickHouse->is_retryable_error($code)` returns true for the
common transient codes (timeouts, network errors, replica catch-up,
keeper exceptions, ...). Inspect `$ch->last_error_code` from
inside your query callback and schedule a retry only when the predicate
fires - permanent errors (auth failures, missing tables) won't qualify.
Sample skeleton:
$ch->query($sql, sub {
my ($r, $err) = @_;
if ($err && EV::ClickHouse->is_retryable_error($ch->last_error_code)) {
schedule_retry($sql);
} elsif ($err) { warn "permanent: $err" }
});
- Idempotent insert silently drops some rows
`idempotent => 1` auto-mints
`insert_deduplication_token`; if your producer issues the SAME logical
batch twice (e.g. retry after a transient network blip) only the first
write lands, by design. To force two distinct logical batches through,
either pass an explicit `idempotent => $token` per batch or
omit the option for fresh inserts. See `eg/idempotent_insert.pl`.
- `on_data` vs `iterate` - which should I pick?
`on_data => sub { }` in the per-query settings is the
lowest-overhead streaming path: each native data block is delivered as
soon as the parser has it, no per-row allocation overhead beyond the
batch arrayref. `iterate` is a synchronous-feeling pull wrapper around
the same machinery - useful when the surrounding code is procedural
(ETL scripts, exporters) and a callback shape doesn't fit. Both are
native-only.
- Connection in front of nginx / reverse proxy strips X-ClickHouse-\* headers
Pass `http_basic_auth => 1` to send the credentials as
`Authorization: Basic ...` instead. Most HTTP gateways forward
Authorization verbatim while filtering proprietary headers.
# TUNING
- Native vs HTTP
Native (port 9000) is typically 2-5x faster for insert and select-of-many-rows
because rows ship as binary columns instead of TSV text. Use HTTP only when
the network path requires HTTPS-only or when you need `raw => 1` CSV /
JSONEachRow / Parquet bodies.
- `compress => 1`
Enables LZ4 (native) or gzip (HTTP). LZ4 cost is small and saves ~50-70%
on text-heavy columns. Gzip is heavier; turn on only if you're bandwidth-bound.
- `insert_streamer` batch\_size
Default 10\_000 is a good baseline. Smaller (1k-2k) reduces memory pressure
on the producer; larger (50k-100k) reduces server-side merge cost on
MergeTree. Match to your row width: ~1 MB per batch is a sweet spot.
- `keepalive`
Enable on long-lived idle connections (HTTP behind a load balancer or
NAT, or a native connection that may sit minutes between queries). 15-30s
is typical.
- `reconnect_max_attempts`
Always set in production. Default is unlimited; a permanent failure
(wrong host, wrong port, dead server) will spin `on_error` forever
otherwise.
- `progress_period`
Coalesce on\_progress packets to one fire per N seconds. Big SELECTs can
emit hundreds per second; throttle to 1-5s for monitoring dashboards.
- Pull-iterator vs `on_data`
`on_data` has lower per-block overhead. `iterate` trades that for a
synchronous-feeling API; use it when the surrounding code is procedural.
- `EV::ClickHouse::Pool`
A Pool fans concurrent queries across N independent connections, so a
slow query on one doesn't head-of-line-block the others. Use it for
read-mostly fan-out; do not use it for queries that depend on
session-level state (temporary tables, `set`) since each query may
( run in 0.905 second using v1.01-cache-2.11-cpan-6aa56a78535 )