ClickHouse-Encoder
view release on metacpan or search on metacpan
* mortal-stack cleanup at scope exit drops one back, leaving the
* net refcount at 1 (owned by `result`). */
(void)hv_stores(result, "columns",
newRV_inc((SV *)cols));
(void)hv_stores(result, "consumed", newSVuv((UV)(p - start)));
RETVAL = newRV_noinc((SV *)result);
}
OUTPUT: RETVAL
# Row-oriented decoder: same wire walk as decode_block, but values are
# distributed into row-major arrayrefs as each column is decoded, then
# the per-column AV is freed. Peak memory holds one column's values
# plus all row AVs (vs decode_rows-via-Perl which holds both
# representations + does the transpose in Perl).
SV *
decode_block_rows(class, bytes, ...)
SV *class
SV *bytes
CODE:
{
PERL_UNUSED_VAR(class);
eg/json_query.pl view on Meta::CPAN
#!/usr/bin/env perl
# json_query.pl - run a select against a JSON column, decode rows via
# decode_blocks, walk the decoded objects in Perl.
#
# Usage:
# json_query.pl --table events --where "j.user.id = 42"
#
# Demonstrates:
# - HTTP select format native returns a concatenated stream of blocks;
# decode_blocks() walks them
# - The decoded JSON values are nested Perl hashrefs (dotted paths
# auto-unflattened on the way back)
use strict;
use warnings;
use Getopt::Long;
use HTTP::Tiny;
use ClickHouse::Encoder;
my ($host, $port, $table, $col, $where, $limit) =
('127.0.0.1', 8123, 'events', 'j', '', 10);
GetOptions(
eg/migrate_table.pl view on Meta::CPAN
#
# Usage:
# migrate_table.pl --src-table events --dst-table events_copy
# migrate_table.pl --src-host db1 --dst-host db2 \
# --src-table events --dst-table events_copy \
# --where "ts >= '2026-05-01'"
#
# The schemas of src and dst must match; this script does not translate
# types. Use alter table ... modify column on the destination first if
# you need a schema change. Uses the decode_blocks callback form so
# decoded blocks don't accumulate. Note: the select response body is
# still fetched in one HTTP GET - for very large tables, slice the
# source query with where / limit or split by date range.
use strict;
use warnings;
use Getopt::Long;
use HTTP::Tiny;
use Encode;
use ClickHouse::Encoder;
my ($src_host, $src_port, $src_table) = ('127.0.0.1', 8123, '');
eg/migrate_table.pl view on Meta::CPAN
my @cols = map [$_->{name}, $_->{type}], @{ $block->{columns} };
$enc = ClickHouse::Encoder->new(columns => \@cols);
$bi = ClickHouse::Encoder->bulk_inserter(
host => $dst_host,
port => $dst_port,
table => $dst_table,
encoder => $enc,
batch_size => $batch_size,
compress => $compress);
}
# Transpose column-major decoded block to row-major for the encoder.
for my $r (0 .. $block->{nrows} - 1) {
$bi->push([map $_->{values}[$r], @{ $block->{columns} }]);
}
$total_rows += $block->{nrows};
});
$bi->finish if $bi;
print STDERR "Migrated $total_rows rows from $src_table to $dst_table\n";
lib/ClickHouse/Encoder.pm view on Meta::CPAN
push @parts, $p if length $p;
$start = $i + 1;
}
}
return @parts;
}
# Row-oriented decode: { ncols, nrows, names, types, rows } where
# rows is an arrayref of arrayrefs. Calls the XS decode_block_rows,
# which distributes column values into per-row arrayrefs as each
# column is decoded and frees the column AV eagerly. Peak memory
# is one column's AV plus the row AVs (vs a Perl-side transpose
# that holds ALL column AVs alongside the half-built row AVs).
# Throughput is similar to the column-major path; the win is the
# tighter memory profile on wide blocks.
sub decode_rows {
my ($class, $bytes, $offset) = @_;
return $class->decode_block_rows($bytes, $offset // 0);
}
# Decode a concatenated stream of Native blocks (the body of a
# `select ... format native` response). Returns an arrayref of the
# same hashref shape as decode_block. Stops cleanly when bytes are
# exhausted; partial trailing bytes raise an error from XS.
# Uses the 3-arg form of decode_block (with offset) to avoid O(N^2)
# substr copies on long streams.
sub decode_blocks {
my ($class, $bytes, $cb, %opts) = @_;
my $keep = $opts{keep};
my $off = 0;
my $len = length $bytes;
# Callback form: hand each block to $cb as it's decoded; never
# accumulate. Useful for streaming selects where the full block
# list would not fit comfortably in memory.
if ($cb) {
while ($off < $len) {
my $block = $class->decode_block($bytes, $off, $keep);
$off += $block->{consumed};
$cb->($block);
}
return;
}
lib/ClickHouse/Encoder.pm view on Meta::CPAN
while ($pos < $len) {
my @row;
for my $col (@$cols) {
push @row, _rb_decode_value($col->[1], \$bytes, \$pos, \%cache);
}
push @rows, \@row;
}
return \@rows;
}
# Post-process a decoded block (or any one column's values arrayref):
# rewrite Date / Date32 / DateTime / DateTime64 integer epochs into
# ISO 8601 strings or Time::Moment instances. Modifies the block in
# place AND returns it so the call can be chained. as => 'iso' (the
# default) emits UTC strings with a 'Z' suffix; as => 'datetime'
# returns Time::Moment objects (requires Time::Moment installed).
# DateTime64 precision is read from the column's type string so each
# tick converts to the correct number of fractional digits.
sub coerce_datetimes {
my ($class_or_self, $block, %opts) = @_;
my $as = $opts{as} // 'iso';
lib/ClickHouse/Encoder.pm view on Meta::CPAN
=head2 decode_blocks
my $blocks = ClickHouse::Encoder->decode_blocks($bytes);
ClickHouse::Encoder->decode_blocks($bytes, sub { my $b = shift; ... });
A C<select ... format native> response is a concatenated stream of
blocks (one per granule of C<max_block_size> rows). With no callback,
C<decode_blocks> walks the stream and returns an arrayref of the same
hashref shape as L</decode_block>. With a callback, each block is
passed to the callback as it's decoded and no list is accumulated -
useful for very long selects where the full block list wouldn't fit
comfortably in memory.
Uses the 3-arg form of L</decode_block> (with explicit offset) to
keep total work O(N) regardless of block count. Stops cleanly when
bytes are exhausted; partial trailing bytes croak.
The optional C<keep =E<gt> \%names> hashref forwards a column filter
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
lib/ClickHouse/Encoder.pm view on Meta::CPAN
semantics are identical to L</encode_row_binary> and to the Native
decoder - C<decode_row_binary> yields the same Perl values
L</decode_block> would for the same data.
=head2 coerce_datetimes
my $blk = ClickHouse::Encoder->decode_block($bytes);
ClickHouse::Encoder->coerce_datetimes($blk); # ISO strings
ClickHouse::Encoder->coerce_datetimes($blk, as => 'datetime'); # Time::Moment
Post-process a decoded block: rewrite every C<Date> / C<Date32> /
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
lib/ClickHouse/Encoder.pm view on Meta::CPAN
summary, progress, ... }>
slot attached (whichever C<X-ClickHouse-*> headers the server sent).
C<undef> until the first flush succeeds.
=back
=head2 decimal128_str
my $s = ClickHouse::Encoder->decimal128_str($lo, $hi, $scale);
Convert a decoded C<Decimal128> low/high uint64 pair (as returned
by L</decode_block>) into a signed decimal string with the given
fractional scale. Uses L<Math::BigInt> for the 128-bit arithmetic.
=head2 decimal256_str
my $s = ClickHouse::Encoder->decimal256_str(\@limbs, $scale);
Convert a decoded C<Decimal256> 4-limb arrayref (low-to-high
uint64s, as returned by L</decode_block>) into a signed decimal
string with the given fractional scale. Uses L<Math::BigInt> for
the 256-bit arithmetic.
=head2 compressed_writer
my $w = ClickHouse::Encoder->compressed_writer('zstd', \&raw_writer);
my $st = $enc->streamer($w, batch_size => 1000);
Class method that wraps a writer coderef so each emitted block is
lib/ClickHouse/Encoder/TCP.pm view on Meta::CPAN
|| $type == SERVER_EXTREMES) {
# Data packet: table_name then a Native block payload.
# We surface the table name and the byte range of the block;
# the caller passes those bytes to ClickHouse::Encoder->decode_block.
($pkt{table_name}, $offset) = unpack_string($bytes, $offset);
$pkt{block_offset} = $offset;
# Caller decodes from here; advancing $offset requires parsing
# the inner block (ncols + nrows + per-column bytes). For the
# caller's convenience we don't attempt that here - they should
# use ClickHouse::Encoder->decode_block on the slice and add
# block_consumed = $decoded->{consumed} to advance.
}
elsif ($type == SERVER_TABLE_COLUMNS) {
($pkt{table_name}, $offset) = unpack_string($bytes, $offset);
($pkt{column_descriptor}, $offset) = unpack_string($bytes, $offset);
}
elsif ($type == SERVER_PROFILE_EVENTS) {
# Same shape as a Data packet.
($pkt{table_name}, $offset) = unpack_string($bytes, $offset);
$pkt{block_offset} = $offset;
}
lib/ClickHouse/Encoder/TCP.pm view on Meta::CPAN
die "unpack_packet: unknown server packet type $type\n";
}
return (\%pkt, $offset);
}
# Blocking read helper for IO::Socket / IO::Handle. Pulls bytes until
# one full packet parses, returns the packet hashref. For
# Data/Totals/Extremes/ProfileEvents packets the inner Native block
# bytes are read into $pkt->{block_bytes} (the raw slice ready for
# ClickHouse::Encoder->decode_block); $pkt->{block} carries the
# pre-decoded block hashref.
#
# Reads whatever sysread returns (up to 4 KiB at a time): the
# protocol packets are size-self-describing, so we let unpack_packet
# tell us when it has enough bytes. Avoids the trap of trying to
# block-read N bytes when the server has only sent K < N.
sub read_packet {
my (undef, $fh, %opts) = @_;
# When `compressed => 1` is passed, the inner Native block of any
# Data-shaped packet is expected to be wrapped in CH's
# compressed-block framing (16-byte CityHash128 + 9-byte header +
lib/ClickHouse/Encoder/TCP.pm view on Meta::CPAN
=head2 read_packet $fh
my $rbuf = '';
my $pkt = ClickHouse::Encoder::TCP->read_packet($fh, buffer => \$rbuf);
my $pkt = ClickHouse::Encoder::TCP->read_packet($fh, compressed => 1);
Blocking read from a filehandle. C<sysread>s in chunks until it has
one whole packet, parses it, and returns the hashref. For Data-shaped
packets, also reads the inner block and surfaces C<block_bytes> +
pre-decoded C<block> hashref. Convenience only; non-blocking
transports should call L</unpack_packet> on their own buffer.
A single C<sysread> may pull in more than one packet. Pass
C<<< buffer =E<gt> \my $buf >>> and thread the same scalar ref
through every C<read_packet> call on the filehandle: read_packet
seeds itself from that buffer and leaves any over-read bytes there
for the next call. B<This is required to read more than one packet>
- without a caller buffer, over-read bytes are dropped and a second
C<read_packet> call may block or misparse. The compression and
buffer options are independent and may be combined.
lib/ClickHouse/Encoder/TCP.pm view on Meta::CPAN
Croaks on truncated input or a varint wider than 64 bits.
=item pack_string $str
Encode a length-prefixed string: a varint length followed by the
UTF-8 bytes of C<$str>. Returns the byte string.
=item unpack_string $bytes, $offset
Decode one length-prefixed string at C<$offset>; returns
C<($string, $new_offset)> with the raw (undecoded) string bytes.
=back
=head1 CONSTANTS
Client packet types: C<CLIENT_HELLO>, C<CLIENT_QUERY>, C<CLIENT_DATA>,
C<CLIENT_CANCEL>, C<CLIENT_PING>.
Server packet types: C<SERVER_HELLO>, C<SERVER_DATA>,
C<SERVER_EXCEPTION>, C<SERVER_PROGRESS>, C<SERVER_PONG>,
lib/ClickHouse/Encoder/TCP.pm view on Meta::CPAN
=item * B<Settings values are strings.> Each setting is emitted with
flags=0 (ordinary) and the value as a string. Numeric/typed-setting
encoding (flexible settings, available at higher revisions) is out
of scope.
=item * B<Wire compression is opt-in.> C<pack_query> still defaults
to C<COMPRESSION_DISABLE>; to negotiate compression pass
C<compression =E<gt> COMPRESSION_ENABLE> in C<pack_query>, then
C<compress =E<gt> 'lz4'> (or C<'zstd'>) to both C<pack_data> and
C<pack_data_end>. Compressed Data packets coming back from the
server are decoded by C<read_packet($fh, compressed =E<gt> 1)>.
The compressed-block framing (16-byte CityHash128 v1.0.2 + 9-byte
header + payload) lives in L<ClickHouse::Encoder/compress_native_block>.
=back
=head1 SEE ALSO
L<ClickHouse::Encoder> - the wire-format encoder these packets carry,
plus L<compress_native_block|ClickHouse::Encoder/compress_native_block>
/ L<decompress_native_block|ClickHouse::Encoder/decompress_native_block>
t/buffer-grow.t view on Meta::CPAN
sub { shift @rows2 },
sub { $chunked .= $_[0] },
batch_size => 1000,
);
cmp_ok(length($chunked), '>', length($bin) * 0.95,
'streamed 50k rows in 50 blocks adds up to similar size (per-block overhead is small)');
# Round-trip: decode the streamed concatenation and confirm we get
# back exactly @rows. This validates that every doubling-cycle of
# buf_grow produced wire-correct output.
my @decoded;
my $off = 0;
my $blen = length $chunked;
while ($off < $blen) {
my $ncols = read_varint(\$chunked, \$off);
my $nrows = read_varint(\$chunked, \$off);
is($ncols, 1, 'block has 1 column');
my $name_len = read_varint(\$chunked, \$off); $off += $name_len;
my $type_len = read_varint(\$chunked, \$off); $off += $type_len;
for (1 .. $nrows) {
my $slen = read_varint(\$chunked, \$off);
push @decoded, substr($chunked, $off, $slen);
$off += $slen;
}
}
is(scalar @decoded, scalar @rows, 'round-trip: 50000 rows decoded');
is($decoded[0], $rows[0][0], 'round-trip: first row matches');
is($decoded[-1], $rows[-1][0], 'round-trip: last row matches');
is($decoded[25_000],$rows[25_000][0], 'round-trip: middle row matches');
done_testing();
t/coerce-datetimes.t view on Meta::CPAN
#!/usr/bin/env perl
# coerce_datetimes: post-process a decoded block to rewrite Date /
# Date32 / DateTime / DateTime64 columns from raw epoch integers into
# ISO 8601 strings or Time::Moment instances.
use strict;
use warnings;
use Test::More;
use lib 'blib/lib', 'blib/arch';
use ClickHouse::Encoder;
# Encode a block with one of every time-column type, decode it, and
# coerce. 1700000000 = 2023-11-14 22:13:20 UTC; 19675 days since
t/compress-native.t view on Meta::CPAN
is($csize, length($framed) - 16,
'compressed_size header covers 9-byte header + payload');
my ($plain, $consumed) = ClickHouse::Encoder->decompress_native_block(
$framed, hasher => \&md5);
is($plain, $bytes, 'round-trip restores original bytes');
is($consumed, length($framed), 'consumed = whole framed block');
# And the decoder still finds three rows in the decompressed block.
my $blk = ClickHouse::Encoder->decode_block($plain);
is($blk->{nrows}, 3, 'decoded block has 3 rows after decompression');
}
# Checksum mismatch detected
{
my $framed = ClickHouse::Encoder->compress_native_block(
'hello ' x 100, hasher => \&md5);
substr($framed, 0, 1, "\xff"); # corrupt first byte of checksum
my $err = eval {
ClickHouse::Encoder->decompress_native_block(
$framed, hasher => \&md5); 1
t/compress-native.t view on Meta::CPAN
{
my @inputs = ('first block payload',
'second' x 200,
'' . ('z' x 100));
my $stream = '';
for my $b (@inputs) {
$stream .= ClickHouse::Encoder->compress_native_block(
$b, hasher => \&md5);
}
my $off = 0;
my @decoded;
while ($off < length $stream) {
my ($plain, $n) = ClickHouse::Encoder->decompress_native_block(
$stream, hasher => \&md5, offset => $off);
push @decoded, $plain;
$off += $n;
}
is_deeply(\@decoded, \@inputs, 'walked a multi-block compressed stream');
}
# mode => 'auto' picks LZ4 when it wins, falls back to method-tag 0x02
# (uncompressed-inside-framing) when LZ4 output >= input size. Mirrors
# what ClickHouse's own CompressedWriteBuffer does for incompressible
# payloads.
{
# Highly compressible (LZ4 wins): expect tag 0x82
my $compressible = 'aaaaaaaa' x 200;
my $framed1 = ClickHouse::Encoder->compress_native_block(
t/decimal-boundary.t view on Meta::CPAN
{
my $enc = ClickHouse::Encoder->new(columns => [['v','Decimal128(0)']]);
# Maximum positive (2^127 - 1)
my $max_pos = '170141183460469231731687303715884105727';
# Maximum negative -2^127
my $max_neg = '-170141183460469231731687303715884105728';
my $bytes = $enc->encode([[$max_pos], [$max_neg], ['0'], ['1'], ['-1']]);
my $blk = ClickHouse::Encoder->decode_block($bytes);
my @rows = @{ $blk->{columns}[0]{values} };
is(scalar @rows, 5, 'Decimal128: all rows decoded');
# Reconstruct strings from limb pairs
is(ClickHouse::Encoder->decimal128_str(@{$rows[0]}, 0), $max_pos,
'Decimal128: max positive round-trip');
is(ClickHouse::Encoder->decimal128_str(@{$rows[1]}, 0), $max_neg,
'Decimal128: max negative round-trip');
is(ClickHouse::Encoder->decimal128_str(@{$rows[2]}, 0), '0',
'Decimal128: zero round-trip');
is(ClickHouse::Encoder->decimal128_str(@{$rows[3]}, 0), '1',
'Decimal128: one round-trip');
is(ClickHouse::Encoder->decimal128_str(@{$rows[4]}, 0), '-1',
t/decimal-boundary.t view on Meta::CPAN
}
# Decimal256(0): integer up to (2^255 - 1) ~= 5.8e76
{
my $enc = ClickHouse::Encoder->new(columns => [['v','Decimal256(0)']]);
my $max_pos = '5789604461865809771178549250434395392663499233282028201972879200395656481996';
my $max_neg = '-' . $max_pos;
my $bytes = $enc->encode([[$max_pos], [$max_neg], ['1234567890' x 5]]);
my $blk = ClickHouse::Encoder->decode_block($bytes);
my @rows = @{ $blk->{columns}[0]{values} };
is(scalar @rows, 3, 'Decimal256: rows decoded');
is(ClickHouse::Encoder->decimal256_str($rows[0], 0), $max_pos,
'Decimal256: large positive');
# Decimal256 max-negative is asymmetric (-2^255), but max_neg here
# is symmetric (-(2^255-1)). Reconstruct via decimal256_str.
is(ClickHouse::Encoder->decimal256_str($rows[1], 0), $max_neg,
'Decimal256: large negative');
is(ClickHouse::Encoder->decimal256_str($rows[2], 0), '1234567890' x 5,
'Decimal256: 50-digit positive');
}
t/decode-blocks.t view on Meta::CPAN
# bounded by max_block_size).
my $enc = ClickHouse::Encoder->new(columns => [['n', 'Int32'], ['s', 'String']]);
# Two blocks of 2 rows each, concatenated.
my $block1 = $enc->encode([[1, "one"], [2, "two"]]);
my $block2 = $enc->encode([[3, "three"], [4, "four"]]);
my $stream = $block1 . $block2;
my $blocks = ClickHouse::Encoder->decode_blocks($stream);
is(scalar(@$blocks), 2, 'two blocks decoded');
is($blocks->[0]{nrows}, 2, 'block 0: 2 rows');
is($blocks->[1]{nrows}, 2, 'block 1: 2 rows');
is_deeply($blocks->[0]{columns}[0]{values}, [1, 2], 'block 0 col 0');
is_deeply($blocks->[1]{columns}[1]{values}, ["three", "four"], 'block 1 col 1');
# Empty stream
{
my $blocks = ClickHouse::Encoder->decode_blocks("");
is_deeply($blocks, [], 'empty stream -> empty list');
}
t/decode-blocks.t view on Meta::CPAN
}
# Direct 3-arg form of decode_block (with offset) - this is the XS path
# decode_blocks relies on; pin it independently.
{
my $first = ClickHouse::Encoder->decode_block($stream, 0);
is($first->{nrows}, 2, '3-arg decode_block at offset 0');
my $second = ClickHouse::Encoder->decode_block($stream, $first->{consumed});
is($second->{nrows}, 2, '3-arg decode_block at non-zero offset');
is_deeply($second->{columns}[1]{values}, ["three", "four"],
'3-arg form decoded correct second block');
# Negative offset rejected
my $err = eval { ClickHouse::Encoder->decode_block($stream, -1); 1 }
? "" : $@;
like($err, qr/non-negative/, 'negative offset rejected');
# Offset past end rejected
$err = eval {
ClickHouse::Encoder->decode_block($stream, length($stream) + 1); 1
} ? "" : $@;
t/json-typed-paths.t view on Meta::CPAN
my $out = rt('JSON(score UInt32, label String)', [
{score => 10, label => 'a'},
{}, # both missing
]);
is($out->[0]{score}, 10, 'row 0 score');
is($out->[0]{label}, 'a', 'row 0 label');
is($out->[1]{score}, 0, 'row 1 missing UInt32 -> 0');
is($out->[1]{label}, '', 'row 1 missing String -> ""');
}
# 5. Nullable typed paths: undef -> NULL (decoded as undef)
{
my $out = rt('JSON(maybe Nullable(Int32))', [
{maybe => 42},
{}, # missing
{maybe => undef},
]);
is($out->[0]{maybe}, 42, 'Nullable: value');
is($out->[1]{maybe}, undef, 'Nullable: missing -> undef');
is($out->[2]{maybe}, undef, 'Nullable: explicit undef -> undef');
}
t/json-typed-paths.t view on Meta::CPAN
# the type-string header) is identical to plain JSON.
{
my $b1 = ClickHouse::Encoder->new(columns => [['j','JSON']])
->encode([[{a => 1}]]);
my $b2 = ClickHouse::Encoder->new(columns => [['j','JSON()']])
->encode([[{a => 1}]]);
# The two buffers differ only in the type-string column header.
# Decode both and compare values.
my $v1 = ClickHouse::Encoder->decode_block($b1)->{columns}[0]{values};
my $v2 = ClickHouse::Encoder->decode_block($b2)->{columns}[0]{values};
is_deeply($v1, $v2, 'JSON() with no typed paths == JSON (decoded)');
}
# 11. Typed path with Boolean
{
my $out = rt('JSON(active Bool)', [
{active => 1},
{active => 0},
]);
is($out->[0]{active}, 1, 'Bool true');
is($out->[1]{active}, 0, 'Bool false');
is($carol_age, '40', 'nested path subcolumn query');
# Read back JSON column via select format native and decode
open my $fh, '-|', @ch_cmd, '--query',
'select j from test_json order by toInt64(j.age) asc nulls last format native'
or die "Cannot spawn clickhouse-client: $!";
binmode $fh;
my $native = do { local $/; <$fh> };
close $fh;
my $block = ClickHouse::Encoder->decode_block($native);
is($block->{ncols}, 1, 'decoded ncols');
is($block->{nrows}, 6, 'decoded nrows');
# All 6 should be hashrefs
is(scalar(grep { ref $_ eq 'HASH' } @{ $block->{columns}[0]{values} }),
6, 'all 6 decoded rows are hashrefs');
# Typed JSON paths
ch_query("drop table if exists test_json_typed");
system(@ch_cmd, @json_flags, '--query',
"create table test_json_typed "
. "(j JSON(name String, age UInt32)) engine = Memory") == 0
or skip "Could not create typed JSON table", 3;
my $te = ClickHouse::Encoder->new(
columns => [['j','JSON(name String, age UInt32)']]);
my $td = $te->encode([
t/parse-create-table.t view on Meta::CPAN
my $p = ClickHouse::Encoder->parse_create_table($sql);
is_deeply($p->{columns}, [['we`ird', 'Int32'], ['ok', 'String']],
'backtick-in-name round-trips format -> parse');
}
# The doubled-backtick escape form is also accepted on input.
{
my $p = ClickHouse::Encoder->parse_create_table(
'CREATE TABLE t (`a``b` Int32) ENGINE = Memory');
is_deeply($p->{columns}, [['a`b', 'Int32']],
'doubled-backtick escape decoded on input');
}
# CREATE OR REPLACE TABLE and CREATE TEMPORARY TABLE header variants.
{
my $p = ClickHouse::Encoder->parse_create_table(
'CREATE OR REPLACE TABLE t (`x` Int32) ENGINE = Memory');
is($p->{table}, 't', 'CREATE OR REPLACE TABLE header parsed');
my $q = ClickHouse::Encoder->parse_create_table(
'CREATE TEMPORARY TABLE tmp (`y` String) ENGINE = Memory');
t/roundtrip.t view on Meta::CPAN
elsif ($code eq 'Variant') {
my $mode = unpack('Q<', substr($$buf, $$off, 8));
$$off += 8;
die "Variant mode != 0" unless $mode == 0;
my @wire_disc;
for (1..$nrows) { push @wire_disc, ord(substr($$buf, $$off, 1)); $$off += 1 }
# wire_disc is in alphabetical-order space; map to declaration idx.
my @parts = @{ $type->{parts} };
my $wire_to_decl = $type->{wire_to_decl};
my @counts; for my $w (@wire_disc) { $counts[$w]++ if $w != 255 }
my @subcols; # subcols[wire_idx] = decoded values for that wire arm
for my $w (0 .. $#parts) {
my $decl = $wire_to_decl->[$w];
$subcols[$w] = _decode_column($buf, $off, $parts[$decl], $counts[$w] // 0);
}
my @cursors = (0) x scalar @parts;
for my $r (0 .. $nrows - 1) {
my $w = $wire_disc[$r];
if ($w == 255) { push @vals, undef; next }
push @vals, [$wire_to_decl->[$w], $subcols[$w][ $cursors[$w]++ ]];
}
pipe(my $r, my $w) or die "pipe: $!";
binmode $r; binmode $w;
print $w $server_pkt;
close $w;
my $got = ClickHouse::Encoder::TCP->read_packet($r, compressed => 1);
is($got->{type},
ClickHouse::Encoder::TCP::SERVER_DATA,
'compressed read_packet: SERVER_DATA type');
is($got->{block}{nrows}, 3,
'compressed read_packet: nrows decoded');
is_deeply($got->{block}{columns}[0]{values}, [42, 43, 44],
'compressed read_packet: values round-trip');
is($got->{compressed_consumed}, length($framed),
'compressed read_packet: compressed_consumed matches framed size');
close $r;
}
# read_packet(buffer => \$buf): a single sysread can pull in several
# packets; the caller-owned buffer must carry the over-read bytes
# forward so a looping reader does not lose them. Write Pong (\x04)
# SELECT count() via the same connection
print $sock ClickHouse::Encoder::TCP->pack_query(
query => 'select count() from ch_tcp_t');
my $got_count;
while (1) {
my $p = ClickHouse::Encoder::TCP->read_packet($sock);
last if $p->{type} == 5;
next if $p->{type} == 3 || $p->{type} == 6 || $p->{type} == 14
|| $p->{type} == 11;
if ($p->{type} == 1) {
# Data: decoded inline by read_packet.
$got_count = $p->{block}{columns}[0]{values}[0]
if $p->{block}{nrows};
}
}
is($got_count, 3, 'SELECT count via TCP returned 3');
close $sock;
}
done_testing();
# Test UTF-8 String
{
use utf8;
my $enc = ClickHouse::Encoder->new(columns => [['v', 'String']]);
my $utf8_str = "Привет мир 日本語 🎉";
my $bin = $enc->encode([[$utf8_str]]);
ok(defined $bin, 'UTF-8 string encodes');
my $off = skip_header($bin);
my ($len, $off2) = read_varint($bin, $off);
my $decoded = substr($bin, $off2, $len);
utf8::decode($decoded);
is($decoded, $utf8_str, 'UTF-8 string content preserved');
}
# Test UTF-8 FixedString (truncation on byte boundary)
{
use utf8;
my $enc = ClickHouse::Encoder->new(columns => [['v', 'FixedString(6)']]);
# "Привет" is 12 bytes in UTF-8, should truncate to 6 bytes
my $bin = $enc->encode([["Привет"]]);
my $off = skip_header($bin);
is(length(substr($bin, $off)), 6, 'UTF-8 FixedString truncates by bytes');
t/variant-all-null.t view on Meta::CPAN
use ClickHouse::Encoder;
my $enc = ClickHouse::Encoder->new(columns =>
[['v','Variant(Int32, String)']]);
# All-NULL Variant: every row is undef
{
my $bytes = $enc->encode([[undef], [undef], [undef]]);
my $blk = ClickHouse::Encoder->decode_block($bytes);
my @rows = @{ $blk->{columns}[0]{values} };
is(scalar @rows, 3, 'three rows decoded');
ok(!defined $rows[0], 'row 0 undef');
ok(!defined $rows[1], 'row 1 undef');
ok(!defined $rows[2], 'row 2 undef');
}
# Single-row all-NULL Variant
{
my $bytes = $enc->encode([[undef]]);
my $blk = ClickHouse::Encoder->decode_block($bytes);
is(scalar @{ $blk->{columns}[0]{values} }, 1, 'one-row block');
ok(!defined $blk->{columns}[0]{values}[0], 'lone row is undef');
}
# Zero-row Variant (just the column header, no body): the streamer
# shouldn't blow up on an empty batch.
{
my $bytes = $enc->encode([]);
my $blk = ClickHouse::Encoder->decode_block($bytes);
is($blk->{nrows}, 0, 'zero-row block decoded');
}
# Mixed NULL and non-NULL: pins that the count-by-variant pass
# correctly handles the disc=255 rows alongside disc=0/1 rows.
{
my $bytes = $enc->encode([
[undef], [[0, 42]], [undef], [[1, 'hi']], [undef],
]);
my $blk = ClickHouse::Encoder->decode_block($bytes);
my @rows = @{ $blk->{columns}[0]{values} };
is(scalar @rows, 5, 'mixed: five rows decoded');
ok(!defined $rows[0], 'mixed: row 0 undef');
ok(!defined $rows[2], 'mixed: row 2 undef');
ok(!defined $rows[4], 'mixed: row 4 undef');
is_deeply($rows[1], [0, 42], 'mixed: row 1 = Int32 variant');
is_deeply($rows[3], [1, 'hi'], 'mixed: row 3 = String variant');
}
done_testing();
t/wide-tables.t view on Meta::CPAN
ok(defined $bytes && length $bytes > 0, "encoded $N_COLS x $N_ROWS block");
cmp_ok($encode_secs, '<', 5,
"encode finishes in under 5s (got ${encode_secs}s) - no quadratic blow-up");
# Decode and verify shape + a sample of values.
$t0 = time();
my $blk = ClickHouse::Encoder->decode_block($bytes);
my $decode_secs = time() - $t0;
is($blk->{ncols}, $N_COLS, 'decoded ncols matches');
is($blk->{nrows}, $N_ROWS, 'decoded nrows matches');
cmp_ok($decode_secs, '<', 5,
"decode finishes in under 5s (got ${decode_secs}s)");
# Sample a few columns from different positions
for my $i (0, 1, 2, 100, 250, $N_COLS - 1) {
my $col = $blk->{columns}[$i];
is($col->{name}, "c$i", "col $i name preserved");
is(scalar @{ $col->{values} }, $N_ROWS,
"col $i has all $N_ROWS values");
# Sample row 0 against canonical
xt/bench-regression.t view on Meta::CPAN
ok(length($bytes) > 0, "encoded $n_rows rows");
cmp_ok($secs, '<', $budget_secs,
"encode of $n_rows narrow rows under ${budget_secs}s "
. "(took ${secs}s; set CHE_BENCH_BUDGET to override)");
# Decode budget: a touch more generous since decode allocates per-row
# SVs whereas encode just writes bytes.
$t0 = time();
my $blk = ClickHouse::Encoder->decode_block($bytes);
my $decode_secs = time() - $t0;
is($blk->{nrows}, $n_rows, 'decoded all rows');
cmp_ok($decode_secs, '<', $budget_secs * 1.5,
"decode of $n_rows rows under "
. sprintf('%.1f', $budget_secs * 1.5) . "s (took ${decode_secs}s)");
done_testing();
my $t1 = [gettimeofday];
my $block = ClickHouse::Encoder->decode_block($bytes);
my $dec_secs = tv_interval($t1);
my $dec_rps = $N / $dec_secs;
note sprintf("decode: %d rows in %.3fs -> %.0f rows/s",
$N, $dec_secs, $dec_rps);
cmp_ok($dec_rps, '>=', $MIN_DECODE_RPS,
"decode throughput >= $MIN_DECODE_RPS rows/s");
# Light correctness check on the decoded output (a perf test shouldn't
# silently pass if the encoder produced garbage).
is($block->{nrows}, $N, 'decoded nrows matches input');
is($block->{columns}[0]{values}[0], 1, 'col 0 row 0');
is($block->{columns}[1]{values}[$N - 1], "user_$N", 'col 1 last row');
done_testing();
xt/spelling.t view on Meta::CPAN
passthrough
scalarref subcolumns SvNOK SvIOK SvIsBOOL
CTEs POSTs SELECTs backoff TLS AVs
CityHash cityhash CompressedReadBuffer
hasher impl compat vendored Digest::CityHash Inline::C
transactionally incompressible CompressedWriteBuffer
uint uint64
Liveness WKT inserter inserters params
RowBinary DDL deduplicated framings interop
jitter resynchronise backQuote
varint varuint LEB codecs undecoded misparse
));
all_pod_files_spelling_ok('lib');
( run in 3.106 seconds using v1.01-cache-2.11-cpan-600a1bdf6e4 )