EV-ClickHouse
view release on metacpan or search on metacpan
lib/EV/ClickHouse.pm view on Meta::CPAN
$args{user} //= _uri_unescape($u) if defined $u && $u ne '';
$args{password} //= _uri_unescape($pw) if defined $pw;
$args{host} //= $h;
$args{port} //= $p if defined $p;
$args{database} //= _uri_unescape($db) if defined $db && $db ne '';
_uri_qs_into($qs, \%args);
} else {
die "EV::ClickHouse: invalid URI '$uri'\n";
}
}
my $loop = delete $args{loop} || EV::default_loop;
my $self = $class->_new($loop);
# Multi-host failover: hosts => ['a', 'b', 'c'] or ['a:9000', 'b:9001'].
# On a connect-phase failure, advance to the next host and reconnect
# via auto_reconnect (or the user calling reset). Falls back to single
# host => '...' when not provided.
my $hosts_list;
if (my $h = delete $args{hosts}) {
die "hosts must be a non-empty arrayref"
unless ref($h) eq 'ARRAY' && @$h;
$hosts_list = $h;
if (!defined $args{host}) {
my ($h0, $p0) = _split_host_port($h->[0], $args{port});
$args{host} = $h0;
$args{port} //= $p0;
}
}
# Failover state + on_failover both live in the connection's C struct
# now: emit_error advances the host ring before firing on_error,
# which keeps the hot error path off the Perl stack.
my $user_on_error = exists $args{on_error}
? delete $args{on_error} : sub { die @_ };
$self->on_error($user_on_error);
if (my $cb = delete $args{on_failover}) { $self->on_failover($cb) }
for my $h (qw(on_connect on_progress on_disconnect on_trace on_query_complete on_query_start on_log)) {
$self->$h(delete $args{$h}) if exists $args{$h};
}
my $host = delete $args{host} // '127.0.0.1';
my $port = delete $args{port};
my $protocol = delete $args{protocol} // 'http';
my $user = delete $args{user} // 'default';
my $password = delete $args{password} // '';
my $database = delete $args{database} // delete $args{db} // 'default';
die "EV::ClickHouse: unknown protocol '$protocol' (expected 'http' or 'native')\n"
unless $protocol eq 'http' || $protocol eq 'native';
$port //= ($protocol eq 'native') ? 9000 : 8123;
$self->_set_protocol($protocol eq 'native' ? 1 : 0);
# Pass-through setters. Skip only when the key was absent — explicit
# 0/'' must reach the setter so e.g. `compress => 0` is honored, not
# ignored. (Use `exists` rather than `defined` so a deliberate undef
# is also passed through and rejected by the setter if invalid.)
for my $opt (qw(compress tls tls_skip_verify auto_reconnect
keepalive reconnect_delay reconnect_max_delay
reconnect_jitter reconnect_max_attempts
progress_period http_basic_auth
connect_timeout query_timeout
max_query_size max_recv_buffer)) {
next unless exists $args{$opt};
my $val = delete $args{$opt};
my $setter = "_set_$opt";
$self->$setter($val);
}
for my $opt (qw(session_id tls_ca_file tls_cert_file tls_key_file)) {
defined(my $val = delete $args{$opt}) or next;
my $setter = "_set_$opt";
$self->$setter($val);
}
# query_log_comment: 1 = auto-generate "ev_ch user=$ENV{USER} pid=$$"; any
# other defined non-empty string (including "0") is taken literally;
# undef / not present / empty string is disabled.
{
my $qlc = delete $args{query_log_comment};
if (defined $qlc && length $qlc) {
my $cmt = (!ref($qlc) && "$qlc" ne '1')
? $qlc
: sprintf 'ev_ch user=%s pid=%d', $ENV{USER} // 'na', $$;
$cmt =~ s{\*/}{*\\/}g;
$self->_set_query_log_comment($cmt);
}
}
# decode_flags bitmask (DT_STR=1, DEC_SCALE=2, ENUM_STR=4, NAMED_ROWS=8)
my $decode_flags = (delete $args{decode_datetime} ? 1 : 0)
| (delete $args{decode_decimal} ? 2 : 0)
| (delete $args{decode_enum} ? 4 : 0)
| (delete $args{named_rows} ? 8 : 0);
$self->_set_decode_flags($decode_flags) if $decode_flags;
if (my $settings = delete $args{settings}) { $self->_set_settings($settings) }
warn "EV::ClickHouse->new: unknown parameter(s): " . join(', ', sort keys %args) . "\n"
if %args;
if ($hosts_list) {
$self->_set_failover($hosts_list, $port);
}
# Async DNS via EV::cares when available — non-IP hostnames are
# resolved off-loop so the constructor returns immediately and the
# main EV loop never blocks on getaddrinfo. Pre-connect-queued
# queries fire once the resolved-address connect completes. Falls
# back to the XS blocking resolver if EV::cares isn't installed
# or the host is already an IP literal.
if ($host !~ /^[\d.]+$|^\[?[0-9a-fA-F:]+\]?$/
&& eval { require EV::cares; 1 }) {
# Stash the resolver in %_failover, keyed by refaddr of the
# resolver itself (NOT of $self). Two reasons:
# - never delete the resolver from inside its own callback —
# ares_destroy from a c-ares cb corrupts the channel heap.
# We defer the delete via EV::timer(0,...) so it runs from
# a clean stack frame.
# - keying by refaddr($self) was racy: A's deferred-delete
# could fire after A's struct was freed and B got the same
# refaddr, dropping B's resolver. refaddr($r) is unique
# while $r is alive in %_failover.
my $r = EV::cares->new;
my $key = refaddr($r);
$_failover{$key} = $r;
my $weak2 = $self; weaken $weak2;
$self->_set_dns_pending(1);
$r->resolve($host, sub {
my ($status, @addrs) = @_;
my $w; $w = EV::timer(0, 0, sub { undef $w; delete $_failover{$key} });
# Skip if the connection has been DESTROYed or the user
# finished it while DNS was in flight (cleanup_connection
# clears dns_pending; if it's 0 here, finish ran already).
return unless $weak2 && $weak2->_take_dns_pending;
if ($status != 0 || !@addrs) {
$weak2->skip_pending;
# Warn if the handler itself throws — matches the XS
# emit_error path (WARN_AND_CLEAR_ERRSV); a bare eval here
# would make a DNS failure vanish under the default
# `on_error => sub { die @_ }`.
eval { $weak2->on_error->("DNS resolution failed for '$host'"); 1 }
or warn "EV::ClickHouse: exception in error handler: $@";
return;
}
my ($v4) = grep /^[\d.]+$/, @addrs;
$weak2->connect($v4 // $addrs[0], $port, $user, $password, $database);
});
} else {
$self->connect($host, $port, $user, $password, $database);
}
$self;
}
sub _split_host_port {
my ($entry, $default_port) = @_;
if ($entry =~ /^\[([^\]]+)\](?::(\d+))?$/) {
return ($1, $2 // $default_port); # IPv6 literal in brackets
}
if ($entry =~ /^([^:]+):(\d+)$/) {
return ($1, $2);
}
return ($entry, $default_port);
}
# Pull-based result iterator: $it = $ch->iterate($sql, [\%settings])
# while (my $batch = $it->next($timeout)) { ... }
# Wraps the native on_data per-block callback in a synchronous-feeling
# pull interface for procedural code. The iterator drives the EV loop
# from inside ->next until the next block arrives, the query completes,
# or the optional timeout (seconds) expires.
sub iterate {
my ($self, $sql, $settings) = @_;
my $it = bless {
ch => $self,
batches => [],
done => 0,
err => undef,
}, 'EV::ClickHouse::Iterator';
my $on_data = sub {
push @{ $it->{batches} }, $_[0];
EV::break;
lib/EV/ClickHouse.pm view on Meta::CPAN
unless defined $query_id && ref($cb) eq 'CODE';
die "kill_query: invalid query_id '$query_id'"
unless $query_id =~ /\A[A-Za-z0-9_\-]+\z/;
my $mode = $opts{async} ? 'async' : 'sync';
$self->query("kill query where query_id = '$query_id' $mode", $cb);
}
# Per-connection latency tracking. EV::ClickHouse is a blessed scalar
# (XS struct pointer) so we can't stash hash slots on it. We keep a
# lexical refaddr-keyed map and hold the entry's lifetime via a guard
# object stored inside the wrapper closure: when the user replaces
# on_query_complete (or XS DESTROY clears it), the closure is freed,
# the guard goes out of scope, and the %DUR_STATE entry is reaped.
my %DUR_STATE;
{
package EV::ClickHouse::_DurGuard;
# During global destruction %DUR_STATE may have already been
# reaped; touching it then is undefined.
sub DESTROY {
return if ${^GLOBAL_PHASE} eq 'DESTRUCT';
delete $DUR_STATE{ ${ $_[0] } };
}
}
sub track_query_durations {
my ($self, $size) = @_;
$size //= 1024;
my $key = refaddr($self);
if ($size == 0) {
if (my $st = delete $DUR_STATE{$key}) {
$self->on_query_complete($st->{prev}); # undef restores no-handler
}
return $self;
}
my $st = $DUR_STATE{$key};
if (!$st) {
$st = $DUR_STATE{$key} = {
size => $size,
buf => [],
pos => 0,
prev => $self->on_query_complete,
};
my $guard = bless \(my $k = $key), 'EV::ClickHouse::_DurGuard';
my $prev = $st->{prev};
$self->on_query_complete(sub {
# Use captured $key (not refaddr at fire time) so we still
# reach the right ring after DESTROY zeroed the SV.
my $ring = $DUR_STATE{$key};
if ($ring) {
my $dur = $_[4];
if (defined $dur && $dur >= 0) {
if (@{ $ring->{buf} } < $ring->{size}) {
push @{ $ring->{buf} }, $dur;
} else {
$ring->{buf}[ $ring->{pos} ] = $dur;
$ring->{pos} = ($ring->{pos} + 1) % $ring->{size};
}
}
}
$prev->(@_) if $prev;
$guard; # keep the guard alive for the closure's lifetime
});
} else {
# Resize: linearize chronological order (oldest at $st->{pos}
# once the ring is full), keep the newest min(N, $size) samples.
# Plain shift would drop by physical index instead of by age.
my $buf = $st->{buf};
if (@$buf >= $st->{size}) {
# Ring full: chronological is buf[pos..end] then buf[0..pos-1].
my $pos = $st->{pos} % $st->{size};
@$buf = (@{$buf}[$pos .. $#$buf], @{$buf}[0 .. $pos - 1]);
}
# Insertion order is now buf[0..n-1] — already chronological.
shift @$buf while @$buf > $size;
$st->{size} = $size;
# After linearization the oldest item is at index 0. While the
# ring is still filling, push appends and pos is ignored; once
# full, the next overwrite must target the oldest (index 0).
$st->{pos} = 0;
}
return $self;
}
sub query_duration_p {
my ($self, $p) = @_;
my $st = $DUR_STATE{ refaddr($self) } or return undef;
my @s = sort { $a <=> $b } @{ $st->{buf} };
return undef unless @s;
$p = 0 if $p < 0;
$p = 1 if $p > 1;
$s[ int($p * (@s - 1) + 0.5) ];
}
sub query_duration_count {
my $st = $DUR_STATE{ refaddr($_[0]) } or return 0;
scalar @{ $st->{buf} };
}
# Local in-flight cancel guarded by query_id match. Only triggers
# cancel() if the connection's current in-flight query (last_query_id)
# matches $query_id, so a race where the intended query has already
# finished and a different one is now running can't silently kill the
# wrong query. Returns 1 if it cancelled, 0 if the id didn't match.
sub cancel_by_query_id {
my ($self, $query_id) = @_;
die "cancel_by_query_id: query_id required" unless defined $query_id && length $query_id;
my $cur = $self->last_query_id;
return 0 unless defined $cur && $cur eq $query_id;
$self->cancel;
return 1;
}
# Retry a query over the same connection with exponential backoff,
# only on retryable server errors. Falls through to the user's $cb
# with the final result (success or last error) — never invokes $cb
# more than once. Per-attempt $settings are honored; pass via the
# settings => \%hash key.
#
# $ch->retry("select * from t",
# retries => 3, backoff => 0.5, jitter => 0.25,
# cb => sub { my ($rows, $err) = @_; ... });
lib/EV/ClickHouse.pm view on Meta::CPAN
my ($self, $method, @rest) = @_;
my $ch = $self->_pick;
$rest[-1] = $self->_cb_observer($ch, $rest[-1]) if ref $rest[-1] eq 'CODE';
$ch->$method(@rest);
}
sub query { shift->_dispatch(query => @_) }
sub insert { shift->_dispatch(insert => @_) }
sub ping { shift->_dispatch(ping => @_) }
sub for_table { shift->_pick->for_table(@_) }
sub iterate { shift->_pick->iterate(@_) }
sub insert_streamer { shift->_pick->insert_streamer(@_) }
# Same as _dispatch but pins the target to $conn[$idx] instead of
# polling _pick. Circuit-breaker observation still applies.
sub _dispatch_to {
my ($self, $method, $idx, @rest) = @_;
die "${method}_to: index $idx out of range"
if $idx < 0 || $idx >= @{ $self->{conns} };
my $ch = $self->{conns}[$idx];
$rest[-1] = $self->_cb_observer($ch, $rest[-1]) if ref $rest[-1] eq 'CODE';
$ch->$method(@rest);
}
sub query_to { shift->_dispatch_to(query => @_) }
sub insert_to { shift->_dispatch_to(insert => @_) }
# Nominate a member: returns its connection object so subsequent calls
# stick to it. The caller is responsible for not abusing this (the pool
# can't apply the circuit breaker to calls it doesn't see).
sub nominate {
my ($self, $idx) = @_;
die "nominate: index $idx out of range" if $idx < 0 || $idx >= @{ $self->{conns} };
$self->{conns}[$idx];
}
# Hedged read: dispatch the same query to N (default 2) distinct
# members and resolve with whichever returns first. Subsequent
# completions are silently dropped. Errors are reported only if every
# member fails. Recommended for tail-latency-sensitive selects on
# replicated tables; do NOT use for insert (would silently double-write
# on dedupe miss). $cb receives ($rows, undef, $member_idx) on success
# or (undef, $err) when every member fails.
sub hedged_query {
my ($self, $sql, @rest) = @_;
my $cb = pop @rest;
my %opts = @rest;
my $hedge_n = delete $opts{hedge} // 2;
my $settings = delete $opts{settings};
die "hedged_query: callback required" unless ref($cb) eq 'CODE';
die "hedged_query: unknown options: " . join(', ', sort keys %opts)
if %opts;
my @c = @{ $self->{conns} };
die "hedged_query: no members" unless @c;
# Filter out circuit-broken members. If the breaker tripped everywhere
# fall back to the full set so the caller still hears something — but
# in that fallback skip _cb_observer too, otherwise every failed hedge
# extends each member's dead_until (resetting cooldown indefinitely
# under load).
my $now = EV::time();
my @alive = grep { $self->{cb_state}[$_]{dead_until} <= $now } 0 .. $#c;
my $all_dead = !@alive;
@alive = (0 .. $#c) if $all_dead;
$hedge_n = @alive if $hedge_n > @alive;
$hedge_n = 1 if $hedge_n < 1;
# Reservoir-style shuffle for distinct random picks.
my @pool = @alive;
my @idx;
while (@idx < $hedge_n) { push @idx, splice(@pool, int(rand(scalar @pool)), 1) }
my $fired = 0;
my $pending = scalar @idx;
my $first_err;
for my $i (@idx) {
my $ch = $c[$i];
my $inner = sub {
my ($rows, $err) = @_;
$pending--;
return if $fired;
if (!$err) {
$fired = 1;
# 3rd arg = winning member index, so callers can attribute
# wins / track per-replica latency without scanning conns.
$cb->($rows, undef, $i);
return;
}
$first_err //= $err;
if (!$pending) {
$fired = 1;
$cb->(undef, $first_err);
}
};
# Pass $i (not $ch) so _cb_observer can index cb_state directly
# instead of walking conns via _slot_for.
my $obs = $self->_cb_observer($i, $inner, !$all_dead);
if ($settings) { $ch->query($sql, $settings, $obs) }
else { $ch->query($sql, $obs) }
}
return;
}
# Circuit breaker introspection: per-member state for monitoring.
# Returns ({ fails => N, dead_until => $epoch_seconds, alive => 0|1 }, ...).
sub circuit_state {
my $self = shift;
my $now = EV::time();
map +{ %$_, alive => $_->{dead_until} <= $now },
@{ $self->{cb_state} };
}
# Aggregate stats
sub size { scalar @{ $_[0]{conns} } }
sub pending_count { my $t = 0; $t += $_->pending_count for @{ $_[0]{conns} }; $t }
sub conns { @{ $_[0]{conns} } }
# Apply a code ref to every pool member. The callback receives
# ($conn, $idx) per call. Useful for warm-up (preload dictionaries,
# set session-level variables, dispatch a probe per member). The
# callback is invoked synchronously in pool order; if it throws,
# subsequent members are still visited (errors silently swallowed,
# matching the broadcast cancel/skip_pending/reset convention).
sub with_each {
my ($self, $cb) = @_;
die "Usage: \$pool->with_each(\$cb)" unless ref($cb) eq 'CODE';
my @c = @{ $self->{conns} };
for my $i (0 .. $#c) { eval { $cb->($c[$i], $i) } }
return;
}
# Broadcast the same SELECT to every member and collect per-member
# results. Useful for `system.replicas`-style diagnostics where each
# shard needs to be queried directly. Callback fires once with an
# arrayref of { member => $i, rows => [...], err => $msg }, ordered
# by member index. Per-query settings are honoured. Dead members are
# included in the result with a "circuit open" error string rather
# than dispatched — the breaker would refuse them anyway.
sub fan_out {
my ($self, $sql, @rest) = @_;
my $cb = pop @rest;
my %opts = @rest;
my $settings = delete $opts{settings};
die "fan_out: callback required" unless ref($cb) eq 'CODE';
my @c = @{ $self->{conns} };
die "fan_out: no members" unless @c;
my @out = map { { member => $_, rows => undef, err => undef } } 0 .. $#c;
my $left = scalar @c;
my $deliver = sub { $cb->(\@out) unless --$left };
my $now = EV::time();
for my $i (0 .. $#c) {
my $ch = $c[$i];
# Short-circuit dead members so a long cooldown doesn't stall fan_out.
if ($self->{cb_thresh}
&& $self->{cb_state}[$i]{dead_until} > $now) {
$out[$i]{err} = "fan_out: member $i circuit open";
$deliver->();
next;
}
my $obs = $self->_cb_observer($i, sub {
($out[$i]{rows}, $out[$i]{err}) = @_;
$deliver->();
});
# Wrap each member's dispatch so a synchronous croak (e.g.
# "not connected" before auto_reconnect catches up) doesn't
# strand the rest of the callbacks waiting on $left.
eval {
if ($settings) { $ch->query($sql, $settings, $obs) }
else { $ch->query($sql, $obs) }
lib/EV/ClickHouse.pm view on Meta::CPAN
my ($rows, $err) = @_;
die $err if $err;
print "row: @$_\n" for @$rows; # row: 0 / row: 1 / row: 2
});
# Per-query settings + parameterized values (no string interpolation)
$ch->query(
"select {x:UInt32} + {y:UInt32} as sum",
{ params => { x => 40, y => 2 }, max_execution_time => 30 },
sub { my ($rows, $err) = @_; print $rows->[0][0], "\n" }, # 42
);
# insert - arrayref of rows (no TSV escaping needed)
$ch->insert("my_table", [
[1, "hello\tworld"], # embedded tab is fine
[2, undef], # null
[3, [10, 20]], # Array column
], sub { my (undef, $err) = @_; warn "insert: $err" if $err });
# insert - pre-formatted TSV string
$ch->insert("my_table", "1\tfoo\n2\tbar\n", sub { ... });
# Raw HTTP response body (HTTP only)
$ch->query("select * from t format CSV", { raw => 1 }, sub {
my ($body, $err) = @_;
print $body;
});
EV::run;
=head1 DESCRIPTION
EV::ClickHouse is an asynchronous ClickHouse client that integrates with
the L<EV> event loop. It speaks both the ClickHouse HTTP protocol
(port 8123) and the native TCP protocol (port 9000) directly in XS, with
no external ClickHouse client library linked. zlib is required; OpenSSL
(for TLS) and liblz4 (for native compression) are optional and detected
at build time.
=head2 Features
=over 4
=item * HTTP and native TCP protocols, with the same Perl API
=item * gzip compression (HTTP) and LZ4 compression with CityHash
checksums (native)
=item * TLS/SSL via OpenSSL, with optional C<tls_skip_verify> for
self-signed certs and C<tls_ca_file> for additional roots
=item * Connection URIs (C<clickhouse[+native]://user:pass@host:port/db>),
including bracketed IPv6 literals
=item * Per-query and connection-level ClickHouse settings; parameterized
queries via C<params>; external tables (native) via C<external>
=item * Auto-reconnect with exponential backoff; queued (unsent) queries
are preserved across reconnects
=item * Keepalive pings for idle native connections; graceful drain;
query cancellation and skip_pending
=item * Streaming results via C<on_data> per-block callback (native);
on_progress for native progress packets
=item * Raw HTTP response mode for CSV / JSONEachRow / Parquet / etc.
=item * 35+ ClickHouse types including Int/UInt 8..256, Float32/64,
BFloat16, Decimal32/64/128/256, UUID, IPv4/IPv6, Nullable, Array,
Tuple, Map, LowCardinality (with cross-block dictionaries),
SimpleAggregateFunction, Nested, Geo (Point/Ring/LineString/Polygon
and the Multi variants), and JSON / Object('json') with auto-flattened
hashref leaves (Int64/Float64/Bool/String + Array variants).
=item * Opt-in decode of Date/DateTime, Decimal, and Enum columns; named-rows
(hashref) mode
=back
=head1 CONSTRUCTOR
=head2 new
my $ch = EV::ClickHouse->new(%args);
The connection is initiated immediately; C<new> returns before it
completes. Queries issued before C<on_connect> fires are queued and
dispatched once the connection is ready.
B<Connection parameters:>
=over 4
=item uri => $uri_string
Single-string connection target:
C<clickhouse[+native]://user:pass@host:port/database?key=value>.
The C<+native> suffix selects the native protocol; otherwise HTTP is used.
Hostnames, IPv4 addresses, and bracketed IPv6 literals are all accepted
(e.g. C<clickhouse://[::1]:9000/db>). Query-string values are merged into
the constructor arguments. Discrete C<host>, C<port>, etc. arguments
override the URI.
=item host => $hostname
Server hostname. Default: C<127.0.0.1>.
B<Note:> DNS resolution is blocking unless L<EV::cares> is installed.
With L<EV::cares> available, hostnames are resolved off-loop at
construct time (the constructor returns immediately, queries queue
until the resolved address is connected). Falls back to blocking
C<getaddrinfo> otherwise.
=item hosts => [$h1, $h2, ...]
Multi-host failover list. Each entry is C<host>, C<host:port>, or a
bracketed-IPv6 literal. On a connect-phase failure (refused, timeout,
ServerHello stall), the client advances to the next host in round-robin
order; pair with C<auto_reconnect =E<gt> 1> for automatic recovery.
lib/EV/ClickHouse.pm view on Meta::CPAN
TCP/TLS connection timeout. C<0> (default) means no timeout. Floating
point allowed.
=item query_timeout => $seconds
Default per-query timeout applied to every query and insert. The query
callback receives a C<timeout> error if exceeded. Override per-call via
the C<query_timeout> key in the settings hashref.
=item max_query_size => $bytes
Client-side guard: croak before sending any query whose SQL text exceeds
this many bytes. C<0> (default) disables the check. Useful as a
last-resort defense against accidentally sending unbounded strings.
=item max_recv_buffer => $bytes
Defensive ceiling on the response. The cap applies to the raw recv
buffer (every protocol), the chunked-decoded body (HTTP), and the
gzip-decompressed body (HTTP), so the same upper bound applies to the
user-visible payload regardless of transport encoding. On overflow the
query callback receives an appropriate error ("recv buffer overflow",
"chunked response too large", or "gzip body exceeds max_recv_buffer")
and the connection is torn down so no subsequent query can slip past
the cap on the same socket. C<0> (default) keeps the historical
no-cap behaviour (still bounded internally by a hard 128 MB ceiling
on compressed paths). Recommended in production when the schema is
constrained and you want a hard upper bound (e.g.
C<128 * 1024 * 1024> for 128 MB).
=item http_basic_auth => 0 | 1
HTTP only. When set, send credentials as
C<Authorization: Basic base64(user:password)> instead of the default
C<X-ClickHouse-User> / C<X-ClickHouse-Key> header pair. Use this when
the connection passes through an HTTP gateway (nginx, Envoy, ...) that
strips the X-ClickHouse-* headers but forwards Basic auth verbatim.
Default: C<0>.
=item auto_reconnect => 0 | 1
Reconnect automatically on connection loss. Default: C<0>. When enabled,
queued (unsent) queries are preserved across reconnects; in-flight queries
receive an error.
The reconnect path covers TCP/TLS connect failures, C<connect_timeout>
or C<query_timeout> expiry, and any clean server-side EOF (idle or
mid-request). Mid-query I/O errors (ECONNRESET / EPIPE) and a malformed
native ServerHello are B<not> retried - they typically indicate a
misconfigured peer or client-side bug that retry would only loop on.
Combine with C<reconnect_max_attempts> for an explicit ceiling.
=item settings => \%hash
ClickHouse settings applied to every query and insert. Per-call settings
(see L</query>, L</insert>) override these.
settings => { async_insert => 1, max_threads => 4 }
=item keepalive => $seconds
Send a keepalive request every N seconds while the connection is idle:
a native CLIENT_PING on the native protocol or a C<GET /ping> on HTTP
(some load balancers / NATs drop idle HTTP connections after a few
seconds; TCP-level keepalive is too coarse). Default: C<0> (disabled).
=item reconnect_delay => $seconds
Initial delay for the C<auto_reconnect> exponential backoff. Each failed
attempt doubles the delay, capped at C<reconnect_max_delay>. Default:
C<0> (immediate retry, no backoff).
=item reconnect_max_delay => $seconds
Backoff ceiling. Default: C<0>, meaning no explicit cap; the implementation
still bounds the backoff exponent at 20 doublings, so with
C<reconnect_delay = 0.5> the worst case is roughly 6 days. Setting an
explicit ceiling is recommended in production.
=item reconnect_jitter => $fraction
Multiplicative jitter applied to each backoff delay: the actual sleep
is uniformly random in C<[delay, delay * (1 + jitter)]>. C<0> (default)
disables. Set to C<0.1>-C<0.5> when many clients reconnect against a
shared cluster - without jitter, every replica restart causes a
synchronised reconnect storm at the same backoff intervals. Jitter is
applied I<after> C<reconnect_max_delay> clamping, then re-clamped, so
the ceiling is never exceeded.
=item reconnect_max_attempts => $N
Cap the total number of reconnect attempts before giving up. Once the
cap is reached, C<on_error> fires with the message
C<"max reconnect attempts exceeded"> and no further attempts are made
(the user can manually call C<reset> later). Default: C<0> (unlimited
retries; be careful with permanent failures like wrong host).
=item progress_period => $seconds
Coalesce C<on_progress> packets so the callback fires at most once per
N seconds, with the per-field counters accumulated over the interval.
Useful for big SELECTs where the server can emit hundreds of progress
packets per second. Default: C<0> (fire on every packet).
=item query_log_comment => 1 | $string
Prepend a SQL block comment to every query for C<system.query_log>
traceability. C<1> auto-generates C<ev_ch user=$ENV{USER} pid=$$>;
a string is taken literally. Omit (or pass a falsy value) to disable.
Embedded C<*/> sequences are escaped to keep the comment well-formed.
=back
B<Decode options (native protocol only):>
These shape how column values are returned. All are opt-in and default
to C<0>, which returns raw numeric forms for stable round-tripping.
=over 4
=item decode_datetime => 0 | 1
Return C<Date>, C<Date32>, C<DateTime>, and C<DateTime64> as formatted
strings (e.g. C<"2024-01-15">, C<"2024-01-15 10:30:00">) instead of raw
integers. Uses UTC; columns with an explicit timezone
lib/EV/ClickHouse.pm view on Meta::CPAN
C<Nested> columns become arrays of tuples. C<LowCardinality> works
correctly across multi-block results with shared dictionaries.
=head2 insert
$ch->insert($table, $data, sub { my (undef, $err) = @_ });
$ch->insert($table, $data, \%settings, sub { my (undef, $err) = @_ });
C<$data> may be either:
=over 4
=item * A pre-formatted TabSeparated string (tabs separate columns,
newlines separate rows, with the standard ClickHouse escapes).
=item * An arrayref of arrayrefs (rows of column values).
=back
When using arrayrefs, no TSV escaping is needed: C<undef> maps to null
and strings may contain tabs and newlines freely.
Nested arrayrefs (Array/Tuple columns) and hashrefs (Map columns) are
supported B<only on the native protocol>, where the encoder has the
column type from the server's sample block. On HTTP the same call
croaks rather than silently produce malformed TSV; use the native
protocol or pre-serialise nested types into ClickHouse TSV literal form.
# Native: nested types encode directly.
$ch->insert("my_table", [
[1, "hello\tworld"], # embedded tab
[2, undef], # null
[3, [10, 20]], # Array column (native only)
[4, { a => 1, b => 2 }], # Map column (native only)
], sub { ... });
The optional C<\%settings> hashref works exactly as in L</query>,
including C<query_id>, C<query_timeout>, and C<params>. Two extra
flags are recognised here:
=over 4
=item C<idempotent =E<gt> 1 | $token>
Auto-mints (or uses the supplied) C<insert_deduplication_token>, so a
reconnect-driven retry of the same insert doesn't double-write. Falsy
values are a no-op.
=item C<async_insert =E<gt> 1>
Enables ClickHouse server-side insert batching by setting
C<async_insert=1, wait_for_async_insert=0>. Both sub-settings can be
overridden by passing them explicitly.
=back
=head2 ping
$ch->ping(sub { my ($result, $err) = @_ });
Send a no-op round trip to verify the connection is alive. On success
C<$result> is true, C<$err> is C<undef>. On error: C<(undef, $error)>.
=head2 is_healthy
$ch->is_healthy(sub { my ($ok, $err) = @_ });
$ch->is_healthy(sub { ... }, $timeout_seconds);
Bounded health probe: wraps L</ping> with a deadline (default 5s). The
callback receives C<(1, undef)> on a successful round trip, or
C<(0, $msg)> on ping error or timeout. Failure does B<not> tear down the
connection; recovery (C<reset>, host rotation, etc.) is the caller's
choice. Useful for L4 load-balancer probes and self-monitoring loops.
=head2 ping_round_trip
$ch->ping_round_trip(sub {
my ($seconds, $err) = @_;
die "ping: $err" if $err;
printf "rtt = %.3fms\n", $seconds * 1000;
});
Issue a single PING and report wall-clock latency in seconds. Lighter
than installing L</track_query_durations> for a one-shot probe;
returns C<(undef, $err)> on transport failure. Pairs well with
L</is_healthy> for health-check endpoints that want both liveness and
latency.
=head2 slow_query_log
my $prev = $ch->slow_query_log(0.1, sub {
my ($qid, $rows, $bytes, $code, $dur, $err) = @_;
warn sprintf("SLOW %.3fs %s\n", $dur, $qid // '?');
});
Filtered variant of L</on_query_complete> that fires only when the
query took at least C<$threshold> seconds. Returns the previous
C<on_query_complete> so the caller can restore it. The previous
handler is also chained on every call, so installing this on top of
existing instrumentation is safe.
=head2 server_setting
$ch->server_setting('max_threads', sub {
my ($value, $err) = @_;
warn "max_threads = $value\n";
});
Looks one value up from C<system.settings>. Convenient one-liner for
"what's the server's effective C<$x>?". Returns C<undef> via the
callback if the setting name isn't present on this server.
=head2 row_count
$ch->row_count('events', sub { ... });
$ch->row_count('events', "ts > now() - interval 1 hour", sub { ... });
C<select count() from $table [where $where]>. C<$where> is interpolated
literally; use parameterized predicates via the L</query> C<params>
mechanism for user-supplied filters. Returns the row count or
C<(undef, $err)>.
lib/EV/ClickHouse.pm view on Meta::CPAN
my ($info, $err) = @_;
die $err if $err;
for my $col (@{ $info->{columns} }) {
printf "%-20s %s\n", $col->{name}, $col->{type};
}
});
Schema introspection: issues C<describe table $name> and delivers
C<{ columns =E<gt> [{name=E<gt>..., type=E<gt>...}, ...] }> to the
callback. Useful for generic insert pipelines that need column types
without hard-coding them. C<$name> may be C<table> or C<db.table>;
non-identifier characters are rejected up-front.
=head2 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;
B<Native protocol only> - relies on the per-block C<on_data> hook and
will croak if invoked on an HTTP connection.
Synchronous-feeling pull iterator over a streaming select. Internally
wraps the native C<on_data> per-block callback and drives the EV loop
from inside C<-E<gt>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.
C<-E<gt>error>, C<-E<gt>is_done>, and C<-E<gt>cancel> are also
available on the returned iterator object.
=head2 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 C<SERVER_LOG>
packet the server emits. Useful for surfacing
C<send_logs_level =E<gt> 'information'> server-side trace events to
the application's own log stream without polling C<system.text_log>.
The row hash keys mirror the server-side log block schema; missing
keys (older revisions) come through as C<undef>.
=head2 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 L</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 C<on_query_complete>. Also accepted as a constructor
argument.
=head2 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 C<\%settings> hashref of
L</query> or L</insert>. When set, it B<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,
);
=head2 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
C<batch_size> is reached, then dispatched as a single C<insert()>.
Dispatches are serialised; push_row keeps buffering while a batch is
in flight (the native protocol cannot pipeline INSERTs). C<finish>
flushes the remaining buffer and fires its callback once all batches
complete; if any batch failed the first error is delivered as
C<$err>. The streamer also offers C<buffered_count> and C<in_flight>
accessors for backpressure logic.
C<<< $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 B<not> touch the underlying
C<$ch> - any batch already on the wire still completes normally. Any
callback registered via C<finish> or C<await_drain> that has not yet
fired is invoked with a C<'streamer reset'> error rather than being
lib/EV/ClickHouse.pm view on Meta::CPAN
async chain.
$pool->with_session(sub {
my ($ch, $release) = @_;
$ch->query("create temporary table t (n UInt32)", sub {
$ch->query("insert into t values (1),(2),(3)", sub {
$ch->query("select sum(n) from t", sub {
my ($rows) = @_;
say $rows->[0][0];
$release->();
});
});
});
});
C<<< $pool->query_to($idx, $sql, $cb) >>> /
C<<< $pool->insert_to($idx, $table, $data, $cb) >>> force-routes a
call to a specific member without going through C<_pick>. Circuit
breaker observation still applies (success/failure is recorded
against that member). Useful for replica-targeted DDL, S3 ingest
that has to land on a chosen node, or sticky-affinity reads.
C<<< $pool->nominate($idx) >>> returns the underlying connection so
subsequent calls bypass the pool entirely. Use sparingly - calls
made directly on the nominated connection don't update the
circuit-breaker state.
C<<< $pool->hedged_query($sql, hedge =E<gt> 2, $cb) >>> dispatches
the same select to C<hedge> distinct random members and resolves
with whichever returns first. The callback receives
C<($rows, undef, $member_idx)> on success (so callers can attribute
wins per member) or C<(undef, $err)> if I<every> member fails.
Extra completions after the winner are silently discarded.
Recommended for tail-latency-sensitive selects on replicated tables.
B<Do not> use for insert - would silently double-write when the
server's dedupe window misses.
C<<< $pool->fan_out($sql, $cb) >>> sends the same select to I<every>
member and collects per-member results into one arrayref:
$pool->fan_out("select hostName(), uptime()", sub {
for my $r (@{ $_[0] }) {
printf "[%d] err=%s rows=%s\n",
$r->{member}, $r->{err} // '-',
$r->{rows} ? scalar @{$r->{rows}} : '-';
}
});
Useful for shard-aware diagnostics (per-replica lag, distinct
C<system.*> values across the pool). Errors are per-member, not
aggregated - the callback always fires with a complete list. Pass
C<settings =E<gt> \%h> for per-query options.
B<Circuit breaker:> pass C<circuit_threshold =E<gt> N> at construction
to enable per-member fail-fast. After N consecutive query/insert/ping
errors on a given member, that member is excluded from C<_pick> for
C<circuit_cooldown> seconds (default 30). A successful callback resets
the per-member fail counter. If every member is dead at pick time the
breaker is bypassed so the next attempt still has a chance to recover.
Inspect with C<$pool-E<gt>circuit_state> which returns one
C<{ fails =E<gt> N, dead_until =E<gt> $epoch, alive =E<gt> 0|1 }>
hashref per member.
B<Graceful shutdown:> C<<< $pool->shutdown($grace_seconds, $cb) >>>
drains every member, then calls C<finish> on each. If C<$grace_seconds>
elapses before every member drains, members still in flight are
force-finished and C<$cb> receives the string
C<"Pool::shutdown timed out after Ns">. On a clean shutdown C<$cb>
receives undef. C<$grace_seconds> may be 0 (or undef) to wait
indefinitely. The callback fires exactly once.
$SIG{TERM} = sub { $pool->shutdown(10, sub { EV::break }) };
=head1 LIFECYCLE
=head2 finish
$ch->finish;
Close the connection. Pending queries receive an error callback. Aliased
as C<disconnect>.
=head2 reset
$ch->reset;
Disconnect and immediately reconnect using the original parameters.
Aliased as C<reconnect>.
=head2 drain
$ch->drain(sub { ... });
Register a callback to fire once all pending queries (queued + in-flight)
have completed. If nothing is pending, the callback fires synchronously.
The classic graceful-shutdown pattern:
$ch->query("select 1", sub { ... });
$ch->query("select 2", sub { ... });
$ch->drain(sub {
$ch->finish;
EV::break;
});
=head2 cancel
$ch->cancel;
Cancel the currently in-flight query. Native protocol sends CLIENT_CANCEL
and waits for the server's EndOfStream/Exception; HTTP closes the connection
(use C<auto_reconnect> or call L</reset> to recover). The query's callback
receives an error.
=head2 skip_pending
$ch->skip_pending;
Drop every pending operation: each queued and in-flight callback is invoked
with C<(undef, $error_message)>. If a request was on the wire, the connection
is torn down; call L</reset> (or rely on C<auto_reconnect>) before issuing
new queries.
lib/EV/ClickHouse.pm view on Meta::CPAN
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" }
});
=item Idempotent insert silently drops some rows
C<<< idempotent =E<gt> 1 >>> auto-mints
C<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 C<<< idempotent =E<gt> $token >>> per batch or
omit the option for fresh inserts. See F<eg/idempotent_insert.pl>.
=item C<on_data> vs C<iterate> - which should I pick?
C<<< on_data =E<gt> 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. C<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.
=item Connection in front of nginx / reverse proxy strips X-ClickHouse-* headers
Pass C<<< http_basic_auth =E<gt> 1 >>> to send the credentials as
C<Authorization: Basic ...> instead. Most HTTP gateways forward
Authorization verbatim while filtering proprietary headers.
=back
=head1 TUNING
=over 4
=item 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 C<raw =E<gt> 1> CSV /
JSONEachRow / Parquet bodies.
=item C<compress =E<gt> 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.
=item C<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.
=item C<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.
=item C<reconnect_max_attempts>
Always set in production. Default is unlimited; a permanent failure
(wrong host, wrong port, dead server) will spin C<on_error> forever
otherwise.
=item C<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.
=item Pull-iterator vs C<on_data>
C<on_data> has lower per-block overhead. C<iterate> trades that for a
synchronous-feeling API; use it when the surrounding code is procedural.
=item C<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, C<set>) since each query may
land on a different connection.
=back
=head2 Performance tuning checklist
=over 4
=item 1. Pick the right protocol
Native (port 9000) beats HTTP (port 8123) for almost all workloads.
HTTP is only required for HTTPS-fronted ingress, the C<raw> mode that
returns C<RowBinary> / C<JSONEachRow> / C<Parquet> bodies unparsed, or
gateway authentication that strips proprietary CH headers (see
C<http_basic_auth>).
=item 2. Tune C<batch_size> for INSERTs
Aim for ~1 MB per batch. ClickHouse merges every block into a part on
disk, so 1k blocks of 1k rows each is dramatically slower than 1 block
of 1M rows because of merge amplification. C<insert_streamer> with
C<batch_size =E<gt> $rows_for_1MB> + C<high_water> backpressure is the
production-grade default.
=item 3. Cap C<max_recv_buffer>
Without a cap, a runaway select (or a buggy upstream that returns
gigabytes) will grow the recv buffer until the process is OOM-killed.
Set C<max_recv_buffer =E<gt> 64 * 1024 * 1024> (64 MB) and let the
parser tear the connection down with a clean error if exceeded - the
caller's on_error can decide whether to retry or surface to the user.
=item 4. Watch for head-of-line blocking
( run in 0.761 second using v1.01-cache-2.11-cpan-bbc515a03b3 )