ClickHouse-Encoder

 view release on metacpan or  search on metacpan

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

}

sub for_table {
    my ($class, $table, %opts) = @_;
    _validate_table_name($table);
    return $class->_for_describe("describe table $table", %opts);
}

sub _for_describe {
    my ($class, $describe_sql, %opts) = @_;

    my $via = $opts{via} // 'client';
    my @lines;

    if ($via eq 'http') {
        my ($url, $hdr) = _http_url_headers(
            "$describe_sql format tabseparated", %opts);
        my $resp = _http_tiny(%opts, timeout => $opts{timeout} // 10)
            ->get($url, { headers => $hdr });
        die "HTTP describe table failed (status $resp->{status}): $resp->{content}\n"
            unless $resp->{success};
        @lines = split /\n/, $resp->{content};
    }
    elsif ($via eq 'client') {
        my $host     = $opts{host}     // 'localhost';
        my $port     = $opts{port}     // 9000;
        my $database = $opts{database} // 'default';
        my $user     = $opts{user}     // 'default';
        my $password = $opts{password} // '';
        my $client   = $opts{client}   // 'clickhouse-client';

        my @cmd = (
            $client,
            '--host', $host,
            '--port', $port,
            '--database', $database,
            '--user', $user,
            '--query', "$describe_sql format tabseparated",
        );

        # Pass password via env so it doesn't show up in `ps`. Always
        # set $password (even empty) so an empty `password => ''` arg
        # also overrides any inherited CLICKHOUSE_PASSWORD.
        local $ENV{CLICKHOUSE_PASSWORD} = $password;

        open my $fh, '-|', @cmd
            or die "Failed to run clickhouse-client: $!";
        @lines = <$fh>;
        close $fh;
        die "clickhouse-client failed: exit code " . ($? >> 8) if $?;
    }
    else {
        die "Unknown via='$via' (expected 'client' or 'http')";
    }

    my @columns;
    # ClickHouse's TabSeparated format escapes \\ \n \t \r \0 in field
    # values. Multi-line type expressions (e.g. named Tuple, Nested) come
    # out as a single TSV line with embedded \n and indentation; un-escape
    # so the XS parser sees the canonical type string. The default branch
    # ($1) handles \\ -> \ and any other backslash-escape forms transparently.
    my $unesc = sub {
        my $s = shift // '';
        $s =~ s{\\(.)}{ $1 eq 'n' ? "\n"
                      : $1 eq 't' ? "\t"
                      : $1 eq 'r' ? "\r"
                      : $1 eq '0' ? "\0"
                      : $1 }ge;
        $s;
    };
    for my $line (@lines) {
        chomp $line;
        next if $line eq '';
        my ($name, $type) = split /\t/, $line;
        push @columns, [$unesc->($name), $unesc->($type)];
    }

    die "No columns found for: $describe_sql" unless @columns;
    return $class->new(columns => \@columns);
}

sub columns {
    my $self = shift;
    return $self->_columns;
}

sub validate_rows {
    my ($self, $rows) = @_;
    my @errors;
    for my $i (0 .. $#$rows) {
        local $@;
        unless (eval { $self->encode([$rows->[$i]]); 1 }) {
            (my $msg = "$@") =~ s/\s*at .+ line \d+\.\s*\z//;
            push @errors, { row => $i, error => $msg };
        }
    }
    return \@errors;
}

sub compressed_writer {
    my ($class, $mode, $writer) = @_;
    return $writer if !defined $mode || $mode eq 'raw';
    if ($mode eq 'zstd') {
        require Compress::Zstd;
        return sub {
            my $out = Compress::Zstd::compress($_[0]);
            defined $out or die "zstd compression failed";
            $writer->($out);
        };
    }
    if ($mode eq 'gzip') {
        require IO::Compress::Gzip;
        return sub {
            my $out;
            IO::Compress::Gzip::gzip(\$_[0], \$out)
                or die "gzip failed: $IO::Compress::Gzip::GzipError";
            $writer->($out);
        };
    }
    die "Unknown compress mode '$mode' (expected 'zstd', 'gzip', or 'raw')";
}

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

=over 4

=item *

Integers: C<Int8>, C<Int16>, C<Int32>, C<Int64>, C<UInt8>, C<UInt16>,
C<UInt32>, C<UInt64>.

=item *

Floats: C<Float32>, C<Float64>, C<BFloat16> (CH 24.x; 2-byte truncated
Float32). C<Inf>, C<-Inf>, and C<NaN> are preserved.

=item *

Strings: C<String> (length-prefixed bytes), C<FixedString(N)> (N bytes,
null-padded). Both pass the SV's bytes through unchanged: a UTF-8 string
encodes its UTF-8 bytes, a binary blob encodes its bytes, and truncation
is by byte not codepoint.

=item *

Dates: C<Date>, C<Date32>, C<DateTime>, C<DateTime('tz')> (timezone is part
of the schema, not the value), C<DateTime64(P)> with C<P> in 0..9.

=item *

Decimals: C<Decimal32(S)> (S in 0..9), C<Decimal64(S)> (0..18),
C<Decimal128(S)> (0..38), C<Decimal256(S)> (0..76), and
C<Decimal(P, S)> with C<P> in 1..38 (auto-routed to 32/64/128).

=item *

Enums: C<Enum8('a' = 1, ...)>, C<Enum16(...)>.

=item *

C<Bool> / C<Boolean> (1 byte; truthy/falsy in Perl sense).

=item *

C<UUID> (16 bytes; accept either the standard
C<xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx> string or 16 raw bytes).

=item *

C<IPv4> (UInt32 LE; accept dotted-quad string or integer),
C<IPv6> (16 bytes network-order; accept colon-hex string or 16 raw bytes).

=item *

C<Map(K, V)> (wire-equivalent to C<Array(Tuple(K, V))>; accept either a
hashref or an arrayref of pairs).

=item *

C<Variant(T1, T2, ...)> (CH 24.1+; tagged union). Each row is either
C<undef> (null) or C<[$variant_idx, $value]>, where C<$variant_idx> is
the 0-based position in the declared C<Variant(...)> type list (so
C<Variant(String, UInt32)> uses 0 for String and 1 for UInt32). Up to 254
variants. ClickHouse stores Variant arms alphabetically by type name on
the wire; the encoder remaps your declared index transparently, so you
always use declaration order in your code. C<describe table> returns the
arms already alphabetized, and C<for_table> uses that form, so indices
match either way.

=item *

C<LowCardinality(String)>, C<LowCardinality(FixedString(N))>,
C<LowCardinality(Nullable(String))>, C<LowCardinality(Nullable(FixedString(N)))>.

=item *

Geo: C<Point>, C<Ring>, C<LineString>, C<MultiLineString>, C<Polygon>,
C<MultiPolygon> (aliases over C<Tuple>/C<Array>). L</parse_wkt> turns
a Well-Known-Text string into the nested-arrayref input these
columns expect.

=item *

Composites: C<Array(T)>, C<Tuple(T1, T2, ...)>, C<Nullable(T)>. Arbitrary
nesting; C<Nullable(Nullable(T))> is rejected, matching ClickHouse. Named
tuple elements (C<Tuple(a Int32, b String)>) are accepted; the wire
format itself ignores names, but for a named tuple the encoder will
also accept rows as hashrefs (C<<< { a => 1, b => "x" } >>>) in addition
to arrayrefs (C<[1, "x"]>). Missing keys in a hashref encode as the
type's null placeholder.

=item *

C<SimpleAggregateFunction(func, T)>: parsed as plain C<T>; the
aggregation function name only affects how readers consume the column,
not the wire-level binary format. Full C<AggregateFunction> (with
function-specific binary state) is not supported; use
C<SimpleAggregateFunction> where the inner type matches the on-the-wire
representation.

=item *

C<JSON> / C<Object('json')> (the stable CH 24.8+ JSON type). Each row
is a hashref of leaf values; nested hashrefs are auto-flattened into
dotted-path subcolumns matching CH's internal storage (so
C<<< { user => { name => "alice" } } >>> stores under the C<user.name>
path on the wire and round-trips back as the same nested structure
on decode). Leaf value typing is inferred per-path from Perl SV flags:

