pg: Add query tracing & prepare/execute time measurements

What I'd really like, in addition to this, is a way to extract a query
from an $st object that can be run in the psql CLI. VNDB has a debugging
feature for that, but it's less trivial to make that work with binary
query parameters.

FU.xs

@@ -1,5 +1,6 @@
 #include <stdio.h>
 #include <errno.h>
+#include <time.h> /* struct timespec & clock_gettime() */
 #include <string.h> /* strerror() */
 #include <arpa/inet.h> /* inet_ntop(), inet_ntoa() */
 #include <sys/socket.h> /* fd passing */
@@ -174,6 +175,12 @@ void _debug_trace(fupg_conn *c, bool on)
     else PQuntrace(c->conn);
     ST(0) = c->self;
 
+void query_trace(fupg_conn *c, SV *cb)
+  CODE:
+    if (c->trace) SvREFCNT_dec(c->trace);
+    SvGETMAGIC(cb);
+    c->trace = SvOK(cb) ? SvREFCNT_inc(cb) : NULL;
+
 void status(fupg_conn *c)
   CODE:
     ST(0) = sv_2mortal(newSVpv(fupg_conn_status(c), 0));
@@ -273,16 +280,6 @@ void cache(fupg_st *x, ...)
     if (ix == 0 && x->prepared) fu_confess("Invalid attempt to change statement configuration after it has already been prepared or executed");
     FUPG_STFLAGS;
 
-void param_types(fupg_st *st)
-  CODE:
-    FUPG_ST_COOKIE;
-    ST(0) = fupg_st_param_types(aTHX_ st);
-
-void columns(fupg_st *st)
-  CODE:
-    FUPG_ST_COOKIE;
-    ST(0) = fupg_st_columns(aTHX_ st);
-
 void exec(fupg_st *st)
   CODE:
     FUPG_ST_COOKIE;
@@ -338,6 +335,48 @@ void kvh(fupg_st *st)
     FUPG_ST_COOKIE;
     ST(0) = fupg_st_kvh(aTHX_ st);
 
+void param_types(fupg_st *st)
+  CODE:
+    FUPG_ST_COOKIE;
+    ST(0) = fupg_st_param_types(aTHX_ st);
+
+void param_values(fupg_st *st);
+  CODE:
+    ST(0) = fupg_st_param_values(aTHX_ st);
+
+void columns(fupg_st *st)
+  CODE:
+    FUPG_ST_COOKIE;
+    ST(0) = fupg_st_columns(aTHX_ st);
+
+void nrows(fupg_st *st)
+  CODE:
+    ST(0) = st->result ? fupg_exec_result(aTHX_ st->result) : &PL_sv_undef;
+
+void query(fupg_st *st)
+  CODE:
+    ST(0) = newSVpvn_utf8(st->query, strlen(st->query), 1);
+
+void exec_time(fupg_st *st)
+  CODE:
+    ST(0) = st->exectime <= 0 ? &PL_sv_undef : sv_2mortal(newSVnv(st->exectime));
+
+void prepare_time(fupg_st *st)
+  CODE:
+    ST(0) = !st->prepared ? &PL_sv_undef : sv_2mortal(newSVnv(st->preptime));
+
+void get_cache(fupg_st *st)
+  CODE:
+    ST(0) = st->stflags & FUPG_CACHE ? &PL_sv_yes : &PL_sv_no;
+
+void get_text_params(fupg_st *st)
+  CODE:
+    ST(0) = st->stflags & FUPG_TEXT_PARAMS ? &PL_sv_yes : &PL_sv_no;
+
+void get_text_results(fupg_st *st)
+  CODE:
+    ST(0) = st->stflags & FUPG_TEXT_RESULTS ? &PL_sv_yes : &PL_sv_no;
+
 void DESTROY(fupg_st *st)
   CODE:
     fupg_st_destroy(aTHX_ st);

FU/Pg.pm

@@ -125,9 +125,32 @@ Prepared statements that still have an active C<$st> object are not counted
 towards this number. The cache works as an LRU: when it's full, the statement
 that hasn't been used for the longest time is reclaimed.
 
