view release on metacpan or  search on metacpan

README.md  view on Meta::CPAN

`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