=over 4

=item *

Numeric SV with the integer flag set but not the float flag
(C<SvIOK && !SvNOK>) -> C<Int64>.

=item *

Numeric SV with the float flag set (C<SvNOK>): non-integer value ->
C<Float64>; a NV that happens to equal an integer (e.g. C<1.0>)
collapses to C<Int64> on the wire, mirroring CH's own C<JSONEachRow>
inference. Round-trip then decodes back as an integer; to keep an
integer-valued float as C<Float64> use a dedicated C<Float64> column
or a C<<< Variant(Float64, ...) >>>. C<NaN> and C<Inf> stay C<Float64>.

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

is integer-only).

=item *

C<Decimal*>: a number goes through C<double> (lossy past 2^53); a
B<string> matching C<[+-]?digits[.digits]?> is parsed digit-by-digit and
scaled exactly. B<If precision matters, pass strings.>

=item *

C<Enum8> / C<Enum16>: accept either the declared name or its integer
value; mixing the two within a column is fine.

=item *

C<Nullable(T)>: C<undef> writes a null-bitmap entry plus a type-shaped
placeholder (zero scalar, empty array, or a recursive zero tuple).

=back

=head1 EXAMPLES

The F<eg/> directory ships runnable scripts:

=over 4

=item F<eg/insert_http.pl>

