/ 
EV-ClickHouse-0.03
/ README.md

NAME

EV::ClickHouse - Async ClickHouse client using EV

SYNOPSIS

use EV;
use EV::ClickHouse;

# Discrete parameters
my $ch = EV::ClickHouse->new(
    host       => '127.0.0.1',
    port       => 9000,
    protocol   => 'native',     # or 'http'
    user       => 'default',
    password   => '',
    database   => 'default',
    settings   => { max_threads => 4 },  # connection-level defaults
    on_connect => sub { print "connected\n" },
    on_error   => sub { warn "error: $_[0]\n" },
);

# Or via URI: clickhouse[+native]://user:pass@host:port/db?key=val
my $ch = EV::ClickHouse->new(
    uri        => 'clickhouse+native://default:@127.0.0.1:9000/default',
    on_connect => sub { ... },
);

# select
$ch->query("select number from system.numbers limit 3", sub {
    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;

DESCRIPTION

EV::ClickHouse is an asynchronous ClickHouse client that integrates with the 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.

Features

CONSTRUCTOR

new

my $ch = EV::ClickHouse->new(%args);

The connection is initiated immediately; new returns before it completes. Queries issued before on_connect fires are queued and dispatched once the connection is ready.

Connection parameters:

Callbacks:

Options:

Decode options (native protocol only):

These shape how column values are returned. All are opt-in and default to 0, which returns raw numeric forms for stable round-tripping.

METHODS

query

$ch->query($sql, sub { my ($rows, $err) = @_ });
$ch->query($sql, \%settings, sub { my ($rows, $err) = @_ });

Executes a SQL statement. The callback receives:

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:

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:

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 supported 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 \%settings hashref works exactly as in "query", including query_id, query_timeout, and params. Two extra flags are recognised here:

ping

$ch->ping(sub { my ($result, $err) = @_ });

Send a no-op round trip to verify the connection is alive. On success $result is true, $err is undef. On error: (undef, $error).

is_healthy

$ch->is_healthy(sub { my ($ok, $err) = @_ });
$ch->is_healthy(sub { ... }, $timeout_seconds);

Bounded health probe: wraps "ping" with a deadline (default 5s). The callback receives (1, undef) on a successful round trip, or (0, $msg) on ping error or timeout. Failure does not tear down the connection; recovery (reset, host rotation, etc.) is the caller's choice. Useful for L4 load-balancer probes and self-monitoring loops.

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 "track_query_durations" for a one-shot probe; returns (undef, $err) on transport failure. Pairs well with "is_healthy" for health-check endpoints that want both liveness and latency.

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 "on_query_complete" that fires only when the query took at least $threshold seconds. Returns the previous 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.

server_setting

$ch->server_setting('max_threads', sub {
    my ($value, $err) = @_;
    warn "max_threads = $value\n";
});

Looks one value up from system.settings. Convenient one-liner for "what's the server's effective $x?". Returns undef via the callback if the setting name isn't present on this server.

row_count

$ch->row_count('events', sub { ... });
$ch->row_count('events', "ts > now() - interval 1 hour", sub { ... });

select count() from $table [where $where]. $where is interpolated literally; use parameterized predicates via the "query" params mechanism for user-supplied filters. Returns the row count or (undef, $err).

table_size

$ch->table_size('events', sub {
    my ($info, $err) = @_;
    # $info = { rows => N, bytes_on_disk => N, data_uncompressed_bytes => N }
});

Sums system.parts for the (optionally database-qualified) table and returns a hashref. Active parts only - does not count detached parts. Suitable for ops dashboards; not authoritative for per-row billing (parts may double-count rows during MERGE).

ddl

$ch->ddl("create table t (n UInt32) engine=Memory", sub {
    my (undef, $err) = @_; die "ddl: $err" if $err;
});

Strict variant of "query" for DDL/DML. Identical wire behaviour; the separate name is a readability marker for migration scripts so the intent of each call is obvious at the call site.

dictionary_reload

$ch->dictionary_reload('my_dict', $cb);

Shortcut for system reload dictionary my_dict. The dictionary name is validated against [A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)? before splicing into the SQL.

refresh_view

$ch->refresh_view('mv_aggregated_hourly', $cb);

Shortcut for system refresh view mv_aggregated_hourly (requires ClickHouse 23.12+). Same name validation as "dictionary_reload".

wait_mutation

$ch->query("alter table events update tag = 'x' where id = 7", sub {
    my (undef, $err) = @_;
    die $err if $err;
    $ch->wait_mutation('events', sub {
        my ($info, $err) = @_;
        die $err if $err;       # a mutation failed, or timed out
        print "mutation done\n";
    }, poll => 0.5, timeout => 30);
});

ALTER TABLE ... UPDATE/DELETE runs asynchronously on the server; the statement returns before the mutation has been applied. wait_mutation polls system.mutations until the table has no incomplete mutations, then fires $cb->({ pending => 0 }).

A mutation that keeps failing stays incomplete with a latest_fail_reason - once that reason persists across consecutive polls (a single transient failure that the mutation's next retry clears is tolerated) it is surfaced as $cb->(undef, "wait_mutation: ..."). Options:

Table-name validation matches "for_table"; a db.table name also filters system.mutations by database.

parse_uri

my $parsed = EV::ClickHouse->parse_uri(
    'clickhouse+native://u:p@host:9000/db?compress=1'
);
# $parsed = {
#   protocol => 'native', host => 'host', port => 9000,
#   user => 'u', password => 'p', database => 'db',
#   compress => '1',
# };
my $ch = EV::ClickHouse->new(%$parsed, on_connect => sub { ... });

Class method that parses a ClickHouse URI into a hash, returning undef if the URI doesn't match the expected shape. Lets tooling validate user-supplied URIs without opening a connection. Query- string keys are flattened to top-level args so the result drops straight into "new"; matches the inline URI parser there.

is_retryable_error

EV::ClickHouse->is_retryable_error($code)   # class method
$ch->is_retryable_error($code)              # also works on instance

Returns true if the given ClickHouse error code (as reported by "last_error_code" or the per-query $err argument's prefix) is a 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"; 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" or "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.

$streamer->columns_from_table($cb) looks up the target table's column list via "for_table" and stores it as the streamer's named-row columns, so callers can construct a streamer without knowing the schema in advance. The callback fires once the lookup completes (undef on success, an error string on failure).

my $s = $ch->insert_streamer('events');
$s->columns_from_table(sub {
    my ($err) = @_;
    die "describe: $err" if $err;
    $s->push_row({ ts => time, user_id => 7, action => 'click' });
    ...
});

insert_iter

$ch->insert_iter('events', sub {
    # producer: return next row (arrayref or hashref) or undef when done
    return undef unless my $row = next_event();
    return $row;
}, sub {
    my (undef, $err) = @_;
    die "ingest: $err" if $err;
}, batch_size => 5_000, columns => [qw(ts user_id action payload)]);

Generator-driven insert. Internally wraps "insert_streamer" with an EV::idle pump that calls $producer repeatedly, respecting high_water for backpressure. undef from the producer signals end-of-stream and triggers finish. %opts is forwarded to "insert_streamer".

kill_query

$ch->kill_query($query_id, sub {
    my ($rows, $err) = @_;
    warn "kill: $err" if $err;
});

Shortcut for kill query where query_id = '...' sync. Validates $query_id against [A-Za-z0-9_-]+ before interpolating into the SQL. Pass async => 1 to send ASYNC instead of SYNC when fire-and-forget semantics are wanted.

cancel_by_query_id

$ch->cancel_by_query_id($qid);

Race-safe local cancel: only triggers "cancel" if the connection's current in-flight query (last_query_id) matches $qid, so the caller can't accidentally kill a different query that has already started in the meantime. Returns 1 if it cancelled, 0 if the id no longer matched.

retry

$ch->retry($sql,
    retries => 3,
    backoff => 0.5,        # initial delay; doubles each attempt
    jitter  => 0.25,       # add 0..25% random jitter to each wait
    cb      => sub { my ($rows, $err) = @_; ... },
);

Retry a select (or any read-only statement) on the same connection over exponential backoff, but only when the server-side error is in the "is_retryable_error" set (timeouts, memory limits, replica catch-up, etc.). The callback fires exactly once with the final result - either the first success or the error from the last attempt. Per-attempt settings => \%h are honoured. jitter => $fraction adds up to $fraction * current_wait randomness on top of each exponential step, so the spread scales with the backoff window.

Not for insert - a partial-write that the server logged but didn't acknowledge will be re-applied; use idempotent => 1 on the underlying "insert" instead.

insert_async

$ch->insert_async('events', \@rows, sub { ... });
$ch->insert_async('events', \@rows, sub { ... }, wait_for_flush => 0);

Ergonomic wrapper around server-side async_insert => 1. Defaults to wait_for_flush => 1 (the callback fires after the batch has flushed); pass 0 for fire-and-forget. Pass additional settings via settings => \%h.

insert_aggregated

$ch->insert_aggregated(
    'metrics_agg',
    agg_col  => { func => 'uniqExact', args => ['UInt64'] },
    key_cols => [qw(ts site)],
    rows     => [[1700000000, 'a', 42], ...],
    cb       => sub { my (undef, $err) = @_; ... },
);

Generic client-side state serialization for AggregateFunction columns isn't feasible (each aggregator has its own binary format), so this helper builds insert into t (cols) select k as k, funcState(cast(v as T)) as agg union all ... - one single-row select per row in rows, wrapping each per-row value with the ${func}State combinator so the server constructs the aggregate state. Validates table, column, and function names; per-row values are quoted via a small SQL-literal helper (numbers raw, strings escape-quoted, undef becomes NULL). Only scalar leaf values are supported (no nested arrays).

bind_ident

my $sql = "select * from " . EV::ClickHouse->bind_ident($table);

Backtick-quote an identifier safely for SQL splicing. Accepts simple or dotted (db.table) identifiers matching [A-Za-z_][A-Za-z0-9_]*; anything else croaks. The regex rejects backticks outright so no escaping is needed inside the result.

track_query_durations

$ch->track_query_durations(1024);
say $ch->query_duration_p(0.95);

Install a fixed-size ring buffer of recent query durations (sourced from on_query_complete's $duration_s argument). Composes with a user-supplied on_query_complete (preserved, called first). Pass 0 to disable and restore the previous handler. Subsequent calls replace the previous ring size.

query_duration_p($p) returns the $p-quantile in seconds ($p in [0,1]); query_duration_count returns the number of samples currently buffered.

pending_queries

for my $q (@{ $ch->pending_queries }) {
    printf "%s %s age=%.3fs\n",
           $q->{state}, $q->{query_id} // '-', $q->{age};
}

Snapshot of pending queries: returns arrayref of hashrefs. The head of the in-flight queue (if any) appears first with state => 'in_flight', query_id => last_query_id, and age => seconds since dispatch. Queued entries follow with state => 'queued' and age => 0 (they have no dispatch time yet). SQL/settings are not retained after enqueue and so are not included.

dump_state

my $h = $ch->dump_state;
# { connected, connecting, dns_pending, pending_count,
#   callback_depth, send_len/pos/cap, recv_len/cap, fd,
#   protocol, server_revision, reconnect_attempts,
#   host, port, send_count, compress, tls }

Read-only diagnostic snapshot of internal struct state. Intended for debugging stuck connections; field set may shift between releases (don't script against it in production).

for_json_paths

$ch->for_json_paths('events', 'payload', sub {
    my ($paths, $err) = @_;
    for my $p (@$paths) { say "$p->{path} : $p->{type}" }
});

Discovers the dynamic JSON path layout of a JSON/Object('json') column. Internally walks the Map(String, String) returned by JSONAllPathsWithTypes(col) with a single arrayJoin(mapKeys(m)) (the map alias is preserved so each path's type is correlated via a second lookup), dedupes, sorts by path, and returns [ { path => 'a.b.c', type => 'Int64' }, ... ]. Useful for monitoring schema drift on weakly-typed columns.

EV::ClickHouse::Pool

my $pool = EV::ClickHouse::Pool->new(
    host => 'ch', port => 9000, protocol => 'native',
    size => 8,                # other %args pass through to ::new
);
$pool->query($sql, $cb);
$pool->insert($table, $data, $cb);
$pool->drain(sub { ... });    # all connections drained
$pool->finish;

Built-in connection pool. Each member is an independent EV::ClickHouse with its own auto_reconnect, send queue, and in-flight callback queue, so a hung query on one connection doesn't block the others. Dispatch picks the least-busy connection; ties are broken round-robin.

The Pool exposes per-pick dispatch via query, insert, ping, for_table, iterate, insert_streamer; aggregate stats via size, pending_count, conns (the underlying connection list); and broadcast lifecycle methods drain, finish, cancel, skip_pending, reset (each affects every member because the state they touch is owned per connection, not per query). The broadcast cancel, skip_pending, and reset methods wrap each per-member call in eval so a member that croaks doesn't abort the broadcast; per-member errors are silently discarded (the surviving members still receive the call). Iterate conns yourself if you need per-member error handling.

$pool->with_each(sub { my ($conn, $idx) = @_; ... }) calls $cb once per member, passing the connection object and its index. Each per-member call is wrapped in eval so a single croak does not abort the iteration; per-member errors are silently discarded - wrap the body yourself if you need them. Useful for one-off per-member work that doesn't justify a new broadcast method (e.g. resetting a counter, asking each member for last_error_code, kicking off a custom probe).

Queries that need server-side state (temporary tables, session variables) must use a single connection, not a Pool, since successive calls may land on different members.

$pool->with_session(sub { my ($conn, $release) = @_; ... }) checks out a least-busy member and "pins" it for the duration of the callback: while pinned, _pick avoids that member when other callers request a connection (it remains selectable as a fallback if every other member is unavailable). The callback must call $release->() when its multi-query sequence completes - typically from the innermost query's callback so the pin lasts across the 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->();
            });
        });
    });
});

