view release on metacpan or  search on metacpan

t/live.t  view on Meta::CPAN

    my $n = ch_query_result("select count() from test_json");
    is($n, '6', '6 rows in JSON table');

    my $nm_alice = ch_query_result(
        "select count() from test_json where j.name = 'alice'");
    is($nm_alice, '1', 'string subcolumn query');
    my $score_count = ch_query_result(
        "select count() from test_json where toFloat64(j.score) > 3");
    is($score_count, '1', 'float subcolumn query');
    my $carol_age = ch_query_result(
        "select toInt64(j.user.age) from test_json where j.user.name = 'carol'");
    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([
        [{name => 'alice', age => 30}],
        [{name => 'bob',   age => 25, extra => 'more'}],
    ]);
    ok(ch_insert_native('test_json_typed', $td),
       'typed JSON paths INSERT accepted');
    my $cnt = ch_query_result("select count() from test_json_typed");
    is($cnt, '2', 'typed JSON paths: 2 rows');
    my $name = ch_query_result(
        "select j.name from test_json_typed where j.age = 30");
    is($name, 'alice', 'typed path subcolumn query');
}

# Test: compressed Native INSERT over HTTP. End-to-end wire-compat
# check for the bundled CityHash128 v1.0.2 port - if our hash diverges
# from cityhash102 in any detail the server rejects with "Checksum
# doesn't match in compressed block" before storing any rows.
SKIP: {
    eval { require HTTP::Tiny; require Compress::LZ4; 1 }
        or skip 'HTTP::Tiny or Compress::LZ4 not installed', 4;

    my $port = $http_port;
    my $http = HTTP::Tiny->new(timeout => 5);
    my $ping = $http->get("http://127.0.0.1:$port/ping");
    skip "HTTP endpoint not reachable on :$port", 4
        unless $ping->{success} && $ping->{content} =~ /Ok/;

    # Drop + create a fresh table via the existing client; then push
    # rows through the HTTP `?decompress=1` path.
    ch_query("drop table if exists test_compressed");
    ch_query("create table test_compressed
                  (id Int32, msg String, ts DateTime) engine = Memory");

    my $enc = ClickHouse::Encoder->new(columns =>
        [['id','Int32'], ['msg','String'], ['ts','DateTime']]);
    my $native = $enc->encode([
        [1, 'alpha',  1700000001],
        [2, 'beta',   1700000002],
        [3, 'gamma',  1700000003],
    ]);

    # Wrap in CH's compressed-block framing - this calls our bundled
    # CityHash128 v1.0.2 implementation. If the hash bytes diverge
    # from CH's cityhash102 the server will refuse the INSERT.
    my $framed = ClickHouse::Encoder->compress_native_block(
        $native, mode => 'lz4');

    my $resp = $http->post(
        "http://127.0.0.1:$port/"
            . "?query=insert+into+test_compressed+format+native"
            . "&decompress=1",
        { content => $framed,
          headers => { 'Content-Type' => 'application/octet-stream' } });
    ok($resp->{success},
       'compressed Native INSERT accepted (CityHash128 validates against CH)')
        or diag("status=$resp->{status} body=$resp->{content}");

    my $cnt = ch_query_result("select count() from test_compressed");
    is($cnt, '3', 'compressed INSERT delivered 3 rows');

    my $sum = ch_query_result(
        "select sum(id) from test_compressed");
    is($sum, '6', 'compressed INSERT row values intact (sum=6)');

    # Round-trip the other way: SELECT the data back uncompressed and
    # confirm the values match (additional pin on the whole pipeline).
    my $first_msg = ch_query_result(
        "select msg from test_compressed where id = 2");
    is($first_msg, 'beta', 'compressed-then-stored row decodes correctly');

    # Round-trip the other direction: SELECT with ?compress=1 and read
    # the response through select_blocks(decompress => 1). End-to-end
    # validation of CityHash128 in the response direction too.
    my @ids;
    ClickHouse::Encoder->select_blocks(
        "select id, msg, ts from test_compressed order by id",
        host => '127.0.0.1', port => $port,
        decompress => 1,
        on_block   => sub {
            my $b = shift;
            for my $col (@{ $b->{columns} }) {
                push @ids, @{ $col->{values} } if $col->{name} eq 'id';
            }
        });
    is_deeply(\@ids, [1, 2, 3],