view release on metacpan or  search on metacpan

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

package EV::ClickHouse;
use strict;
use warnings;

use EV;
use Scalar::Util qw(refaddr weaken);

# Identifier validation regexes — single source of truth for the table
# / column / function names that get spliced into SQL via the helpers.
my $RE_IDENT = qr/\A[A-Za-z_][A-Za-z0-9_]*\z/;
my $RE_TABLE = qr/\A[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)?\z/;

BEGIN {
    our $VERSION = '0.05';
    use XSLoader;
    XSLoader::load __PACKAGE__, $VERSION;
}

# Holds in-flight EV::cares resolvers so they aren't garbage-collected
# before their callback fires. Keyed by refaddr of the resolver itself
# (not the connection) — see new() for the rationale. Each entry is
# deleted from inside its own resolved callback via a deferred timer.
# Plain package hash (not Hash::Util::FieldHash) so module load doesn't
# add tied/magic SVs that Test::LeakTrace would flag.
our %_failover;

*q          = \&query;
*reconnect  = \&reset;
*disconnect = \&finish;
*ddl        = \&query;  # readability at call sites for DDL/DML

sub _uri_unescape { my $s = $_[0]; $s =~ s/%([0-9A-Fa-f]{2})/chr(hex($1))/ge; $s }

# Parse a URI query string into a hash. Bare keys (no `=`) are stored
# as 1 (standard URL flag convention); existing keys in $h are preserved
# (path-derived host/port/etc win over query-string overrides).
sub _uri_qs_into {
    my ($qs, $h) = @_;
    return unless defined $qs && length $qs;
    for my $pair (split /&/, $qs) {
        my ($k, $v) = split /=/, $pair, 2;
        next unless defined $k && length $k;
        $h->{$k} //= defined $v ? _uri_unescape($v) : 1;
    }
}