+=item $conn->query_trace($sub)
+
+Set a subroutine to be called on every query executed on this connection. The
+subroutine is given a statement object, refer to the C<$st> methods below for
+the fields that can be inspected. C<$sub> can be set to C<undef> to disable
+query tracing.
+
+It is important to not hold on to the given C<$st> any longer than strictly
+necessary, because the prepared statement is not closed or reclaimed while the
+object remains alive. If you need information to remain around for longer than
+the duration of the subroutine call, it's best to grab the relevant information
+from the C<$st> methods and save that for later.
+
+Also worth noting that the subroutine is called from the context of the code
+executing the query, but I<before> the query results have been returned.
+
+The subroutine is (currently) only called for queries executed through C<<
+$conn->exec >>, C<< $conn->q >>, C<< $conn->Q >> and their C<$txn> variants;
+internal queries performed by this module (such as for transaction management,
+querying type information, etc) do not trigger the callback. Statements that
+result in an error being thrown during or before execution are also not
+traceable this way. This behavior might change in the future.
+
 =item $conn->disconnect
 
-Close the connection. Any active transactions are rolled back and any further
+Close the connection. Any active transactions are rolled back and further
 attempts to use C<$conn> throw an error.
 
 =back
@@ -154,19 +177,19 @@ placeholders, as is common with L<DBI>, is not supported. An error is thrown
 when attempting to execute a query where the number of C<@params> does not
 match the number of placeholders in C<$sql>.
 
-Note that this method just creates a statement object, the given query is not
+Note that this method just creates a statement object, the query is not
 prepared or executed until the appropriate statement methods (see below) are
 used.
 
 =item $conn->Q(@args)
 
-Same as C<< $conn->q() >> but uses L<FU::SQL> to construct the SQL query and
+Same as C<< $conn->q() >> but uses L<FU::SQL> to construct the query and bind
-bind parameters.
+parameters.
 
 =back
 
 Statement objects returned by C<< $conn->q() >> support the following
-configuration parameters:
+configuration parameters, which can be set before the statement is executed:
 
 =over
 
@@ -190,37 +213,8 @@ Shorthand for setting C<text_params> and C<text_results> at the same time.
 
 =back
 
-Statement objects can be inspected with the following two methods:
+To execute the statement, call one (and exactly one) of the following methods,
-
+depending on how you'd like to obtain the results:
-=over
-
-=item $st->param_types
-
-Returns an arrayref of integers indicating the type (as I<oid>) of each
-parameter in the given C<$sql> string. Example:
-
-  my $oids = $conn->q('SELECT id FROM books WHERE id = $1 AND title = $2')->param_types;
-  # $oids = [23,25]
-
-  my $oids = $conn->q('SELECT id FROM books')->params;
-  # $oids = []
-
-=item $st->columns
-
-Returns an arrayref of hashrefs describing each column that the statement
-returns.
-
-  my $cols = $conn->q('SELECT id, title FROM books')->columns;
-  # $cols = [
-  #   { name => 'id', oid => 23 },
-  #   { name => 'title', oid => 25 },
-  # ]
-
-
-=back
-
-The statement can be executed with one of the following methods, depending on
-how you'd like to obtain the results:
 
 =over
 
@@ -333,11 +327,79 @@ columns are stored as hashref.
 =back
 
 The only time you actually need to assign a statement object to a variable is
-when you want to inspect C<param_types> or C<columns>, in all other cases you
+when you want to inspect the statement using one of the methods below, in all
-can chain the methods for more concise code. For example:
+other cases you can chain the methods for more concise code. For example:
 
   my $data = $conn->q('SELECT a, b FROM table')->cache(0)->text->alla;
 
