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.02';
require XSLoader;
XSLoader::load('ClickHouse::Encoder', $VERSION);
# An encoder is a blessed IV holding a raw Encoder* (likewise Streamer).
# ithreads clone SVs verbatim, so without this both interpreters own the
# same pointer and each frees it - a double free on any thread spawn.
# The child's copies become plain undef instead. See THREADS in the docs.
sub CLONE_SKIP { return 1 }
# 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.
# '@' counts: it opens the userinfo section, so 'evil.com@internal.db'
# would send the request to internal.db. IPv6 literals are the one host
# form that legitimately holds colons; accept them bracketed ('[::1]'),
# which is also the form the URL needs to keep the port unambiguous.
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';
my $is_ipv6 = $host =~ /\A\[[0-9A-Fa-f:.]+\]\z/;
die "endpoint: host must not contain URL-structural characters "
. "(got '$host')\n"
if !length $host
|| $host =~ m{[/?#&\@\s]}
|| ($host =~ m{:} && !$is_ipv6);
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 $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. At EOF the
# threshold is ignored - the buffer gets a last chance before it
# is called trailing junk.
while (length($inner) > 0 && ($at_eof || length($inner) >= $need_at)) {
my $block = eval { $class->decode_block($inner, 0, $keep) };
if ($@) {
# Both shapes mean "not enough bytes yet". The prologue's
# "exceeds remaining buffer" guard compares a header count
# against what has arrived, so a block of N rows trips it
# whenever a chunk boundary leaves fewer than N bytes
# buffered - ordinary on real data. Reading more cannot
# loop: EOF ends the outer scan.
if ($@ =~ /buffer truncated \(need (\d+) more/i) {
$need_at = length($inner) + $1;
last;
}
last if $@ =~ /buffer truncated|exceeds remaining buffer/i;
die $@; # real error
}
$cb->($block);
substr($inner, 0, $block->{consumed}, '');
$need_at = 0;
}
# 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. Re-run the scan ungated once, then anything still left
# in either buffer is a truncated final block; raise rather
# than swallow.
if (!$at_eof && length $inner) { $at_eof = 1; next }
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) ? 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 = '';
# As in decode_stream: after a truncation, wait for exactly the byte
# count the decoder said it was short before re-parsing. The forced
# final drain below still attempts the last partial block.
my $need_at = 0;
my $drain = sub {
my ($force) = @_;
# 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, '');
}
lib/ClickHouse/Encoder.pm view on Meta::CPAN
}
};
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->(1); # forced: the response is complete, ignore $retry_at
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::Streamer; ## no critic (ProhibitMultiplePackages)
# Body is XS; declared here only to carry CLONE_SKIP (see main package).
sub CLONE_SKIP { return 1 }
package ClickHouse::Encoder::BulkInserter; ## no critic (ProhibitMultiplePackages)
# Holds an encoder, which does not survive a thread spawn (see the main
# package). Skip cloning the whole inserter rather than handing the child
# one whose {enc} has silently become undef.
sub CLONE_SKIP { return 1 }
sub new {
my ($class, %args) = @_;
my $origin_raw = delete $args{_origin};
my $origin = (ref($origin_raw) ? 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 THREADS
Encoders and streamers are per-thread objects: construct them inside the
thread that uses them.
Both hold a C-level structure behind a blessed scalar, so they set
C<CLONE_SKIP>. When a thread is spawned they are B<not> cloned into it -
the child interpreter sees a plain unblessed value, and calling a method
on it raises C<Can't call method ... on unblessed reference> rather than
sharing (and eventually double-freeing) the parent's structure. The
parent's own encoder is unaffected and stays usable across the spawn.
Nothing else in the distribution keeps mutable global state, so any
number of threads may each build and use their own encoder concurrently.
Processes forked with C<fork> are unaffected by any of this and inherit a
working encoder as usual.
=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.
( run in 0.947 second using v1.01-cache-2.11-cpan-a49fcb8fa48 )