vndb/fu
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

pg: Add support for domain types

Browse source 
+ refactor things a bit so that send & recv functions use the same
context struct, because the way they're setup is pretty much the same
for both. This also adds recursive type resolution for bind parameters.
This commit is contained in:
Yorhel 2025-02-10 14:22:38 +01:00
parent 7515032261
commit d95ff76d43
3 changed files with 83 additions and 82 deletions
Download patch file Download diff file Expand all files Collapse all files

115
c/pgconn.c
View file

@ -46,7 +46,8 @@ typedef struct {
const char **param_values; /* Points into conn->buf or st->bind SVs, may be invalid after exec */ const char **param_values; /* Points into conn->buf or st->bind SVs, may be invalid after exec */
int *param_lengths; int *param_lengths;
int *param_formats; int *param_formats;
fupg_recv *recv; fupg_tio send;
fupg_tio *recv;
PGresult *result; PGresult *result;
} fupg_st; } fupg_st;
@ -272,7 +273,9 @@ static void fupg_refresh_types(pTHX_ fupg_conn *c) {
const char *sql = const char *sql =
"SELECT oid, typname, typtype" "SELECT oid, typname, typtype"
", CASE WHEN typcategory = 'A' THEN typelem ELSE 0 END" ", CASE WHEN typtype = 'd' THEN typbasetype"
" WHEN typcategory = 'A' THEN typelem"
" ELSE 0 END"
" FROM pg_type" " FROM pg_type"
" ORDER BY oid"; " ORDER BY oid";
PGresult *r = PQexecParams(c->conn, sql, 0, NULL, NULL, NULL, NULL, 1); PGresult *r = PQexecParams(c->conn, sql, 0, NULL, NULL, NULL, NULL, 1);
@ -290,15 +293,19 @@ static void fupg_refresh_types(pTHX_ fupg_conn *c) {
t->elemoid = fu_frombeU(32, PQgetvalue(r, i, 3)); t->elemoid = fu_frombeU(32, PQgetvalue(r, i, 3));
if (t->elemoid) { if (t->elemoid) {
/* array */ if (typ == 'd') { /* domain */
t->send = fupg_send_array; t->send = fupg_send_domain;
t->recv = fupg_recv_array; t->recv = fupg_recv_domain;
} else { /* array */
t->send = fupg_send_array;
t->recv = fupg_recv_array;
}
} else if (typ == 'e') { } else if (typ == 'e') {
/* enum, can use text send/recv */ /* enum, can use text send/recv */
t->send = fupg_send_text; t->send = fupg_send_text;
t->recv = fupg_recv_text; t->recv = fupg_recv_text;
} else { } else {
/* TODO: records, domain types, (multi)ranges, custom overrides, by-name lookup for dynamic-oid types */ /* TODO: records, (multi)ranges, custom overrides, by-name lookup for dynamic-oid types */
const fupg_type *builtin = fupg_builtin_byoid(t->oid); const fupg_type *builtin = fupg_builtin_byoid(t->oid);
if (builtin) { if (builtin) {
t->send = builtin->send; t->send = builtin->send;
@ -310,6 +317,7 @@ static void fupg_refresh_types(pTHX_ fupg_conn *c) {
} }
static const fupg_type *fupg_lookup_type(pTHX_ fupg_conn *c, int *refresh_done, Oid oid) { static const fupg_type *fupg_lookup_type(pTHX_ fupg_conn *c, int *refresh_done, Oid oid) {
if (oid == 0) return NULL;
const fupg_type *t = NULL; const fupg_type *t = NULL;
if (c->types && (t = fupg_type_byoid(c->types, c->ntypes, oid))) return t; if (c->types && (t = fupg_type_byoid(c->types, c->ntypes, oid))) return t;
if ((t = fupg_builtin_byoid(oid))) return t; if ((t = fupg_builtin_byoid(oid))) return t;
@ -446,25 +454,40 @@ static void fupg_st_check_dupcols(pTHX_ PGresult *r) {
SvREFCNT_dec((SV *)hv); SvREFCNT_dec((SV *)hv);
} }
static void fupg_params_send(pTHX_ fupg_st *st, Oid oid, SV *val, fustr *out, int *refresh_done) { static void fupg_tio_setup(pTHX_ fupg_st *st, fupg_tio *tio, int issend, Oid oid, int *refresh_done) {
fupg_send send, elem; tio->oid = oid;
const fupg_type *t = fupg_lookup_type(aTHX_ st->conn, refresh_done, oid); if (st->stflags & (issend ? FUPG_TEXT_PARAMS : FUPG_TEXT_RESULTS)) {
if (!t) fu_confess("No type found with oid %u", oid); tio->name = "{textfmt}";
if (!t->send) fu_confess("Unable to use type '%s' (oid %u) as bind parameter", t->name, t->oid); tio->send = fupg_send_text;
send.oid = oid; tio->recv = fupg_recv_text;
send.name = t->name; return;
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, *t = fupg_lookup_type(aTHX_ st->conn, refresh_done, oid);
const fupg_type *e = fupg_lookup_type(aTHX_ st->conn, refresh_done, t->elemoid); if (!t) fu_confess("No type found with oid %u", oid);
if (!e) fu_confess("No type found with oid %u", t->elemoid); if (!t->send || !t->recv) fu_confess("Unable to send or receive type '%s' (oid %u)", t->name, oid);
send.arrayelem = &elem; tio->name = t->name;
elem.oid = e->oid;
elem.name = e->name; if (issend ? t->send == fupg_send_domain : t->recv == fupg_recv_domain) {
elem.fn = e->send; e = fupg_lookup_type(aTHX_ st->conn, refresh_done, t->elemoid);
assert(e->send != fupg_send_array); /* TODO: might as well fix this, we'll need recursion for record types anyway */ if (!e) fu_confess("Base type %u not found for domain '%s' (oid %u)", t->elemoid, t->name, t->oid);
t = e;
}
tio->send = t->send;
tio->recv = t->recv;
if (issend ? tio->send == fupg_send_array : tio->recv == fupg_recv_array) {
tio->arrayelem = safecalloc(1, sizeof(*tio->arrayelem));
fupg_tio_setup(aTHX_ st, tio->arrayelem, issend, t->elemoid, refresh_done);
}
}
static void fupg_tio_free(fupg_tio *tio) {
if (!tio) return;
if (tio->send == fupg_send_array) {
fupg_tio_free(tio->arrayelem);
safefree(tio->arrayelem);
} }
send.fn(aTHX_ &send, val, out);
} }
static void fupg_params_setup(pTHX_ fupg_st *st, int *refresh_done) { static void fupg_params_setup(pTHX_ fupg_st *st, int *refresh_done) {
@ -486,8 +509,12 @@ static void fupg_params_setup(pTHX_ fupg_st *st, int *refresh_done) {
st->param_values[i] = NULL; st->param_values[i] = NULL;
continue; continue;
} }
fupg_tio_setup(aTHX_ st, &st->send, 1, PQparamtype(st->describe, i), refresh_done);
off = fustr_len(buf); off = fustr_len(buf);
fupg_params_send(aTHX_ st, PQparamtype(st->describe, i), st->bind[i], buf, refresh_done); 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_lengths[i] = fustr_len(buf) - off;
st->param_formats[i] = 1; st->param_formats[i] = 1;
st->param_values[i] = ""; st->param_values[i] = "";
@ -503,35 +530,6 @@ static void fupg_params_setup(pTHX_ fupg_st *st, int *refresh_done) {
} }
} }
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) { static void fupg_st_execute(pTHX_ fupg_st *st) {
/* Disallow fetching the results more than once. I don't see a reason why /* 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 * someone would need that and disallowing it leaves room for fetching the
@ -574,14 +572,14 @@ static void fupg_st_execute(pTHX_ fupg_st *st) {
st->recv = safecalloc(st->nfields, sizeof(*st->recv)); st->recv = safecalloc(st->nfields, sizeof(*st->recv));
int i; int i;
for (i=0; i<st->nfields; i++) for (i=0; i<st->nfields; i++)
fupg_recv_setup(aTHX_ st, st->recv + i, PQftype(st->result, i), &refresh_done); fupg_tio_setup(aTHX_ st, st->recv + i, 0, PQftype(st->result, i), &refresh_done);
} }
static SV *fupg_st_getval(pTHX_ fupg_st *st, int row, int col) { static SV *fupg_st_getval(pTHX_ fupg_st *st, int row, int col) {
PGresult *r = st->result; PGresult *r = st->result;
if (PQgetisnull(r, row, col)) return newSV(0); if (PQgetisnull(r, row, col)) return newSV(0);
const fupg_recv *ctx = st->recv+col; const fupg_tio *ctx = st->recv+col;
return ctx->fn(aTHX_ ctx, PQgetvalue(r, row, col), PQgetlength(r, row, col)); return ctx->recv(aTHX_ ctx, PQgetvalue(r, row, col), PQgetlength(r, row, col));
} }
static SV *fupg_st_exec(pTHX_ fupg_st *st) { static SV *fupg_st_exec(pTHX_ fupg_st *st) {
@ -653,7 +651,8 @@ static void fupg_st_destroy(fupg_st *st) {
safefree(st->param_values); safefree(st->param_values);
safefree(st->param_lengths); safefree(st->param_lengths);
safefree(st->param_formats); safefree(st->param_formats);
if (st->recv) for (i=0; i<st->nfields; i++) fupg_recv_free(st->recv + i); if (st->recv) for (i=0; i<st->nfields; i++) fupg_tio_free(st->recv + i);
fupg_tio_free(&st->send);
safefree(st->recv); safefree(st->recv);
PQclear(st->describe); PQclear(st->describe);
PQclear(st->result); PQclear(st->result);

41
c/pgtypes.c
View file

@ -1,34 +1,26 @@
typedef struct fupg_send fupg_send; typedef struct fupg_tio fupg_tio;
typedef struct fupg_recv fupg_recv;
/* Send function, takes a Perl value and should write the binary encoded /* Send function, takes a Perl value and should write the binary encoded
* format into the given fustr. */ * format into the given fustr. */
typedef void (*fupg_send_fn)(pTHX_ const fupg_send *, SV *, fustr *); typedef void (*fupg_send_fn)(pTHX_ const fupg_tio *, SV *, fustr *);
/* Receive function, takes a binary string and should return a Perl value. */ /* Receive function, takes a binary string and should return a Perl value. */
typedef SV *(*fupg_recv_fn)(pTHX_ const fupg_recv *, const char *, int); typedef SV *(*fupg_recv_fn)(pTHX_ const fupg_tio *, const char *, int);
struct fupg_send { /* Type I/O context */
struct fupg_tio {
Oid oid; Oid oid;
const char *name; const char *name;
fupg_send_fn fn; fupg_send_fn send;
fupg_recv_fn recv;
union { union {
fupg_send *arrayelem; fupg_tio *arrayelem;
};
};
struct fupg_recv {
Oid oid;
const char *name;
fupg_recv_fn fn;
union {
fupg_recv *arrayelem;
}; };
}; };
typedef struct { typedef struct {
Oid oid; Oid oid;
Oid elemoid; /* For arrays */ Oid elemoid; /* For arrays & domain types */
char name[64]; char name[64];
fupg_send_fn send; fupg_send_fn send;
fupg_recv_fn recv; fupg_recv_fn recv;
@ -36,8 +28,8 @@ typedef struct {
#define RECVFN(name) static SV *fupg_recv_##name(pTHX_ const fupg_recv *ctx __attribute__((unused)), const char *buf, int len) #define RECVFN(name) static SV *fupg_recv_##name(pTHX_ const fupg_tio *ctx __attribute__((unused)), const char *buf, int len)
#define SENDFN(name) static void fupg_send_##name(pTHX_ const fupg_send *ctx __attribute__((unused)), SV *val, fustr *out) #define SENDFN(name) static void fupg_send_##name(pTHX_ const fupg_tio *ctx __attribute__((unused)), SV *val, fustr *out)
#define RERR(msg, ...) fu_confess("Error parsing value for type '%s' (oid %u): "msg, ctx->name, ctx->oid __VA_OPT__(,) __VA_ARGS__) #define RERR(msg, ...) fu_confess("Error parsing value for type '%s' (oid %u): "msg, ctx->name, ctx->oid __VA_OPT__(,) __VA_ARGS__)
#define SERR(msg, ...) fu_confess("Error converting Perl value '%s' to type '%s' (oid %u): "msg, SvPV_nolen(val), ctx->name, ctx->oid __VA_OPT__(,) __VA_ARGS__) #define SERR(msg, ...) fu_confess("Error converting Perl value '%s' to type '%s' (oid %u): "msg, SvPV_nolen(val), ctx->name, ctx->oid __VA_OPT__(,) __VA_ARGS__)
#define RLEN(l) if (l != len) RERR("expected %d bytes but got %d", l, len) #define RLEN(l) if (l != len) RERR("expected %d bytes but got %d", l, len)
@ -62,6 +54,9 @@ typedef struct {
} else SERR("expected integer");\ } else SERR("expected integer");\
if (iv < min || iv > max) SERR("integer out of range") if (iv < min || iv > max) SERR("integer out of range")
/* These are simply marker functions, not supposed to be called directly */
RECVFN(domain) { (void)buf; (void)len; RERR("domain type should not be handled by this function"); }
SENDFN(domain) { (void)out; SERR("domain type should not be handled by this function"); }
RECVFN(bool) { RECVFN(bool) {
RLEN(1); RLEN(1);
@ -216,7 +211,7 @@ SENDFN(jsonpath) {
#define ARRAY_MAXDIM 100 #define ARRAY_MAXDIM 100
static SV *fupg_recv_array_elem(pTHX_ const fupg_recv *elem, const char *header, U32 dim, U32 ndim, const char **buf, const char *end) { static SV *fupg_recv_array_elem(pTHX_ const fupg_tio *elem, const char *header, U32 dim, U32 ndim, const char **buf, const char *end) {
SV *r = &PL_sv_undef; SV *r = &PL_sv_undef;
if (dim == ndim) { if (dim == ndim) {
if (end - *buf < 4) fu_confess("Invalid array format"); if (end - *buf < 4) fu_confess("Invalid array format");
@ -225,7 +220,7 @@ static SV *fupg_recv_array_elem(pTHX_ const fupg_recv *elem, const char *header,
if (end - *buf < len) fu_confess("Invalid array format"); if (end - *buf < len) fu_confess("Invalid array format");
if (len >= 0) { if (len >= 0) {
r = elem->fn(aTHX_ elem, *buf, len); r = elem->recv(aTHX_ elem, *buf, len);
*buf += len; *buf += len;
} }
@ -257,7 +252,7 @@ RECVFN(array) {
return fupg_recv_array_elem(aTHX_ ctx->arrayelem, header, 0, ndim, &data, buf+len); return fupg_recv_array_elem(aTHX_ ctx->arrayelem, header, 0, ndim, &data, buf+len);
} }
void fupg_send_array_elem(aTHX_ const fupg_send *elem, const U32 *dims, U32 dim, U32 ndim, SV *v, fustr *out, int *hasnull) { void fupg_send_array_elem(aTHX_ const fupg_tio *elem, const U32 *dims, U32 dim, U32 ndim, SV *v, fustr *out, int *hasnull) {
SvGETMAGIC(v); SvGETMAGIC(v);
if (dim == ndim) { if (dim == ndim) {
if (!SvOK(v)) { if (!SvOK(v)) {
@ -267,7 +262,7 @@ void fupg_send_array_elem(aTHX_ const fupg_send *elem, const U32 *dims, U32 dim,
} }
size_t lenoff = fustr_len(out); size_t lenoff = fustr_len(out);
fustr_write(out, "\0\0\0\0", 4); fustr_write(out, "\0\0\0\0", 4);
elem->fn(elem, v, out); elem->send(elem, v, out);
fu_tobeU(32, fustr_start(out) + lenoff, fustr_len(out) - lenoff - 4); fu_tobeU(32, fustr_start(out) + lenoff, fustr_len(out) - lenoff - 4);
return; return;
} }

9
t/pgtypes-dynamic.t
View file

@ -8,7 +8,7 @@ plan skip_all => 'Please set FU_TEST_DB to a PostgreSQL connection string to run
my $conn = FU::PG->connect($ENV{FU_TEST_DB}); my $conn = FU::PG->connect($ENV{FU_TEST_DB});
ok !eval { $conn->q('SELECT $1::aclitem', '')->exec; 1 }; ok !eval { $conn->q('SELECT $1::aclitem', '')->exec; 1 };
like $@, qr/Unable to use type/; like $@, qr/Unable to send or receive/;
{ {
my $txn = $conn->txn; my $txn = $conn->txn;
@ -18,6 +18,13 @@ like $@, qr/Unable to use type/;
is_deeply $txn->q("SELECT '{a,b,null}'::fupg_test_enum[]")->val, ['a','b',undef]; is_deeply $txn->q("SELECT '{a,b,null}'::fupg_test_enum[]")->val, ['a','b',undef];
is $txn->q('SELECT $1::fupg_test_enum[]', ['a','b',undef])->text_results->val, '{a,b,NULL}'; is $txn->q('SELECT $1::fupg_test_enum[]', ['a','b',undef])->text_results->val, '{a,b,NULL}';
$txn->exec("CREATE DOMAIN fupg_test_domain AS fupg_test_enum CHECK(value IN('a','b'))");
is $txn->q("SELECT 'a'::fupg_test_domain")->val, 'a';
is $txn->q('SELECT $1::fupg_test_domain', 'b')->val, 'b';
is_deeply $txn->q("SELECT '{a,b,null}'::fupg_test_domain[]")->val, ['a','b',undef];
is $txn->q('SELECT $1::fupg_test_domain[]', ['a','b',undef])->text_results->val, '{a,b,NULL}';
} }
done_testing; done_testing;