vndb/fu
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
fu/c/pgconn.c
Yorhel d5f593387a Abstract and cleanup ugly byte swapping code 
These macros don't assume alignment and may be somewhat inefficient with
all that copying. I'm hoping GCC is able to optimize that crap somewhat.

Also the pg type receive functions can not, in fact, assume that their
input buffers are properly aligned. That won't necessarily be the case
for array elements.
2025-02-10 07:35:34 +01:00

666 lines
24 KiB
C
Raw Blame History

#define FUPG_CACHE 1
#define FUPG_TEXT_PARAMS 2
#define FUPG_TEXT_RESULTS 4
#define FUPG_TEXT (FUPG_TEXT_PARAMS|FUPG_TEXT_RESULTS)
typedef struct {
SV *self;
PGconn *conn;
UV prep_counter;
UV cookie_counter;
UV cookie; /* currently active transaction object; 0 = none active */
int stflags;
int ntypes;
fupg_type *types;
fustr buf; /* Scratch space for query params */
} fupg_conn;
typedef struct fupg_txn fupg_txn;
struct fupg_txn {
SV *self;
fupg_txn *parent;
fupg_conn *conn;
UV cookie; /* 0 means done */
int stflags;
char rollback_cmd[64];
};
typedef struct {
/* Set on creation */
SV *self; /* (unused, but whatever) */
fupg_conn *conn; /* has a refcnt on conn->self */
UV cookie;
char *query;
SV **bind;
int nbind;
int stflags;
/* Set during prepare */
int prepared;
char name[32];
PGresult *describe;
/* Set during execute */
int nfields;
const char **param_values; /* Points into conn->buf or st->bind SVs, may be invalid after exec */
int *param_lengths;
int *param_formats;
fupg_recv *recv;
PGresult *result;
} fupg_st;
/* Utilities */
static SV *fupg_conn_errsv(PGconn *conn, const char *action) {
dTHX;
HV *hv = newHV();
hv_stores(hv, "action", newSVpv(action, 0));
hv_stores(hv, "severity", newSVpvs("FATAL")); /* Connection-related errors are always fatal */
hv_stores(hv, "message", newSVpv(PQerrorMessage(conn), 0));
return fu_croak_hv(hv, "FU::PG::error", "FATAL: %s", PQerrorMessage(conn));
}
__attribute__((noreturn))
static void fupg_conn_croak(fupg_conn *c, const char *action) {
dTHX;
croak_sv(fupg_conn_errsv(c->conn, action));
}
/* Takes ownership of the PGresult and croaks. */
__attribute__((noreturn))
static void fupg_result_croak(PGresult *r, const char *action, const char *query) {
dTHX;
HV *hv = newHV();
hv_stores(hv, "action", newSVpv(action, 0));
char *s = PQresultErrorField(r, PG_DIAG_SEVERITY_NONLOCALIZED);
hv_stores(hv, "severity", newSVpv(s ? s : "FATAL", 0));
if (query) hv_stores(hv, "query", newSVpv(query, 0));
/* If the PGresult is not an error, assume it's an unexpected resultStatus */
s = PQresultErrorField(r, PG_DIAG_MESSAGE_PRIMARY);
hv_stores(hv, "message", s ? newSVpv(s, 0) : newSVpvf("unexpected status code '%s'", PQresStatus(PQresultStatus(r))));
/* I like the verbose error messages. Doesn't include anything that's not
* also fetched below, but saves me from having to do the formatting
* manually. */
char *verbose = NULL;
if (s) {
verbose = PQresultVerboseErrorMessage(r, PQERRORS_VERBOSE, PQSHOW_CONTEXT_ERRORS);
if (s) {
hv_stores(hv, "verbose_message", newSVpv(verbose, 0));
PQfreemem(verbose);
}
}
if ((s = PQresultErrorField(r, PG_DIAG_MESSAGE_DETAIL))) hv_stores(hv, "detail", newSVpv(s, 0));
if ((s = PQresultErrorField(r, PG_DIAG_MESSAGE_HINT))) hv_stores(hv, "hint", newSVpv(s, 0));
if ((s = PQresultErrorField(r, PG_DIAG_STATEMENT_POSITION))) hv_stores(hv, "statement_position", newSVpv(s, 0));
if ((s = PQresultErrorField(r, PG_DIAG_INTERNAL_POSITION))) hv_stores(hv, "internal_position", newSVpv(s, 0));
if ((s = PQresultErrorField(r, PG_DIAG_INTERNAL_QUERY))) hv_stores(hv, "internal_query", newSVpv(s, 0));
if ((s = PQresultErrorField(r, PG_DIAG_CONTEXT))) hv_stores(hv, "context", newSVpv(s, 0));
if ((s = PQresultErrorField(r, PG_DIAG_SCHEMA_NAME))) hv_stores(hv, "schema_name", newSVpv(s, 0));
if ((s = PQresultErrorField(r, PG_DIAG_TABLE_NAME))) hv_stores(hv, "table_name", newSVpv(s, 0));
if ((s = PQresultErrorField(r, PG_DIAG_COLUMN_NAME))) hv_stores(hv, "column_name", newSVpv(s, 0));
if ((s = PQresultErrorField(r, PG_DIAG_DATATYPE_NAME))) hv_stores(hv, "datatype_name", newSVpv(s, 0));
if ((s = PQresultErrorField(r, PG_DIAG_CONSTRAINT_NAME))) hv_stores(hv, "constraint_name", newSVpv(s, 0));
if ((s = PQresultErrorField(r, PG_DIAG_SOURCE_FILE))) hv_stores(hv, "source_file", newSVpv(s, 0));
if ((s = PQresultErrorField(r, PG_DIAG_SOURCE_LINE))) hv_stores(hv, "source_line", newSVpv(s, 0));
if ((s = PQresultErrorField(r, PG_DIAG_SOURCE_FUNCTION))) hv_stores(hv, "source_function", newSVpv(s, 0));
PQclear(r);
croak_sv(verbose
? fu_croak_hv(hv, "FU::PG::error", "%s", SvPV_nolen(*hv_fetchs(hv, "verbose_message", 0)))
: fu_croak_hv(hv, "FU::PG::error", "%s: %s",
SvPV_nolen(*hv_fetchs(hv, "severity", 0)),
SvPV_nolen(*hv_fetchs(hv, "message", 0))
)
);
}
static SV *fupg_exec_result(pTHX_ PGresult *r) {
SV *ret = &PL_sv_undef;
char *tup = PQcmdTuples(r);
if (tup && *tup) {
ret = sv_2mortal(newSVpv(tup, 0));
SvIV(ret);
SvIOK_only(ret);
}
return ret;
}
static void fupg_exec_ok(pTHX_ fupg_conn *c, const char *sql) {
PGresult *r = PQexec(c->conn, sql);
if (!r) fupg_conn_croak(c, "exec");
if (PQresultStatus(r) != PGRES_COMMAND_OK) fupg_result_croak(r, "exec", sql);
PQclear(r);
}
/* Connection & transaction handling */
static SV *fupg_connect(pTHX_ const char *str) {
PGconn *conn = PQconnectdb(str);
if (PQstatus(conn) != CONNECTION_OK) {
SV *sv = fupg_conn_errsv(conn, "connect");
PQfinish(conn);
croak_sv(sv);
}
fupg_conn *c = safemalloc(sizeof(fupg_conn));
c->conn = conn;
c->prep_counter = c->cookie_counter = c->cookie = 0;
c->stflags = 0;
c->ntypes = 0;
c->types = NULL;
fustr_init(&c->buf, NULL, SIZE_MAX);
return fu_selfobj(c, "FU::PG::conn");
}
static const char *fupg_conn_status(fupg_conn *c) {
if (PQstatus(c->conn) == CONNECTION_BAD) return "bad";
switch (PQtransactionStatus(c->conn)) {
case PQTRANS_IDLE: return c->cookie ? "txn_done" : "idle";
case PQTRANS_ACTIVE: return "active"; /* can't happen, we don't do async */
case PQTRANS_INTRANS: return "txn_idle";
case PQTRANS_INERROR: return "txn_error";
default: return "unknown";
}
}
static void fupg_conn_disconnect(fupg_conn *c) {
PQfinish(c->conn);
c->conn = NULL;
}
static void fupg_conn_destroy(fupg_conn *c) {
PQfinish(c->conn);
if (c->buf.sv) SvREFCNT_dec(c->buf.sv);
safefree(c->types);
safefree(c);
}
static SV *fupg_conn_txn(pTHX_ fupg_conn *c) {
fupg_exec_ok(c, "BEGIN");
fupg_txn *t = safecalloc(1, sizeof(fupg_txn));
t->conn = c;
t->cookie = c->cookie = ++c->cookie_counter;
t->stflags = c->stflags;
strcpy(t->rollback_cmd, "ROLLBACK");
SvREFCNT_inc(c->self);
return fu_selfobj(t, "FU::PG::txn");
}
static SV *fupg_txn_txn(pTHX_ fupg_txn *t) {
char cmd[64];
UV cookie = ++t->conn->cookie_counter;
snprintf(cmd, sizeof(cmd), "SAVEPOINT fupg_%"UVuf, cookie);
fupg_exec_ok(t->conn, cmd);
fupg_txn *n = safecalloc(1, sizeof(fupg_txn));
n->conn = t->conn;
n->parent = t;
n->cookie = t->conn->cookie = cookie;
n->stflags = t->stflags;
snprintf(n->rollback_cmd, sizeof(n->rollback_cmd), "ROLLBACK TO SAVEPOINT fupg_%"UVuf, cookie);
SvREFCNT_inc(t->self);
return fu_selfobj(n, "FU::PG::txn");
}
static const char *fupg_txn_status(fupg_txn *t) {
if (PQstatus(t->conn->conn) == CONNECTION_BAD) return "bad";
if (!t->cookie) return "done";
int a = t->cookie == t->conn->cookie;
switch (PQtransactionStatus(t->conn->conn)) {
case PQTRANS_IDLE: return "done";
case PQTRANS_ACTIVE: return "active";
case PQTRANS_INTRANS: return a ? "idle" : "txn_idle";
case PQTRANS_INERROR: return a ? "error" : "txn_error";
default: return "unknown";
}
}
static void fupg_txn_commit(pTHX_ fupg_txn *t) {
char cmd[64];
if (t->parent) snprintf(cmd, sizeof(cmd), "RELEASE SAVEPOINT fupg_%"UVuf, t->cookie);
else strcpy(cmd, "COMMIT");
t->cookie = 0;
fupg_exec_ok(t->conn, cmd);
}
static void fupg_txn_rollback(pTHX_ fupg_txn *t) {
t->cookie = 0;
fupg_exec_ok(t->conn, t->rollback_cmd);
}
static void fupg_txn_destroy(pTHX_ fupg_txn *t) {
if (t->cookie) {
PGresult *r = PQexec(t->conn->conn, t->rollback_cmd);
/* Can't really throw an error in DESTROY. If a rollback command fails,
* we're sufficiently screwed that the only sensible recourse is to
* disconnect and let any further operations throw an error. */
if (!r || PQresultStatus(r) != PGRES_COMMAND_OK)
fupg_conn_disconnect(t->conn);
PQclear(r);
}
if (t->parent) {
t->conn->cookie = t->parent->cookie;
SvREFCNT_dec(t->parent->self);
} else {
t->conn->cookie = 0;
SvREFCNT_dec(t->conn->self);
}
safefree(t);
}
/* XXX: It feels a bit wasteful to load *all* types; even on an empty database
* that's ~55k of data, but it's easier and (potentially) faster than fetching
* each type seperately as we encounter them.
* Perhaps an easier optimization is to filter out all table-based composites
* and their array types by default, I've never seen anyone use those types for
* I/O and that would shrink the data by nearly a factor 5.
*/
static void fupg_refresh_types(pTHX_ fupg_conn *c) {
safefree(c->types);
c->types = 0;
c->ntypes = 0;
const char *sql =
"SELECT oid, typname, typtype"
", CASE WHEN typcategory = 'A' THEN typelem ELSE 0 END"
" FROM pg_type"
" ORDER BY oid";
PGresult *r = PQexecParams(c->conn, sql, 0, NULL, NULL, NULL, NULL, 1);
if (!r) fupg_conn_croak(c, "exec");
if (PQresultStatus(r) != PGRES_TUPLES_OK) fupg_result_croak(r, "exec", sql);
c->ntypes = PQntuples(r);
c->types = safecalloc(c->ntypes, sizeof(*c->types));
int i;
for (i=0; i<c->ntypes; i++) {
fupg_type *t = c->types + i;
t->oid = fu_frombeU(32, PQgetvalue(r, i, 0));
snprintf(t->name, sizeof(t->name), "%s", PQgetvalue(r, i, 1));
char typ = *PQgetvalue(r, i, 2);
t->elemoid = fu_frombeU(32, PQgetvalue(r, i, 3));
if (t->elemoid) {
/* array */
t->send = fupg_send_array;
t->recv = fupg_recv_array;
} else if (typ == 'e') {
/* enum, can use text send/recv */
t->send = fupg_send_text;
t->recv = fupg_recv_text;
} else {
/* TODO: records, domain types, (multi)ranges, custom overrides, by-name lookup for dynamic-oid types */
const fupg_type *builtin = fupg_builtin_byoid(t->oid);
if (builtin) {
t->send = builtin->send;
t->recv = builtin->recv;
}
}
}
PQclear(r);
}
static const fupg_type *fupg_lookup_type(pTHX_ fupg_conn *c, int *refresh_done, Oid oid) {
const fupg_type *t = NULL;
if (c->types && (t = fupg_type_byoid(c->types, c->ntypes, oid))) return t;
if ((t = fupg_builtin_byoid(oid))) return t;
if (*refresh_done) return NULL;
*refresh_done = 1;
fupg_refresh_types(c);
return fupg_type_byoid(c->types, c->ntypes, oid);
}
/* Querying */
static SV *fupg_exec(pTHX_ fupg_conn *c, const char *sql) {
PGresult *r = PQexec(c->conn, sql);
if (!r) fupg_conn_croak(c, "exec");
switch (PQresultStatus(r)) {
case PGRES_EMPTY_QUERY:
case PGRES_COMMAND_OK:
case PGRES_TUPLES_OK: break;
default: fupg_result_croak(r, "exec", sql);
}
SV *ret = fupg_exec_result(r);
PQclear(r);
return ret;
}
static SV *fupg_q(pTHX_ fupg_conn *c, int stflags, const char *query, I32 ax, I32 argc) {
fupg_st *st = safecalloc(1, sizeof(fupg_st));
st->conn = c;
st->cookie = c->cookie;
st->stflags = stflags;
SvREFCNT_inc(c->self);
st->query = savepv(query);
if (argc > 2) {
st->bind = safemalloc((argc-2) * sizeof(SV *));
I32 i;
for (i=2; i < argc; i++) {
SvGETMAGIC(ST(i));
st->bind[st->nbind] = SvREFCNT_inc(ST(i));
st->nbind++;
}
}
return fu_selfobj(st, "FU::PG::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");
/* TODO: This is where we check for any cached prepared statements */
snprintf(st->name, sizeof(st->name), "fupg%"UVuf, ++st->conn->prep_counter);
/* 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);
PQsendDescribePrepared(st->conn->conn, st->name);
PQpipelineSync(st->conn->conn);
PGresult *prep = PQgetResult(st->conn->conn); PQgetResult(st->conn->conn); /* NULL */
PGresult *desc = PQgetResult(st->conn->conn); PQgetResult(st->conn->conn); /* NULL */
PGresult *sync = PQgetResult(st->conn->conn);
PQexitPipelineMode(st->conn->conn);
if (!prep) {
PQclear(desc); PQclear(sync);
fupg_conn_croak(st->conn , "prepare");
}
if (PQresultStatus(prep) != PGRES_COMMAND_OK) {
PQclear(desc); PQclear(sync);
fupg_result_croak(prep, "prepare", st->query);
}
PQclear(prep);
st->prepared = 1;
if (!desc) {
PQclear(sync);
fupg_conn_croak(st->conn , "prepare");
}
if (PQresultStatus(desc) != PGRES_COMMAND_OK) {
PQclear(sync);
fupg_result_croak(desc, "prepare", st->query);
}
st->describe = desc;
if (!sync) fupg_conn_croak(st->conn , "prepare");
if (PQresultStatus(sync) != PGRES_PIPELINE_SYNC)
fupg_result_croak(sync, "prepare", st->query);
PQclear(sync);
}
static SV *fupg_st_params(pTHX_ fupg_st *st) {
fupg_st_prepare(aTHX_ st);
int i, nparams = PQnparams(st->describe);
AV *av = newAV_alloc_x(nparams);
for (i=0; i<nparams; i++) {
HV *hv = newHV();
hv_stores(hv, "oid", newSViv(PQparamtype(st->describe, i)));
av_push_simple(av, newRV_noinc((SV *)hv));
}
return sv_2mortal(newRV_noinc((SV *)av));
}
static SV *fupg_st_columns(pTHX_ fupg_st *st) {
fupg_st_prepare(aTHX_ st);
int i, ncols = PQnfields(st->describe);
AV *av = newAV_alloc_x(ncols);
for (i=0; i<ncols; i++) {
HV *hv = newHV();
const char *name = PQfname(st->describe, i);
hv_stores(hv, "name", newSVpvn_utf8(name, strlen(name), 1));
hv_stores(hv, "oid", newSViv(PQftype(st->describe, i)));
int tmod = PQfmod(st->describe, 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_st_check_dupcols(pTHX_ PGresult *r) {
HV *hv = newHV();
int i, nfields = PQnfields(r);
for (i=0; i<nfields; i++) {
const char *key = PQfname(r, i);
int len = -strlen(key);
if (hv_exists(hv, key, len)) {
SvREFCNT_dec((SV *)hv);
fu_confess("Query returns multiple columns with the same name ('%s')", key);
}
hv_store(hv, key, len, &PL_sv_yes, 0);
}
SvREFCNT_dec((SV *)hv);
}
static void fupg_params_send(pTHX_ fupg_st *st, Oid oid, SV *val, fustr *out, int *refresh_done) {
fupg_send send, elem;
const fupg_type *t = fupg_lookup_type(aTHX_ st->conn, refresh_done, oid);
if (!t) fu_confess("No type found with oid %u", oid);
if (!t->send) fu_confess("Unable to use type '%s' (oid %u) as bind parameter", t->name, t->oid);
send.oid = oid;
send.name = t->name;
send.fn = t->send;
if (t->send == fupg_send_array) {
if (!t->elemoid) fu_confess("Type '%s' (oid %u) is marked as an array type, but element type is unknown", t->name, t->oid);
const fupg_type *e = fupg_lookup_type(aTHX_ st->conn, refresh_done, t->elemoid);
if (!e) fu_confess("No type found with oid %u", t->elemoid);
send.arrayelem = &elem;
elem.oid = e->oid;
elem.name = e->name;
elem.fn = e->send;
assert(e->send != fupg_send_array); /* TODO: might as well fix this, we'll need recursion for record types anyway */
}
send.fn(aTHX_ &send, val, out);
}
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;
}
off = fustr_len(buf);
fupg_params_send(aTHX_ st, PQparamtype(st->describe, i), st->bind[i], buf, refresh_done);
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_recv_setup(pTHX_ fupg_st *st, fupg_recv *r, Oid oid, int *refresh_done) {
r->oid = oid;
if (st->stflags & FUPG_TEXT_RESULTS) {
r->name = "{textfmt}";
r->fn = fupg_recv_text;
return;
}
const fupg_type *t = fupg_lookup_type(aTHX_ st->conn, refresh_done, r->oid);
if (!t) fu_confess("No type found with oid %u", r->oid);
if (!t->recv) fu_confess("Unable to receive data of type '%s' (oid %u)", t->name, t->oid);
r->name = t->name;
r->fn = t->recv;
if (r->fn == fupg_recv_array) {
if (!t->elemoid) fu_confess("Type '%s' (oid %u) is marked as an array type, but element type is unknown", t->name, t->oid);
r->arrayelem = safecalloc(1, sizeof(*r->arrayelem));
fupg_recv_setup(aTHX_ st, r->arrayelem, t->elemoid, refresh_done);
}
}
static void fupg_recv_free(fupg_recv *r) {
if (!r) return;
if (r->fn == fupg_recv_array) {
fupg_recv_free(r->arrayelem);
safefree(r->arrayelem);
}
}
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");
/* TODO: prepare can be skipped when prepared statement caching is disabled and (text-format queries or no bind params) */
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 */
PGresult *r = 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);
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_recv_setup(aTHX_ st, st->recv + i, PQftype(st->result, i), &refresh_done);
}
static SV *fupg_st_getval(pTHX_ fupg_st *st, int row, int col) {
PGresult *r = st->result;
if (PQgetisnull(r, row, col)) return newSV(0);
const fupg_recv *ctx = st->recv+col;
return ctx->fn(aTHX_ ctx, PQgetvalue(r, row, col), PQgetlength(r, row, col));
}
static SV *fupg_st_exec(pTHX_ fupg_st *st) {
fupg_st_execute(aTHX_ st);
return fupg_exec_result(st->result);
}
static SV *fupg_st_val(pTHX_ fupg_st *st) {
fupg_st_prepare(aTHX_ st);
if (PQnfields(st->describe) > 1) fu_confess("Invalid use of $st->val() on query returning more than one column");
if (PQnfields(st->describe) == 0) fu_confess("Invalid use of $st->val() on query returning no data");
fupg_st_execute(aTHX_ st);
if (PQntuples(st->result) > 1) fu_confess("Invalid use of $st->val() on query returning more than one row");
SV *sv = PQntuples(st->result) == 0 ? newSV(0) : fupg_st_getval(aTHX_ st, 0, 0);
return sv_2mortal(sv);
}
static I32 fupg_st_rowl(pTHX_ fupg_st *st, I32 ax) {
dSP;
fupg_st_execute(aTHX_ st);
if (PQntuples(st->result) == 0) fu_confess("Invalid use of $st->rowl() on query returning zero rows");
if (PQntuples(st->result) > 1) fu_confess("Invalid use of $st->rowl() on query returning more than one row");
if (GIMME_V != G_LIST) {
ST(0) = sv_2mortal(newSViv(st->nfields));
return 1;
}
(void)POPs;
EXTEND(SP, st->nfields);
int i;
for (i=0; i<st->nfields; i++) mPUSHs(fupg_st_getval(aTHX_ st, 0, i));
return st->nfields;
}
static SV *fupg_st_rowa(pTHX_ fupg_st *st) {
fupg_st_execute(aTHX_ st);
if (PQntuples(st->result) == 0) fu_confess("Invalid use of $st->rowl() on query returning zero rows");
if (PQntuples(st->result) > 1) fu_confess("Invalid use of $st->rowl() on query returning more than one row");
AV *av = newAV_alloc_x(st->nfields);
SV *sv = sv_2mortal(newRV_noinc((SV *)av));
int i;
for (i=0; i<st->nfields; i++) av_push_simple(av, fupg_st_getval(aTHX_ st, 0, i));
return sv;
}
static SV *fupg_st_rowh(pTHX_ fupg_st *st) {
fupg_st_prepare(aTHX_ st);
fupg_st_check_dupcols(aTHX_ st->describe);
fupg_st_execute(aTHX_ st);
if (PQntuples(st->result) == 0) fu_confess("Invalid use of $st->rowh() on query returning zero rows");
if (PQntuples(st->result) > 1) fu_confess("Invalid use of $st->rowh() on query returning more than one row");
HV *hv = newHV();
SV *sv = sv_2mortal(newRV_noinc((SV *)hv));
int i;
for (i=0; i<st->nfields; i++) {
const char *key = PQfname(st->result, i);
hv_store(hv, key, -strlen(key), fupg_st_getval(aTHX_ st, 0, i), 0);
}
return sv;
}
static void fupg_st_destroy(fupg_st *st) {
int i;
/* Ignore failure, this is just a best-effort attempt to free up resources on the backend */
if (st->prepared) PQclear(PQclosePrepared(st->conn->conn, st->name));
safefree(st->query);
for (i=0; i < st->nbind; i++) SvREFCNT_dec(st->bind[i]);
safefree(st->bind);
safefree(st->param_values);
safefree(st->param_lengths);
safefree(st->param_formats);
if (st->recv) for (i=0; i<st->nfields; i++) fupg_recv_free(st->recv + i);
safefree(st->recv);
PQclear(st->describe);
PQclear(st->result);
SvREFCNT_dec(st->conn->self);
safefree(st);
}
/* TODO: $st->alla, allh, flat, kvv, kva, kvh */
/* TODO: Prepared statement caching */
Reference in a new issue View git blame Copy permalink