File Coverage

blib/lib/EV/ClickHouse.pm

Criterion	Covered	Total	%
statement	23	778	2.9
branch	0	418	0.0
condition	0	334	0.0
subroutine	8	123	6.5
pod	25	27	92.5
total	56	1680	3.3

package EV::ClickHouse;
use strict;
use warnings;

use EV;
use Scalar::Util qw(refaddr weaken);

# Identifier validation regexes — single source of truth for the table
# / column / function names that get spliced into SQL via the helpers.
my $RE_IDENT = qr/\A[A-Za-z_][A-Za-z0-9_]*\z/;
my $RE_TABLE = qr/\A[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)?\z/;

BEGIN {
    our $VERSION = '0.03';
    use XSLoader;
    XSLoader::load __PACKAGE__, $VERSION;
}

# Holds in-flight EV::cares resolvers so they aren't garbage-collected
# before their callback fires. Keyed by refaddr of the resolver itself
# (not the connection) — see new() for the rationale. Each entry is
# deleted from inside its own resolved callback via a deferred timer.
# Plain package hash (not Hash::Util::FieldHash) so module load doesn't
# add tied/magic SVs that Test::LeakTrace would flag.
our %_failover;

*q          = \&query;
*reconnect  = \&reset;
*disconnect = \&finish;
*ddl        = \&query;  # readability at call sites for DDL/DML

sub _uri_unescape { my $s = $_[0]; $s =~ s/%([0-9A-Fa-f]{2})/chr(hex($1))/ge; $s }

# Parse a URI query string into a hash. Bare keys (no `=`) are stored
# as 1 (standard URL flag convention); existing keys in $h are preserved
# (path-derived host/port/etc win over query-string overrides).
sub _uri_qs_into {
    my ($qs, $h) = @_;
    return unless defined $qs && length $qs;
    for my $pair (split /&/, $qs) {
        my ($k, $v) = split /=/, $pair, 2;
        next unless defined $k && length $k;
        $h->{$k} //= defined $v ? _uri_unescape($v) : 1;
    }
}