End-to-end insert over HTTP via L<HTTP::Tiny>. The shortest path to "insert
real data into a real ClickHouse".

=item F<eg/insert_streaming.pl>

Reuse one encoder across many batches, piping each batch to
C<clickhouse-client>. Demonstrates the intended one-encoder-many-batches
pattern.

=item F<eg/for_table.pl>

Schema discovery via C<for_table>.

=item F<eg/from_csv.pl>

Read a CSV with L<Text::CSV_XS>, map columns to a ClickHouse schema, and
insert via HTTP.

=item F<eg/insert_clickhouse_local.pl>

Server-less ETL: encode rows, pipe Native bytes into C<clickhouse-local>,
have it write a Parquet (or ORC, etc.) file.

=item F<eg/etl_dbi.pl>

Read rows from a source database via L<DBI>, encode to Native, insert into
ClickHouse via HTTP. Reuses one encoder across all fetched batches.

=item F<eg/insert_compressed.pl>

insert with on-the-wire compression (zstd via L<Compress::Zstd>, falling
back to gzip via core L<IO::Compress::Gzip>). Sets C<Content-Encoding>
so ClickHouse decompresses transparently.

=item F<eg/insert_async_ev.pl>

Non-blocking concurrent inserts using L<EV>'s event loop with raw HTTP
sockets, paired with this encoder's L</streamer>. Demonstrates the
"many in-flight inserts without blocking on each round-trip" pattern.

=item F<eg/insert_with_lowcardinality.pl>

Measures the wire-size reduction (~50% on event/log data) and encoding
throughput of C<LowCardinality(String)> versus plain C<String> for the
typical case where a column has few distinct values that repeat across
many rows.

=item F<eg/json_lines_ingest.pl>

Reads NDJSON from STDIN or a file, maps each object's fields onto a
ClickHouse table's columns (discovered via C<for_table>), and inserts
batched blocks over HTTP.

=item F<eg/streaming_aggregate.pl>

Pre-aggregates an event stream in Perl (per minute, per key) and
flushes rolled-up counters to a C<SummingMergeTree> on a wall-clock
timer. The classic pattern when the firehose is too high-cardinality
to store as raw rows.

=item F<eg/postgres_to_clickhouse.pl>

Replicates a PostgreSQL table to ClickHouse using L<DBD::Pg> on the
source side and this encoder's streamer on the destination side.
Memory is bounded by the batch size, so it scales to hundreds of
millions of rows.

=item F<eg/clickhouse_replication.pl>

Replicates one ClickHouse table to another (potentially on a different
server) by streaming Native bytes end-to-end via a temp-file spool.

=item F<eg/parallel_loader.pl>

Forks N worker processes, each ingesting one slice of the input.
Workers share nothing; each opens its own HTTP connection. Scales
network-bound ingestion linearly with worker count.

=item F<eg/redis_to_clickhouse.pl>

Drains a Redis stream (XREADGROUP) or list (BRPOP) into a ClickHouse
table, with idle-flush so the destination doesn't see arbitrarily
delayed batches when the source is quiet.

=item F<eg/syslog_ingest.pl>

Reads RFC 5424 syslog lines from STDIN, parses lossily (any
unparseable line still goes through with the raw text in C<msg>),
and inserts into a fixed schema.

=item F<eg/json_streaming.pl>

NDJSON from STDIN into a C<JSON> column via the encoder's streaming