$pool->query_to($idx, $sql, $cb) / $pool->insert_to($idx, $table, $data, $cb) force-routes a call to a specific member without going through _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.

$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.

$pool->hedged_query($sql, hedge => 2, $cb) dispatches the same select to hedge distinct random members and resolves with whichever returns first. The callback receives ($rows, undef, $member_idx) on success (so callers can attribute wins per member) or (undef, $err) if every member fails. Extra completions after the winner are silently discarded. Recommended for tail-latency-sensitive selects on replicated tables. Do not use for insert - would silently double-write when the server's dedupe window misses.

$pool->fan_out($sql, $cb) sends the same select to 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 system.* values across the pool). Errors are per-member, not aggregated - the callback always fires with a complete list. Pass settings => \%h for per-query options.

Circuit breaker: pass circuit_threshold => 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 _pick for 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 $pool->circuit_state which returns one { fails => N, dead_until => $epoch, alive => 0|1 } hashref per member.

Graceful shutdown: $pool->shutdown($grace_seconds, $cb) drains every member, then calls finish on each. If $grace_seconds elapses before every member drains, members still in flight are force-finished and $cb receives the string "Pool::shutdown timed out after Ns". On a clean shutdown $cb receives undef. $grace_seconds may be 0 (or undef) to wait indefinitely. The callback fires exactly once.

