630 lines
22 KiB
C
630 lines
22 KiB
C
#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 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 = calloc(c->ntypes, sizeof(*c->types));
|
|
int i;
|
|
for (i=0; i<c->ntypes; i++) {
|
|
fupg_type *t = c->types + i;
|
|
t->oid = __builtin_bswap32(*((Oid *)PQgetvalue(r, i, 0)));
|
|
snprintf(t->name, sizeof(t->name), "%s", PQgetvalue(r, i, 1));
|
|
char typ = *PQgetvalue(r, i, 2);
|
|
|
|
/* enum, can use text send/recv */
|
|
if (typ == 'e') {
|
|
t->send = fupg_send_text;
|
|
t->recv = fupg_recv_text;
|
|
continue;
|
|
}
|
|
/* TODO: Array types, records, 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_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_send send;
|
|
send.oid = PQparamtype(st->describe, i);
|
|
const fupg_type *t = fupg_lookup_type(aTHX_ st->conn, refresh_done, send.oid);
|
|
if (!t) fu_confess("No type found with oid %u", send.oid);
|
|
if (!t->send) fu_confess("Unable to use type '%s' (oid %u) as bind parameter", t->name, t->oid);
|
|
send.name = t->name;
|
|
send.fn = t->send;
|
|
off = fustr_len(buf);
|
|
send.fn(aTHX_ &send, st->bind[i], buf);
|
|
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_results_setup(pTHX_ fupg_st *st, int *refresh_done) {
|
|
int i;
|
|
st->recv = safecalloc(st->nfields, sizeof(*st->recv));
|
|
if (st->stflags & FUPG_TEXT_RESULTS) {
|
|
for (i=0; i<st->nfields; i++)
|
|
st->recv[i].fn = fupg_recv_text;
|
|
return;
|
|
}
|
|
|
|
for (i=0; i<st->nfields; i++) {
|
|
fupg_recv *r = st->recv + i;
|
|
r->oid = PQftype(st->result, i);
|
|
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;
|
|
}
|
|
}
|
|
|
|
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);
|
|
fupg_results_setup(aTHX_ st, &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);
|
|
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 */
|