view release on metacpan or  search on metacpan

t/tcp.t  view on Meta::CPAN

    # Verify the framed empty block decompresses to the canonical
    # empty-Native-block bytes (ncols=0, nrows=0).
    my $framed = substr($end, 2);
    my $plain  = ClickHouse::Encoder->decompress_native_block($framed);
    is($plain, "\x00\x00",
       'compressed pack_data_end framing wraps the 2-byte empty block');
}

# pack_data / pack_data_end also accept compress => 'zstd'; the only
# wire difference from lz4 is the method tag (0x90 vs 0x82). Same
# round-trip check.
SKIP: {
    eval { require Compress::Zstd; 1 }
        or skip 'Compress::Zstd not installed', 3;

    my $enc   = ClickHouse::Encoder->new(columns => [['x','Int32']]);
    my $block = $enc->encode([[1],[2],[3]]);

    my $pkt    = ClickHouse::Encoder::TCP->pack_data($block,
                                                     compress => 'zstd');
    my $framed = substr($pkt, 2);
    is(ord(substr($framed, 16, 1)), 0x90,
       'zstd pack_data uses ZSTD method tag 0x90');
    is(ClickHouse::Encoder->decompress_native_block($framed), $block,
       'zstd-compressed Data packet round-trips through decompress');

    my $end = ClickHouse::Encoder::TCP->pack_data_end(compress => 'zstd');
    is(ClickHouse::Encoder->decompress_native_block(substr($end, 2)),
       "\x00\x00",
       'zstd pack_data_end framing wraps the empty block');
}

# read_packet(compressed => 1) decompresses inner Data blocks.
# read_packet is for SERVER-side packets; CLIENT_DATA collides with
# SERVER_EXCEPTION (both =2), so build a SERVER_DATA (=1) packet by
# hand: type varint + empty table_name varint + compressed-framed
# Native block bytes.
SKIP: {
    eval { require Compress::LZ4; 1 }
        or skip 'Compress::LZ4 not installed', 4;

    my $enc = ClickHouse::Encoder->new(columns => [['x','Int32']]);
    my $block = $enc->encode([[42], [43], [44]]);
    my $framed = ClickHouse::Encoder->compress_native_block(
        $block, mode => 'lz4');
    my $server_pkt =
        "\x01"          # SERVER_DATA type
      . "\x00"          # empty table_name varint
      . $framed;        # compressed-framed Native block

    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)
# and EndOfStream (\x05) back-to-back, then read both.
{
    pipe(my $r, my $w) or die "pipe: $!";
    binmode $r; binmode $w;
    print $w "\x04\x05";   # Pong + EndOfStream in one segment
    close $w;

    my $buf = '';
    my $p1 = ClickHouse::Encoder::TCP->read_packet($r, buffer => \$buf);
    is($p1->{type}, 4, 'buffered read_packet: first call returns Pong');
    is($buf, "\x05", 'buffered read_packet: over-read byte kept in buffer');

    # Second call is satisfied entirely from the buffer - the pipe is
    # already at EOF, so without the carry-forward this would die.
    my $p2 = ClickHouse::Encoder::TCP->read_packet($r, buffer => \$buf);
    is($p2->{type}, 5, 'buffered read_packet: second call returns EndOfStream');
    is($buf, '', 'buffered read_packet: buffer drained');
    close $r;
}

# Without a buffer, read_packet is one-shot: the over-read EndOfStream
# byte is dropped, so a second call hits EOF and croaks.
{
    pipe(my $r, my $w) or die "pipe: $!";
    binmode $r; binmode $w;
    print $w "\x04\x05";
    close $w;

    my $p1 = ClickHouse::Encoder::TCP->read_packet($r);
    is($p1->{type}, 4, 'one-shot read_packet: first call returns Pong');
    my $err = eval {
        ClickHouse::Encoder::TCP->read_packet($r); 1
    } ? '' : $@;
    like($err, qr/connection closed/,
         'one-shot read_packet: over-read byte was dropped, 2nd call hits EOF');
    close $r;
}

# Live INSERT end-to-end. Opt-in: this requires a ClickHouse server
# at protocol revision <= 54474 (older than the chunking-negotiation
# extension introduced around CH 24.10). Modern servers (rev >=
# 54475) advertise a chunking offer right after Hello that this
# subset does not respond to, so the connection ends with a fast
# protocol-mismatch error rather than a real handshake. The unit
# tests above fully cover the wire-format encoding correctness.
SKIP: {
    skip "set TEST_CLICKHOUSE_TCP=1 (default port 9000) to enable "
       . "live TCP tests; requires a server at protocol revision "
       . "<= 54474 - newer servers add chunking negotiation past "
       . "this subset's scope", 4

t/tcp.t  view on Meta::CPAN

    # Create + INSERT
    print $sock ClickHouse::Encoder::TCP->pack_query(
        query => 'drop table if exists ch_tcp_t');
    # Consume to EndOfStream
    while (1) {
        my $p = ClickHouse::Encoder::TCP->read_packet($sock);
        last if $p->{type} == 5;  # EOS
        next if $p->{type} == 3;  # progress
        die "drop: unexpected $p->{type}" if $p->{type} == 2;
    }
    print $sock ClickHouse::Encoder::TCP->pack_query(
        query => 'create table ch_tcp_t (x Int32, s String) engine=Memory');
    while (1) {
        my $p = ClickHouse::Encoder::TCP->read_packet($sock);
        last if $p->{type} == 5;
        next if $p->{type} == 3;
        die "create: unexpected $p->{type}" if $p->{type} == 2;
    }

    # INSERT
    print $sock ClickHouse::Encoder::TCP->pack_query(
        query => 'insert into ch_tcp_t format native');
    # Server replies with TableColumns + empty Data sample block.
    # Drain until first non-progress non-table-columns packet.
    while (1) {
        my $p = ClickHouse::Encoder::TCP->read_packet($sock);
        last if $p->{type} == 1;       # SERVER_DATA (sample)
        next if $p->{type} == 11;      # TableColumns
        next if $p->{type} == 3;       # Progress
        die "insert: unexpected $p->{type}" if $p->{type} == 2;
    }

    my $enc = ClickHouse::Encoder->new(columns =>
        [['x','Int32'],['s','String']]);
    my $block = $enc->encode([[1,'a'],[2,'b'],[3,'c']]);
    print $sock ClickHouse::Encoder::TCP->pack_data($block);
    print $sock ClickHouse::Encoder::TCP->pack_data_end();

    # Read until EOS
    while (1) {
        my $p = ClickHouse::Encoder::TCP->read_packet($sock);
        last if $p->{type} == 5;
        next if $p->{type} == 3;
        next if $p->{type} == 6;       # ProfileInfo
        next if $p->{type} == 14;      # ProfileEvents
        die "insert flush: type $p->{type} $p->{message}\n"
            if $p->{type} == 2;
    }
    ok(1, 'INSERT via TCP completed without exception');

    # 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();

# helpers ----------------------------------------------------------
sub _varint {
    my $v = shift;
    my $s = '';
    while ($v >= 0x80) { $s .= chr(($v & 0x7f) | 0x80); $v >>= 7 }
    return $s . chr($v);
}
sub _len_str {
    my $s = shift;
    return _varint(length $s) . $s;
}