$SIG{TERM} = sub { $pool->shutdown(10, sub { EV::break }) };

LIFECYCLE

finish

$ch->finish;

Close the connection. Pending queries receive an error callback. Aliased as disconnect.

reset

$ch->reset;

Disconnect and immediately reconnect using the original parameters. Aliased as reconnect.

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;
});

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 auto_reconnect or call "reset" to recover). The query's callback receives an error.

skip_pending

$ch->skip_pending;

Drop every pending operation: each queued and in-flight callback is invoked with (undef, $error_message). If a request was on the wire, the connection is torn down; call "reset" (or rely on auto_reconnect) before issuing new queries.

ACCESSORS

All per-query accessors (column_names, column_types, last_query_id, last_error_code, last_totals, last_extremes, profile_rows, profile_bytes, profile_rows_before_limit) are reset at the moment a new query is dispatched (queued or sent), not when its callback fires. It is always safe to read them inside the query's own callback. Reading them after dispatching a subsequent query but before its callback fires returns the initial state (0 or undef), never the previous query's data. Connection-level accessors (is_connected, server_info, server_version, server_timezone, pending_count) are unaffected.

ALIASES

q          -> query
ddl        -> query
reconnect  -> reset
disconnect -> finish

REQUIREMENTS

TROUBLESHOOTING

