view release on metacpan or  search on metacpan

lib/ClickHouse/Encoder/TCP.pm  view on Meta::CPAN

            ($pkt{version_patch}, $offset)
                = unpack_varint($bytes, $offset);
        }
    }
    elsif ($type == SERVER_EXCEPTION) {
        # ExceptionPacket: code(Int32 LE), name, message, stack_trace, has_nested(byte)
        die "unpack_packet: truncated at offset $offset"
            if $offset + 4 > length $bytes;
        $pkt{code} = unpack 'l<', substr($bytes, $offset, 4);
        $offset += 4;
        ($pkt{name},        $offset) = unpack_string($bytes, $offset);
        ($pkt{message},     $offset) = unpack_string($bytes, $offset);
        ($pkt{stack_trace}, $offset) = unpack_string($bytes, $offset);
        die "unpack_packet: truncated at offset $offset"
            if $offset >= length $bytes;
        $pkt{has_nested} = ord substr($bytes, $offset++, 1);
    }
    elsif ($type == SERVER_PROGRESS) {
        # At the targeted protocol revision (54429) a Progress packet
        # is exactly these five varints. Later revisions append fields
        # (elapsed_ns, ...), but the Progress packet carries no
        # revision of its own, so the only safe stopping point without
        # threading the negotiated revision through is the end of the
        # 54429 layout. Reading a trailing field "if bytes remain"
        # would, on a buffer that already holds the next packet,
        # consume that packet's bytes instead.
        ($pkt{rows},          $offset) = unpack_varint($bytes, $offset);
        ($pkt{bytes},         $offset) = unpack_varint($bytes, $offset);
        ($pkt{total_rows},    $offset) = unpack_varint($bytes, $offset);
        ($pkt{written_rows},  $offset) = unpack_varint($bytes, $offset);
        ($pkt{written_bytes}, $offset) = unpack_varint($bytes, $offset);
    }
    elsif ($type == SERVER_PONG || $type == SERVER_END_OF_STREAM) {
        # No payload.
    }
    elsif ($type == SERVER_PROFILE_INFO) {
        ($pkt{rows},            $offset) = unpack_varint($bytes, $offset);
        ($pkt{blocks},          $offset) = unpack_varint($bytes, $offset);
        ($pkt{rows_bytes},      $offset) = unpack_varint($bytes, $offset);
        die "unpack_packet: truncated at offset $offset"
            if $offset >= length $bytes;
        $pkt{applied_limit} = ord substr($bytes, $offset++, 1);
        ($pkt{rows_before_limit}, $offset)
            = unpack_varint($bytes, $offset);
        die "unpack_packet: truncated at offset $offset"
            if $offset >= length $bytes;
        $pkt{calculated_rows_before_limit}
            = ord substr($bytes, $offset++, 1);
    }
    elsif ($type == SERVER_DATA || $type == SERVER_TOTALS
        || $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;
    }
    else {
        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 +
    # LZ4/ZSTD/raw payload). This must match what the caller negotiated
    # in pack_query(... compression => COMPRESSION_ENABLE); reading
    # without `compressed => 1` from a connection that DOES use
    # compression will misparse the Native block and croak with a
    # decode error rather than a clean diagnostic.
    my $compressed = $opts{compressed};
    # Optional caller-owned buffer: pass `buffer => \my $buf` and thread
    # the same ref through every read_packet call on this filehandle.
    # A fast server can pack several packets into one TCP segment;
    # read_packet then leaves the bytes it over-read in that buffer for
    # the next call. Without it, read_packet is a one-shot helper and
    # over-read bytes are lost - unsafe to call in a loop.
    my $bufref = $opts{buffer};
    my $buf = $bufref ? $$bufref : '';
    my $read_some = sub {
        my $got = sysread $fh, my $chunk, 4096;
        die "read_packet: read error: $!\n" if !defined $got;
        die "read_packet: connection closed mid-packet\n" if $got == 0;
        $buf .= $chunk;
    };

    my $pkt;
    my $end;   # offset just past the consumed packet
    while (1) {
        my $ok = eval {
            ($pkt, $end) = __PACKAGE__->unpack_packet($buf, 0);
            1;
        };
        last if $ok;
        die $@ unless $@ =~ /truncated/;
        $read_some->();
    }

    # For Data-shaped packets, also pull the inner Native block bytes.
    # decode_block tells us how many bytes it consumed; if the block
    # is still truncated, read more and retry.
    if (exists $pkt->{block_offset}) {
        require ClickHouse::Encoder;
        if ($compressed) {
            # Step 1: pull the compressed-block frame (16 hash + 9 hdr
            # + payload). Read more bytes until the frame is complete.
            while (1) {
                my $ok = eval {
                    my ($plain, $consumed) =
                        ClickHouse::Encoder->decompress_native_block(
                            $buf, offset => $pkt->{block_offset});
                    # Decode the inner Native block out of $plain.
                    my $b = ClickHouse::Encoder->decode_block($plain);
                    $pkt->{block}       = $b;
                    $pkt->{block_bytes} = $plain;   # decompressed bytes

lib/ClickHouse/Encoder/TCP.pm  view on Meta::CPAN

Wraps an encoded Native block (from
L<ClickHouse::Encoder/encode>) in a Data packet with an optional
C<table_name> (usually empty for inserts).

With C<compress =E<gt> 'lz4'> (or C<'zstd'>, or C<'auto'> - any mode
L<ClickHouse::Encoder/compress_native_block> accepts) the block is
first wrapped in ClickHouse's compressed-block framing (16-byte
CityHash128 + 9-byte header + compressed payload) before being placed
inside the Data packet. The server must already be expecting
compressed data via the corresponding C<pack_query(... compression
=E<gt> COMPRESSION_ENABLE)>; sending compressed Data without
negotiating compression in the Query packet will be rejected as a
parse error by C<CompressedReadBuffer>. C<compress> absent, or
C<'none'> / C<'raw'>, emits the bare uncompressed block (the default).

=head2 pack_data_end %opts

    my $bytes = ClickHouse::Encoder::TCP->pack_data_end();
    my $bytes = ClickHouse::Encoder::TCP->pack_data_end(
        compress => 'lz4');

Sends an empty Data packet; this is how an insert pipeline tells
the server "no more data". When compression was negotiated in the
Query, the empty block must be sent through the same compressed
framing so C<CompressedReadBuffer> parses it the same way -
C<compress> takes the same values as L</pack_data>.

=head2 pack_ping, pack_cancel

    my $b = ClickHouse::Encoder::TCP->pack_ping;

Trivial control packets.

=head1 PACKET DECODER

=head2 unpack_packet $bytes, $offset

    my ($pkt, $new_offset) = ClickHouse::Encoder::TCP->unpack_packet(
        $buffer, $offset);

Parse one server packet from C<$bytes> starting at C<$offset>.
Returns a hashref with at least C<type> (numeric, one of the
C<SERVER_*> constants) and packet-specific fields. Croaks on
truncated input - catch with C<eval> on a sliding-buffer reader.

For Data/Totals/Extremes/ProfileEvents packets the hashref carries
C<block_offset>: the byte offset where the inner Native block begins.
Pass C<<< substr($bytes, $block_offset) >>> to
C<<< ClickHouse::Encoder->decode_block >>> to extract rows.

=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.

When the caller has negotiated compression (via C<pack_query(...
compression =E<gt> COMPRESSION_ENABLE)>), pass C<compressed =E<gt>
1> so C<read_packet> peels the inner compressed-block framing
(16-byte CityHash128 + 9-byte header + LZ4/ZSTD payload) via
L<ClickHouse::Encoder/decompress_native_block> before decoding the
Native block. C<block_bytes> on the returned hashref is the
decompressed inner block; C<compressed_consumed> reports how many
on-the-wire bytes the framed block occupied.

=head1 WIRE CODECS

The varint and length-prefixed-string codecs the packet builders use
are XS-backed and callable directly. They are a semi-public surface:
stable and reusable, but secondary to the packet-level API above.
They are plain functions (not class methods) - invoke them fully
qualified, e.g. C<ClickHouse::Encoder::TCP::pack_varint($n)>.

=over 4

=item pack_varint $uint

Encode a non-negative integer in ClickHouse's LEB128 varuint form;
returns the byte string.

=item unpack_varint $bytes, $offset

Decode one varint at C<$offset>; returns C<($value, $new_offset)>.
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>,
C<SERVER_END_OF_STREAM>, C<SERVER_PROFILE_INFO>, C<SERVER_TOTALS>,
C<SERVER_EXTREMES>, C<SERVER_TABLE_COLUMNS>, C<SERVER_PROFILE_EVENTS>.

Other: C<STAGE_COMPLETE>, C<STAGE_WITH_MERGEABLE>,
C<STAGE_FETCH_COLUMNS>, C<COMPRESSION_DISABLE>, C<COMPRESSION_ENABLE>,
C<DEFAULT_REVISION>.

All constants live in the package namespace and are not exported;
reference them as C<ClickHouse::Encoder::TCP::SERVER_DATA> etc.

=head1 CAVEATS

=over 4

=item * B<Modern server cutoff.> The default revision (54429) predates
the chunking-negotiation extension introduced in ClickHouse 24.10
(protocol revision E<gt>= 54475). Newer servers send a chunking offer
right after C<SERVER_HELLO> that this subset does not respond to;
the connection then fails with a fast protocol-mismatch error.
For integration with recent servers, prefer HTTP transport.

=item * B<String encoding.> Inputs to C<pack_string> (any string
field: query, names, settings) are encoded as UTF-8 bytes; passing a
byte-mode string with non-ASCII bytes will be reinterpreted as
Latin-1 by Perl's UTF-8 upgrade rules. If you need raw bytes, encode
to UTF-8 yourself first. C<unpack_string> conversely returns the
raw byte string the server sent - it does not set the UTF-8 flag,
so callers wanting characters should C<decode_utf8> the result.

=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>
for the matching block-framing helpers.

L<EV::ClickHouse> - full async ClickHouse client (TCP + HTTP) for
select result streaming, prepared queries with parameter binding,
and chunking-negotiation against modern CH revisions.

=head1 AUTHOR

vividsnow

=head1 LICENSE

Same terms as Perl itself.

=cut