EV-ClickHouse
view release on metacpan or search on metacpan
- max\_query\_size => $bytes
Client-side guard: croak before sending any query whose SQL text exceeds
this many bytes. `0` (default) disables the check. Useful as a
last-resort defense against accidentally sending unbounded strings.
- max\_recv\_buffer => $bytes
Defensive ceiling on the response. The cap applies to the raw recv
buffer (every protocol), the chunked-decoded body (HTTP), and the
gzip-decompressed body (HTTP), so the same upper bound applies to the
user-visible payload regardless of transport encoding. On overflow the
query callback receives an appropriate error ("recv buffer overflow",
"chunked response too large", or "gzip body exceeds max\_recv\_buffer")
and the connection is torn down so no subsequent query can slip past
the cap on the same socket. `0` (default) keeps the historical
no-cap behaviour (still bounded internally by a hard 128 MB ceiling
on compressed paths). Recommended in production when the schema is
constrained and you want a hard upper bound (e.g.
`128 * 1024 * 1024` for 128 MB).
empty `data` arrayref is a valid zero-row table. Several external tables
may be supplied at once. Croaks on the HTTP protocol or on a malformed
spec (odd structure list, non-arrayref row, or a column type that
cannot be encoded).
**Native protocol type notes:** values come back as typed Perl scalars.
By default `Date`/`DateTime` are integers (days since epoch / Unix
timestamps); enable `decode_datetime` for strings. `Enum` values are
numeric codes; `decode_enum` returns labels. `Decimal` values are
unscaled integers; `decode_decimal` scales them to floats.
`SimpleAggregateFunction` is transparently decoded as its inner type.
`Nested` columns become arrays of tuples. `LowCardinality` works
correctly across multi-block results with shared dictionaries.
## insert
$ch->insert($table, $data, sub { my (undef, $err) = @_ });
$ch->insert($table, $data, \%settings, sub { my (undef, $err) = @_ });
`$data` may be either:
return decimal string representations on platforms with `__int128` (Int128/UInt128)
or always for the 256-bit forms.
- Floats
Float32/Float64 round-trip exactly within IEEE-754 limits. `NaN`/`+Inf`/
`-Inf` are preserved.
- BFloat16
Top 16 bits of a Float32. Encoded by truncation; decoded by zero-extension.
Suitable for ML feature columns; not for accounting.
- Decimal32/64/128
Decoded as IV (raw integer) or NV (scaled to N decimal digits if
`decode_decimal => 1`). Decimal128 over very long precision may lose
trailing digits in the NV form; pass `decode_decimal => 0` and divide
yourself with [Math::BigInt](https://metacpan.org/pod/Math%3A%3ABigInt) for exact arithmetic.
- Decimal256
lib/EV/ClickHouse.pm view on Meta::CPAN
=item max_query_size => $bytes
Client-side guard: croak before sending any query whose SQL text exceeds
this many bytes. C<0> (default) disables the check. Useful as a
last-resort defense against accidentally sending unbounded strings.
=item max_recv_buffer => $bytes
Defensive ceiling on the response. The cap applies to the raw recv
buffer (every protocol), the chunked-decoded body (HTTP), and the
gzip-decompressed body (HTTP), so the same upper bound applies to the
user-visible payload regardless of transport encoding. On overflow the
query callback receives an appropriate error ("recv buffer overflow",
"chunked response too large", or "gzip body exceeds max_recv_buffer")
and the connection is torn down so no subsequent query can slip past
the cap on the same socket. C<0> (default) keeps the historical
no-cap behaviour (still bounded internally by a hard 128 MB ceiling
on compressed paths). Recommended in production when the schema is
constrained and you want a hard upper bound (e.g.
C<128 * 1024 * 1024> for 128 MB).
lib/EV/ClickHouse.pm view on Meta::CPAN
spec (odd structure list, non-arrayref row, or a column type that
cannot be encoded).
=back
B<Native protocol type notes:> values come back as typed Perl scalars.
By default C<Date>/C<DateTime> are integers (days since epoch / Unix
timestamps); enable C<decode_datetime> for strings. C<Enum> values are
numeric codes; C<decode_enum> returns labels. C<Decimal> values are
unscaled integers; C<decode_decimal> scales them to floats.
C<SimpleAggregateFunction> is transparently decoded as its inner type.
C<Nested> columns become arrays of tuples. C<LowCardinality> works
correctly across multi-block results with shared dictionaries.
=head2 insert
$ch->insert($table, $data, sub { my (undef, $err) = @_ });
$ch->insert($table, $data, \%settings, sub { my (undef, $err) = @_ });
C<$data> may be either:
lib/EV/ClickHouse.pm view on Meta::CPAN
return decimal string representations on platforms with C<__int128> (Int128/UInt128)
or always for the 256-bit forms.
=item Floats
Float32/Float64 round-trip exactly within IEEE-754 limits. C<NaN>/C<+Inf>/
C<-Inf> are preserved.
=item BFloat16
Top 16 bits of a Float32. Encoded by truncation; decoded by zero-extension.
Suitable for ML feature columns; not for accounting.
=item Decimal32/64/128
Decoded as IV (raw integer) or NV (scaled to N decimal digits if
C<decode_decimal =E<gt> 1>). Decimal128 over very long precision may lose
trailing digits in the NV form; pass C<decode_decimal =E<gt> 0> and divide
yourself with L<Math::BigInt> for exact arithmetic.
=item Decimal256
t/21_edge_types.t view on Meta::CPAN
});
});
# 15-16: Map(String, UInt32)
with_native(sub {
$ch->query("select map('a', 1, 'b', 2)", sub {
my ($rows, $err) = @_;
ok(!$err, "Map: no error") or diag $err;
my $r = first_row($rows, $err);
is_deeply($r ? $r->[0] : undef, { a => 1, b => 2 },
"Map: hashref decoded");
EV::break;
});
});
t/24_review_gaps.t view on Meta::CPAN
host => $host, port => $nat_port, protocol => 'native',
on_connect => sub {
$ch->query("select toUUID('6ba7b810-9dad-11d1-80b4-00c04fd430c8')", sub {
($rows) = @_; EV::break;
});
},
);
run_with_timeout(10);
is($rows && @$rows ? $rows->[0][0] : undef,
'6ba7b810-9dad-11d1-80b4-00c04fd430c8',
"UUID decoded as canonical text");
$ch->finish if $ch->is_connected;
}
# 8: Date32 column round-trip with explicit decode (text form).
SKIP: {
skip "Native port not reachable", 1 unless $nat_ok;
my ($ch, $rows);
$ch = EV::ClickHouse->new(
host => $host, port => $nat_port, protocol => 'native',
decode_datetime => 1,
on_connect => sub {
$ch->query("select toDate32('2099-12-31')", sub {
($rows) = @_; EV::break;
});
},
);
run_with_timeout(10);
is($rows && @$rows ? $rows->[0][0] : undef, '2099-12-31',
"Date32 decoded as text past 2038 (no time_t overflow)");
$ch->finish if $ch->is_connected;
}
# 9: Enum16 column round-trip.
SKIP: {
skip "Native port not reachable", 1 unless $nat_ok;
my ($ch, $rows);
$ch = EV::ClickHouse->new(
host => $host, port => $nat_port, protocol => 'native', decode_enum => 1,
on_connect => sub {
$ch->query("select CAST('b' as Enum16('a' = 1, 'b' = 2, 'c' = 3))", sub {
($rows) = @_; EV::break;
});
},
);
run_with_timeout(10);
is($rows && @$rows ? $rows->[0][0] : undef, 'b', "Enum16 decoded with decode_enum");
$ch->finish if $ch->is_connected;
}
# 10-11: reset() explicit re-handshake. After reset(), the connection should
# be reconnect-able and the next query must succeed.
SKIP: {
skip "Native port not reachable", 2 unless $nat_ok;
my ($ch, $rows);
my $second_connects = 0;
$ch = EV::ClickHouse->new(
t/26_json.t view on Meta::CPAN
$ch->query("select j.name, j.age, j FROM $table order by j.name", sub {
($rows, $err) = @_; EV::break;
});
});
});
},
);
run_with_timeout(15);
ok(!$err, "typed-path JSON: no error") or diag "err=$err";
is($rows && @$rows ? $rows->[0][0] : undef, 'alice',
"typed path j.name decoded as String column");
is($rows && @$rows ? $rows->[0][1] : undef, 30,
"typed path j.age decoded as UInt32 column");
$ch->finish if $ch->is_connected;
}
t/28_pass2_coverage.t view on Meta::CPAN
EV::break;
});
},
on_error => sub { $err //= $_[0]; EV::break },
);
run_with_timeout(5);
like $err, qr/recv buffer/i, 'max_recv_buffer tears down on overflow';
eval { $ch->finish };
}
# max_recv_buffer also applies to chunked HTTP and gzip-decoded bodies
# (not only raw recv_buf). A small cap on a large select must error out.
SKIP: {
skip 'HTTP not reachable', 1
unless IO::Socket::INET->new(PeerAddr => $host, PeerPort => $hport, Timeout => 2);
my $err;
my $ch; $ch = EV::ClickHouse->new(
host => $host, port => $hport, protocol => 'http',
max_recv_buffer => 4096,
on_connect => sub {
$ch->query("select number from numbers(50000) format TabSeparated", sub {
t/47_extra_types.t view on Meta::CPAN
});
# Decimal256: no native int256, so the decoder delivers the raw 32-byte
# LE value (documented contract - hand to Math::BigInt for arithmetic).
typed_case("Decimal256", "select toDecimal256('123456789.123456789', 9) as d", 1, sub {
my ($rows) = @_;
is(length($rows->[0][0] // ''), 32,
"decimal256: decoder delivers the raw 32-byte value");
});
# Interval: decoded as an integer count of units.
typed_case("Interval", "select toIntervalDay(7) as iv", 2, sub {
my ($rows) = @_;
ok(defined $rows->[0][0], "interval: value decoded");
is($rows->[0][0], 7, "interval: toIntervalDay(7) decodes to 7");
});
# Geo Point: Tuple(Float64, Float64). Uses the ::Point cast (older
# servers lack the toPoint() function but accept the cast).
typed_case("Geo Point", "select (10, 20)::Point as pt", 2, sub {
my ($rows) = @_;
ok(ref $rows->[0][0] eq 'ARRAY', "geo point: decodes as a tuple/arrayref");
is_deeply($rows->[0][0], [10, 20], "geo point: [x, y] values");
});
xs/proto_http.c view on Meta::CPAN
static void process_http_response(ev_clickhouse_t *self) {
size_t hdr_end;
int status;
const char *val;
size_t val_len;
size_t content_length = 0;
int chunked = 0;
int is_gzip = 0;
const char *body;
size_t body_len;
char *decoded = NULL;
size_t decoded_len = 0;
size_t decoded_cap = 0;
if (self->recv_len == 0 || self->send_count == 0) return;
/* find headers end */
hdr_end = find_header_end(self->recv_buf, self->recv_len);
if (hdr_end == 0) return; /* need more data */
/* parse status */
status = parse_http_status(self->recv_buf, hdr_end);
xs/proto_http.c view on Meta::CPAN
/* guard against overflow and unbounded growth —
* close connection since remaining chunks would
* corrupt the stream for subsequent requests.
* Apply both the hard ceiling and the user's
* opt-in max_recv_buffer (when set) to the
* post-decode body size. */
size_t cap = CH_MAX_DECOMPRESS_SIZE;
if (self->max_recv_buffer > 0
&& self->max_recv_buffer < cap)
cap = self->max_recv_buffer;
if (decoded_len + chunk_size < decoded_len
|| decoded_len + chunk_size > cap) {
if (decoded) Safefree(decoded);
self->send_count--;
teardown_after_deliver(self,
"chunked response too large", "connection closed");
return;
}
if (decoded == NULL) {
decoded_cap = chunk_size + 256;
Newx(decoded, decoded_cap, char);
} else if (decoded_len + chunk_size > decoded_cap) {
decoded_cap = (decoded_len + chunk_size) * 2;
Renew(decoded, decoded_cap, char);
}
Copy(cp, decoded + decoded_len, chunk_size, char);
decoded_len += chunk_size;
cp += chunk_size + 2; /* skip chunk data + \r\n */
}
if (!chunked_complete) goto need_more;
}
body = decoded;
body_len = decoded_len;
consumed = cp - self->recv_buf;
} else {
/* Content-Length based — bound content_length before any pointer
* arithmetic so a malicious huge value cannot overflow
* hdr_end + content_length and slip past the need-more check. */
size_t cl_cap = CH_MAX_DECOMPRESS_SIZE;
if (self->max_recv_buffer > 0 && self->max_recv_buffer < cl_cap)
cl_cap = self->max_recv_buffer;
if (content_length > cl_cap) {
self->send_count--;
teardown_after_deliver(self,
"Content-Length exceeds limit", "connection closed");
return;
}
if (self->recv_len < hdr_end + content_length) return; /* need more data */
body = self->recv_buf + hdr_end;
body_len = content_length;
consumed = hdr_end + content_length;
}
/* deliver response (body, body_len, consumed are set; decoded may be NULL) */
self->send_count--;
if (status == 200) {
char *final_body = (char *)body;
size_t final_len = body_len;
if (is_gzip && body_len > 0) {
size_t dec_len;
char *dec = gzip_decompress(body, body_len, &dec_len);
if (!dec) {
if (decoded) Safefree(decoded);
recv_consume(self, consumed);
int destroyed = deliver_error(self, "gzip decompression failed");
if (destroyed) return;
goto done;
}
/* Apply user's opt-in max_recv_buffer to the gzip-decoded
* body too — same trip-wire semantics as the chunked path
* and the on_readable raw-recv check. Tear the connection
* down on overflow so subsequent queries can't slip past
* the cap on the same socket (matches the chunked path's
* behaviour and the POD contract). */
if (self->max_recv_buffer > 0
&& dec_len > self->max_recv_buffer) {
Safefree(dec);
if (decoded) Safefree(decoded);
teardown_after_deliver(self,
"gzip body exceeds max_recv_buffer", "connection closed");
return;
}
final_body = dec;
final_len = dec_len;
}
{
int is_raw = peek_cb_raw(self);
xs/proto_http.c view on Meta::CPAN
if (is_raw) {
/* raw mode — deliver body as scalar, skip TSV parsing */
destroyed = deliver_raw_body(self, final_body, final_len);
} else {
AV *rows = NULL;
if (final_len > 0)
rows = parse_tab_separated(final_body, final_len);
destroyed = deliver_rows(self, rows);
}
if (final_body != body) Safefree(final_body);
if (decoded) Safefree(decoded);
recv_consume(self, consumed);
if (destroyed) return;
}
} else {
/* error */
char *errmsg = format_http_error(self, status, body, body_len, is_gzip);
if (decoded) Safefree(decoded);
recv_consume(self, consumed);
int destroyed = deliver_error(self, errmsg);
Safefree(errmsg);
if (destroyed) return;
}
if (self->magic != EV_CH_MAGIC) return;
done:
/* Stop query timeout timer on response */
stop_timing(self);
pipeline_advance(self);
return;
need_more:
/* incomplete response — keep reading */
if (decoded) Safefree(decoded);
return;
}
xs/proto_native_parse.c view on Meta::CPAN
if (self->server_revision >= DBMS_MIN_REVISION_WITH_BLOCK_INFO) {
rc = skip_block_info(bbuf, blen, &bpos);
if (rc <= 0) _BAIL_LOG(rc);
}
uint64_t nc, nr;
rc = read_varuint(bbuf, blen, &bpos, &nc);
if (rc <= 0) _BAIL_LOG(rc);
rc = read_varuint(bbuf, blen, &bpos, &nr);
if (rc <= 0) _BAIL_LOG(rc);
/* Collect column name + decoded values, then assemble per-row HVs. */
char **names = NULL;
SV ***data = NULL; /* data[col][row] */
if (nc > 0) {
/* A column occupies at least one wire byte (its name-length varint),
* so more columns than remaining bytes is malformed — reject before
* the allocation rather than letting Newxz attempt a huge size. */
if (nc > (uint64_t)(blen - bpos)) _BAIL_LOG(-1);
Newxz(names, nc, char *);
Newxz(data, nc, SV **);
}
( run in 0.427 second using v1.01-cache-2.11-cpan-f4a522933cf )