TUNING

Performance tuning checklist

ARCHITECTURE

The client is a single state machine driven by an EV event loop. Each connection holds: a TCP fd (non-blocking), a send buffer, a receive buffer, a callback queue (next-in-line per protocol), and a pending send queue (buffered before connect).

State transitions:

Connect TCP --> [TLS handshake] --> [Native ServerHello]
    --> Connected --> { dispatch from send_queue;
                        parse response; deliver via cb_queue }

The connect_timeout timer covers all three pre-Connected stages. auto_reconnect re-runs the chain via schedule_reconnect.

Two key invariants:

For deeper detail (state-machine table, queue semantics) see CLAUDE.md in the source distribution.

TYPES

Per-column wire format and Perl-side gotchas. All numeric types round-trip stable raw values by default; opt into string forms via decode_datetime, decode_decimal, decode_enum.

COOKBOOK

The eg/ directory in the source distribution carries runnable patterns for the common production shapes. Each one is self-contained and reads top-to-bottom.

EV::ClickHouse::Error

Lightweight error class. Callbacks always receive ($result, $err_msg) plain strings (preserved for compatibility); this object is opt-in via EV::ClickHouse::Error->from_ch($ch, $err) when callers want structured access to the code, symbolic name, and retryability of the last server-side exception.

my $e = EV::ClickHouse::Error->from_ch($ch, $err) or return;
if ($e->is_retryable) { schedule_retry() }
elsif ($e->code == 60) {                              # UNKNOWN_TABLE
    warn "table missing: $e";                          # stringifies to msg
}

The object overloads stringification to return the message, so legacy callsites that string-compare or interpolate $err keep working verbatim when the error is wrapped.

SEE ALSO

AUTHOR

vividsnow

LICENSE

This library is free software; you can redistribute it and/or modify it under the same terms as Perl itself.