view release on metacpan or  search on metacpan

t/decode-blocks.t  view on Meta::CPAN

#!/usr/bin/env perl
use strict;
use warnings;
use Test::More;
use lib 'blib/lib', 'blib/arch';
use ClickHouse::Encoder;

# decode_blocks reads a concatenated stream of Native blocks - which is
# what a select ... format native response over HTTP returns when the
# result spans multiple blocks (the server emits one block per N rows,
# 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');
}

# Single block also works
{
    my $blocks = ClickHouse::Encoder->decode_blocks($block1);
    is(scalar(@$blocks), 1, 'single block via decode_blocks');
    is($blocks->[0]{nrows}, 2);
}

# Three blocks
{
    my $three = $block1 . $block1 . $block2;
    my $blocks = ClickHouse::Encoder->decode_blocks($three);
    is(scalar(@$blocks), 3, 'three blocks');
    is_deeply($blocks->[0]{columns}[0]{values}, [1, 2]);
    is_deeply($blocks->[1]{columns}[0]{values}, [1, 2]);
    is_deeply($blocks->[2]{columns}[0]{values}, [3, 4]);
}

# Trailing garbage croaks
{
    my $err = eval {
        ClickHouse::Encoder->decode_blocks($block1 . "xxx"); 1
    } ? "" : $@;
    like($err, qr/truncated|exceeds remaining/i, 'trailing garbage croaks');
}

# 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
    } ? "" : $@;
    like($err, qr/past end/, 'offset past end rejected');
}

# decode_blocks callback form
{
    my $three = $block1 . $block1 . $block2;
    my @nrows;
    ClickHouse::Encoder->decode_blocks($three, sub {
        push @nrows, $_[0]{nrows};
    });
    is_deeply(\@nrows, [2, 2, 2], 'callback form invoked per block');
}

# decode_blocks_iter returns coderef yielding one block per call
{
    my $three = $block1 . $block1 . $block2;
    my $iter = ClickHouse::Encoder->decode_blocks_iter($three);
    isa_ok($iter, 'CODE', 'iter returns coderef');
    my @blocks;
    while (my $b = $iter->()) { push @blocks, $b }
    is(scalar(@blocks), 3, 'iter yielded 3 blocks');
    is($iter->(), undef, 'iter exhausted returns undef');
}

# Empty input via iter
{
    my $iter = ClickHouse::Encoder->decode_blocks_iter("");
    is($iter->(), undef, 'empty input iter -> undef immediately');
}

# decode_stream: pull-style from a filehandle, exercises chunk-by-chunk
# buffer growth and the "need more bytes" retry path.
{
    require File::Temp;
    my ($fh, $path) = File::Temp::tempfile(UNLINK => 1);
    binmode $fh;
    print $fh $block1 . $block2;
    close $fh;
    open my $in, '<', $path or die "open $path: $!";
    binmode $in;
    my @nrows;
    ClickHouse::Encoder->decode_stream($in,
        sub { push @nrows, $_[0]{nrows} },
        chunk_size => 4);  # tiny chunks force the buffer-grow path
    close $in;
    is_deeply(\@nrows, [2, 2],
              'decode_stream yields blocks across small reads');
}

# Truncated tail -> croak with clear message.