ClickHouse-Encoder
view release on metacpan or search on metacpan
lib/ClickHouse/Encoder.pm view on Meta::CPAN
package ClickHouse::Encoder;
use strict;
use warnings;
our $VERSION = '0.01';
require XSLoader;
XSLoader::load('ClickHouse::Encoder', $VERSION);
# Validate a [db.]table identifier as ASCII word characters only.
# Rejects anything that could inject SQL via the --query argument.
sub _validate_table_name {
my $table = shift;
$table =~ /\A[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)?\z/
or die "Invalid table name '$table': expected [db.]name with [A-Za-z0-9_]";
return;
}
# Build (url, headers) for a ClickHouse HTTP request with the given SQL
# in the `query` parameter. Pulls connection params from %opts using
# the same defaults as for_table / insert_http. UTF-8 encodes before
# percent-escaping so non-ASCII (caf%C3%A9, emoji) round-trips correctly.
sub _http_url_headers {
my ($sql, %opts) = @_;
require Encode;
my $esc = sub {
my $s = Encode::encode('UTF-8', $_[0], 0);
$s =~ s/([^A-Za-z0-9\-_.~])/sprintf('%%%02X', ord($1))/ge;
$s;
};
my ($scheme, $host, $port) = _check_endpoint(\%opts);
my $database = $opts{database} // 'default';
my $user = $opts{user} // 'default';
my $password = $opts{password} // '';
my $url = "$scheme://$host:$port/?database=" . $esc->($database);
$url .= "&query=" . $esc->($sql) if length $sql;
# Per-query settings: { max_memory_usage => '...', max_execution_time => 30 }
if (my $s = $opts{settings}) {
for my $k (sort keys %$s) {
$url .= "&" . $esc->($k) . "=" . $esc->($s->{$k});
}
}
# Insert-side idempotency token: identical token + payload is rejected.
if (defined(my $tok = $opts{dedup_token})) {
$url .= "&insert_deduplication_token=" . $esc->($tok);
}
my %hdr = ('X-ClickHouse-User' => $user);
$hdr{'X-ClickHouse-Key'} = $password if $password ne '';
return ($url, \%hdr);
}
# Validate the host/port/scheme triple shared by every HTTP entry point.
# Rejects anything other than http/https, ensures the port is a positive
# integer, and refuses host strings that contain URL-structural characters
# (':/?#&'). Centralised here so insert_http, bulk_inserter, ping, and
# select_blocks share a single allow-list and identical error messages.
sub _check_endpoint {
my ($opts) = @_;
my $scheme = $opts->{scheme} // 'http';
my $host = $opts->{host} // 'localhost';
my $port = $opts->{port} // 8123;
die "endpoint: scheme must be 'http' or 'https' (got '$scheme')\n"
unless $scheme eq 'http' || $scheme eq 'https';
die "endpoint: host must not contain URL-structural characters "
. "(got '$host')\n"
if $host =~ m{[:/?#&\s]} || !length $host;
die "endpoint: port must be a positive integer (got '$port')\n"
unless $port =~ /\A[1-9]\d{0,4}\z/ && $port < 65536;
return ($scheme, $host, $port);
}
# Build an HTTP::Tiny instance honoring ssl_options (verify_SSL, SSL_ca_file,
# etc.) and keep_alive. Shared by insert_http, bulk_inserter, server_version,
# ping, select_blocks; callers pass %opts unchanged. Loading HTTP::Tiny here
# keeps the require local to HTTP code paths.
sub _http_tiny {
my (%opts) = @_;
require HTTP::Tiny;
my @args = (timeout => $opts{timeout} // 60);
push @args, keep_alive => 1 if $opts{keep_alive};
push @args, SSL_options => $opts{ssl_options} if $opts{ssl_options};
push @args, verify_SSL => $opts{verify_SSL} if exists $opts{verify_SSL};
return HTTP::Tiny->new(@args);
}
# Parse a flat CH JSON object string (X-ClickHouse-Summary /
# X-ClickHouse-Progress) without depending on JSON::PP. Both are small
# flat objects of stringified integers (read_rows, written_rows,
# total_rows_to_read, elapsed_ns, ...). Returns a hashref or undef.
sub _parse_ch_kv {
my ($str) = @_;
return unless defined $str && length $str;
my %h;
# NB: stash $1/$2 before the digit-test regex - that regex resets
# capture variables and would silently turn every key into ''.
while ($str =~ /"([^"\\]+)"\s*:\s*"([^"\\]*)"/g) {
my ($k, $v) = ($1, $2);
$h{$k} = ($v =~ /\A-?\d+\z/) ? $v + 0 : $v;
}
return scalar(keys %h) ? \%h : undef;
}
# Lift a few ClickHouse response headers into a small hashref. Returns
# undef when none are present; otherwise carries query_id, server
# (revision), format, a parsed summary, and the final progress snapshot
# so callers don't reparse the same headers. X-ClickHouse-Progress is
# sent repeatedly while a query runs (with send_progress_in_http_headers
# =1); HTTP::Tiny collapses repeats into an arrayref, so we take the
# last - the most complete - snapshot.
sub _decorate_response {
my ($resp) = @_;
return $resp unless ref $resp eq 'HASH';
my $h = $resp->{headers} or return $resp;
my %ch;
for my $k (qw(query-id server format exception-code)) {
my $hv = $h->{"x-clickhouse-$k"};
$ch{$k} = $hv if defined $hv;
}
if (my $sum = _parse_ch_kv($h->{'x-clickhouse-summary'})) {
$ch{summary} = $sum;
}
if (defined(my $pv = $h->{'x-clickhouse-progress'})) {
$pv = $pv->[-1] if ref $pv eq 'ARRAY';
if (my $prog = _parse_ch_kv($pv)) {
$ch{progress} = $prog;
}
}
lib/ClickHouse/Encoder.pm view on Meta::CPAN
my $chunk_size = $opts{chunk_size} // 64 * 1024;
my $keep = $opts{keep};
my $decompress = $opts{decompress};
my $buf = ''; # raw bytes from the filehandle
my $inner = ''; # decompressed Native bytes (== $buf when !decompress)
my $done = 0;
until ($done) {
# Phase 1: peel compressed-block frames out of $buf into $inner.
if ($decompress) {
while (length($buf) >= 25) {
my $csize = unpack 'V', substr($buf, 17, 4);
last if length($buf) < 16 + $csize;
my ($plain, $consumed) = eval {
$class->decompress_native_block($buf)
};
die $@ if $@;
$inner .= $plain;
substr($buf, 0, $consumed, '');
}
} else {
$inner = $buf;
}
# Phase 2: decode complete Native blocks out of $inner.
while (length($inner) > 0) {
my $block = eval { $class->decode_block($inner, 0, $keep) };
if ($@) {
# Truncation or malformed mid-block; need more bytes.
# Only "buffer truncated" means "data ran short, read
# more"; "exceeds remaining" indicates a malformed wire
# value (e.g. a corrupted varint count) and should die
# rather than spin reading more bytes.
last if $@ =~ /buffer truncated/i;
die $@; # real error
}
$cb->($block);
substr($inner, 0, $block->{consumed}, '');
}
# When not decompressing, $inner aliases $buf - carry the residual
# back so the next read sees the unconsumed tail.
$buf = $inner unless $decompress;
# Read more bytes. read() returns 0 on EOF, undef on error.
my $more;
my $n = read $fh, $more, $chunk_size;
die "decode_stream: read error: $!" if !defined $n;
if ($n == 0) {
# EOF. If anything is left in either buffer it's a truncated
# final block; raise rather than swallow.
die "decode_stream: " . length($buf) . " trailing bytes "
. "after last complete compressed block"
if $decompress && length $buf;
die "decode_stream: " . length($inner) . " trailing bytes "
. "after last complete block" if length $inner;
$done = 1;
} else {
$buf .= $more;
}
}
return;
}
# Query the ClickHouse HTTP endpoint for its version. Returns a list
# of (major, minor, patch, build) integers and the raw string. Useful
# for capability gating in user code (e.g. only use JSON columns if
# the server is at least 24.8). HTTP-only; native TCP not supported.
sub server_version {
my ($class, %opts) = @_;
my ($url, $hdr) = _http_url_headers('select version()', %opts);
my $resp = _http_tiny(%opts, timeout => $opts{timeout} // 10)
->get($url, { headers => $hdr });
die "HTTP select version() failed (status $resp->{status}): "
. "$resp->{content}\n"
unless $resp->{success};
(my $raw = $resp->{content}) =~ s/\s+\z//;
my @parts = ($raw =~ /(\d+)/g);
return wantarray
? (@parts, $raw)
: { major => $parts[0] // 0, minor => $parts[1] // 0,
patch => $parts[2] // 0, build => $parts[3] // 0,
raw => $raw };
}
# Lightweight liveness check via CH's /ping endpoint. Returns 1 on
# success; croaks on HTTP failure (which includes connection refused,
# timeout, or non-2xx response). Use to gate on server availability in
# bootstrap scripts and bulk-load orchestration.
sub ping {
my ($class, %opts) = @_;
my ($scheme, $host, $port) = _check_endpoint(\%opts);
my $url = "$scheme://$host:$port/ping";
my $resp = _http_tiny(%opts, timeout => $opts{timeout} // 5)->get($url);
die "ping: HTTP $resp->{status}: $resp->{content}\n"
unless $resp->{success};
return 1;
}
# Parse a Well-Known-Text (WKT) geometry string into the nested-arrayref
# representation that the Geo column encoders accept. Supported geometries:
# POINT, LINESTRING, MULTILINESTRING, POLYGON, MULTIPOLYGON. Coordinates
# are decimal numbers separated by whitespace; rings/parts are
# parenthesized. The Ring CH type has no WKT name (it is a single closed
# LINESTRING); accept LINESTRING for Ring as well. Returns the structure
# only; the caller chooses which Geo column to feed it into.
sub parse_wkt {
my ($class, $wkt) = @_;
die "parse_wkt: input required\n" unless defined $wkt;
$wkt =~ s/\A\s+//;
$wkt =~ s/\s+\z//;
my ($kind, $rest) = $wkt =~ /\A([A-Za-z]+)\s*(.*)\z/s
or die "parse_wkt: not a WKT geometry: $wkt\n";
$kind = uc $kind;
# Strip the outermost parens (which every geometry has) once for
# uniform downstream parsing. EMPTY is rejected because CH Geo
# columns have no null/empty representation other than zero-length.
$rest =~ s/\A\(\s*//
or die "parse_wkt: $kind missing '(': $wkt\n";
$rest =~ s/\s*\)\z//
or die "parse_wkt: $kind unmatched parens: $wkt\n";
if ($kind eq 'POINT') {
return _wkt_point($rest);
}
if ($kind eq 'LINESTRING') {
return [ _wkt_points($rest) ];
}
if ($kind eq 'MULTILINESTRING' || $kind eq 'POLYGON') {
# Same shape on the wire (CH Polygon = ring + holes; outer ring
# comes first; MultiLineString is parallel parts). Parse as a
# list of paren-wrapped point-lists.
my @parts;
while ($rest =~ /\G\s*,?\s*\(\s*([^()]+?)\s*\)/gc) {
push @parts, [ _wkt_points($1) ];
}
die "parse_wkt: $kind: no parts parsed in $wkt\n" unless @parts;
return \@parts;
}
if ($kind eq 'MULTIPOLYGON') {
my @polys;
# MULTIPOLYGON(((...),(...)), ((...))): the outer level is
# polygons, each polygon is a list of rings.
while ($rest =~ /\G\s*,?\s*\(\s*(.*?)\s*\)\s*(?=,|\z)/gcs) {
my $poly_body = $1;
my @rings;
while ($poly_body =~ /\G\s*,?\s*\(\s*([^()]+?)\s*\)/gc) {
push @rings, [ _wkt_points($1) ];
}
die "parse_wkt: MULTIPOLYGON ring parse failed in $wkt\n"
unless @rings;
push @polys, \@rings;
}
lib/ClickHouse/Encoder.pm view on Meta::CPAN
$sum += _type_byte_estimate($p, $avg_str);
}
return $sum;
}
# LowCardinality: dict + per-row 1-byte index (typical low cardinality).
# Variant: 1 disc byte + (heuristically) inner avg.
return 1 + $avg_str if $base eq 'LowCardinality' || $base eq 'Variant';
# JSON/Object/Dynamic: shape-dependent. Heuristic: roughly two
# avg_string_size payloads per row (one for path machinery + one
# for value bytes), so the caller's avg_string_size override
# actually moves the estimate.
return 2 * $avg_str if $base eq 'JSON' || $base eq 'Object' || $base eq 'Dynamic';
# SimpleAggregateFunction(func, T) -> inner T.
if ($base eq 'SimpleAggregateFunction') {
my ($inner) = $type =~ /^SimpleAggregateFunction\([^,]+,\s*(.+)\)$/;
return _type_byte_estimate($inner, $avg_str) if defined $inner;
}
return $avg_str; # unknown -> conservative
}
# Coarse byte-size estimate for an encoded block, parameterized on
# row count (an integer) or arrayref (counted). Uses per-type byte
# budgets; variable-length types use a 16-byte average per value
# (override via $avg_str). For batch-size decisions ("is this 1 MiB
# or 100 MiB before I compress?"). NOT byte-exact: a String row with
# a 10 KiB blob will be undercounted. Run encode() for the real size.
sub estimate_size {
my ($self, $rows_or_n, %opts) = @_;
my $n = ref $rows_or_n eq 'ARRAY' ? scalar @$rows_or_n : $rows_or_n;
my $avg_str = $opts{avg_string_size} // 16;
my $cols = $self->columns;
my $total = 4; # block header (ncols + nrows varints, ~tiny)
for my $c (@$cols) {
my ($name, $type) = @$c;
$total += length($name) + length($type) + 2; # lenstr headers
$total += $n * _type_byte_estimate($type, $avg_str);
}
return $total;
}
# Return a configured encoder for the column shape produced by an
# arbitrary select. Runs `describe ($sql)` via the same `via=>...`
# transport as for_table. Useful when the schema isn't a real table.
sub for_query {
my ($class, $sql, %opts) = @_;
# ClickHouse's describe accepts a subquery, but the SQL must not
# contain unmatched parentheses; let the server reject malformed
# queries rather than re-implementing SQL validation here.
my $describe = "describe ($sql)";
return $class->_for_describe($describe, %opts);
}
# Issue an insert ... format native over HTTP using HTTP::Tiny. Returns
# the response hashref from HTTP::Tiny (->{success}, ->{status},
# ->{content}). Compresses with zstd/gzip if `compress` is set; takes
# whatever encoder produces (so `for_table` + rows is the typical
# combination). Does not retry; the caller does HTTP-level error policy.
# Set up URL + headers for an insert ... format native HTTP request.
# Shared by insert_http and BulkInserter::new. Validates the table name
# and stamps the Content-Type / Content-Encoding headers as needed.
sub _build_insert_endpoint {
my ($table, $compress, %args) = @_;
_validate_table_name($table);
die "unknown compress='$compress' "
. "(expected 'raw', 'zstd', or 'gzip')\n"
unless $compress eq 'raw' || $compress eq 'zstd'
|| $compress eq 'gzip';
my ($url, $hdr) = _http_url_headers(
"insert into $table format native", %args);
$hdr->{'Content-Type'} = 'application/octet-stream';
$hdr->{'Content-Encoding'} = $compress
if $compress eq 'zstd' || $compress eq 'gzip';
return ($url, $hdr);
}
# Apply zstd/gzip compression to $body in place (or pass through for 'raw').
# $compress is validated upstream by _build_insert_endpoint; we trust it
# here. $origin is the class used to resolve compressed_writer (so the
# helper works for class-method and instance callers alike).
sub _apply_compression {
my ($origin, $compress, $body) = @_;
return $body if $compress eq 'raw';
my $compressed;
my $wrap = $origin->compressed_writer(
$compress, sub { $compressed = $_[0] });
$wrap->($body);
return $compressed;
}
sub insert_http {
my ($class_or_self, %args) = @_;
my $enc = $args{encoder} // do {
my $cols = $args{columns} or die "insert_http needs columns or encoder";
$class_or_self->new(columns => $cols);
};
my $rows = $args{rows} or die "insert_http needs rows arrayref";
my $table = $args{table} or die "insert_http needs table";
my $timeout = $args{timeout} // 60;
my $compress = $args{compress} // 'raw';
my $origin = ref $class_or_self || $class_or_self;
my ($url, $hdr) = _build_insert_endpoint($table, $compress, %args);
my $body = _apply_compression($origin, $compress, $enc->encode($rows));
my $resp = _http_tiny(%args, timeout => $timeout)
->post($url, { headers => $hdr, content => $body });
return _decorate_response($resp);
}
# Stream a select response: POST the SQL with default_format=Native,
# feed the response chunks into a sliding buffer, decode complete blocks
# as they arrive, and pass each one to $opts{on_block}. Memory stays
# bounded by chunk_size + one block, so this is the right entry point
# for selects that won't fit in memory. The user's $sql must NOT include
# a format clause - this helper always requests format Native.
sub select_blocks {
my ($class, $sql, %opts) = @_;
my $cb = $opts{on_block}
or die "select_blocks: 'on_block' coderef required\n";
die "select_blocks: \$sql should not include a format clause "
. "(select_blocks always requests format Native)\n"
if $sql =~ /\bformat\s+\w+\s*\z/i;
my $keep = $opts{keep};
my $decompress = $opts{decompress};
# Build URL+headers via the same helper insert_http uses, but with an
# empty SQL placeholder; we POST the SQL as the request body. Adding
# default_format=Native ensures the response is Native bytes even if
# the user's SQL doesn't terminate with format.
my %h_opts = %opts;
# Drop keys this method consumes; also drop dedup_token, which is
# meaningful only on insert (would be silently dead weight on the
# select URL otherwise and could mask a typo by the caller).
delete @h_opts{qw(on_block keep timeout decompress dedup_token)};
my ($url, $hdr) = _http_url_headers('', %h_opts);
$url .= '&default_format=Native';
# When decompress=1 is requested the server wraps each Native block
# in its compressed-block framing (X-ClickHouse-Compressed header).
# Add ?compress=1 to the URL so CH knows to compress the response.
$url .= '&compress=1' if $decompress;
my $buf = '';
# Block walker: when decompress is set, walk through compressed-block-
# framing entries and feed the decompressed bytes into a second
# accumulator that decode_block reads. Otherwise feed buf directly.
my $inner_buf = '';
my $drain = sub {
# Phase 1: pull compressed-block frames out of $buf into $inner_buf
if ($decompress) {
while (length($buf) >= 25) { # 16 hash + 9 header minimum
my $csize = unpack 'V', substr($buf, 17, 4);
last if length($buf) < 16 + $csize;
my ($plain, $consumed) =
$class->decompress_native_block($buf);
$inner_buf .= $plain;
substr($buf, 0, $consumed, '');
}
} else {
$inner_buf = $buf;
}
# Phase 2: decode whole Native blocks out of $inner_buf
while (length($inner_buf) > 0) {
lib/ClickHouse/Encoder.pm view on Meta::CPAN
if ($@) {
last if $@ =~ /buffer truncated/i;
die $@;
}
$cb->($block);
substr($inner_buf, 0, $block->{consumed}, '');
}
# When not decompressing, inner_buf IS buf; carry the residual
# back so the next data_callback append sees the unconsumed tail.
if (!$decompress) {
$buf = $inner_buf;
}
};
my $resp = _http_tiny(%opts, timeout => $opts{timeout} // 60)->post(
$url,
{ content => $sql,
headers => { %$hdr, 'Content-Type' => 'text/plain' },
data_callback => sub { $buf .= $_[0]; $drain->() },
});
die "select_blocks: HTTP $resp->{status}: $resp->{content}\n"
unless $resp->{success};
$drain->();
die "select_blocks: " . length($buf) . " trailing bytes "
. "after last complete compressed block\n"
if $decompress && length $buf;
die "select_blocks: " . length($inner_buf) . " trailing bytes "
. "after last complete block\n"
if length $inner_buf;
return;
}
# Returns a bulk-inserter object: ->push($row), ->push_many(\@rows),
# ->flush (idempotent), ->finish. Holds a single HTTP::Tiny instance
# across batches (so keepalive applies) and auto-flushes when the
# accumulated row count crosses batch_size. Transient HTTP failures
# (5xx, network errors) are retried up to retries times with linear
# backoff; 4xx errors die immediately.
sub bulk_inserter {
my ($class_or_self, %args) = @_;
return ClickHouse::Encoder::BulkInserter->new(%args,
_origin => $class_or_self);
}
package ClickHouse::Encoder::BulkInserter; ## no critic (ProhibitMultiplePackages)
sub new {
my ($class, %args) = @_;
my $origin_raw = delete $args{_origin};
my $origin = (ref $origin_raw || $origin_raw) || 'ClickHouse::Encoder';
my $enc = $args{encoder} // do {
my $cols = $args{columns} or die "bulk_inserter needs columns or encoder";
$origin->new(columns => $cols);
};
my $table = $args{table} or die "bulk_inserter needs table";
my $compress = $args{compress} // 'raw';
my $timeout = $args{timeout} // 60;
my ($url, $hdr) = ClickHouse::Encoder::_build_insert_endpoint(
$table, $compress, %args);
return bless {
enc => $enc,
url => $url,
hdr => $hdr,
rows => [],
batch_size => $args{batch_size} // 10_000,
retries => $args{retries} // 3,
retry_wait => $args{retry_wait} // 0.5,
retry_max_wait => $args{retry_max_wait} // 30,
compress => $compress,
http => ClickHouse::Encoder::_http_tiny(
%args, timeout => $timeout, keep_alive => 1),
origin => $origin,
sent_rows => 0,
sent_batches => 0,
last_response => undef,
summary => {},
}, $class;
}
sub push :method { ## no critic (ProhibitBuiltinHomonyms)
my ($self, $row) = @_;
CORE::push @{ $self->{rows} }, $row;
$self->flush if @{ $self->{rows} } >= $self->{batch_size};
return $self;
}
sub push_many {
my ($self, $rows) = @_;
CORE::push @{ $self->{rows} }, @{$rows};
# Slice exactly batch_size rows per flush so we never POST one
# oversized body when push_many is called with N >> batch_size.
# `local` restores $self->{rows} to the remainder arrayref even
# when `flush` croaks mid-batch (so caller's eval{} sees the
# untried rows still buffered for a retry).
while (@{ $self->{rows} } > $self->{batch_size}) {
my @batch = splice @{ $self->{rows} }, 0, $self->{batch_size};
local $self->{rows} = \@batch;
$self->flush;
}
$self->flush if @{ $self->{rows} } >= $self->{batch_size};
return $self;
}
sub flush {
my $self = shift;
my $rows = $self->{rows};
return $self if !@{$rows};
my $body = ClickHouse::Encoder::_apply_compression(
$self->{origin}, $self->{compress}, $self->{enc}->encode($rows));
my $resp;
my $last_err;
for my $attempt (0 .. $self->{retries}) {
$resp = $self->{http}->post($self->{url},
{ headers => $self->{hdr}, content => $body });
last if $resp->{success};
# 4xx errors are not retryable - the request is malformed.
die "bulk_inserter: HTTP $resp->{status}: $resp->{content}\n"
lib/ClickHouse/Encoder.pm view on Meta::CPAN
to L</decode_block> for every block in the stream, matching the
same semantics: present keys are decoded, absent ones still have
their bytes consumed (to keep the cursor aligned) but their values
are not materialized and their column hash carries C<skipped =E<gt> 1>.
Useful for big-fan-out select responses where only a few columns
of a wide row matter.
ClickHouse::Encoder->decode_blocks($bytes, $cb,
keep => { id => 1, event => 1 });
=head2 decode_blocks_iter
my $iter = ClickHouse::Encoder->decode_blocks_iter($bytes);
while (my $block = $iter->()) { ... }
my $iter = ClickHouse::Encoder->decode_blocks_iter($bytes,
keep => { id => 1 });
Returns a coderef that yields one block per call (C<undef> when
exhausted). Same per-block payload as L</decode_block>; useful when
you want pull-style iteration without committing to a callback.
Accepts the same C<keep> filter as L</decode_blocks>.
=head2 decode_stream
ClickHouse::Encoder->decode_stream($fh, sub { my $block = shift; ... },
chunk_size => 65536);
ClickHouse::Encoder->decode_stream($fh, $cb,
keep => { id => 1, event => 1 });
Pull bytes incrementally from a filehandle (or any read-able IO
handle), yielding each complete block to the callback as it
arrives. Uses a sliding buffer; on a truncated decode it reads more
bytes and retries. Memory stays bounded by C<chunk_size> + one
block, so this is the right entry point for select responses too
large to buffer in full. Croaks on partial trailing bytes.
The C<keep> filter is the same one L</decode_block> accepts:
unwanted columns still have their bytes consumed (so the cursor
stays aligned) but their values are not materialized into an SV
array, so peak memory stays bounded by the kept columns.
With C<decompress =E<gt> 1>, C<$fh> is expected to deliver a stream
of compressed-block-framed Native blocks (the format CH's HTTP
C<?compress=1> response uses, or a captured native-TCP Data stream
under compression). C<decode_stream> peels each compressed block
via L</decompress_native_block> before feeding the resulting raw
Native bytes into L</decode_block>.
C<$fh> must support Perl's C<read()> builtin (any plain filehandle
or L<IO::Handle> subclass). Raw socket descriptors that only
support C<sysread> need to be wrapped via L<IO::Socket> or read
into a buffer that is then fed to L</decode_block> directly.
=head2 ping
ClickHouse::Encoder->ping(host => 'db', port => 8123);
ClickHouse::Encoder->ping(scheme => 'https', host => 'db', port => 8443);
Liveness check via CH's C</ping> endpoint. Returns C<1> on success;
croaks on connection refused, timeout, or any non-2xx HTTP status.
Accepts the same C<scheme>/C<host>/C<port>/C<timeout>/C<ssl_options>
options as the rest of the HTTP entry points.
=head2 server_version
my $v = ClickHouse::Encoder->server_version(
host => 'db', port => 8123);
if ($v->{major} >= 24) { ... }
Fetches C<select version()> over HTTP. In scalar context returns
C<<< { major, minor, patch, build, raw } >>>; in list context
returns C<($major, $minor, $patch, $build, $raw)>. Useful for
capability gating in user code. Accepts the same
C<scheme>/C<host>/C<port>/C<database>/C<user>/C<password>/C<timeout>/
C<ssl_options>/C<verify_SSL>/C<settings> options as the rest of the
HTTP entry points.
=head2 types
my @t = ClickHouse::Encoder->types;
Returns the list of supported ClickHouse type names (parametric
types as their syntactic prefix, e.g. C<Decimal>, C<Array>). For
runtime feature detection and tooling that wants to introspect
supported types without parsing POD.
=head2 schema_diff
my $d = ClickHouse::Encoder->schema_diff(\@cols_a, \@cols_b);
# $d = {
# added => [[name, type], ...], # in $b but not $a
# removed => [[name, type], ...], # in $a but not $b
# changed => [[name, type_a, type_b], ...],
# }
Compare two column lists (each an arrayref of C<[$name, $type]>
pairs, the shape L</new> takes). Useful for migration scripts and
detecting schema drift between source and destination in CH-to-CH
replication pipelines.
=head2 format_create_table
my $sql = ClickHouse::Encoder->format_create_table(
table => 'events',
columns => [['id','Int32'], ['msg','String']],
engine => 'MergeTree', # default
order_by => '(id)',
partition_by => 'toYYYYMM(ts)', # optional
primary_key => '(id)', # optional
sample_by => 'id', # optional
ttl => 'event_date + INTERVAL 90 DAY', # optional
settings => 'index_granularity=8192', # optional
);
Emits a C<create table> statement string from a column list of the
same shape L</new> takes. The C<table> name is validated by the
same regex L</for_table> / L</insert_http> use, and column names
are backtick-quoted with embedded backticks escaped. The
C<engine> / C<partition_by> / C<primary_key> / C<order_by> /
lib/ClickHouse/Encoder.pm view on Meta::CPAN
C<DateTime> / C<DateTime(tz)> / C<DateTime64(p)> column's values
from raw epoch integers into either ISO 8601 strings (the default;
UTC with a C<Z> suffix) or L<Time::Moment> instances
(C<as =E<gt> 'datetime'>). The block is mutated in place and also
returned, so the call can be chained.
C<Nullable()>-wrapped time columns are handled too; C<undef>
values pass through unchanged. Non-time columns are untouched.
C<DateTime64(p)> precision is honored - the fractional part is
emitted as exactly C<p> digits in ISO form, or as nanoseconds
widened from C<p> ticks in C<Time::Moment> form.
This is a separate post-decode step (rather than an option on
L</decode_block>) so the cost is only paid when the caller wants
formatted values. Raw integer epochs are still cheaper to compare,
filter, or pass back into a re-encode.
=head2 parse_wkt
my $point = ClickHouse::Encoder->parse_wkt('POINT(1.5 2.5)');
my $poly = ClickHouse::Encoder->parse_wkt(
'POLYGON((0 0, 4 0, 4 4, 0 4, 0 0))');
# round-trip into a CH Geo column:
my $enc = ClickHouse::Encoder->new(columns => [
['p', 'Point'], ['poly', 'Polygon']]);
$enc->encode([[ $point, $poly ]]);
Parse a Well-Known-Text geometry string into the nested-arrayref
shape that the Geo column encoders accept. Supports C<POINT>,
C<LINESTRING>, C<MULTILINESTRING>, C<POLYGON>, and C<MULTIPOLYGON>.
The CH C<Ring> type has no WKT name; feed a C<LINESTRING> result
into a Ring column directly. Geometry names are case-insensitive
and surrounding whitespace is tolerated. Malformed input croaks
with a message identifying the offending geometry.
=head2 estimate_size
my $bytes = $enc->estimate_size(\@rows);
my $bytes = $enc->estimate_size($n_rows,
avg_string_size => 64);
Coarse byte-size estimate for an encoded block, parameterized on row
count (an integer or arrayref-of-rows; only the count is used).
Returns an order-of-magnitude figure for batch-split decisions
("should I split this into two POSTs?") without paying the encode
cost. Fixed-width types are byte-exact; variable types
(C<String>, C<Array>, etc.) use a configurable 16-byte average
heuristic. For byte-exact size, call C<length($enc-E<gt>encode(...))>.
=head2 select_blocks
ClickHouse::Encoder->select_blocks(
'select id, event, ts from events where date = today()',
host => 'db.example', port => 8123,
database => 'default', user => 'default',
on_block => sub { my $block = shift; ... },
keep => { id => 1, event => 1 }, # optional projection
);
Streaming counterpart to L</insert_http>: POSTs C<$sql> to the
ClickHouse HTTP endpoint with C<default_format=Native>, feeds the
response chunks into a sliding buffer, and invokes C<on_block> for
every complete L</decode_block>-shaped block as it arrives. Memory
stays bounded by one HTTP::Tiny chunk plus one block; this is the
right entry point for selects that return more than fits in
process memory.
C<$sql> must NOT end with a C<format ...> clause - C<select_blocks>
appends C<format Native> at the URL level and croaks when the SQL
trails with a different format pin. A C<FORMAT> token inside the
query body (for example a column literal) is fine.
The optional C<keep =E<gt> \%names> hashref forwards a column
filter to L</decode_block>: skipped columns still have their bytes
consumed (so the cursor stays aligned) but their values are not
materialized into SVs. Useful when you only need a few of many
select-list columns.
With C<decompress =E<gt> 1> the URL is augmented with
C<?compress=1> so ClickHouse wraps each response Native block in
its compressed-block framing (16-byte CityHash128 + 9-byte header
+ LZ4 payload). The HTTP body is then a stream of compressed blocks
which C<select_blocks> peels and decompresses block-by-block via
L</decompress_native_block> before feeding the result to
L</decode_block>. Memory stays bounded by one HTTP chunk plus one
compressed block plus one decompressed block.
Recognised options (besides C<on_block> / C<keep> /
C<decompress>): C<scheme>, C<host>, C<port>, C<database>, C<user>,
C<password>, C<timeout>, C<ssl_options>, C<verify_SSL>, and
C<settings> (per-query CH settings hashref, useful for
C<max_execution_time> and similar). C<dedup_token> is meaningful
only on insert and is ignored if passed here.
=head2 bulk_inserter
my $bi = ClickHouse::Encoder->bulk_inserter(
host => 'db.example', port => 8123, table => 'events',
columns => \@cols, batch_size => 5000, compress => 'zstd',
retries => 3);
$bi->push([$row]) for @rows;
$bi->finish;
Holds an L<HTTP::Tiny> instance with keep-alive across batches,
accumulates rows, auto-flushes at C<batch_size>, retries transient
HTTP failures (5xx and 599 network errors) with exponential backoff
and jitter. 4xx errors die immediately. Options:
=over 4
=item C<host>, C<port>, C<database>, C<user>, C<password>, C<scheme>, C<timeout>
Same as L</for_table>; passed to L<HTTP::Tiny>.
=item C<table>
Required; same identifier rule as L</for_table>.
=item C<encoder> or C<columns>
Pass either an existing encoder or a column list (used to build one).
lib/ClickHouse/Encoder.pm view on Meta::CPAN
optional column projection via C<--keep>.
=item F<eg/json_path_projection.pl>
Demo of C<keep =E<gt> {...}> projection on top of L</select_blocks>:
decodes only the requested columns and prints one row per line.
=item F<eg/csv_export.pl>
select to CSV: counterpart to F<from_csv.pl>. Drives a CSV writer
from a streaming select, emitting the header row from the first
block's column names.
=item F<eg/migrate_with_transform.pl>
CH-to-CH migration with a row-level transform between read and
write. Discovers source schema via L</for_table>, streams the rows
via L</select_blocks>, applies a user-supplied transform coderef,
and forwards survivors through L</bulk_inserter>.
=item F<eg/replay_pcap.pl>
Replay a captured Native byte stream (e.g. saved from
C<curl --output> of a C<select ... format native> response) and
print a block-by-block summary. Off-line debugging tool.
=item F<eg/tcp_compressed_pipeline.pl>
End-to-end TCP insert pipeline that negotiates compression in
C<pack_query>, then wraps every C<pack_data> / C<pack_data_end>
in CH's compressed-block framing. Showcases the matched-pair
convention against a real ClickHouse server (protocol revision
E<lt>= 54474; see L<ClickHouse::Encoder::TCP/CAVEATS>).
=item F<eg/rowbinary_insert.pl>
insert using the C<RowBinary> format via L</encode_row_binary>,
with a local L</decode_row_binary> round-trip check. For interop
with pipelines that speak RowBinary rather than Native.
=item F<eg/async_insert.pl>
Server-side async insert - C<async_insert=1> (and optionally
C<wait_for_async_insert>) passed through the C<settings> option,
so the server buffers and background-flushes the batch.
=item F<eg/geo_from_wkt.pl>
Ingest geometry given as Well-Known-Text into C<Point> / C<Polygon>
columns using L</parse_wkt> to convert each WKT string.
=item F<eg/insert_with_settings.pl>
insert with per-query CH C<settings> and an C<insert_deduplication_token>,
showing how an identical retry under the same token is deduplicated
server-side.
=item F<eg/ping_healthcheck.pl>
A wait-for-server readiness gate built on L</ping> - retry until the
C</ping> endpoint answers, then proceed.
=item F<eg/observability.pl>
Reads server-side stats from an insert pipeline: per-batch
C<< $bi->last_response->{ch} >> detail plus the cumulative
C<< $bi->summary >> rollup of C<X-ClickHouse-Summary> counters.
=item F<eg/schema_migrate.pl>
Fetches C<show create table>, parses it with L</parse_create_table>,
diffs the columns against a desired schema, and emits the
C<alter table> migration via L</apply_schema_diff>.
=back
For a working reference of the ClickHouse Native binary format that this
module emits, see F<doc/wire-format.md>.
=head1 PERFORMANCE
The encoder is written so that the dominant cost on most workloads is the
data-generation Perl code, not the encoding step. Some indicative numbers
from F<bench/local_insert_benchmark.pl> (500_000 rows, 5 columns including
C<Array(String)>, in-process via C<clickhouse-local>):
Native (this module) encode 0.29s + ingest 0.13s = 0.42s end-to-end
TabSeparated (Perl) encode 0.79s + ingest 0.11s = 0.90s end-to-end
-> Native ~2x faster end-to-end, payload ~18% smaller
For wide tables with many string columns the gap widens (the XS encoder
pulls further ahead of plain-Perl TSV serialization). See F<bench/> for
reproducible scripts and additional scenarios.
=head1 CAVEATS
=over 4
=item *
A 64-bit Perl is required (C<< $Config{ivsize} >= 8 >>). On a 32-bit perl
F<Makefile.PL> exits with C<OS unsupported> (CPAN Testers reports NA, not
FAIL) because C<Int64> / C<UInt64> would otherwise silently truncate.
=item *
Output is little-endian; this matches every supported ClickHouse server
build. The encoder relies on the host's native float byte order matching
its native integer byte order, which holds on every IEEE 754 platform in
practice.
=item *
C<encode> builds the whole block in memory. For very large batches you
typically want to chunk into multiple C<encode> calls and send each block
sequentially.
=item *
L</compress_native_block> and L</decompress_native_block> bundle a
port of CityHash128 v1.0.2 (the "cityhash102" variant ClickHouse uses
( run in 0.484 second using v1.01-cache-2.11-cpan-f4a522933cf )