view release on metacpan or  search on metacpan

c/pgst.c  view on Meta::CPAN

    }
    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(r, i);
        hv_stores(hv, "name", newSVpvn_utf8(name, strlen(name), 1));
        hv_stores(hv, "oid", newSViv(PQftype(r, i)));
        int tmod = PQfmod(r, i);
        if (tmod >= 0) hv_stores(hv, "typemod", newSViv(tmod));
        av_push_simple(av, newRV_noinc((SV *)hv));
    }
    return sv_2mortal(newRV_noinc((SV *)av));
}

static void fupg_params_setup(pTHX_ fupg_st *st, int *refresh_done) {
    int i;
    st->param_values = safecalloc(st->nbind, sizeof(*st->param_values));
    if (st->stflags & FUPG_TEXT_PARAMS) {
        for (i=0; i<st->nbind; i++)
            st->param_values[i] = !SvOK(st->bind[i]) ? NULL : SvPVutf8_nolen(st->bind[i]);
        return;
    }

    fustr *buf = &st->conn->buf;
    buf->cur = fustr_start(buf);
    st->param_lengths = safecalloc(st->nbind, sizeof(*st->param_lengths));
    st->param_formats = safecalloc(st->nbind, sizeof(*st->param_formats));
    size_t off = 0;
    for (i=0; i<st->nbind; i++) {
        if (!SvOK(st->bind[i])) {
            st->param_values[i] = NULL;
            continue;
        }
        fupg_tio_setup(aTHX_ st->conn, &st->send,
                FUPGT_SEND | (st->stflags & FUPG_TEXT_PARAMS ? FUPGT_TEXT : 0),
                PQparamtype(st->describe, i), refresh_done);
        off = fustr_len(buf);
        st->send.send(aTHX_ &st->send, st->bind[i], buf);
        fupg_tio_free(&st->send);
        memset(&st->send, 0, sizeof(st->send));

        st->param_lengths[i] = fustr_len(buf) - off;
        st->param_formats[i] = 1;
        st->param_values[i] = "";
        /* Don't write param_values here, the buffer may be invalidated when writing the next param */
    }
    off = 0;
    buf->cur = fustr_start(buf);
    for (i=0; i<st->nbind; i++) {
        if (st->param_values[i]) {
            st->param_values[i] = buf->cur + off;
            off += st->param_lengths[i];
        }
    }
}

static void fupg_st_execute(pTHX_ fupg_st *st) {
    /* Disallow fetching the results more than once. I don't see a reason why
     * someone would need that and disallowing it leaves room for fetching the
     * results in a streaming fashion without breaking API compat. */
    if (st->result) fu_confess("Invalid attempt to execute statement multiple times");

    /* Whether we can do a direct call or need to prepare first */
    int direct = !st->describe && (st->nbind == 0 || st->stflags & FUPG_TEXT_PARAMS) && !(st->stflags & FUPG_CACHE);
    if (!direct) {
        fupg_st_prepare(aTHX_ st);
        if (PQnparams(st->describe) != st->nbind)
            fu_confess("Statement expects %d bind parameters but %d were given", PQnparams(st->describe), st->nbind);
    }
    int refresh_done = 0;
    fupg_params_setup(aTHX_ st, &refresh_done);

    /* I'm not super fond of this approach. Storing the full query results in a
     * PGresult involves unnecessary parsing, memory allocation and copying.
     * The wire protocol is sufficiently simple that I could parse the query
     * results directly from the network buffers without much additional code,
     * and that would be much more efficient. Alas, libpq doesn't let me do
     * that.
     * There is the option of fetching results in chunked mode, but from what I
     * 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 = direct ? PQexecParams(st->conn->conn,
            st->query, st->nbind, NULL,
            (const char * const *)st->param_values,
            st->param_lengths, st->param_formats,
            st->stflags & FUPG_TEXT_RESULTS ? 0 : 1
        ) : PQexecPrepared(st->conn->conn,
            st->name, st->nbind,
            (const char * const *)st->param_values,
            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:
        case PGRES_TUPLES_OK: break;
        default: fupg_result_croak(r, "exec", st->query);
    }
    st->result = r;

    st->nfields = PQnfields(r);
    st->recv = safecalloc(st->nfields, sizeof(*st->recv));
    int i;
    for (i=0; i<st->nfields; i++)
        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) {
    PGresult *r = st->result;