view release on metacpan or  search on metacpan

t/roundtrip.t  view on Meta::CPAN

    elsif ($code eq 'FixedString') {
        my $n = $type->{n};
        for (1..$nrows) { push @vals, substr($$buf, $$off, $n); $$off += $n }
    }
    elsif ($code eq 'Enum8')   { for (1..$nrows) { push @vals, unpack('c', substr($$buf, $$off, 1)); $$off += 1 } }
    elsif ($code eq 'Enum16')  { for (1..$nrows) { push @vals, unpack('s<', substr($$buf, $$off, 2)); $$off += 2 } }
    elsif ($code eq 'Decimal128') {
        for (1..$nrows) {
            my $lo = unpack('Q<', substr($$buf, $$off, 8));
            my $hi = unpack('q<', substr($$buf, $$off + 8, 8));
            push @vals, [$lo, $hi];
            $$off += 16;
        }
    }
    elsif ($code eq 'Decimal256') {
        for (1..$nrows) {
            push @vals, [map {
                unpack('Q<', substr($$buf, $$off + 8 * $_, 8))
            } 0 .. 3];
            $$off += 32;
        }
    }
    elsif ($code eq 'Bool' || $code eq 'Boolean') {
        for (1..$nrows) { push @vals, unpack('C', substr($$buf, $$off, 1)); $$off += 1 }
    }
    elsif ($code eq 'UUID') {
        for (1..$nrows) { push @vals, substr($$buf, $$off, 16); $$off += 16 }
    }
    elsif ($code eq 'IPv4') {
        for (1..$nrows) {
            my $v = unpack('V', substr($$buf, $$off, 4));
            $$off += 4;
            push @vals, sprintf('%d.%d.%d.%d',
                ($v>>24)&0xff, ($v>>16)&0xff, ($v>>8)&0xff, $v&0xff);
        }
    }
    elsif ($code eq 'IPv6') {
        for (1..$nrows) { push @vals, substr($$buf, $$off, 16); $$off += 16 }
    }
    elsif ($code eq 'Map') {
        # Map(K, V) on the wire is Array(Tuple(K, V)).
        my $array_t = { code => 'Array',
            inner => { code => 'Tuple', parts => [$type->{key}, $type->{val}] } };
        @vals = @{ _decode_column($buf, $off, $array_t, $nrows) };
    }
    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]++ ]];
        }
    }
    elsif ($code eq 'LowCardinality') {
        my $version = unpack('Q<', substr($$buf, $$off, 8));   $$off += 8;
        my $flags   = unpack('Q<', substr($$buf, $$off, 8));   $$off += 8;
        my $dict_n  = unpack('Q<', substr($$buf, $$off, 8));   $$off += 8;
        die "LC version != 1" unless $version == 1;
        my $idx_type = $flags & 0xff;
        my $inner    = $type->{inner};
        $inner = $inner->{inner} if $inner->{code} eq 'Nullable';
        my $dict = _decode_column($buf, $off, $inner, $dict_n);
        my $idx_n = unpack('Q<', substr($$buf, $$off, 8));     $$off += 8;
        for my $r (1 .. $idx_n) {
            my $i;
            if    ($idx_type == 0) { $i = ord(substr($$buf, $$off, 1)); $$off += 1 }
            elsif ($idx_type == 1) { $i = unpack('v',  substr($$buf, $$off, 2)); $$off += 2 }
            elsif ($idx_type == 2) { $i = unpack('V',  substr($$buf, $$off, 4)); $$off += 4 }
            else                    { $i = unpack('Q<', substr($$buf, $$off, 8)); $$off += 8 }
            # For Nullable LC, slot 0 is the null sentinel.
            if ($type->{inner}{code} eq 'Nullable' && $i == 0) {
                push @vals, undef;
            } else {
                push @vals, $dict->[$i];
            }
        }
    }
    elsif ($code eq 'Array') {
        my @offsets;
        for (1..$nrows) { push @offsets, unpack('Q<', substr($$buf, $$off, 8)); $$off += 8 }
        my $total = @offsets ? $offsets[-1] : 0;
        my $flat = _decode_column($buf, $off, $type->{inner}, $total);
        my $prev = 0;
        for my $end (@offsets) {
            push @vals, [@$flat[$prev .. $end - 1]];
            $prev = $end;
        }
    }
    elsif ($code eq 'Tuple') {
        my @cols = map { _decode_column($buf, $off, $_, $nrows) } @{$type->{parts}};
        for my $r (0 .. $nrows - 1) {
            push @vals, [map { $_->[$r] } @cols];
        }
    }
    elsif ($code eq 'Nullable') {
        my @nulls;
        for (1..$nrows) { push @nulls, ord(substr($$buf, $$off, 1)); $$off += 1 }
        my $inner = _decode_column($buf, $off, $type->{inner}, $nrows);
        for my $r (0 .. $nrows - 1) {
            push @vals, $nulls[$r] ? undef : $inner->[$r];
        }
    }
    else {
        die "Decoder: unsupported $code";
    }
    return \@vals;
}

sub decode_block {
    my ($buf) = @_;
    my $off = 0;