+Statement objects can be inspected with the following methods (many of which
+only make sense after the query has been executed):
+
+=over
+
+=item $st->query
+
+Returns the SQL query that the statement was created with.
+
+=item $st->param_values
+
+Returns the provided bind parameters as an arrayref.
+
+=item $st->param_types
+
+Returns an arrayref of integers indicating the type (as I<oid>) of each
+parameter in the given C<$sql> string. Example:
+
+  my $oids = $conn->q('SELECT id FROM books WHERE id = $1 AND title = $2')->param_types;
+  # $oids = [23,25]
+
+  my $oids = $conn->q('SELECT id FROM books')->params;
+  # $oids = []
+
+This method can be called before the query has been executed, but will then
+trigger a prepare operation.
+
+=item $st->columns
+
+Returns an arrayref of hashrefs describing each column that the statement
+returns.
+
+  my $cols = $conn->q('SELECT id, title FROM books')->columns;
+  # $cols = [
+  #   { name => 'id', oid => 23 },
+  #   { name => 'title', oid => 25 },
+  # ]
+
+=item $st->nrows
+
+Returns the number of rows the query returned, may return C<undef> for
+C<exec()>-style queries.
+
+=item $st->exec_time
+
+Observed query execution time, in seconds. Includes network round-trip and
+fetching the full query results. Does not include conversion of the query
+results into Perl values.
+
+=item $st->prepare_time
+
+Observed query preparation time, in seconds, including network round-trip.
+Returns 0 if a cached prepared statement was used or C<undef> if the query was
+executed without a separate preparation phase (currently only happens with C<<
+$conn->exec() >>, but support for direct query execution may be added for other
+queries in the future as well).
+
+=item $st->get_cache
+
+=item $st->get_text_params
+
+=item $st->get_text_results
+
+Returns the respective configuration parameters.
+
+=back
+
+
 
 =head2 Transactions
 

c/common.c