sub new {
    my ($class, %args) = @_;

    # Connection URI: clickhouse://user:pass@host:port/database
    # host accepts a bracketed IPv6 literal (e.g. clickhouse://[::1]:9000/db)
    if (my $uri = delete $args{uri}) {
        if ($uri =~ m{^clickhouse(?:\+(\w+))?://(?:([^:@]*?)(?::([^@]*))?\@)?(\[[^\]]+\]|[^/:?]+)(?::(\d+))?(?:/([^?]*))?(?:\?(.*))?$}) {
            my ($proto, $u, $pw, $h, $p, $db, $qs) = ($1, $2, $3, $4, $5, $6, $7);
            $h =~ s/^\[(.*)\]$/$1/;
            $args{protocol} //= $proto if $proto;
            $args{user}     //= _uri_unescape($u)  if defined $u && $u ne '';
            $args{password} //= _uri_unescape($pw) if defined $pw;
            $args{host}     //= $h;
            $args{port}     //= $p     if defined $p;
            $args{database} //= _uri_unescape($db) if defined $db && $db ne '';
            _uri_qs_into($qs, \%args);
        } else {
            die "EV::ClickHouse: invalid URI '$uri'\n";
        }
    }

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

        my $setter = "_set_$opt";
        $self->$setter($val);
    }
    for my $opt (qw(session_id tls_ca_file tls_cert_file tls_key_file)) {
        defined(my $val = delete $args{$opt}) or next;
        my $setter = "_set_$opt";
        $self->$setter($val);
    }

    # query_log_comment: 1 = auto-generate "ev_ch user=$ENV{USER} pid=$$"; any
    # other defined non-empty string (including "0") is taken literally;
    # undef / not present / empty string is disabled.
    {
        my $qlc = delete $args{query_log_comment};
        if (defined $qlc && length $qlc) {
            my $cmt = (!ref($qlc) && "$qlc" ne '1')
                    ? $qlc
                    : sprintf 'ev_ch user=%s pid=%d', $ENV{USER} // 'na', $$;
            $cmt =~ s{\*/}{*\\/}g;
            $self->_set_query_log_comment($cmt);
        }
    }

    # decode_flags bitmask (DT_STR=1, DEC_SCALE=2, ENUM_STR=4, NAMED_ROWS=8)
    my $decode_flags = (delete $args{decode_datetime} ? 1 : 0)
                     | (delete $args{decode_decimal}  ? 2 : 0)
                     | (delete $args{decode_enum}     ? 4 : 0)
                     | (delete $args{named_rows}      ? 8 : 0);
    $self->_set_decode_flags($decode_flags) if $decode_flags;

    if (my $settings = delete $args{settings}) { $self->_set_settings($settings) }

    warn "EV::ClickHouse->new: unknown parameter(s): " . join(', ', sort keys %args) . "\n"
        if %args;

    if ($hosts_list) {
        $self->_set_failover($hosts_list, $port);
    }

    # Async DNS via EV::cares when available — non-IP hostnames are
    # resolved off-loop so the constructor returns immediately and the
    # main EV loop never blocks on getaddrinfo. Pre-connect-queued
    # queries fire once the resolved-address connect completes. Falls
    # back to the XS blocking resolver if EV::cares isn't installed
    # or the host is already an IP literal.
    if ($host !~ /^[\d.]+$|^\[?[0-9a-fA-F:]+\]?$/
        && eval { require EV::cares; 1 }) {
        # Stash the resolver in %_failover, keyed by refaddr of the
        # resolver itself (NOT of $self). Two reasons:
        #   - never delete the resolver from inside its own callback —
        #     ares_destroy from a c-ares cb corrupts the channel heap.
        #     We defer the delete via EV::timer(0,...) so it runs from
        #     a clean stack frame.
        #   - keying by refaddr($self) was racy: A's deferred-delete
        #     could fire after A's struct was freed and B got the same
        #     refaddr, dropping B's resolver. refaddr($r) is unique
        #     while $r is alive in %_failover.
        my $r = EV::cares->new;
        my $key = refaddr($r);
        $_failover{$key} = $r;
        my $weak2 = $self; weaken $weak2;
        $self->_set_dns_pending(1);
        $r->resolve($host, sub {
            my ($status, @addrs) = @_;
            my $w; $w = EV::timer(0, 0, sub { undef $w; delete $_failover{$key} });
            # Skip if the connection has been DESTROYed or the user
            # finished it while DNS was in flight (cleanup_connection
            # clears dns_pending; if it's 0 here, finish ran already).
            return unless $weak2 && $weak2->_take_dns_pending;
            if ($status != 0 || !@addrs) {
                $weak2->skip_pending;
                # Warn if the handler itself throws — matches the XS
                # emit_error path (WARN_AND_CLEAR_ERRSV); a bare eval here
                # would make a DNS failure vanish under the default
                # `on_error => sub { die @_ }`.
                eval { $weak2->on_error->("DNS resolution failed for '$host'"); 1 }
                    or warn "EV::ClickHouse: exception in error handler: $@";
                return;
            }
            my ($v4) = grep /^[\d.]+$/, @addrs;
            $weak2->connect($v4 // $addrs[0], $port, $user, $password, $database);
        });
    } else {
        $self->connect($host, $port, $user, $password, $database);
    }

    $self;
}

sub _split_host_port {
    my ($entry, $default_port) = @_;
    if ($entry =~ /^\[([^\]]+)\](?::(\d+))?$/) {
        return ($1, $2 // $default_port);   # IPv6 literal in brackets
    }
    if ($entry =~ /^([^:]+):(\d+)$/) {
        return ($1, $2);
    }
    return ($entry, $default_port);
}

# Pull-based result iterator: $it = $ch->iterate($sql, [\%settings])
#   while (my $batch = $it->next($timeout)) { ... }
# Wraps the native on_data per-block callback in a synchronous-feeling
# pull interface for procedural code. The iterator drives the EV loop
# from inside ->next until the next block arrives, the query completes,
# or the optional timeout (seconds) expires.
sub iterate {
    my ($self, $sql, $settings) = @_;
    my $it = bless {
        ch       => $self,
        batches  => [],
        done     => 0,
        err      => undef,
    }, 'EV::ClickHouse::Iterator';
    my $on_data = sub {
        push @{ $it->{batches} }, $_[0];
        EV::break;
    };
    my %s = $settings ? %$settings : ();
    $s{on_data} = $on_data;
    $self->query($sql, \%s, sub {