EV-ClickHouse
view release on metacpan or search on metacpan
ClickHouse.xs view on Meta::CPAN
uint64_t progress_acc[5]; /* coalesced totals since last dispatch */
/* LowCardinality cross-block dictionary state */
SV ***lc_dicts; /* array of dictionaries, one per column */
uint64_t *lc_dict_sizes; /* size of each dictionary */
int lc_num_cols; /* number of columns with LC state */
};
struct ev_ch_cb_s {
SV *cb;
int raw; /* return raw response body instead of parsed rows */
SV *on_data; /* per-query streaming callback (fires per block) */
SV *on_complete;/* per-query on_query_complete override (or NULL) */
double query_timeout; /* per-query timeout (0=use default) */
ngx_queue_t queue;
};
struct ev_ch_send_s {
char *data; /* full HTTP request or native packet */
size_t data_len;
SV *cb;
char *insert_data; /* deferred TSV data for native INSERT */
size_t insert_data_len;
SV *insert_av; /* deferred AV* data for native INSERT */
int raw; /* return raw response body */
SV *on_data; /* per-query streaming callback */
SV *on_complete; /* per-query on_query_complete override */
double query_timeout; /* per-query timeout */
char *query_id; /* query_id for tracking */
ngx_queue_t queue;
};
/* Forward declarations for helpers defined further down (or in xs/io.c)
* but called from earlier code in this file or from xs/*.c included
* before the definition site. */
static void timer_cb(EV_P_ ev_timer *w, int revents);
Changes
cityhash.h
ClickHouse.xs
cpanfile
eg/async_dns.pl
eg/async_insert.pl
eg/auth_proxy.pl
eg/auto_reconnect.pl
eg/auto_reconnect_resilient.pl
eg/cancel_streaming.pl
eg/circuit_breaker.pl
eg/connection_pool.pl
eg/csv_export.pl
eg/csv_import.pl
eg/dashboard_metrics.pl
eg/decimal_bigmath.pl
eg/decode_options.pl
eg/distributed_pool.pl
eg/drain.pl
eg/error_handling.pl
eg/external_tables.pl
eg/failover.pl
eg/fan_out.pl
eg/geo.pl
eg/graceful_shutdown.pl
eg/health_dashboard.pl
eg/health_probe.pl
eg/hedged_pool.pl
eg/idempotent_insert.pl
eg/insert.pl
eg/insert_streaming.pl
eg/ipv6.pl
eg/iterate.pl
eg/json.pl
eg/keepalive.pl
eg/log_tail.pl
eg/migration_runner.pl
eg/named_rows.pl
eg/native.pl
eg/native_compress.pl
eg/on_progress.pl
eg/params.pl
eg/ping.pl
eg/pool.pl
eg/query.pl
eg/query_comment.pl
eg/queue.pl
eg/settings.pl
eg/slow_query_log.pl
eg/streaming.pl
eg/tls.pl
eg/totals.pl
eg/types_zoo.pl
eg/uri.pl
eg/with_totals.pl
lib/EV/ClickHouse.pm
LICENSE
Makefile.PL
MANIFEST This list of files
MANIFEST.SKIP
t/05_native.t
t/06_native_types.t
t/07_native_compress.t
t/08_settings.t
t/09_insert_arrayref.t
t/10_raw_query.t
t/11_new_features.t
t/12_advanced.t
t/13_params_uri.t
t/14_new_accessors.t
t/15_streaming.t
t/16_totals_extremes.t
t/17_progress.t
t/18_cancel.t
t/19_timeouts.t
t/20_reconnect.t
t/21_edge_types.t
t/22_named_rows.t
t/23_more_coverage.t
t/24_review_gaps.t
t/25_features.t
Croaks if used with the native protocol.
- `query_timeout => $seconds`
Per-query timeout, overriding the connection-level `query_timeout`.
- `on_data => 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
**not** accumulated, so the final callback receives `(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 },
);
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
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
`idempotent => 1` auto-mints
`insert_deduplication_token`; if your producer issues the SAME logical
batch twice (e.g. retry after a transient network blip) only the first
write lands, by design. To force two distinct logical batches through,
either pass an explicit `idempotent => $token` per batch or
omit the option for fresh inserts. See `eg/idempotent_insert.pl`.
- `on_data` vs `iterate` - which should I pick?
`on_data => sub { }` in the per-query settings is the
lowest-overhead streaming path: each native data block is delivered as
soon as the parser has it, no per-row allocation overhead beyond the
batch arrayref. `iterate` is a synchronous-feeling pull wrapper around
the same machinery - useful when the surrounding code is procedural
(ETL scripts, exporters) and a callback shape doesn't fit. Both are
native-only.
- Connection in front of nginx / reverse proxy strips X-ClickHouse-\* headers
Pass `http_basic_auth => 1` to send the credentials as
`Authorization: Basic ...` instead. Most HTTP gateways forward
eg/cancel_streaming.pl view on Meta::CPAN
#!/usr/bin/env perl
# Cancel a streaming select mid-flight from inside the on_data callback,
# once a condition is met. The connection stays alive for follow-up queries.
use strict;
use warnings;
use EV;
use EV::ClickHouse;
my $ch;
my $blocks_seen = 0;
$ch = EV::ClickHouse->new(
host => $ENV{CLICKHOUSE_HOST} // '127.0.0.1',
eg/csv_export.pl view on Meta::CPAN
my $host = $ENV{CLICKHOUSE_HOST} // '127.0.0.1';
my $nport = $ENV{CLICKHOUSE_NATIVE_PORT} // 9000;
my $rows = $ENV{ROWS} // 1_000_000;
my $out = $ENV{OUT} // 'export.csv';
open my $fh, '>', $out or die "open $out: $!";
$fh->autoflush(0);
my $written = 0;
# Path 1: callback-driven streaming via on_data.
my $ch; $ch = EV::ClickHouse->new(
host => $host, port => $nport, protocol => 'native',
on_connect => sub {
$ch->query(
"select number, toString(now() + number) from numbers($rows)",
{ on_data => sub {
my ($batch) = @_;
# Each $batch is an arrayref of arrayrefs. CSV-quote the
# second field (timestamp string) to handle any commas.
for my $row (@$batch) {
eg/health_dashboard.pl view on Meta::CPAN
my $probe = EV::timer(0, 5, sub {
for my $i (0 .. $#conns) {
$conns[$i]->ping_round_trip(sub {
my ($s, $err) = @_;
$rtt[$i] = $err ? undef : $s;
});
}
});
# Tiny HTTP server. Single-shot, no keep-alive, no streaming - just
# enough to demonstrate the JSON shape.
my $listener = IO::Socket::INET->new(
Listen => 16, LocalAddr => '0.0.0.0', LocalPort => $dash_port,
ReuseAddr => 1, Blocking => 0,
) or die "listen $dash_port: $!";
my $accept_io = EV::io($listener->fileno, EV::READ, sub {
while (my $cli = $listener->accept) {
$cli->blocking(0);
my $buf = '';
eg/iterate.pl view on Meta::CPAN
#!/usr/bin/env perl
# Pull-iterator: synchronous-feeling consumption of a streaming select.
# Useful for procedural ETL / export pipelines where callback-driven
# code doesn't fit the rest of the program. Native protocol only:
# iterate() relies on the per-block on_data hook and croaks on HTTP.
use strict;
use warnings;
use EV;
use EV::ClickHouse;
my $ch = EV::ClickHouse->new(
host => $ENV{CLICKHOUSE_HOST} // '127.0.0.1',
lib/EV/ClickHouse.pm view on Meta::CPAN
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 },
);
lib/EV/ClickHouse.pm view on Meta::CPAN
my $it = $ch->iterate("select number from numbers(1_000_000)");
while (my $batch = $it->next($timeout)) {
process($_) for @$batch;
}
die $it->error if $it->error;
B<Native protocol only> - relies on the per-block C<on_data> hook and
will croak if invoked on an HTTP connection.
Synchronous-feeling pull iterator over a streaming select. Internally
wraps the native C<on_data> per-block callback and drives the EV loop
from inside C<-E<gt>next> until the next block arrives, the query
completes, or the optional timeout (seconds) expires. Useful for
procedural ETL / export code that doesn't fit a callback shape.
C<-E<gt>error>, C<-E<gt>is_done>, and C<-E<gt>cancel> are also
available on the returned iterator object.
=head2 on_log
lib/EV/ClickHouse.pm view on Meta::CPAN
on_batch_error => sub { warn "batch err: $_[0]" }, # per-failure
);
while (my $row = next_event()) {
$s->push_row($row);
}
$s->finish(sub {
my (undef, $err) = @_;
die "ingest failed: $err" if $err;
});
Buffered streaming insert for ETL workloads. Rows are buffered until
C<batch_size> is reached, then dispatched as a single C<insert()>.
Dispatches are serialised; push_row keeps buffering while a batch is
in flight (the native protocol cannot pipeline INSERTs). C<finish>
flushes the remaining buffer and fires its callback once all batches
complete; if any batch failed the first error is delivered as
C<$err>. The streamer also offers C<buffered_count> and C<in_flight>
accessors for backpressure logic.
C<<< $streamer->reset >>> discards any rows still in the local buffer
and clears the sticky error so the streamer can be reused after a
lib/EV/ClickHouse.pm view on Meta::CPAN
C<<< idempotent =E<gt> 1 >>> auto-mints
C<insert_deduplication_token>; if your producer issues the SAME logical
batch twice (e.g. retry after a transient network blip) only the first
write lands, by design. To force two distinct logical batches through,
either pass an explicit C<<< idempotent =E<gt> $token >>> per batch or
omit the option for fresh inserts. See F<eg/idempotent_insert.pl>.
=item C<on_data> vs C<iterate> - which should I pick?
C<<< on_data =E<gt> sub { } >>> in the per-query settings is the
lowest-overhead streaming path: each native data block is delivered as
soon as the parser has it, no per-row allocation overhead beyond the
batch arrayref. C<iterate> is a synchronous-feeling pull wrapper around
the same machinery - useful when the surrounding code is procedural
(ETL scripts, exporters) and a callback shape doesn't fit. Both are
native-only.
=item Connection in front of nginx / reverse proxy strips X-ClickHouse-* headers
Pass C<<< http_basic_auth =E<gt> 1 >>> to send the credentials as
C<Authorization: Basic ...> instead. Most HTTP gateways forward
t/11_new_features.t view on Meta::CPAN
cb => sub {
$ch->query("select nonexistent_column from system.one", sub {
my ($rows, $err) = @_;
ok($err, 'error_code: got error');
like($err, qr/Code: \d+/, 'error_code: contains Code: N');
EV::break;
});
},
);
# Test 24-26: streaming on_data callback (native)
with_native(
tests => 3,
cb => sub {
my @blocks;
$ch->query(
"select number from numbers(100)",
{ on_data => sub { push @blocks, $_[0] } },
sub {
my ($rows, $err) = @_;
ok(!$err, 'on_data: no error');
t/15_streaming.t view on Meta::CPAN
use strict;
use warnings;
use Test::More;
use EV;
use EV::ClickHouse;
# on_data streaming callback (native protocol): rows are delivered per-block
# as they arrive; the final callback gets undef rows (no accumulation).
my $host = $ENV{TEST_CLICKHOUSE_HOST} || '127.0.0.1';
my $port = $ENV{TEST_CLICKHOUSE_NATIVE_PORT} || 9000;
require IO::Socket::INET;
plan skip_all => "ClickHouse native port not reachable"
unless IO::Socket::INET->new(PeerAddr => $host, PeerPort => $port, Timeout => 2);
plan tests => 9;
t/15_streaming.t view on Meta::CPAN
port => $port,
protocol => 'native',
on_connect => sub { $cb->() },
on_error => sub { diag("error: $_[0]"); EV::break },
);
my $t = EV::timer(15, 0, sub { EV::break });
EV::run;
$ch->finish if $ch && $ch->is_connected;
}
# 1-4: streaming a moderate result set produces multiple blocks.
with_native(sub {
my $blocks = 0;
my $streamed_rows = 0;
$ch->query(
"select number from numbers(100000)",
{
on_data => sub {
my ($rows) = @_;
$blocks++;
$streamed_rows += scalar @$rows;
},
},
sub {
my ($rows, $err) = @_;
ok(!$err, "streaming: no error") or diag $err;
ok($blocks > 1, "streaming: got multiple blocks ($blocks)");
is($streamed_rows, 100000, "streaming: total rows match");
ok(!defined $rows, "streaming: final callback gets undef rows");
EV::break;
},
);
});
# 5-7: with on_data, last_query_id and column_names still work.
with_native(sub {
my $blocks = 0;
$ch->query(
"select number, toString(number) as s from numbers(50000)",
{
query_id => 'streaming-test',
on_data => sub { $blocks++ },
},
sub {
my (undef, $err) = @_;
ok(!$err, "streaming with metadata: no error") or diag $err;
is($ch->last_query_id, 'streaming-test', "query_id set");
is_deeply($ch->column_names, ['number', 's'], "column_names captured");
EV::break;
},
);
});
# 8-9: empty result still fires the final callback (and on_data may or may
# not fire, depending on whether the server emits an empty data block).
with_native(sub {
my $blocks = 0;
$ch->query(
"select number from numbers(1) where number > 999",
{ on_data => sub { $blocks++; } },
sub {
my ($rows, $err) = @_;
ok(!$err, "streaming empty: no error") or diag $err;
ok(!defined $rows, "streaming empty: final rows undef");
EV::break;
},
);
});
t/24_review_gaps.t view on Meta::CPAN
$ch->cancel; # nothing pending — must not crash
$ch->query("select 1", sub { ($rows) = @_; EV::break });
},
);
run_with_timeout(10);
is($rows && @$rows ? $rows->[0][0] : undef, 1,
"cancel() with no in-flight is a no-op; subsequent query succeeds");
$ch->finish if $ch->is_connected;
}
# 13: on_data on HTTP protocol must croak (or error-deliver) — streaming is
# native-only.
SKIP: {
skip "HTTP port not reachable", 1 unless $http_ok;
my $ch;
$ch = EV::ClickHouse->new(
host => $host, port => $http_port,
on_connect => sub { EV::break },
);
run_with_timeout(5);
t/25_features.t view on Meta::CPAN
use warnings;
use Test::More;
use EV;
use EV::ClickHouse;
# Tests for the 0.03 feature batch:
# - max_reconnect_attempts
# - HTTP keepalive PING
# - progress_period coalescing
# - for_table schema helper
# - insert_streamer streaming insert
# - cancel during on_data
# - on_disconnect not firing on connect-phase failures
my $host = $ENV{TEST_CLICKHOUSE_HOST} || '127.0.0.1';
my $http_port = $ENV{TEST_CLICKHOUSE_PORT} || 8123;
my $nat_port = $ENV{TEST_CLICKHOUSE_NATIVE_PORT} || 9000;
require IO::Socket::INET;
my $http_ok = IO::Socket::INET->new(PeerAddr => $host, PeerPort => $http_port, Timeout => 2) ? 1 : 0;
my $nat_ok = IO::Socket::INET->new(PeerAddr => $host, PeerPort => $nat_port, Timeout => 2) ? 1 : 0;
t/25_features.t view on Meta::CPAN
});
},
);
run_with_timeout(10);
ok($rows && @$rows, "progress test: query completed");
cmp_ok(scalar @ticks, '<=', 5,
"progress_period throttles to <=5 fires for a sub-second query") or diag "got " . scalar(@ticks) . " ticks";
$ch->finish if $ch->is_connected;
}
# 9-10: cancel during on_data (mid-stream cancel from inside the streaming cb).
# Native CLIENT_CANCEL doesn't raise an error — what matters is (a) the query
# callback fires (no hang) and (b) the connection survives for follow-up.
SKIP: {
skip "Native port not reachable", 2 unless $nat_ok;
my ($ch, $blocks, $cb_fired, $follow_ok);
$ch = EV::ClickHouse->new(
host => $host, port => $nat_port, protocol => 'native',
on_connect => sub {
$blocks = 0;
$ch->query(
t/33_progress_accuracy.t view on Meta::CPAN
my $ch; $ch = EV::ClickHouse->new(
host => $host, port => $nport, protocol => 'native',
on_progress => sub {
my ($rows) = @_; # ($rows, $bytes, $total_rows, ...)
$progress_total += $rows;
},
on_connect => sub {
$ch->query(
"select number from numbers($rows_target)",
{ on_data => sub { } }, # streaming so progress fires
sub {
$profile_rows = $ch->profile_rows;
EV::break;
},
);
},
);
my $bail = EV::timer(20, 0, sub { EV::break });
EV::run;
undef $bail;
xs/proto_native_parse.c view on Meta::CPAN
if (num_cols > 0)
av_extend(row, num_cols - 1);
for (c = 0; c < num_cols; c++) {
av_push(row, columns[c][r]);
}
av_push(*target, newRV_noinc((SV*)row));
}
}
}
/* Fire on_data streaming callback if set (only for DATA, not TOTALS/EXTREMES) */
{
SV *on_data = (ptype == SERVER_DATA) ? peek_cb_on_data(self) : NULL;
if (on_data && self->native_rows) {
/* Hold a reference across call_sv: a reentrant
* skip_pending() / cancel() in the handler would
* otherwise pop the cb_queue entry and free this
* callback while we're still invoking it. */
SvREFCNT_inc(on_data);
self->callback_depth++;
{
( run in 1.005 second using v1.01-cache-2.11-cpan-600a1bdf6e4 )