EV-ClickHouse

 view release on metacpan or  search on metacpan

lib/EV/ClickHouse.pm  view on Meta::CPAN

Called on native protocol progress packets. Not fired for HTTP.

=item on_disconnect => sub { }

Called when an established connection closes (by C<finish>, server
disconnect, or mid-flight error). Only fires if C<on_connect> had
previously fired - it does B<not> fire for connect-phase failures
(refused, timeout, ServerHello stall) since no connection was ever
established. Fires after internal state has been reset, so it is safe
to queue new queries or call C<reset> from inside the handler.

=item on_trace => sub { my ($message) = @_ }

Debug trace callback. Called with internal state-machine messages
(connect, dispatch, disconnect). Useful for diagnosing protocol issues.

=item on_failover => sub { my ($old_host, $old_port, $new_host, $new_port, $msg) = @_ }

Multi-host only. Fires after the failover wrapper rotates to the next
host in the C<hosts =E<gt> [...]> list, with the old and new (host, port)
pair plus the triggering error message. Use it for metrics ("which host
am I on?") or to log host transitions. Fires before the user's C<on_error>.

=back

B<Options:>

=over 4

=item compress => 0 | 1

Enable compression: gzip on HTTP (request and response), LZ4 with CityHash
checksums on the native protocol. Default: C<0>. Native compression
requires liblz4 at build time.

=item session_id => $id

HTTP session id for stateful operations (temporary tables, SET, etc.).
Native protocol has stateful sessions intrinsically; this option is HTTP-only.

=item connect_timeout => $seconds

TCP/TLS connection timeout. C<0> (default) means no timeout. Floating
point allowed.

=item query_timeout => $seconds

Default per-query timeout applied to every query and insert. The query
callback receives a C<timeout> error if exceeded. Override per-call via
the C<query_timeout> key in the settings hashref.

=item max_query_size => $bytes

Client-side guard: croak before sending any query whose SQL text exceeds
this many bytes. C<0> (default) disables the check. Useful as a
last-resort defense against accidentally sending unbounded strings.

=item max_recv_buffer => $bytes

Defensive ceiling on the response. The cap applies to the raw recv
buffer (every protocol), the chunked-decoded body (HTTP), and the
gzip-decompressed body (HTTP), so the same upper bound applies to the
user-visible payload regardless of transport encoding. On overflow the
query callback receives an appropriate error ("recv buffer overflow",
"chunked response too large", or "gzip body exceeds max_recv_buffer")
and the connection is torn down so no subsequent query can slip past
the cap on the same socket. C<0> (default) keeps the historical
no-cap behaviour (still bounded internally by a hard 128 MB ceiling
on compressed paths). Recommended in production when the schema is
constrained and you want a hard upper bound (e.g.
C<128 * 1024 * 1024> for 128 MB).

=item http_basic_auth => 0 | 1

HTTP only. When set, send credentials as
C<Authorization: Basic base64(user:password)> instead of the default
C<X-ClickHouse-User> / C<X-ClickHouse-Key> header pair. Use this when
the connection passes through an HTTP gateway (nginx, Envoy, ...) that
strips the X-ClickHouse-* headers but forwards Basic auth verbatim.
Default: C<0>.

=item auto_reconnect => 0 | 1

Reconnect automatically on connection loss. Default: C<0>. When enabled,
queued (unsent) queries are preserved across reconnects; in-flight queries
receive an error.

The reconnect path covers TCP/TLS connect failures, C<connect_timeout>
or C<query_timeout> expiry, and any clean server-side EOF (idle or
mid-request). Mid-query I/O errors (ECONNRESET / EPIPE) and a malformed
native ServerHello are B<not> retried - they typically indicate a
misconfigured peer or client-side bug that retry would only loop on.
Combine with C<reconnect_max_attempts> for an explicit ceiling.

=item settings => \%hash

ClickHouse settings applied to every query and insert. Per-call settings
(see L</query>, L</insert>) override these.

    settings => { async_insert => 1, max_threads => 4 }

=item keepalive => $seconds

Send a keepalive request every N seconds while the connection is idle:
a native CLIENT_PING on the native protocol or a C<GET /ping> on HTTP
(some load balancers / NATs drop idle HTTP connections after a few
seconds; TCP-level keepalive is too coarse). Default: C<0> (disabled).

=item reconnect_delay => $seconds

Initial delay for the C<auto_reconnect> exponential backoff. Each failed
attempt doubles the delay, capped at C<reconnect_max_delay>. Default:
C<0> (immediate retry, no backoff).

=item reconnect_max_delay => $seconds

Backoff ceiling. Default: C<0>, meaning no explicit cap; the implementation
still bounds the backoff exponent at 20 doublings, so with
C<reconnect_delay = 0.5> the worst case is roughly 6 days. Setting an
explicit ceiling is recommended in production.

lib/EV/ClickHouse.pm  view on Meta::CPAN

HTTP only. The callback receives the raw response body as a scalar string
instead of parsed rows. Use with an explicit C<format> clause:

    $ch->query("select * from t format CSV", { raw => 1 }, sub {
        my ($body, $err) = @_;
    });

Croaks if used with the native protocol.

=item C<query_timeout =E<gt> $seconds>

Per-query timeout, overriding the connection-level C<query_timeout>.

=item C<on_data =E<gt> sub { my ($rows) = @_; ... }>

Native protocol only. A code ref called for each data block as it arrives,
for streaming large result sets. Rows are delivered incrementally and
B<not> accumulated, so the final callback receives C<(undef)> rather than
all rows. The final callback always fires on completion or error, even if
no data block was emitted (empty result, server-side error before the
first block).

    $ch->query("select * from big_table",
        { on_data => sub { my ($rows) = @_; process_batch($rows) } },
        sub { my (undef, $err) = @_; warn $err if $err },
    );

=item C<external =E<gt> \%tables>

Native protocol only. Ships one or more in-memory data blocks that the
query can reference as tables, JOIN against, or filter with C<IN> -
without creating a server-side temporary table. Each entry maps a table
name to C<{ structure =E<gt> [...], data =E<gt> [...] }>:

    $ch->query(
        "select u.id, u.name from users u where u.id in _wanted",
        { external => {
            _wanted => {
                structure => [ id => 'UInt64' ],
                data      => [ [7], [42], [911] ],
            },
        } },
        sub { my ($rows, $err) = @_; ... },
    );

C<structure> is a flat list of C<name =E<gt> type> pairs (ClickHouse type
names, e.g. C<UInt64>, C<String>, C<Float64>); C<data> is an arrayref of
row arrayrefs, encoded with the same type machinery as L</insert>. An
empty C<data> arrayref is a valid zero-row table. Several external tables
may be supplied at once. Croaks on the HTTP protocol or on a malformed
spec (odd structure list, non-arrayref row, or a column type that
cannot be encoded).

=back

B<Native protocol type notes:> values come back as typed Perl scalars.
By default C<Date>/C<DateTime> are integers (days since epoch / Unix
timestamps); enable C<decode_datetime> for strings. C<Enum> values are
numeric codes; C<decode_enum> returns labels. C<Decimal> values are
unscaled integers; C<decode_decimal> scales them to floats.
C<SimpleAggregateFunction> is transparently decoded as its inner type.
C<Nested> columns become arrays of tuples. C<LowCardinality> works
correctly across multi-block results with shared dictionaries.

=head2 insert

    $ch->insert($table, $data, sub { my (undef, $err) = @_ });
    $ch->insert($table, $data, \%settings, sub { my (undef, $err) = @_ });

C<$data> may be either:

=over 4

=item * A pre-formatted TabSeparated string (tabs separate columns,
newlines separate rows, with the standard ClickHouse escapes).

=item * An arrayref of arrayrefs (rows of column values).

=back

When using arrayrefs, no TSV escaping is needed: C<undef> maps to null
and strings may contain tabs and newlines freely.

Nested arrayrefs (Array/Tuple columns) and hashrefs (Map columns) are
supported B<only on the native protocol>, where the encoder has the
column type from the server's sample block. On HTTP the same call
croaks rather than silently produce malformed TSV; use the native
protocol or pre-serialise nested types into ClickHouse TSV literal form.

    # Native: nested types encode directly.
    $ch->insert("my_table", [
        [1, "hello\tworld"],   # embedded tab
        [2, undef],            # null
        [3, [10, 20]],         # Array column   (native only)
        [4, { a => 1, b => 2 }],  # Map column  (native only)
    ], sub { ... });

The optional C<\%settings> hashref works exactly as in L</query>,
including C<query_id>, C<query_timeout>, and C<params>. Two extra
flags are recognised here:

=over 4

=item C<idempotent =E<gt> 1 | $token>

Auto-mints (or uses the supplied) C<insert_deduplication_token>, so a
reconnect-driven retry of the same insert doesn't double-write. Falsy
values are a no-op.

=item C<async_insert =E<gt> 1>

Enables ClickHouse server-side insert batching by setting
C<async_insert=1, wait_for_async_insert=0>. Both sub-settings can be
overridden by passing them explicitly.

=back

=head2 ping

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

lib/EV/ClickHouse.pm  view on Meta::CPAN

=back

=head1 ARCHITECTURE

The client is a single state machine driven by an L<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 C<schedule_reconnect>.

Two key invariants:

=over 4

=item *

Native protocol is strictly request/response. Only one query is
in-flight per connection at a time. C<insert_streamer> serialises
batches against this constraint.

=item *

C<callback_depth> guards against C<self> being freed mid-callback.
Every callback dispatch increments it; C<check_destroyed> defers the
final C<Safefree> until depth returns to zero.

=back

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

=head1 TYPES

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

=over 4

=item Integers

Int8..Int64 / UInt8..UInt64: native Perl IV/UV. Int128/UInt128/Int256/UInt256
return decimal string representations on platforms with C<__int128> (Int128/UInt128)
or always for the 256-bit forms.

=item Floats

Float32/Float64 round-trip exactly within IEEE-754 limits. C<NaN>/C<+Inf>/
C<-Inf> are preserved.

=item BFloat16

Top 16 bits of a Float32. Encoded by truncation; decoded by zero-extension.
Suitable for ML feature columns; not for accounting.

=item Decimal32/64/128

Decoded as IV (raw integer) or NV (scaled to N decimal digits if
C<decode_decimal =E<gt> 1>). Decimal128 over very long precision may lose
trailing digits in the NV form; pass C<decode_decimal =E<gt> 0> and divide
yourself with L<Math::BigInt> for exact arithmetic.

=item Decimal256

Returns raw 32 LE bytes. Decode with L<Math::BigInt> (see
C<eg/decimal_bigmath.pl>).

=item Date / Date32 / DateTime / DateTime64

Default: integer (days since epoch / Unix seconds). With C<decode_datetime>:
C<YYYY-MM-DD> or C<YYYY-MM-DD HH:MM:SS> or C<YYYY-MM-DD HH:MM:SS.ffffff>.
DateTime carries a timezone string; the formatted output uses it.

=item Bool

Decoded as 0/1. Encoded from any truthy/falsy SV. ClickHouse stores
internally as UInt8 0/1.

=item String / FixedString

Bytes-in, bytes-out. No UTF-8 transformation.

=item UUID

Canonical hex form C<xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx>. Encode
accepts the same.

=item IPv4 / IPv6

Dotted-quad / canonical IPv6 strings.

=item Enum8 / Enum16

Default: integer code. With C<decode_enum =E<gt> 1>: label string.

=item Nullable(T)

C<undef> in Perl maps to null; otherwise the inner type's encoding.

=item Array(T)

Perl arrayref of inner-type values.

=item Tuple(T1, T2, ...)

Perl arrayref ordered as the type declaration. Named tuples
(C<Tuple(a Int32, b String)>) are still arrayref-positional;
parse the name from C<column_types> if you need it.

=item Map(K, V)

Perl hashref. Keys are stringified.