@@ -170,3 +170,9 @@ static SV *fustr_done_(pTHX_ fustr *s) {
 #define fustr_writebeT(T, bits, s, in) fu_tobeT(T, bits, fustr_write_buf(s, bits>>3), in)
 #define fustr_writebeI(bits, s, in) fustr_writebeT(I##bits, bits, s, in)
 #define fustr_writebeU(bits, s, in) fustr_writebeT(U##bits, bits, s, in)
+
+
+/* Return the difference between two struct timespecs as fractional seconds. */
+static double fu_timediff(const struct timespec *a, const struct timespec *b) {
+    return ((double)(a->tv_sec - b->tv_sec)) + (double)(a->tv_nsec - b->tv_nsec) / 1000000000.0;
+}

c/pgconn.c

@@ -29,6 +29,7 @@ static void fupg_prep_destroy(fupg_prep *p) {
 typedef struct {
     SV *self;
     PGconn *conn;
+    SV *trace;
     UV prep_counter;
     UV cookie_counter;
     UV cookie; /* currently active transaction object; 0 = none active */
@@ -159,6 +160,7 @@ static SV *fupg_connect(pTHX_ const char *str) {
 
     fupg_conn *c = safemalloc(sizeof(fupg_conn));
     c->conn = conn;
+    c->trace = NULL;
     c->prep_counter = c->cookie_counter = c->cookie = 0;
     c->stflags = FUPG_CACHE;
     c->ntypes = 0;

c/pgst.c

@@ -12,6 +12,7 @@ typedef struct {
     int prepared;
     char name[32];
     fupg_prep *prep;
+    double preptime;
     PGresult *describe; /* shared with prep->describe if prep is set */
 
     /* Set during execute */
@@ -19,14 +20,36 @@ typedef struct {
     const char **param_values; /* Points into conn->buf or st->bind SVs, may be invalid after exec */
     int *param_lengths;
     int *param_formats;
+    double exectime;
     fupg_tio send;
     fupg_tio *recv;
     PGresult *result;
 } fupg_st;
 
 
+static void fupg_tracecb(pTHX_ fupg_st *st) {
+    if (!st->conn->trace) return;
+    dSP;
+
+    ENTER;
+    SAVETMPS;
+
+    PUSHMARK(SP);
+    mXPUSHs(sv_bless(newRV_inc(st->self), gv_stashpv("FU::Pg::st", GV_ADD)));
+    PUTBACK;
+
+    call_sv(st->conn->trace, G_DISCARD);
+    FREETMPS;
+    LEAVE;
+}
+
 static SV *fupg_exec(pTHX_ fupg_conn *c, const char *sql) {
+    struct timespec t_start;
+    clock_gettime(CLOCK_MONOTONIC, &t_start);
     PGresult *r = PQexec(c->conn, sql);
+    struct timespec t_end;
+    clock_gettime(CLOCK_MONOTONIC, &t_end);
+
     if (!r) fupg_conn_croak(c, "exec");
     switch (PQresultStatus(r)) {
         case PGRES_EMPTY_QUERY:
@@ -34,8 +57,21 @@ static SV *fupg_exec(pTHX_ fupg_conn *c, const char *sql) {
         case PGRES_TUPLES_OK: break;
         default: fupg_result_croak(r, "exec", sql);
     }
+
     SV *ret = fupg_exec_result(aTHX_ r);
-    PQclear(r);
+    if (c->trace) {
+        fupg_st *st = safecalloc(1, sizeof(*st));
+        st->conn = c;
+        SvREFCNT_inc(c->self);
+        st->query = savepv(sql);
+        st->stflags = c->stflags;
+        st->result = r;
+        st->exectime = fu_timediff(&t_end, &t_start);
+        fu_selfobj(st, "FU::Pg::st");
+        fupg_tracecb(aTHX_ st);
+    } else {
+        PQclear(r);
+    }
     return ret;
 }
 
@@ -87,6 +123,7 @@ static void fupg_st_destroy(pTHX_ fupg_st *st) {
 static void fupg_st_prepare(pTHX_ fupg_st *st) {
     if (st->describe) return;
     if (st->prepared) fu_confess("invalid attempt to re-prepare invalid statement");
+    if (st->result) fu_confess("invalid attempt to prepare already executed statement");
 
     if (st->stflags & FUPG_CACHE)
         st->prep = fupg_prepared_ref(aTHX_ st->conn, st->query);
@@ -101,6 +138,9 @@ static void fupg_st_prepare(pTHX_ fupg_st *st) {
     if (st->prep) st->prep->name = st->conn->prep_counter;
     snprintf(st->name, sizeof(st->name), "fupg%"UVuf, st->conn->prep_counter);
 
+    struct timespec t_start;
+    clock_gettime(CLOCK_MONOTONIC, &t_start);
+
     /* Send prepare + describe in a pipeline to avoid a double round-trip with the server */
     PQenterPipelineMode(st->conn->conn);
     PQsendPrepare(st->conn->conn, st->name, st->query, 0, NULL);
@@ -122,6 +162,10 @@ static void fupg_st_prepare(pTHX_ fupg_st *st) {
     PQclear(prep);
     st->prepared = 1;
 
+    struct timespec t_end;
+    clock_gettime(CLOCK_MONOTONIC, &t_end);
+    st->preptime = fu_timediff(&t_end, &t_start);
+
     if (!desc) {
         PQclear(sync);
         fupg_conn_croak(st->conn , "prepare");
@@ -140,6 +184,8 @@ static void fupg_st_prepare(pTHX_ fupg_st *st) {
 }
 
 static SV *fupg_st_param_types(pTHX_ fupg_st *st) {
+    if (st->result && !st->describe)
+        return sv_2mortal(newRV_noinc((SV *)newAV()));
     fupg_st_prepare(aTHX_ st);
     int i, nparams = PQnparams(st->describe);
     AV *av = nparams == 0 ? newAV() : newAV_alloc_x(nparams);
@@ -148,16 +194,28 @@ static SV *fupg_st_param_types(pTHX_ fupg_st *st) {
     return sv_2mortal(newRV_noinc((SV *)av));
 }
 
+static SV *fupg_st_param_values(pTHX_ fupg_st *st) {
+    int i;
+    AV *av = st->nbind == 0 ? newAV() : newAV_alloc_x(st->nbind);
+    for (i=0; i<st->nbind; i++)
+        av_push_simple(av, SvREFCNT_inc(st->bind[i]));
+    return sv_2mortal(newRV_noinc((SV *)av));
+}
+
 static SV *fupg_st_columns(pTHX_ fupg_st *st) {
-    fupg_st_prepare(aTHX_ st);
+    PGresult *r = st->result;
-    int i, nfields = PQnfields(st->describe);
+    if (!r) {
+        fupg_st_prepare(aTHX_ st);
+        r = st->describe;
+    }
+    int i, nfields = PQnfields(r);
     AV *av = nfields == 0 ? newAV() : newAV_alloc_x(nfields);
     for (i=0; i<nfields; i++) {
         HV *hv = newHV();
-        const char *name = PQfname(st->describe, i);
+        const char *name = PQfname(r, i);
         hv_stores(hv, "name", newSVpvn_utf8(name, strlen(name), 1));
-        hv_stores(hv, "oid", newSViv(PQftype(st->describe, i)));
+        hv_stores(hv, "oid", newSViv(PQftype(r, i)));
-        int tmod = PQfmod(st->describe, i);
+        int tmod = PQfmod(r, i);
         if (tmod >= 0) hv_stores(hv, "typemod", newSViv(tmod));
         av_push_simple(av, newRV_noinc((SV *)hv));
     }
@@ -229,6 +287,8 @@ static void fupg_st_execute(pTHX_ fupg_st *st) {
      * gather that just saves a bit of memory in exchange for more and smaller
      * malloc()/free()'s. Performance-wise, it probably won't be much of an
      * improvement */
+    struct timespec t_start;
+    clock_gettime(CLOCK_MONOTONIC, &t_start);
     PGresult *r = PQexecPrepared(st->conn->conn,
             st->name,
             st->nbind,
@@ -236,6 +296,10 @@ static void fupg_st_execute(pTHX_ fupg_st *st) {
             st->param_lengths,
             st->param_formats,
             st->stflags & FUPG_TEXT_RESULTS ? 0 : 1);
+    struct timespec t_end;
+    clock_gettime(CLOCK_MONOTONIC, &t_end);
+    st->exectime = fu_timediff(&t_end, &t_start);
+
     if (!r) fupg_conn_croak(st->conn , "exec");
     switch (PQresultStatus(r)) {
         case PGRES_COMMAND_OK:
@@ -251,6 +315,8 @@ static void fupg_st_execute(pTHX_ fupg_st *st) {
         fupg_tio_setup(aTHX_ st->conn, st->recv + i,
                 FUPGT_RECV | (st->stflags & FUPG_TEXT_RESULTS ? FUPGT_TEXT : 0),
                 PQftype(st->result, i), &refresh_done);
+
+    fupg_tracecb(aTHX_ st);
 }
 
 static SV *fupg_st_getval(pTHX_ fupg_st *st, int row, int col) {

t/pgconnect.t

@@ -401,6 +401,46 @@ subtest 'Prepared statement cache', sub {
     ok !defined numexec('SELECT 2');
 };
 
+
+subtest 'Tracing', sub {
+    my @log;
+    $conn->query_trace(sub($st) { push @log, $st });
+
+    is_deeply $conn->q('SELECT 1 AS a, $1 AS b', 123)->rowa, [ 1, 123 ];
+    is scalar @log, 1;
+    my $st = shift @log;
+    is ref $st, 'FU::Pg::st';
+    is_deeply $st->param_types, [ 25 ];
+    is_deeply $st->param_values, [ 123 ];
+    is_deeply $st->columns, [{ name => 'a', oid => 23 }, { name => 'b', oid => 25 }];
+    is $st->nrows, 1;
+    is $st->query, 'SELECT 1 AS a, $1 AS b';
+    ok $st->exec_time > 0 && $st->exec_time < 1;
+    ok $st->prepare_time > 0 && $st->prepare_time < 1;
+    ok !$st->get_cache;
+    ok $st->text_params;
+    ok $st->text_results;
+
+    $conn->exec('SET client_encoding=UTF8');
+    is scalar @log, 1;
+    $st = shift @log;
+    is ref $st, 'FU::Pg::st';
+    is_deeply $st->param_types, [];
+    is_deeply $st->param_values, [];
+    is_deeply $st->columns, [];
+    ok !defined $st->nrows;
+    is $st->query, 'SET client_encoding=UTF8';
+    ok $st->exec_time > 0 && $st->exec_time < 1;
+    ok !defined $st->prepare_time;
+    ok !$st->get_cache;
+    ok $st->text_params;
+    ok $st->text_results;
+
+    $conn->query_trace(undef);
+    $conn->exec('SELECT 1');
+    is scalar @log, 0;
+};
+
 {
     my $st = $conn->q("SELECT 1");
     undef $conn; # statement keeps the connection alive