sub new {
    my ($class, %args) = @_;

    # Connection URI: clickhouse://user:pass@host:port/database
    # host accepts a bracketed IPv6 literal (e.g. clickhouse://[::1]:9000/db)
    if (my $uri = delete $args{uri}) {
        if ($uri =~ m{^clickhouse(?:\+(\w+))?://(?:([^:@]*?)(?::([^@]*))?\@)?(\[[^\]]+\]|[^/:?]+)(?::(\d+))?(?:/([^?]*))?(?:\?(.*))?$}) {
            my ($proto, $u, $pw, $h, $p, $db, $qs) = ($1, $2, $3, $4, $5, $6, $7);
            $h =~ s/^\[(.*)\]$/$1/;
            $args{protocol} //= $proto if $proto;
            $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;
    };
    my %s = $settings ? %$settings : ();
    $s{on_data} = $on_data;
    $self->query($sql, \%s, sub {
        my (undef, $err) = @_;
        $it->{done} = 1;
        $it->{err}  = $err if $err;
        EV::break;
    });
    $it;
}

# Streaming insert: $s = $ch->insert_streamer($table, %opts)
#   $s->push_row([...]); ...; $s->finish(sub { my (undef, $err) = @_ });
# Buffers rows in batches of `batch_size` (default 10_000) and dispatches
# each batch as an insert(). Dispatches are serialised (the native protocol
# cannot pipeline INSERTs); push_row keeps buffering while a batch is in
# flight and the next batch fires from the in-flight callback.
sub insert_streamer {
    my ($self, $table, %opts) = @_;
    return EV::ClickHouse::Streamer->_new($self, $table, %opts);
}

# Generator-driven insert. $producer is a code ref returning one row
# per call (arrayref or hashref) and undef when exhausted. Rows pump
# into an insert_streamer with backpressure: when buffered_count
# crosses high_water (default 50_000) the producer is paused until
# the streamer drains below the watermark. $cb fires once with
# (undef) on success or (undef, $err) on first failure (same shape
# as Streamer::finish).
sub insert_iter {
    my ($self, $table, $producer, $cb, %opts) = @_;
    die "Usage: \$ch->insert_iter(\$table, \$producer, \$cb, [\%opts])"
        unless defined $table && ref($producer) eq 'CODE' && ref($cb) eq 'CODE';
    my $hi = $opts{high_water} //= 50_000;
    my $s  = $self->insert_streamer($table, %opts);
    my $pump; $pump = EV::idle(sub {
        # Abort early on sticky error so we don't keep pumping rows into
        # a streamer that will never accept them (and that would otherwise
        # accumulate the buffer indefinitely if the producer is fast).
        if (my $err = $s->sticky_error) {
            $pump->stop; undef $pump;
            $s->finish($cb);
            return;
        }
        # Backpressure: skip this tick if the buffer is at the watermark.
        # In-flight batches are handled inside the streamer (push_row keeps
        # buffering while one is on the wire), so we only gate on buffer
        # size here. $hi == 0 means the caller disabled backpressure — the
        # gate must then never fire (>= 0 would be permanently true).
        return if $hi && $s->buffered_count >= $hi;
        my $row = $producer->();
        if (defined $row) {
            $s->push_row($row);
        } else {
            $pump->stop;
            undef $pump;
            $s->finish($cb);
        }
    });
    return;
}

# Ergonomic wrapper around server-side async_insert. Defaults to
# wait_for_flush => 1 so the callback fires after the asynchronous batch
# has been committed, matching the wait_for_async_insert=1 semantics in
# clickhouse-server. Set wait_for_flush => 0 for fire-and-forget (the
# callback then resolves as soon as the server has accepted the row
# into the in-memory async batch).
#
#   $ch->insert_async('events', \@rows, sub {
#       my (undef, $err) = @_;
#       die "async insert: $err" if $err;
#   });
sub insert_async {
    my ($self, $table, $data, $cb, %opts) = @_;
    die "Usage: \$ch->insert_async(\$table, \$data, \$cb, [%opts])"
        unless defined $table && defined $data && ref($cb) eq 'CODE';
    my $wait     = exists $opts{wait_for_flush} ? delete($opts{wait_for_flush}) : 1;
    my %extra    = %{ delete($opts{settings}) // {} };
    my %settings = (
        async_insert          => 1,
        wait_for_async_insert => $wait ? 1 : 0,
        %extra,
    );
    $self->insert($table, $data, \%settings, $cb);
}

# Server-side insert into AggregateFunction columns. Per-aggregator
# state binary format isn't replicable client-side, so each per-row
# value is wrapped in the ${func}State combinator inside a single-row
# select, all union all'd into one insert.
sub insert_aggregated {
    my ($self, $table, %opts) = @_;
    my $cb       = delete $opts{cb}   // die "insert_aggregated: cb required";
    my $rows     = delete $opts{rows} // die "insert_aggregated: rows required";
    die "insert_aggregated: cb must be a coderef" if ref($cb) ne 'CODE';
    die "insert_aggregated: rows must be a non-empty arrayref"
        if ref($rows) ne 'ARRAY' || !@$rows;
    my $key_cols = delete $opts{key_cols} || [];
    die "insert_aggregated: key_cols must be an arrayref"
        if ref($key_cols) ne 'ARRAY';
    my @agg = grep { ref $opts{$_} eq 'HASH' } sort keys %opts;
    die "insert_aggregated: no aggregate columns specified" unless @agg;
    die "insert_aggregated: invalid table '$table'"
        unless $table =~ $RE_TABLE;
    for my $c (@$key_cols, @agg) {
        die "insert_aggregated: invalid column '$c'"
            unless $c =~ $RE_IDENT;
    }
    my @agg_meta;
    for my $ac (@agg) {
        my $spec = $opts{$ac};
        my $func = $spec->{func} // die "insert_aggregated: $ac.func required";
        my $args = $spec->{args} // die "insert_aggregated: $ac.args required";
        die "insert_aggregated: $ac.func must be a simple identifier"
            unless $func =~ $RE_IDENT;
        for my $t (@$args) {
            die "insert_aggregated: $ac arg type contains illegal chars"
                unless $t =~ /\A[A-Za-z0-9_(),\s]+\z/;
        }
        push @agg_meta, { col => $ac, func => $func, types => [@$args] };
    }
    my $expected_cols = @$key_cols;
    $expected_cols += @{ $_->{types} } for @agg_meta;
    for my $r (@$rows) {
        die sprintf("insert_aggregated: row width mismatch (got %d, want %d)",
                    scalar @$r, $expected_cols)
            if @$r != $expected_cols;
    }
    my $col_list = join ', ', @$key_cols, @agg;
    # VALUES doesn't permit aggregate combinators (treated as non-constant),
    # so build a UNION ALL of single-row SELECTs.
    my @selects;
    for my $r (@$rows) {
        my $offset = 0;
        my @parts;
        for my $kc (@$key_cols) {
            push @parts, _sql_quote_value($r->[$offset++]) . " as $kc";
        }
        for my $m (@agg_meta) {
            my @args;
            for my $t (@{ $m->{types} }) {
                push @args, "cast(" . _sql_quote_value($r->[$offset++]) . " as $t)";
            }
            push @parts, "$m->{func}State(" . join(', ', @args) . ") as $m->{col}";
        }
        push @selects, "select " . join(', ', @parts);
    }
    my $sql = "insert into $table ($col_list) " . join(' union all ', @selects);
    $self->query($sql, $cb);
}

# Minimal SQL literal quoter for insert_aggregated. Numbers go raw,
# strings get single-quoted with embedded ' and \ escaped, undef → NULL.
sub _sql_quote_value {
    my ($v) = @_;
    return 'null' unless defined $v;
    die "insert_aggregated: cannot quote a reference (got " . ref($v) . ")"
        if ref $v;
    return $v if $v =~ /\A-?(?:0|[1-9][0-9]*)(?:\.[0-9]+)?(?:[eE][-+]?[0-9]+)?\z/;
    my $s = $v;
    $s =~ s/\\/\\\\/g;
    $s =~ s/'/\\'/g;
    return "'$s'";
}

sub kill_query {
    my ($self, $query_id, $cb, %opts) = @_;
    die "Usage: \$ch->kill_query(\$query_id, \$cb, [%opts])"
        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) = @_; ... });
sub retry {
    my ($self, $sql, %opts) = @_;
    my $cb       = delete $opts{cb}       // die "retry: cb required";
    my $tries    = delete $opts{retries}  // 3;
    my $delay    = delete $opts{backoff}  // 0.25;
    my $jitter   = delete $opts{jitter}   // 0;
    my $settings = delete $opts{settings};
    die "retry: cb must be a coderef" if ref($cb) ne 'CODE';

    my $attempt = 0;
    my $timer;
    my $dispatch;
    $dispatch = sub {
        $attempt++;
        my $on_done = sub {
            my ($rows, $err) = @_;
            if (!$err || $attempt > $tries
                || !EV::ClickHouse->is_retryable_error($self->last_error_code)) {
                undef $dispatch;     # collapse the cycle on terminal call
                return $cb->($rows, $err);
            }
            my $wait = $delay * (2 ** ($attempt - 1));
            $wait += rand($wait * $jitter) if $jitter > 0;
            $timer = EV::timer($wait, 0, sub { undef $timer; $dispatch->() });
        };
        # Synchronous query() can croak (e.g. "not connected" if the
        # connection was finish()d between attempts). Catch and route to
        # the user cb instead of letting it escape into the event loop.
        my $ok = eval {
            if ($settings) { $self->query($sql, $settings, $on_done) }
            else           { $self->query($sql, $on_done) }
            1;
        };
        if (!$ok) {
            my $e = $@ || 'unknown error';
            undef $dispatch;
            $cb->(undef, $e);
        }
    };
    $dispatch->();
    return;
}

# Schema introspection: $ch->for_table('db.t', sub { my ($info, $err) = @_; ... })
# Delivers { columns => [{name=>..., type=>...}, ...] } or (undef, $err).
sub for_table {
    my ($self, $table, $cb) = @_;
    die "Usage: \$ch->for_table(\$table, \$cb)" unless defined $table && ref($cb) eq 'CODE';
    die "for_table: invalid table name '$table'"
        unless $table =~ $RE_TABLE;
    # HTTP needs an explicit FORMAT; native ignores it and returns
    # typed rows that respect the connection's named_rows setting.
    my $sql = "describe table $table";
    $sql .= " format TabSeparated" if !$self->server_revision;
    $self->query($sql, sub {
        my ($rows, $err) = @_;
        return $cb->(undef, $err) if $err;
        # Connection may be in named_rows mode, in which case each row
        # is a hashref keyed by the DESCRIBE column names. Handle both.
        my @cols = map {
            ref $_ eq 'HASH'
                ? { name => $_->{name}, type => $_->{type} }
                : { name => $_->[0],    type => $_->[1] };
        } @{ $rows // [] };
        $cb->({ columns => \@cols });
    });
}

# Discover the dynamic JSON path layout for a JSON/Object('json') column.
# Walks the Map(String, String) returned by JSONAllPathsWithTypes, deduped
# across the table.
sub for_json_paths {
    my ($self, $table, $column, $cb) = @_;
    die "Usage: \$ch->for_json_paths(\$table, \$column, \$cb)"
        unless defined $table && defined $column && ref($cb) eq 'CODE';
    die "for_json_paths: invalid table '$table'"
        unless $table  =~ $RE_TABLE;
    die "for_json_paths: invalid column '$column'"
        unless $column =~ $RE_IDENT;
    # Single arrayJoin to avoid Cartesian; reference the alias to look
    # up the type so each (path, type) pair stays correlated.
    my $sql = "select distinct path, m[path] as type from ("
            . " select m, arrayJoin(mapKeys(m)) as path"
            . " from (select JSONAllPathsWithTypes($column) as m from $table)"
            . ") order by path";
    # HTTP needs explicit format; native returns typed rows directly.
    $sql .= " format TabSeparated" if !$self->server_revision;
    $self->query($sql, sub {
        my ($rows, $err) = @_;
        return $cb->(undef, $err) if $err;
        # Handle both arrayref and named_rows hashref shapes — same idiom
        # as for_table.
        $cb->([ map {
            ref $_ eq 'HASH'
                ? { path => $_->{path}, type => $_->{type} }
                : { path => $_->[0],    type => $_->[1] }
        } @{ $rows // [] } ]);
    });
}

# is_healthy: ping with a deadline. Callback receives (1, undef) on
# success or (0, $err_msg) on timeout or ping error. The connection
# itself isn't disturbed - failure does not call finish/reset; the
# caller decides recovery.
sub is_healthy {
    my ($self, $cb, $timeout) = @_;
    die "Usage: \$ch->is_healthy(\$cb, [\$timeout])" unless ref($cb) eq 'CODE';
    $timeout //= 5;
    my $done = 0;
    my $t = EV::timer($timeout, 0, sub {
        return if $done;
        $done = 1;
        $cb->(0, "health check timeout after ${timeout}s");
    });
    $self->ping(sub {
        return if $done;
        $done = 1;
        undef $t;
        my (undef, $err) = @_;
        $cb->(!$err, $err);
    });
    return;
}

# Single PING + measure wall-clock latency. Callback receives
# ($seconds, undef) on success or (undef, $err) on failure.
# Cheaper than installing track_query_durations for a one-shot probe.
sub ping_round_trip {
    my ($self, $cb) = @_;
    die "Usage: \$ch->ping_round_trip(\$cb)" unless ref($cb) eq 'CODE';
    my $start = EV::time();
    $self->ping(sub {
        my (undef, $err) = @_;
        return $cb->(undef, $err) if $err;
        $cb->(EV::time() - $start, undef);
    });
    return;
}

# Filter callback that fires only when a query exceeds $threshold
# seconds. Installs an on_query_complete observer that composes with
# any existing handler. Returns the previous on_query_complete so the
# caller can restore it later.
sub slow_query_log {
    my ($self, $threshold, $cb) = @_;
    die "Usage: \$ch->slow_query_log(\$threshold, \$cb)"
        unless defined $threshold && ref($cb) eq 'CODE';
    my $prev = $self->on_query_complete;
    $self->on_query_complete(sub {
        # on_query_complete args: ($query_id, $rows, $bytes, $code, $dur, $err)
        my $dur = $_[4];
        $cb->(@_) if defined $dur && $dur >= $threshold;
        $prev->(@_) if $prev;
    });
    return $prev;
}

# Fetch one value from system.settings. Identifier-safe ($name is
# validated; the lookup uses parameter binding so a bad name produces
# a server-side error rather than SQL injection).
sub server_setting {
    my ($self, $name, $cb) = @_;
    die "Usage: \$ch->server_setting(\$name, \$cb)"
        unless defined $name && ref($cb) eq 'CODE';
    die "server_setting: invalid setting name '$name'"
        unless $name =~ $RE_IDENT;
    $self->query(
        "select value from system.settings where name = {n:String}",
        { params => { n => $name } },
        sub {
            my ($rows, $err) = @_;
            return $cb->(undef, $err) if $err;
            $cb->($rows && @$rows ? $rows->[0][0] : undef);
        },
    );
}

# Count rows in a table, optionally with a server-side WHERE filter.
# $where is interpolated literally so the caller is responsible for
# its safety (use parameterized queries via settings.params for
# user-supplied predicates). Returns the integer count or (undef, $err).
sub row_count {
    my ($self, $table, $where, $cb) = @_;
    ($where, $cb) = (undef, $where) if ref($where) eq 'CODE';
    die "Usage: \$ch->row_count(\$table, [\$where], \$cb)"
        unless defined $table && ref($cb) eq 'CODE';
    die "row_count: invalid table '$table'" unless $table =~ $RE_TABLE;
    my $tbl = EV::ClickHouse->bind_ident($table);
    my $sql = "select count() from $tbl";
    $sql .= " where $where" if defined $where && length $where;
    $self->query($sql, sub {
        my ($rows, $err) = @_;
        return $cb->(undef, $err) if $err;
        $cb->($rows && @$rows ? $rows->[0][0] : 0);
    });
}

# Approx on-disk + uncompressed sizes from system.parts. Returns
# { rows => N, bytes_on_disk => N, data_uncompressed_bytes => N }.
sub table_size {
    my ($self, $table, $cb) = @_;
    die "Usage: \$ch->table_size(\$table, \$cb)"
        unless defined $table && ref($cb) eq 'CODE';
    die "table_size: invalid table '$table'" unless $table =~ $RE_TABLE;
    my ($db, $name) = $table =~ /\./ ? split(/\./, $table, 2)
                                     : (undef, $table);
    my $sql = "select sum(rows), sum(bytes_on_disk),"
            . " sum(data_uncompressed_bytes)"
            . " from system.parts where active and table = {t:String}"
            . (defined $db ? " and database = {d:String}" : "");
    $self->query($sql, {
        params => {
            t => $name,
            (defined $db ? (d => $db) : ()),
        },
    }, sub {
        my ($rows, $err) = @_;
        return $cb->(undef, $err) if $err;
        my $r = $rows && @$rows ? $rows->[0] : [0, 0, 0];
        $cb->({
            rows                    => $r->[0] // 0,
            bytes_on_disk           => $r->[1] // 0,
            data_uncompressed_bytes => $r->[2] // 0,
        });
    });
}

# SYSTEM RELOAD DICTIONARY shortcut. Validates the dictionary name.
sub dictionary_reload {
    my ($self, $name, $cb) = @_;
    die "Usage: \$ch->dictionary_reload(\$name, \$cb)"
        unless defined $name && ref($cb) eq 'CODE';
    die "dictionary_reload: invalid name '$name'" unless $name =~ $RE_TABLE;
    $self->query("system reload dictionary " . EV::ClickHouse->bind_ident($name),
                 $cb);
}

# REFRESH MATERIALIZED VIEW (server >= 23.12). Validates the view name.
sub refresh_view {
    my ($self, $name, $cb) = @_;
    die "Usage: \$ch->refresh_view(\$name, \$cb)"
        unless defined $name && ref($cb) eq 'CODE';
    die "refresh_view: invalid name '$name'" unless $name =~ $RE_TABLE;
    $self->query("system refresh view " . EV::ClickHouse->bind_ident($name),
                 $cb);
}

# Poll system.mutations until a table's incomplete mutations finish.
# ALTER ... UPDATE/DELETE is asynchronous; this resolves once the
# mutation(s) reach is_done=1. $cb->({ pending => 0 }) on success,
# $cb->(undef, $err) on a failed mutation / query error / timeout.
#   poll        => seconds between polls (default 1)
#   timeout     => give up after N seconds (optional)
#   mutation_id => wait only for this specific mutation (optional)
sub wait_mutation {
    my ($self, $table, $cb, %opts) = @_;
    die "Usage: \$ch->wait_mutation(\$table, \$cb, [%opts])"
        unless defined $table && ref($cb) eq 'CODE';
    die "wait_mutation: invalid table '$table'" unless $table =~ $RE_TABLE;
    my $poll    = $opts{poll} // 1;   # defined-or: an explicit poll => 0 is honored
    my $timeout = $opts{timeout};
    my $mid     = $opts{mutation_id};
    my ($db, $name) = $table =~ /\./ ? split(/\./, $table, 2) : (undef, $table);

    # A mutation that keeps failing stays is_done=0 with latest_fail_reason
    # set, so filtering on is_done=0 surfaces both running and failing ones.
    my $sql = "select count() as pending,"
            . " anyIf(latest_fail_reason, latest_fail_reason != '') as fail"
            . " from system.mutations"
            . " where table = {t:String} and is_done = 0"
            . (defined $db  ? " and database = {d:String}"   : "")
            . (defined $mid ? " and mutation_id = {m:String}" : "");
    my %params = (t => $name);
    $params{d} = $db  if defined $db;
    $params{m} = $mid if defined $mid;

    my $started = EV::time();
    my $fail_streak = 0;          # consecutive polls that saw a fail reason
    my $timer; my $poll_once;
    $poll_once = sub {
        # query() can croak synchronously (e.g. "not connected" if the
        # connection was finished between polls); route that to $cb
        # instead of letting it escape into the event loop.
        my $ok = eval {
            $self->query($sql, { params => \%params }, sub {
                my ($rows, $err) = @_;
                if ($err) { undef $poll_once; return $cb->(undef, $err) }
                my $r = $rows && @$rows ? $rows->[0] : [0, ''];
                my ($pending, $fail) = @$r;
                # Require a fail reason to persist across polls: a single
                # transient latest_fail_reason can clear on the mutation's
                # next retry, so one sighting is not a terminal failure.
                if (defined $fail && length $fail) {
                    if (++$fail_streak >= 2) {
                        undef $poll_once;
                        return $cb->(undef, "wait_mutation: $fail");
                    }
                } else {
                    $fail_streak = 0;
                }
                if (!$pending) {
                    undef $poll_once;
                    return $cb->({ pending => 0 });
                }
                if (defined $timeout && EV::time() - $started >= $timeout) {
                    undef $poll_once;
                    return $cb->(undef,
                                 "wait_mutation: timed out after ${timeout}s");
                }
                $timer = EV::timer($poll, 0, sub { undef $timer; $poll_once->() });
            });
            1;
        };
        if (!$ok) {
            my $e = $@ || 'unknown error';
            undef $poll_once;
            $cb->(undef, $e);
        }
    };
    $poll_once->();
    return;
}

# Parse a clickhouse[+native]:// URI into a hash without opening a
# connection. Lets tooling validate user-supplied URIs ahead of
# instantiation. Returns undef on a malformed URI.
sub parse_uri {
    my ($class, $uri) = @_;
    return undef unless defined $uri;
    return undef unless $uri =~ m{
        ^clickhouse(?:\+(\w+))?://
        (?:([^:@]*?)(?::([^@]*))?\@)?
        (\[[^\]]+\]|[^/:?]+)
        (?::(\d+))?
        (?:/([^?]*))?
        (?:\?(.*))?$
    }x;
    my ($proto, $user, $pass, $host, $port, $db, $qs) =
        ($1, $2, $3, $4, $5, $6, $7);
    $host =~ s/^\[(.*)\]$/$1/;
    my %out = (host => $host);
    $out{protocol} = $proto                if defined $proto;
    $out{user}     = _uri_unescape($user)  if defined $user && $user ne '';
    $out{password} = _uri_unescape($pass)  if defined $pass;
    $out{port}     = $port + 0             if defined $port;
    $out{database} = _uri_unescape($db)    if defined $db && $db ne '';
    # Query-string keys land at the top level (path-derived host/port/etc
    # win on collision) so the resulting hash can be passed verbatim to new().
    _uri_qs_into($qs, \%out);
    $out{protocol} //= 'http';         # apply default after QS merge
    return \%out;
}

# is_retryable_error($code) -> bool. Class method (no $self required).
# Identifies the common transient ClickHouse error codes that warrant
# an automatic retry. Source: ClickHouse src/Common/ErrorCodes.cpp.
my %RETRYABLE = map { $_ => 1 } (
    159,   # TIMEOUT_EXCEEDED
    202,   # TOO_MANY_SIMULTANEOUS_QUERIES
    203,   # NO_FREE_CONNECTION
    209,   # SOCKET_TIMEOUT
    210,   # NETWORK_ERROR
    241,   # MEMORY_LIMIT_EXCEEDED       (often transient under contention)
    242,   # TABLE_IS_READ_ONLY          (replica catching up)
    252,   # TOO_MANY_PARTS              (merge backlog, retry after backoff)
    285,   # TOO_FEW_LIVE_REPLICAS
    319,   # UNKNOWN_STATUS_OF_INSERT    (idempotent insert salvages this)
    373,   # SESSION_IS_LOCKED
    # NB: deliberately NOT including 394 (QUERY_WAS_CANCELLED). A
    # cancellation expresses caller intent; auto-retrying would loop
    # and burn server resources for no benefit.
    439,   # CANNOT_SCHEDULE_TASK
    999,   # KEEPER_EXCEPTION
);
sub is_retryable_error {
    my ($self_or_class, $code) = @_;
    return 0 unless defined $code;
    $RETRYABLE{ $code + 0 } ? 1 : 0;
}

# bind_ident($name) -> backtick-quoted identifier safe for SQL splicing.
# ClickHouse identifier rules: alnum + _, no leading digit, optional
# dotted form ("db.table"). We accept that subset, croak otherwise.
# The $RE_IDENT validation rejects backticks outright, so the quoter
# can just wrap each part without escaping.
sub bind_ident {
    my ($self_or_class, $name) = @_;
    die "bind_ident: identifier must be defined and non-empty"
        unless defined $name && length $name;
    my @parts = split /\./, $name, -1;
    die "bind_ident: empty component in '$name'"
        if grep { !length } @parts;
    for my $p (@parts) {
        die "bind_ident: invalid identifier component '$p' in '$name'"
            unless $p =~ $RE_IDENT;
    }
    join '.', map { "`$_`" } @parts;
}

# Capability table: feature name → minimum native server revision.
# Lets user code branch cleanly on protocol features instead of
# hard-coding numeric revisions all over the place. HTTP connections
# have no protocol revision (server_revision == 0), so server_supports
# returns false for any non-trivial feature on HTTP - by design.
my %FEATURES = (
    block_info          => 51903,   # DBMS_MIN_REVISION_WITH_BLOCK_INFO
    server_display_name => 54372,
    version_patch       => 54401,
    progress_writes     => 54420,
    server_timezone     => 54423,
    addendum            => 54458,
);
sub server_supports {
    my ($self, $feature) = @_;
    return 0 unless ref $self;
    return 0 unless defined $feature;
    my $required = $FEATURES{$feature};
    return 0 unless defined $required;
    my $have = $self->server_revision or return 0;
    $have >= $required;
}

package EV::ClickHouse::Streamer;

sub _new {
    my ($class, $ch, $table, %opts) = @_;
    # `columns => [@names]` enables named-row mode: push_row({}) hashes
    # are reordered into arrayref by the streamer instead of the caller
    # having to know column position. `named => 1` is a tolerated alias
    # but `columns` is what actually drives the lookup.
    my $columns = $opts{columns};
    die "insert_streamer: columns must be a non-empty arrayref"
        if defined $columns && (ref($columns) ne 'ARRAY' || !@$columns);
    bless {
        ch             => $ch,
        table          => $table,
        settings       => $opts{settings},                  # per-insert hashref
        batch_size     => $opts{batch_size}     || 10_000,
        on_batch_error => $opts{on_batch_error} || sub { }, # per-failure cb
        high_water     => $opts{high_water}     || 0,       # 0 = disabled
        on_high_water  => $opts{on_high_water}  || sub { }, # ($buffered, $in_flight)
        # low_water: threshold at which await_drain callbacks fire.
        # Defaults to half of high_water; honoured only when high_water
        # is set (otherwise await_drain fires whenever buffer is empty).
        # Use defined() so an explicit `low_water => 0` (fire only when
        # the buffer is fully empty) is respected, not treated as unset.
        low_water      => defined($opts{low_water})
                            ? $opts{low_water}
                            : (($opts{high_water} || 0) / 2),
        columns        => $columns,
        high_water_active => 0,
        buffer         => [],
        in_flight      => 0,
        pending_finish => undef,
        drain_waiters  => [],
        sticky_err     => undef,
    }, $class;
}

sub push_row { EV::ClickHouse::_streamer_push_row(@_) }

sub _flush {
    my $self = shift;
    return if $self->{in_flight} || !@{ $self->{buffer} };
    my $batch = $self->{buffer};
    $self->{buffer} = [];
    $self->{in_flight} = 1;
    my $cb = sub {
        my (undef, $err) = @_;
        $self->{in_flight} = 0;
        if ($err) {
            $self->{sticky_err} //= $err;
            $self->{on_batch_error}->($err);
        }
        # Drain any buffered rows queued during the batch
        $self->_flush;
        # Reset high_water latch once we drop below the threshold
        $self->{high_water_active} = 0
            if $self->{high_water_active}
            && @{ $self->{buffer} } < $self->{high_water};
        # Fire await_drain waiters only when the buffer is at/below
        # low_water AND nothing is in flight. _flush above may have
        # immediately re-dispatched (in_flight=1, buffer=[]); waking
        # the waiter then would lie about the streamer being idle.
        my $low = $self->{low_water};
        if (!$self->{in_flight}
            && @{ $self->{drain_waiters} }
            && @{ $self->{buffer} } <= ($low || 0)) {
            my @w = @{ $self->{drain_waiters} };
            $self->{drain_waiters} = [];
            $_->(undef) for @w;     # undef err = normal drain
        }
        # Notify finish() if no work remains
        if ($self->{pending_finish}
            && !$self->{in_flight}
            && !@{ $self->{buffer} }) {
            my $fcb = delete $self->{pending_finish};
            $fcb->(undef, $self->{sticky_err});
        }
    };
    my @opt = $self->{settings} ? ($self->{settings}) : ();
    $self->{ch}->insert($self->{table}, $batch, @opt, $cb);
}

sub finish {
    my ($self, $cb) = @_;
    die "Usage: \$streamer->finish(\$cb)" unless ref($cb) eq 'CODE';
    $self->_flush;
    if (!$self->{in_flight} && !@{ $self->{buffer} }) {
        $cb->(undef, $self->{sticky_err});
    } else {
        $self->{pending_finish} = $cb;
    }
    return;
}

sub buffered_count { scalar @{ $_[0]{buffer} } }
sub in_flight      { $_[0]{in_flight} }
sub sticky_error   { $_[0]{sticky_err} }

# Register a callback that fires once the buffered row count drops
# to low_water (defaults to high_water/2; 0 if high_water not set).
# Pairs with on_high_water to close the backpressure loop:
#   on_high_water => sub { $producer->pause },
#   $streamer->await_drain(sub { $producer->resume });
# Fires synchronously if the buffer is already at/below the threshold.
sub await_drain {
    my ($self, $cb) = @_;
    die "Usage: \$streamer->await_drain(\$cb)" unless ref($cb) eq 'CODE';
    my $low = $self->{low_water} || 0;
    # Fire synchronously when nothing is in flight AND the buffer is
    # at/below low_water: there's no flush pending so no waiter would
    # ever fire otherwise.
    return $cb->(undef) if !$self->{in_flight} && @{ $self->{buffer} } <= $low;
    push @{ $self->{drain_waiters} }, $cb;
    return;
}

# Discard buffered rows + sticky error without finishing. Useful for
# "retry after permanent error" patterns where the producer wants to
# wipe the slate clean (typically after a schema-level fix) and keep
# pushing into the same streamer object. The underlying $ch is NOT
# touched - any in-flight batch already on the wire still completes.
sub reset {
    my ($self) = @_;
    $self->{buffer}            = [];
    $self->{sticky_err}        = undef;
    $self->{high_water_active} = 0;
    # Deliver any pending finish/drain callbacks with a reset error
    # rather than silently dropping them - quiet loss of a finish cb
    # leaves the producer waiting forever.
    my $pf = delete $self->{pending_finish};
    my @dw = @{ delete $self->{drain_waiters} || [] };
    $self->{drain_waiters} = [];
    $pf->(undef, 'streamer reset') if $pf;
    $_->('streamer reset') for @dw;
    return $self;
}

# Discover column names from the target table via for_table, then
# enable named-rows mode by populating $self->{columns}. Callback
# receives undef on success, error string on failure. Useful when
# the producer side doesn't know (or shouldn't care about) the
# schema in advance.
sub columns_from_table {
    my ($self, $cb) = @_;
    die "Usage: \$streamer->columns_from_table(\$cb)" unless ref($cb) eq 'CODE';
    $self->{ch}->for_table($self->{table}, sub {
        my ($info, $err) = @_;
        return $cb->($err) if $err;
        $self->{columns} = [ map { $_->{name} } @{ $info->{columns} } ];
        $cb->(undef);
    });
    return;
}

package EV::ClickHouse::Pool;
use Scalar::Util qw(refaddr);

# Built-in connection pool. Round-robin dispatch with least-busy fallback;
# each connection is independent (own auto_reconnect, own send_queue),
# so a hung query on one doesn't block the others. Pass any EV::ClickHouse
# constructor option in %args; it's applied to every pool member.
#
#   my $pool = EV::ClickHouse::Pool->new(host => 'ch', size => 10, ...);
#   $pool->query($sql, $cb);
#   $pool->insert($table, $data, $cb);
#   $pool->drain(sub { ... });   # all connections drained
#   $pool->finish;
sub new {
    my ($class, %args) = @_;
    my $size = delete $args{size} || 4;
    die "Pool size must be >= 1" if $size < 1;
    # Circuit breaker per member. After `circuit_threshold` consecutive
    # query/insert failures, mark the member dead for `circuit_cooldown`
    # seconds; _pick skips dead members. 0 disables.
    my $threshold = delete $args{circuit_threshold} || 0;
    my $cooldown  = delete $args{circuit_cooldown}  || 30;
    my @conns;
    for (1 .. $size) {
        push @conns, EV::ClickHouse->new(%args);
    }
    bless {
        conns      => \@conns,
        idx        => 0,
        cb_thresh  => $threshold,
        cb_cool    => $cooldown,
        cb_state   => [ map { { fails => 0, dead_until => 0 } } @conns ],
    }, $class;
}

# Pick the connection with the fewest in-flight queries; ties broken by
# round-robin. With circuit_threshold > 0, dead members are skipped
# (unless all are dead - then the breaker is bypassed). Hot path - the
# implementation lives in XS for ~5x lower per-pick cost.
sub _pick { EV::ClickHouse::_pool_pick($_[0]) }

# Find the {fails,dead_until} slot for a given $ch (object identity match).
sub _slot_for {
    my ($self, $ch) = @_;
    my $r = refaddr $ch;
    for my $i (0 .. $#{ $self->{conns} }) {
        return $self->{cb_state}[$i] if refaddr($self->{conns}[$i]) == $r;
    }
    return undef;
}

# Wrap the user callback so the circuit breaker can observe success/failure.
# The user's $cb is the LAST argument of query/insert/ping (per the public
# API). The slot update itself is in XS (_breaker_observe) so the wrapper
# closure body is one XSUB call rather than a handful of Perl ops.
sub _cb_observer {
    my ($self, $ch, $user_cb, $observe_failures) = @_;
    $observe_failures //= 1;
    return $user_cb unless $self->{cb_thresh} && ref($user_cb) eq 'CODE';
    my $slot = ref($ch) ? $self->_slot_for($ch) : $self->{cb_state}[$ch];
    return $user_cb unless $slot;
    my $thresh = $self->{cb_thresh};
    my $cool   = $self->{cb_cool};
    sub {
        # all-dead fallback: observe successes only (resets the breaker
        # on recovery), skip failures (so under-load loss doesn't extend
        # dead_until repeatedly).
        EV::ClickHouse::_breaker_observe($slot, $_[1], $thresh, $cool)
            if $observe_failures || !$_[1];
        $user_cb->(@_);
    };
}

# Dispatch a method on a picked connection, wrapping the trailing CODE arg
# (the user callback) with the circuit-breaker observer so success/failure
# updates the per-member slot.
sub _dispatch {
    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) }
            1;
        } or do {
            $out[$i]{err} = "$@" || "fan_out: dispatch failed";
            $deliver->();
        };
    }
    return;
}

