EV-Pg

 view release on metacpan or  search on metacpan

Pg.xs  view on Meta::CPAN

        else if (st == PGRES_TUPLES_OK || st == PGRES_SINGLE_TUPLE
#ifdef LIBPQ_HAS_CHUNK_MODE
                 || st == PGRES_TUPLES_CHUNK
#endif
                ) {
            int nrows = PQntuples(res);
            int ncols = PQnfields(res);
            AV *rows = newAV();
            int r, c;
            /* Defer metadata building until result_meta is called.
             * In streaming mode, capture only the first result per query
             * (meta_fresh is cleared by advance_cb_queue). */
            if (st == PGRES_TUPLES_OK || !self->meta_fresh) {
                RELEASE_LAST_HV(self->last_result_meta);
                self->meta_fresh = 1;
            }
            if (nrows > 0) av_extend(rows, nrows - 1);
            for (r = 0; r < nrows; r++) {
                AV *row = newAV();
                if (ncols > 0) av_extend(row, ncols - 1);
                for (c = 0; c < ncols; c++) {

README  view on Meta::CPAN

    companion to "send_pipeline_sync". Alias: "flush".

  pipeline_status
        my $st = $pg->pipeline_status;

    One of "PQ_PIPELINE_OFF", "PQ_PIPELINE_ON", or "PQ_PIPELINE_ABORTED".

COPY METHODS
    A "COPY" command runs in two phases: the query callback first fires with
    a string tag ("COPY_IN" / "COPY_OUT" / "COPY_BOTH") to signal that
    streaming has started, then fires a second time with the final command
    result (or error) when the stream ends. See eg/copy_in.pl and
    eg/copy_out.pl.

  put_copy_data
        my $rc = $pg->put_copy_data($data);

    Sends a chunk during COPY IN. Returns 1 on success (data buffered or
    flushed), 0 if the send buffer is full (wait for writability via
    "on_drain", then retry), or -1 on error.

eg/copy_out.pl  view on Meta::CPAN

#!/usr/bin/env perl
use strict;
use warnings;
use EV;
use EV::Pg;

# Demonstrates the multi-phase COPY OUT callback protocol:
#   1. callback fires with ("COPY_OUT") -- streaming has begun
#   2. caller drains rows by looping get_copy_data until it returns -1
#   3. callback fires AGAIN with ($cmd_tuples) on completion

my $conninfo = shift || $ENV{TEST_PG_CONNINFO} || 'dbname=postgres';

my $pg; $pg = EV::Pg->new(
    conninfo => $conninfo,
    on_error => sub { die "connection error: $_[0]\n" },
    on_connect => sub {
        $pg->query("create temp table nums (n int)", sub {

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

    my $st = $pg->pipeline_status;

One of C<PQ_PIPELINE_OFF>, C<PQ_PIPELINE_ON>, or
C<PQ_PIPELINE_ABORTED>.

=head1 COPY METHODS

A C<COPY> command runs in two phases: the query callback first fires
with a string tag (C<"COPY_IN"> / C<"COPY_OUT"> / C<"COPY_BOTH">) to
signal that streaming has started, then fires a second time with the
final command result (or error) when the stream ends.  See
F<eg/copy_in.pl> and F<eg/copy_out.pl>.

=head2 put_copy_data

    my $rc = $pg->put_copy_data($data);

Sends a chunk during COPY IN.  Returns 1 on success (data buffered or
flushed), 0 if the send buffer is full (wait for writability via
C<on_drain>, then retry), or -1 on error.

t/08_features.t  view on Meta::CPAN

        },
        on_error => sub { diag "Error: $_[0]"; EV::break },
    );
    my $t = EV::timer(5, 0, sub { EV::break });
    EV::run;
    is(scalar @rows, 1, 'single_row reset: got only 1 row before abort');
    ok($reconnected, 'single_row reset: reconnected');
    ok($followup_ok, 'single_row reset: post-reconnect query succeeded');
}

# single row mode: DESTROY during single-row streaming
{
    my @rows;
    my $destroyed = 0;
    my $pg;
    $pg = EV::Pg->new(
        conninfo => $conninfo,
        on_connect => sub {
            $pg->query("select generate_series(1,100) as n", sub {
                my ($data, $err) = @_;
                if (ref $data eq 'ARRAY' && @$data > 0) {

t/08_features.t  view on Meta::CPAN

});

# result_meta during single-row mode
with_pg(cb => sub {
    my ($pg) = @_;
    my $meta_seen = 0;
    $pg->query("select 42 as val, 'x' as tag", sub {
        my ($data, $err) = @_;
        if (ref $data eq 'ARRAY' && @$data > 0) {
            my $m = $pg->result_meta;
            is(ref $m, 'HASH', 'result_meta single_row: hashref during streaming');
            is($m->{nfields}, 2, 'result_meta single_row: nfields = 2');
            is($m->{fields}[0]{name}, 'val', 'result_meta single_row: field 0 name');
            $meta_seen = 1;
            return;
        }
        ok($meta_seen, 'result_meta single_row: meta was available during streaming');
        EV::break;
    });
    $pg->set_single_row_mode;
});

# --- set_chunked_rows_mode (libpq >= 17) ---
SKIP: {
    skip 'requires libpq >= 17', 3 unless EV::Pg->lib_version >= 170000;
    with_pg(cb => sub {
        my ($pg) = @_;