# Checkout-style pin: hand the user a least-busy member, run their cb
# with ($conn, $release). Until $release->() is called, the pool will
# avoid handing the same member to other callers via _pick — useful
# for temp tables / SET / session-state work that must land on the
# same connection across multiple queries.
sub with_session {
    my ($self, $cb) = @_;
    die "Usage: \$pool->with_session(\$cb)" unless ref($cb) eq 'CODE';
    my $ch  = $self->_pick;
    my $r   = refaddr $ch;
    $self->{_pinned}{$r} = ($self->{_pinned}{$r} // 0) + 1;
    my $released = 0;
    my $release = sub {
        return if $released++;
        if (--$self->{_pinned}{$r} <= 0) { delete $self->{_pinned}{$r} }
    };
    eval { $cb->($ch, $release); 1 } or do {
        my $err = $@;
        $release->();
        die $err;
    };
    return;
}

# Drain when ALL connections have completed pending work.
sub drain {
    my ($self, $cb) = @_;
    my $left = scalar @{ $self->{conns} };
    my $err;
    for my $c (@{ $self->{conns} }) {
        $c->drain(sub {
            $err //= $_[0] if $_[0];
            $cb->($err) unless --$left;
        });
    }
}

sub finish { $_->finish for @{ $_[0]{conns} } }

# Coordinated graceful shutdown: drain every member, then finish. If
# the optional $grace_seconds elapses before all members drain, force
# finish and report a timeout in the callback. Callback receives undef
# on clean shutdown, an error string on per-member drain error or
# timeout. $cb is optional.
#
#   $pool->shutdown(10, sub {
#       my ($err) = @_;
#       warn "shutdown: $err" if $err;
#       EV::break;
#   });
sub shutdown {
    my ($self, $grace, $cb) = @_;
    # Two-arg form: $pool->shutdown($cb). Treat the coderef as the cb
    # with no grace timer rather than silently dropping it.
    ($grace, $cb) = (undef, $grace) if ref($grace) eq 'CODE';
    $cb //= sub { };
    die "Usage: \$pool->shutdown([\$grace_seconds], \$cb)" unless ref($cb) eq 'CODE';
    my $left  = scalar @{ $self->{conns} };
    my $err;
    my $timer;
    my $fired = 0;
    my $finalize = sub {
        return if $fired++;
        undef $timer;
        $self->finish;
        $cb->($err);
    };
    for my $c (@{ $self->{conns} }) {
        $c->drain(sub {
            $err //= $_[0] if $_[0];
            $finalize->() if --$left == 0;
        });
    }
    if ($grace && $grace > 0) {
        $timer = EV::timer($grace, 0, sub {
            $err //= "Pool::shutdown timed out after ${grace}s";
            $finalize->();
        });
    }
    return;
}

# Broadcast-to-all helpers — these touch every member because they affect
# state owned per connection (queued queries, in-flight cancellation, the
# socket itself). Picking a single member would silently leave the other
# pool connections untouched.
sub cancel       { for (@{ $_[0]{conns} }) { eval { $_->cancel       } } }
sub skip_pending { for (@{ $_[0]{conns} }) { eval { $_->skip_pending } } }
sub reset        { for (@{ $_[0]{conns} }) { eval { $_->reset        } } }

package EV::ClickHouse::Iterator;

sub next { EV::ClickHouse::_iterator_next(@_) }

sub error     { $_[0]{err} }
sub is_done   { $_[0]{done} && !@{ $_[0]{batches} } }
sub cancel    { $_[0]{ch}->cancel }

package EV::ClickHouse::Error;

# Lightweight error object — wraps the (message, code) pair callers
# already get on $err, plus a symbolic name and is_retryable boolean.
# Stringifies to the message so legacy callsites that string-compare
# against $err keep working.
use overload '""' => sub { $_[0]{message} }, fallback => 1;

# Symbolic names for ClickHouse error codes that user code is likely
# to want to branch on by name. Sourced from src/Common/ErrorCodes.cpp;
# extend liberally — the table is informational only.
my %CODE_NAME = (
    0   => 'OK',
    27  => 'CANNOT_PARSE_INPUT_ASSERTION_FAILED',
    32  => 'ATTEMPT_TO_READ_AFTER_EOF',
    33  => 'CANNOT_READ_ALL_DATA',
    44  => 'ILLEGAL_COLUMN',
    47  => 'UNKNOWN_IDENTIFIER',
    60  => 'UNKNOWN_TABLE',
    62  => 'SYNTAX_ERROR',
    81  => 'UNKNOWN_DATABASE',
    86  => 'RECEIVED_ERROR_FROM_REMOTE_IO_SERVER',
    113 => 'UNKNOWN_SETTING',
    159 => 'TIMEOUT_EXCEEDED',
    164 => 'READONLY',
    192 => 'UNKNOWN_USER',
    193 => 'WRONG_PASSWORD',
    194 => 'REQUIRED_PASSWORD',
    202 => 'TOO_MANY_SIMULTANEOUS_QUERIES',
    203 => 'NO_FREE_CONNECTION',
    209 => 'SOCKET_TIMEOUT',
    210 => 'NETWORK_ERROR',
    225 => 'NO_ZOOKEEPER',
    236 => 'ABORTED',
    241 => 'MEMORY_LIMIT_EXCEEDED',
    242 => 'TABLE_IS_READ_ONLY',
    252 => 'TOO_MANY_PARTS',
    285 => 'TOO_FEW_LIVE_REPLICAS',
    319 => 'UNKNOWN_STATUS_OF_INSERT',
    341 => 'UNFINISHED',
    373 => 'SESSION_IS_LOCKED',
    389 => 'INSERT_WAS_DEDUPLICATED',
    394 => 'QUERY_WAS_CANCELLED',
    439 => 'CANNOT_SCHEDULE_TASK',
    497 => 'ACCESS_DENIED',
    516 => 'AUTHENTICATION_FAILED',
    999 => 'KEEPER_EXCEPTION',
);

sub new {
    my ($class, %args) = @_;
    bless {
        message => $args{message} // '',
        code    => $args{code}    // 0,
    }, $class;
}

# Convenience: build from ($ch, $err) — the typical pair that callbacks
# receive. The ClickHouse-side code comes from $ch->last_error_code.
sub from_ch {
    my ($class, $ch, $err) = @_;
    return undef unless defined $err && length $err;
    $class->new(
        message => "$err",
        code    => (eval { $ch->last_error_code } || 0),
    );
}

sub message      { $_[0]{message} }
sub code         { $_[0]{code} }
sub name         { $CODE_NAME{ $_[0]{code} } }
sub is_retryable { EV::ClickHouse->is_retryable_error($_[0]{code}) }

# Class-level table introspection.
sub code_name    { $CODE_NAME{ $_[1] // 0 } }
sub known_codes  { sort { $a <=> $b } keys %CODE_NAME }

1;

__END__