Branch data Line data Source code
1 : : /*
2 : : * dblink.c
3 : : *
4 : : * Functions returning results from a remote database
5 : : *
6 : : * Joe Conway <mail@joeconway.com>
7 : : * And contributors:
8 : : * Darko Prenosil <Darko.Prenosil@finteh.hr>
9 : : * Shridhar Daithankar <shridhar_daithankar@persistent.co.in>
10 : : *
11 : : * contrib/dblink/dblink.c
12 : : * Copyright (c) 2001-2026, PostgreSQL Global Development Group
13 : : * ALL RIGHTS RESERVED;
14 : : *
15 : : * Permission to use, copy, modify, and distribute this software and its
16 : : * documentation for any purpose, without fee, and without a written agreement
17 : : * is hereby granted, provided that the above copyright notice and this
18 : : * paragraph and the following two paragraphs appear in all copies.
19 : : *
20 : : * IN NO EVENT SHALL THE AUTHOR OR DISTRIBUTORS BE LIABLE TO ANY PARTY FOR
21 : : * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING
22 : : * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS
23 : : * DOCUMENTATION, EVEN IF THE AUTHOR OR DISTRIBUTORS HAVE BEEN ADVISED OF THE
24 : : * POSSIBILITY OF SUCH DAMAGE.
25 : : *
26 : : * THE AUTHOR AND DISTRIBUTORS SPECIFICALLY DISCLAIMS ANY WARRANTIES,
27 : : * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
28 : : * AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS
29 : : * ON AN "AS IS" BASIS, AND THE AUTHOR AND DISTRIBUTORS HAS NO OBLIGATIONS TO
30 : : * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
31 : : *
32 : : */
33 : : #include "postgres.h"
34 : :
35 : : #include <limits.h>
36 : :
37 : : #include "access/htup_details.h"
38 : : #include "access/relation.h"
39 : : #include "access/reloptions.h"
40 : : #include "access/table.h"
41 : : #include "catalog/namespace.h"
42 : : #include "catalog/pg_foreign_data_wrapper.h"
43 : : #include "catalog/pg_foreign_server.h"
44 : : #include "catalog/pg_type.h"
45 : : #include "catalog/pg_user_mapping.h"
46 : : #include "commands/defrem.h"
47 : : #include "common/base64.h"
48 : : #include "executor/spi.h"
49 : : #include "foreign/foreign.h"
50 : : #include "funcapi.h"
51 : : #include "lib/stringinfo.h"
52 : : #include "libpq-fe.h"
53 : : #include "libpq/libpq-be.h"
54 : : #include "libpq/libpq-be-fe-helpers.h"
55 : : #include "mb/pg_wchar.h"
56 : : #include "miscadmin.h"
57 : : #include "parser/scansup.h"
58 : : #include "utils/acl.h"
59 : : #include "utils/builtins.h"
60 : : #include "utils/fmgroids.h"
61 : : #include "utils/guc.h"
62 : : #include "utils/hsearch.h"
63 : : #include "utils/lsyscache.h"
64 : : #include "utils/memutils.h"
65 : : #include "utils/rel.h"
66 : : #include "utils/tuplestore.h"
67 : : #include "utils/varlena.h"
68 : : #include "utils/wait_event.h"
69 : :
70 : 21 : PG_MODULE_MAGIC_EXT(
71 : : .name = "dblink",
72 : : .version = PG_VERSION
73 : : );
74 : :
75 : : typedef struct remoteConn
76 : : {
77 : : PGconn *conn; /* Hold the remote connection */
78 : : int openCursorCount; /* The number of open cursors */
79 : : bool newXactForCursor; /* Opened a transaction for a cursor */
80 : : } remoteConn;
81 : :
82 : : typedef struct storeInfo
83 : : {
84 : : FunctionCallInfo fcinfo;
85 : : Tuplestorestate *tuplestore;
86 : : AttInMetadata *attinmeta;
87 : : MemoryContext tmpcontext;
88 : : char **cstrs;
89 : : /* temp storage for results to avoid leaks on exception */
90 : : PGresult *last_res;
91 : : PGresult *cur_res;
92 : : } storeInfo;
93 : :
94 : : /*
95 : : * Internal declarations
96 : : */
97 : : static Datum dblink_record_internal(FunctionCallInfo fcinfo, bool is_async);
98 : : static void prepTuplestoreResult(FunctionCallInfo fcinfo);
99 : : static void materializeResult(FunctionCallInfo fcinfo, PGconn *conn,
100 : : PGresult *res);
101 : : static void materializeQueryResult(FunctionCallInfo fcinfo,
102 : : PGconn *conn,
103 : : const char *conname,
104 : : const char *sql,
105 : : bool fail);
106 : : static PGresult *storeQueryResult(storeInfo *sinfo, PGconn *conn, const char *sql);
107 : : static void storeRow(storeInfo *sinfo, PGresult *res, bool first);
108 : : static remoteConn *getConnectionByName(const char *name);
109 : : static HTAB *createConnHash(void);
110 : : static remoteConn *createNewConnection(const char *name);
111 : : static void deleteConnection(const char *name);
112 : : static char **get_pkey_attnames(Relation rel, int16 *indnkeyatts);
113 : : static char **get_text_array_contents(ArrayType *array, int *numitems);
114 : : static char *get_sql_insert(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals, char **tgt_pkattvals);
115 : : static char *get_sql_delete(Relation rel, int *pkattnums, int pknumatts, char **tgt_pkattvals);
116 : : static char *get_sql_update(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals, char **tgt_pkattvals);
117 : : static char *quote_ident_cstr(char *rawstr);
118 : : static int get_attnum_pk_pos(int *pkattnums, int pknumatts, int key);
119 : : static HeapTuple get_tuple_of_interest(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals);
120 : : static Relation get_rel_from_relname(text *relname_text, LOCKMODE lockmode, AclMode aclmode);
121 : : static char *generate_relation_name(Relation rel);
122 : : static void dblink_connstr_check(const char *connstr);
123 : : static bool dblink_connstr_has_pw(const char *connstr);
124 : : static void dblink_security_check(PGconn *conn, const char *connname,
125 : : const char *connstr);
126 : : static void dblink_res_error(PGconn *conn, const char *conname, PGresult *res,
127 : : bool fail, const char *fmt, ...) pg_attribute_printf(5, 6);
128 : : static char *get_connect_string(const char *servername);
129 : : static char *escape_param_str(const char *str);
130 : : static void validate_pkattnums(Relation rel,
131 : : int2vector *pkattnums_arg, int32 pknumatts_arg,
132 : : int **pkattnums, int *pknumatts);
133 : : static bool is_valid_dblink_option(const PQconninfoOption *options,
134 : : const char *option, Oid context);
135 : : static int applyRemoteGucs(PGconn *conn);
136 : : static void restoreLocalGucs(int nestlevel);
137 : : static bool UseScramPassthrough(ForeignServer *foreign_server, UserMapping *user);
138 : : static void appendSCRAMKeysInfo(StringInfo buf);
139 : : static bool is_valid_dblink_fdw_option(const PQconninfoOption *options, const char *option,
140 : : Oid context);
141 : : static bool dblink_connstr_has_required_scram_options(const char *connstr);
142 : :
143 : : /* Global */
144 : : static remoteConn *pconn = NULL;
145 : : static HTAB *remoteConnHash = NULL;
146 : :
147 : : /* custom wait event values, retrieved from shared memory */
148 : : static uint32 dblink_we_connect = 0;
149 : : static uint32 dblink_we_get_conn = 0;
150 : : static uint32 dblink_we_get_result = 0;
151 : :
152 : : /*
153 : : * Following is hash that holds multiple remote connections.
154 : : * Calling convention of each dblink function changes to accept
155 : : * connection name as the first parameter. The connection hash is
156 : : * much like ecpg e.g. a mapping between a name and a PGconn object.
157 : : *
158 : : * To avoid potentially leaking a PGconn object in case of out-of-memory
159 : : * errors, we first create the hash entry, then open the PGconn.
160 : : * Hence, a hash entry whose rconn.conn pointer is NULL must be
161 : : * understood as a leftover from a failed create; it should be ignored
162 : : * by lookup operations, and silently replaced by create operations.
163 : : */
164 : :
165 : : typedef struct remoteConnHashEnt
166 : : {
167 : : char name[NAMEDATALEN];
168 : : remoteConn rconn;
169 : : } remoteConnHashEnt;
170 : :
171 : : /* initial number of connection hashes */
172 : : #define NUMCONN 16
173 : :
174 : : pg_noreturn static void
175 : 0 : dblink_res_internalerror(PGconn *conn, PGresult *res, const char *p2)
176 : : {
177 : 0 : char *msg = pchomp(PQerrorMessage(conn));
178 : :
179 : 0 : PQclear(res);
180 [ # # ]: 0 : elog(ERROR, "%s: %s", p2, msg);
181 : : }
182 : :
183 : : pg_noreturn static void
184 : 3 : dblink_conn_not_avail(const char *conname)
185 : : {
186 [ + + ]: 3 : if (conname)
187 [ + - ]: 1 : ereport(ERROR,
188 : : (errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST),
189 : : errmsg("connection \"%s\" not available", conname)));
190 : : else
191 [ + - ]: 2 : ereport(ERROR,
192 : : (errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST),
193 : : errmsg("connection not available")));
194 : : }
195 : :
196 : : static void
197 : 38 : dblink_get_conn(char *conname_or_str,
198 : : PGconn *volatile *conn_p, char **conname_p, volatile bool *freeconn_p)
199 : : {
200 : 38 : remoteConn *rconn = getConnectionByName(conname_or_str);
201 : : PGconn *conn;
202 : : char *conname;
203 : : bool freeconn;
204 : :
205 [ + + ]: 38 : if (rconn)
206 : : {
207 : 27 : conn = rconn->conn;
208 : 27 : conname = conname_or_str;
209 : 27 : freeconn = false;
210 : : }
211 : : else
212 : : {
213 : : const char *connstr;
214 : :
215 : 11 : connstr = get_connect_string(conname_or_str);
216 [ + + ]: 11 : if (connstr == NULL)
217 : 6 : connstr = conname_or_str;
218 : 11 : dblink_connstr_check(connstr);
219 : :
220 : : /* first time, allocate or get the custom wait event */
221 [ + + ]: 9 : if (dblink_we_get_conn == 0)
222 : 5 : dblink_we_get_conn = WaitEventExtensionNew("DblinkGetConnect");
223 : :
224 : : /* OK to make connection */
225 : 9 : conn = libpqsrv_connect_start(connstr);
226 : 9 : PQsetNoticeReceiver(conn, libpqsrv_notice_receiver,
227 : : "received message via remote connection");
228 : 9 : libpqsrv_connect_complete(conn, dblink_we_get_conn);
229 : :
230 [ + + ]: 9 : if (PQstatus(conn) == CONNECTION_BAD)
231 : : {
232 : 3 : char *msg = pchomp(PQerrorMessage(conn));
233 : :
234 : 3 : libpqsrv_disconnect(conn);
235 [ + - ]: 3 : ereport(ERROR,
236 : : (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
237 : : errmsg("could not establish connection"),
238 : : errdetail_internal("%s", msg)));
239 : : }
240 : :
241 : 6 : dblink_security_check(conn, NULL, connstr);
242 [ - + ]: 6 : if (PQclientEncoding(conn) != GetDatabaseEncoding())
243 : 0 : PQsetClientEncoding(conn, GetDatabaseEncodingName());
244 : 6 : freeconn = true;
245 : 6 : conname = NULL;
246 : : }
247 : :
248 : 33 : *conn_p = conn;
249 : 33 : *conname_p = conname;
250 : 33 : *freeconn_p = freeconn;
251 : 33 : }
252 : :
253 : : static PGconn *
254 : 17 : dblink_get_named_conn(const char *conname)
255 : : {
256 : 17 : remoteConn *rconn = getConnectionByName(conname);
257 : :
258 [ + - ]: 17 : if (rconn)
259 : 17 : return rconn->conn;
260 : :
261 : 0 : dblink_conn_not_avail(conname);
262 : : return NULL; /* keep compiler quiet */
263 : : }
264 : :
265 : : static void
266 : 125 : dblink_init(void)
267 : : {
268 [ + + ]: 125 : if (!pconn)
269 : : {
270 [ + - ]: 8 : if (dblink_we_get_result == 0)
271 : 8 : dblink_we_get_result = WaitEventExtensionNew("DblinkGetResult");
272 : :
273 : 8 : pconn = (remoteConn *) MemoryContextAlloc(TopMemoryContext, sizeof(remoteConn));
274 : 8 : pconn->conn = NULL;
275 : 8 : pconn->openCursorCount = 0;
276 : 8 : pconn->newXactForCursor = false;
277 : : }
278 : 125 : }
279 : :
280 : : /*
281 : : * Create a persistent connection to another database
282 : : */
283 : 13 : PG_FUNCTION_INFO_V1(dblink_connect);
284 : : Datum
285 : 16 : dblink_connect(PG_FUNCTION_ARGS)
286 : : {
287 : 16 : char *conname_or_str = NULL;
288 : 16 : char *connstr = NULL;
289 : 16 : char *connname = NULL;
290 : : char *msg;
291 : 16 : PGconn *conn = NULL;
292 : 16 : remoteConn *rconn = NULL;
293 : :
294 : 16 : dblink_init();
295 : :
296 [ + + ]: 16 : if (PG_NARGS() == 2)
297 : : {
298 : 11 : conname_or_str = text_to_cstring(PG_GETARG_TEXT_PP(1));
299 : 11 : connname = text_to_cstring(PG_GETARG_TEXT_PP(0));
300 : : }
301 [ + - ]: 5 : else if (PG_NARGS() == 1)
302 : 5 : conname_or_str = text_to_cstring(PG_GETARG_TEXT_PP(0));
303 : :
304 : : /* first check for valid foreign data server */
305 : 16 : connstr = get_connect_string(conname_or_str);
306 [ + + ]: 16 : if (connstr == NULL)
307 : 14 : connstr = conname_or_str;
308 : :
309 : : /* check password in connection string if not superuser */
310 : 16 : dblink_connstr_check(connstr);
311 : :
312 : : /* first time, allocate or get the custom wait event */
313 [ + + ]: 15 : if (dblink_we_connect == 0)
314 : 2 : dblink_we_connect = WaitEventExtensionNew("DblinkConnect");
315 : :
316 : : /* if we need a hashtable entry, make that first, since it might fail */
317 [ + + ]: 15 : if (connname)
318 : : {
319 : 10 : rconn = createNewConnection(connname);
320 : : Assert(rconn->conn == NULL);
321 : : }
322 : :
323 : : /* OK to make connection */
324 : 14 : conn = libpqsrv_connect_start(connstr);
325 [ + - ]: 14 : if (conn != NULL)
326 : 14 : PQsetNoticeReceiver(conn, libpqsrv_notice_receiver,
327 : : "received message via remote connection");
328 : 14 : libpqsrv_connect_complete(conn, dblink_we_connect);
329 : :
330 [ - + ]: 14 : if (PQstatus(conn) == CONNECTION_BAD)
331 : : {
332 : 0 : msg = pchomp(PQerrorMessage(conn));
333 : 0 : libpqsrv_disconnect(conn);
334 [ # # ]: 0 : if (connname)
335 : 0 : deleteConnection(connname);
336 : :
337 [ # # ]: 0 : ereport(ERROR,
338 : : (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
339 : : errmsg("could not establish connection"),
340 : : errdetail_internal("%s", msg)));
341 : : }
342 : :
343 : : /* check password actually used if not superuser */
344 : 14 : dblink_security_check(conn, connname, connstr);
345 : :
346 : : /* attempt to set client encoding to match server encoding, if needed */
347 [ - + ]: 14 : if (PQclientEncoding(conn) != GetDatabaseEncoding())
348 : 0 : PQsetClientEncoding(conn, GetDatabaseEncodingName());
349 : :
350 : : /* all OK, save away the conn */
351 [ + + ]: 14 : if (connname)
352 : : {
353 : 9 : rconn->conn = conn;
354 : : }
355 : : else
356 : : {
357 [ + + ]: 5 : if (pconn->conn)
358 : 1 : libpqsrv_disconnect(pconn->conn);
359 : 5 : pconn->conn = conn;
360 : : }
361 : :
362 : 14 : PG_RETURN_TEXT_P(cstring_to_text("OK"));
363 : : }
364 : :
365 : : /*
366 : : * Clear a persistent connection to another database
367 : : */
368 : 8 : PG_FUNCTION_INFO_V1(dblink_disconnect);
369 : : Datum
370 : 13 : dblink_disconnect(PG_FUNCTION_ARGS)
371 : : {
372 : 13 : char *conname = NULL;
373 : 13 : remoteConn *rconn = NULL;
374 : 13 : PGconn *conn = NULL;
375 : :
376 : 13 : dblink_init();
377 : :
378 [ + + ]: 13 : if (PG_NARGS() == 1)
379 : : {
380 : 9 : conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
381 : 9 : rconn = getConnectionByName(conname);
382 [ + + ]: 9 : if (rconn)
383 : 8 : conn = rconn->conn;
384 : : }
385 : : else
386 : 4 : conn = pconn->conn;
387 : :
388 [ + + ]: 13 : if (!conn)
389 : 1 : dblink_conn_not_avail(conname);
390 : :
391 : 12 : libpqsrv_disconnect(conn);
392 [ + + ]: 12 : if (rconn)
393 : 8 : deleteConnection(conname);
394 : : else
395 : 4 : pconn->conn = NULL;
396 : :
397 : 12 : PG_RETURN_TEXT_P(cstring_to_text("OK"));
398 : : }
399 : :
400 : : /*
401 : : * opens a cursor using a persistent connection
402 : : */
403 : 13 : PG_FUNCTION_INFO_V1(dblink_open);
404 : : Datum
405 : 9 : dblink_open(PG_FUNCTION_ARGS)
406 : : {
407 : 9 : PGresult *res = NULL;
408 : : PGconn *conn;
409 : 9 : char *curname = NULL;
410 : 9 : char *sql = NULL;
411 : 9 : char *conname = NULL;
412 : : StringInfoData buf;
413 : 9 : remoteConn *rconn = NULL;
414 : 9 : bool fail = true; /* default to backward compatible behavior */
415 : :
416 : 9 : dblink_init();
417 : 9 : initStringInfo(&buf);
418 : :
419 [ + + ]: 9 : if (PG_NARGS() == 2)
420 : : {
421 : : /* text,text */
422 : 2 : curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
423 : 2 : sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
424 : 2 : rconn = pconn;
425 : : }
426 [ + + ]: 7 : else if (PG_NARGS() == 3)
427 : : {
428 : : /* might be text,text,text or text,text,bool */
429 [ + + ]: 6 : if (get_fn_expr_argtype(fcinfo->flinfo, 2) == BOOLOID)
430 : : {
431 : 1 : curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
432 : 1 : sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
433 : 1 : fail = PG_GETARG_BOOL(2);
434 : 1 : rconn = pconn;
435 : : }
436 : : else
437 : : {
438 : 5 : conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
439 : 5 : curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
440 : 5 : sql = text_to_cstring(PG_GETARG_TEXT_PP(2));
441 : 5 : rconn = getConnectionByName(conname);
442 : : }
443 : : }
444 [ + - ]: 1 : else if (PG_NARGS() == 4)
445 : : {
446 : : /* text,text,text,bool */
447 : 1 : conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
448 : 1 : curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
449 : 1 : sql = text_to_cstring(PG_GETARG_TEXT_PP(2));
450 : 1 : fail = PG_GETARG_BOOL(3);
451 : 1 : rconn = getConnectionByName(conname);
452 : : }
453 : :
454 [ + - - + ]: 9 : if (!rconn || !rconn->conn)
455 : 0 : dblink_conn_not_avail(conname);
456 : :
457 : 9 : conn = rconn->conn;
458 : :
459 : : /* If we are not in a transaction, start one */
460 [ + + ]: 9 : if (PQtransactionStatus(conn) == PQTRANS_IDLE)
461 : : {
462 : 7 : res = libpqsrv_exec(conn, "BEGIN", dblink_we_get_result);
463 [ - + ]: 7 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
464 : 0 : dblink_res_internalerror(conn, res, "begin error");
465 : 7 : PQclear(res);
466 : 7 : rconn->newXactForCursor = true;
467 : :
468 : : /*
469 : : * Since transaction state was IDLE, we force cursor count to
470 : : * initially be 0. This is needed as a previous ABORT might have wiped
471 : : * out our transaction without maintaining the cursor count for us.
472 : : */
473 : 7 : rconn->openCursorCount = 0;
474 : : }
475 : :
476 : : /* if we started a transaction, increment cursor count */
477 [ + - ]: 9 : if (rconn->newXactForCursor)
478 : 9 : (rconn->openCursorCount)++;
479 : :
480 : 9 : appendStringInfo(&buf, "DECLARE %s CURSOR FOR %s", curname, sql);
481 : 9 : res = libpqsrv_exec(conn, buf.data, dblink_we_get_result);
482 [ + - + + ]: 9 : if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
483 : : {
484 : 2 : dblink_res_error(conn, conname, res, fail,
485 : : "while opening cursor \"%s\"", curname);
486 : 2 : PG_RETURN_TEXT_P(cstring_to_text("ERROR"));
487 : : }
488 : :
489 : 7 : PQclear(res);
490 : 7 : PG_RETURN_TEXT_P(cstring_to_text("OK"));
491 : : }
492 : :
493 : : /*
494 : : * closes a cursor
495 : : */
496 : 10 : PG_FUNCTION_INFO_V1(dblink_close);
497 : : Datum
498 : 5 : dblink_close(PG_FUNCTION_ARGS)
499 : : {
500 : : PGconn *conn;
501 : 5 : PGresult *res = NULL;
502 : 5 : char *curname = NULL;
503 : 5 : char *conname = NULL;
504 : : StringInfoData buf;
505 : 5 : remoteConn *rconn = NULL;
506 : 5 : bool fail = true; /* default to backward compatible behavior */
507 : :
508 : 5 : dblink_init();
509 : 5 : initStringInfo(&buf);
510 : :
511 [ - + ]: 5 : if (PG_NARGS() == 1)
512 : : {
513 : : /* text */
514 : 0 : curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
515 : 0 : rconn = pconn;
516 : : }
517 [ + - ]: 5 : else if (PG_NARGS() == 2)
518 : : {
519 : : /* might be text,text or text,bool */
520 [ + + ]: 5 : if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
521 : : {
522 : 2 : curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
523 : 2 : fail = PG_GETARG_BOOL(1);
524 : 2 : rconn = pconn;
525 : : }
526 : : else
527 : : {
528 : 3 : conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
529 : 3 : curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
530 : 3 : rconn = getConnectionByName(conname);
531 : : }
532 : : }
533 [ - + ]: 5 : if (PG_NARGS() == 3)
534 : : {
535 : : /* text,text,bool */
536 : 0 : conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
537 : 0 : curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
538 : 0 : fail = PG_GETARG_BOOL(2);
539 : 0 : rconn = getConnectionByName(conname);
540 : : }
541 : :
542 [ + - - + ]: 5 : if (!rconn || !rconn->conn)
543 : 0 : dblink_conn_not_avail(conname);
544 : :
545 : 5 : conn = rconn->conn;
546 : :
547 : 5 : appendStringInfo(&buf, "CLOSE %s", curname);
548 : :
549 : : /* close the cursor */
550 : 5 : res = libpqsrv_exec(conn, buf.data, dblink_we_get_result);
551 [ + - + + ]: 5 : if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
552 : : {
553 : 1 : dblink_res_error(conn, conname, res, fail,
554 : : "while closing cursor \"%s\"", curname);
555 : 1 : PG_RETURN_TEXT_P(cstring_to_text("ERROR"));
556 : : }
557 : :
558 : 4 : PQclear(res);
559 : :
560 : : /* if we started a transaction, decrement cursor count */
561 [ + - ]: 4 : if (rconn->newXactForCursor)
562 : : {
563 : 4 : (rconn->openCursorCount)--;
564 : :
565 : : /* if count is zero, commit the transaction */
566 [ + + ]: 4 : if (rconn->openCursorCount == 0)
567 : : {
568 : 2 : rconn->newXactForCursor = false;
569 : :
570 : 2 : res = libpqsrv_exec(conn, "COMMIT", dblink_we_get_result);
571 [ - + ]: 2 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
572 : 0 : dblink_res_internalerror(conn, res, "commit error");
573 : 2 : PQclear(res);
574 : : }
575 : : }
576 : :
577 : 4 : PG_RETURN_TEXT_P(cstring_to_text("OK"));
578 : : }
579 : :
580 : : /*
581 : : * Fetch results from an open cursor
582 : : */
583 : 13 : PG_FUNCTION_INFO_V1(dblink_fetch);
584 : : Datum
585 : 13 : dblink_fetch(PG_FUNCTION_ARGS)
586 : : {
587 : 13 : PGresult *res = NULL;
588 : 13 : char *conname = NULL;
589 : 13 : remoteConn *rconn = NULL;
590 : 13 : PGconn *conn = NULL;
591 : : StringInfoData buf;
592 : 13 : char *curname = NULL;
593 : 13 : int howmany = 0;
594 : 13 : bool fail = true; /* default to backward compatible */
595 : :
596 : 13 : prepTuplestoreResult(fcinfo);
597 : :
598 : 13 : dblink_init();
599 : :
600 [ + + ]: 13 : if (PG_NARGS() == 4)
601 : : {
602 : : /* text,text,int,bool */
603 : 1 : conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
604 : 1 : curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
605 : 1 : howmany = PG_GETARG_INT32(2);
606 : 1 : fail = PG_GETARG_BOOL(3);
607 : :
608 : 1 : rconn = getConnectionByName(conname);
609 [ + - ]: 1 : if (rconn)
610 : 1 : conn = rconn->conn;
611 : : }
612 [ + + ]: 12 : else if (PG_NARGS() == 3)
613 : : {
614 : : /* text,text,int or text,int,bool */
615 [ + + ]: 8 : if (get_fn_expr_argtype(fcinfo->flinfo, 2) == BOOLOID)
616 : : {
617 : 2 : curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
618 : 2 : howmany = PG_GETARG_INT32(1);
619 : 2 : fail = PG_GETARG_BOOL(2);
620 : 2 : conn = pconn->conn;
621 : : }
622 : : else
623 : : {
624 : 6 : conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
625 : 6 : curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
626 : 6 : howmany = PG_GETARG_INT32(2);
627 : :
628 : 6 : rconn = getConnectionByName(conname);
629 [ + - ]: 6 : if (rconn)
630 : 6 : conn = rconn->conn;
631 : : }
632 : : }
633 [ + - ]: 4 : else if (PG_NARGS() == 2)
634 : : {
635 : : /* text,int */
636 : 4 : curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
637 : 4 : howmany = PG_GETARG_INT32(1);
638 : 4 : conn = pconn->conn;
639 : : }
640 : :
641 [ - + ]: 13 : if (!conn)
642 : 0 : dblink_conn_not_avail(conname);
643 : :
644 : 13 : initStringInfo(&buf);
645 : 13 : appendStringInfo(&buf, "FETCH %d FROM %s", howmany, curname);
646 : :
647 : : /*
648 : : * Try to execute the query. Note that since libpq uses malloc, the
649 : : * PGresult will be long-lived even though we are still in a short-lived
650 : : * memory context.
651 : : */
652 : 13 : res = libpqsrv_exec(conn, buf.data, dblink_we_get_result);
653 [ + - + - ]: 26 : if (!res ||
654 [ + + ]: 26 : (PQresultStatus(res) != PGRES_COMMAND_OK &&
655 : 13 : PQresultStatus(res) != PGRES_TUPLES_OK))
656 : : {
657 : 5 : dblink_res_error(conn, conname, res, fail,
658 : : "while fetching from cursor \"%s\"", curname);
659 : 3 : return (Datum) 0;
660 : : }
661 [ - + ]: 8 : else if (PQresultStatus(res) == PGRES_COMMAND_OK)
662 : : {
663 : : /* cursor does not exist - closed already or bad name */
664 : 0 : PQclear(res);
665 [ # # ]: 0 : ereport(ERROR,
666 : : (errcode(ERRCODE_INVALID_CURSOR_NAME),
667 : : errmsg("cursor \"%s\" does not exist", curname)));
668 : : }
669 : :
670 : 8 : materializeResult(fcinfo, conn, res);
671 : 7 : return (Datum) 0;
672 : : }
673 : :
674 : : /*
675 : : * Note: this is the new preferred version of dblink
676 : : */
677 : 20 : PG_FUNCTION_INFO_V1(dblink_record);
678 : : Datum
679 : 30 : dblink_record(PG_FUNCTION_ARGS)
680 : : {
681 : 30 : return dblink_record_internal(fcinfo, false);
682 : : }
683 : :
684 : 4 : PG_FUNCTION_INFO_V1(dblink_send_query);
685 : : Datum
686 : 6 : dblink_send_query(PG_FUNCTION_ARGS)
687 : : {
688 : : PGconn *conn;
689 : : char *sql;
690 : : int retval;
691 : :
692 [ + - ]: 6 : if (PG_NARGS() == 2)
693 : : {
694 : 6 : conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
695 : 6 : sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
696 : : }
697 : : else
698 : : /* shouldn't happen */
699 [ # # ]: 0 : elog(ERROR, "wrong number of arguments");
700 : :
701 : : /* async query send */
702 : 6 : retval = PQsendQuery(conn, sql);
703 [ - + ]: 6 : if (retval != 1)
704 [ # # ]: 0 : elog(NOTICE, "could not send query: %s", pchomp(PQerrorMessage(conn)));
705 : :
706 : 6 : PG_RETURN_INT32(retval);
707 : : }
708 : :
709 : 6 : PG_FUNCTION_INFO_V1(dblink_get_result);
710 : : Datum
711 : 8 : dblink_get_result(PG_FUNCTION_ARGS)
712 : : {
713 : 8 : return dblink_record_internal(fcinfo, true);
714 : : }
715 : :
716 : : static Datum
717 : 38 : dblink_record_internal(FunctionCallInfo fcinfo, bool is_async)
718 : : {
719 : 38 : PGconn *volatile conn = NULL;
720 : 38 : volatile bool freeconn = false;
721 : :
722 : 38 : prepTuplestoreResult(fcinfo);
723 : :
724 : 38 : dblink_init();
725 : :
726 [ + + ]: 38 : PG_TRY();
727 : : {
728 : 38 : char *sql = NULL;
729 : 38 : char *conname = NULL;
730 : 38 : bool fail = true; /* default to backward compatible */
731 : :
732 [ + + ]: 38 : if (!is_async)
733 : : {
734 [ + + ]: 30 : if (PG_NARGS() == 3)
735 : : {
736 : : /* text,text,bool */
737 : 1 : conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
738 : 1 : sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
739 : 1 : fail = PG_GETARG_BOOL(2);
740 : 1 : dblink_get_conn(conname, &conn, &conname, &freeconn);
741 : : }
742 [ + + ]: 29 : else if (PG_NARGS() == 2)
743 : : {
744 : : /* text,text or text,bool */
745 [ + + ]: 22 : if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
746 : : {
747 : 1 : sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
748 : 1 : fail = PG_GETARG_BOOL(1);
749 : 1 : conn = pconn->conn;
750 : : }
751 : : else
752 : : {
753 : 21 : conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
754 : 21 : sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
755 : 21 : dblink_get_conn(conname, &conn, &conname, &freeconn);
756 : : }
757 : : }
758 [ + - ]: 7 : else if (PG_NARGS() == 1)
759 : : {
760 : : /* text */
761 : 7 : conn = pconn->conn;
762 : 7 : sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
763 : : }
764 : : else
765 : : /* shouldn't happen */
766 [ # # ]: 0 : elog(ERROR, "wrong number of arguments");
767 : : }
768 : : else /* is_async */
769 : : {
770 : : /* get async result */
771 : 8 : conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
772 : :
773 [ - + ]: 8 : if (PG_NARGS() == 2)
774 : : {
775 : : /* text,bool */
776 : 0 : fail = PG_GETARG_BOOL(1);
777 : 0 : conn = dblink_get_named_conn(conname);
778 : : }
779 [ + - ]: 8 : else if (PG_NARGS() == 1)
780 : : {
781 : : /* text */
782 : 8 : conn = dblink_get_named_conn(conname);
783 : : }
784 : : else
785 : : /* shouldn't happen */
786 [ # # ]: 0 : elog(ERROR, "wrong number of arguments");
787 : : }
788 : :
789 [ + + ]: 33 : if (!conn)
790 : 2 : dblink_conn_not_avail(conname);
791 : :
792 [ + + ]: 31 : if (!is_async)
793 : : {
794 : : /* synchronous query, use efficient tuple collection method */
795 : 23 : materializeQueryResult(fcinfo, conn, conname, sql, fail);
796 : : }
797 : : else
798 : : {
799 : : /* async result retrieval, do it the old way */
800 : 8 : PGresult *res = libpqsrv_get_result(conn, dblink_we_get_result);
801 : :
802 : : /* NULL means we're all done with the async results */
803 [ + + ]: 8 : if (res)
804 : : {
805 [ + - ]: 5 : if (PQresultStatus(res) != PGRES_COMMAND_OK &&
806 [ - + ]: 5 : PQresultStatus(res) != PGRES_TUPLES_OK)
807 : : {
808 : 0 : dblink_res_error(conn, conname, res, fail,
809 : : "while executing query");
810 : : /* if fail isn't set, we'll return an empty query result */
811 : : }
812 : : else
813 : : {
814 : 5 : materializeResult(fcinfo, conn, res);
815 : : }
816 : : }
817 : : }
818 : : }
819 : 7 : PG_FINALLY();
820 : : {
821 : : /* if needed, close the connection to the database */
822 [ + + ]: 38 : if (freeconn)
823 : 5 : libpqsrv_disconnect(conn);
824 : : }
825 [ + + ]: 38 : PG_END_TRY();
826 : :
827 : 31 : return (Datum) 0;
828 : : }
829 : :
830 : : /*
831 : : * Verify function caller can handle a tuplestore result, and set up for that.
832 : : *
833 : : * Note: if the caller returns without actually creating a tuplestore, the
834 : : * executor will treat the function result as an empty set.
835 : : */
836 : : static void
837 : 51 : prepTuplestoreResult(FunctionCallInfo fcinfo)
838 : : {
839 : 51 : ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
840 : :
841 : : /* check to see if query supports us returning a tuplestore */
842 [ + - - + ]: 51 : if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
843 [ # # ]: 0 : ereport(ERROR,
844 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
845 : : errmsg("set-valued function called in context that cannot accept a set")));
846 [ - + ]: 51 : if (!(rsinfo->allowedModes & SFRM_Materialize))
847 [ # # ]: 0 : ereport(ERROR,
848 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
849 : : errmsg("materialize mode required, but it is not allowed in this context")));
850 : :
851 : : /* let the executor know we're sending back a tuplestore */
852 : 51 : rsinfo->returnMode = SFRM_Materialize;
853 : :
854 : : /* caller must fill these to return a non-empty result */
855 : 51 : rsinfo->setResult = NULL;
856 : 51 : rsinfo->setDesc = NULL;
857 : 51 : }
858 : :
859 : : /*
860 : : * Copy the contents of the PGresult into a tuplestore to be returned
861 : : * as the result of the current function.
862 : : * The PGresult will be released in this function.
863 : : */
864 : : static void
865 : 13 : materializeResult(FunctionCallInfo fcinfo, PGconn *conn, PGresult *res)
866 : : {
867 : 13 : ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
868 : : TupleDesc tupdesc;
869 : : bool is_sql_cmd;
870 : : int ntuples;
871 : : int nfields;
872 : :
873 : : /* prepTuplestoreResult must have been called previously */
874 : : Assert(rsinfo->returnMode == SFRM_Materialize);
875 : :
876 [ - + ]: 13 : if (PQresultStatus(res) == PGRES_COMMAND_OK)
877 : : {
878 : 0 : is_sql_cmd = true;
879 : :
880 : : /*
881 : : * need a tuple descriptor representing one TEXT column to return the
882 : : * command status string as our result tuple
883 : : */
884 : 0 : tupdesc = CreateTemplateTupleDesc(1);
885 : 0 : TupleDescInitEntry(tupdesc, (AttrNumber) 1, "status",
886 : : TEXTOID, -1, 0);
887 : 0 : TupleDescFinalize(tupdesc);
888 : 0 : ntuples = 1;
889 : 0 : nfields = 1;
890 : : }
891 : : else
892 : : {
893 : : Assert(PQresultStatus(res) == PGRES_TUPLES_OK);
894 : :
895 : 13 : is_sql_cmd = false;
896 : :
897 : : /* get a tuple descriptor for our result type */
898 [ + - - ]: 13 : switch (get_call_result_type(fcinfo, NULL, &tupdesc))
899 : : {
900 : 13 : case TYPEFUNC_COMPOSITE:
901 : : /* success */
902 : 13 : break;
903 : 0 : case TYPEFUNC_RECORD:
904 : : /* failed to determine actual type of RECORD */
905 [ # # ]: 0 : ereport(ERROR,
906 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
907 : : errmsg("function returning record called in context "
908 : : "that cannot accept type record")));
909 : : break;
910 : 0 : default:
911 : : /* result type isn't composite */
912 [ # # ]: 0 : elog(ERROR, "return type must be a row type");
913 : : break;
914 : : }
915 : :
916 : : /* make sure we have a persistent copy of the tupdesc */
917 : 13 : tupdesc = CreateTupleDescCopy(tupdesc);
918 : 13 : ntuples = PQntuples(res);
919 : 13 : nfields = PQnfields(res);
920 : : }
921 : :
922 : : /*
923 : : * check result and tuple descriptor have the same number of columns
924 : : */
925 [ - + ]: 13 : if (nfields != tupdesc->natts)
926 [ # # ]: 0 : ereport(ERROR,
927 : : (errcode(ERRCODE_DATATYPE_MISMATCH),
928 : : errmsg("remote query result rowtype does not match "
929 : : "the specified FROM clause rowtype")));
930 : :
931 [ + - ]: 13 : if (ntuples > 0)
932 : : {
933 : : AttInMetadata *attinmeta;
934 : 13 : int nestlevel = -1;
935 : : Tuplestorestate *tupstore;
936 : : MemoryContext oldcontext;
937 : : int row;
938 : : char **values;
939 : :
940 : 13 : attinmeta = TupleDescGetAttInMetadata(tupdesc);
941 : :
942 : : /* Set GUCs to ensure we read GUC-sensitive data types correctly */
943 [ + - ]: 13 : if (!is_sql_cmd)
944 : 13 : nestlevel = applyRemoteGucs(conn);
945 : :
946 : 13 : oldcontext = MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory);
947 : 13 : tupstore = tuplestore_begin_heap(true, false, work_mem);
948 : 13 : rsinfo->setResult = tupstore;
949 : 13 : rsinfo->setDesc = tupdesc;
950 : 13 : MemoryContextSwitchTo(oldcontext);
951 : :
952 : 13 : values = palloc_array(char *, nfields);
953 : :
954 : : /* put all tuples into the tuplestore */
955 [ + + ]: 49 : for (row = 0; row < ntuples; row++)
956 : : {
957 : : HeapTuple tuple;
958 : :
959 [ + - ]: 37 : if (!is_sql_cmd)
960 : : {
961 : : int i;
962 : :
963 [ + + ]: 138 : for (i = 0; i < nfields; i++)
964 : : {
965 [ - + ]: 101 : if (PQgetisnull(res, row, i))
966 : 0 : values[i] = NULL;
967 : : else
968 : 101 : values[i] = PQgetvalue(res, row, i);
969 : : }
970 : : }
971 : : else
972 : : {
973 : 0 : values[0] = PQcmdStatus(res);
974 : : }
975 : :
976 : : /* build the tuple and put it into the tuplestore. */
977 : 37 : tuple = BuildTupleFromCStrings(attinmeta, values);
978 : 36 : tuplestore_puttuple(tupstore, tuple);
979 : : }
980 : :
981 : : /* clean up GUC settings, if we changed any */
982 : 12 : restoreLocalGucs(nestlevel);
983 : : }
984 : :
985 : 12 : PQclear(res);
986 : 12 : }
987 : :
988 : : /*
989 : : * Execute the given SQL command and store its results into a tuplestore
990 : : * to be returned as the result of the current function.
991 : : *
992 : : * This is equivalent to PQexec followed by materializeResult, but we make
993 : : * use of libpq's single-row mode to avoid accumulating the whole result
994 : : * inside libpq before it gets transferred to the tuplestore.
995 : : */
996 : : static void
997 : 23 : materializeQueryResult(FunctionCallInfo fcinfo,
998 : : PGconn *conn,
999 : : const char *conname,
1000 : : const char *sql,
1001 : : bool fail)
1002 : : {
1003 : 23 : ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
1004 : :
1005 : : /* prepTuplestoreResult must have been called previously */
1006 : : Assert(rsinfo->returnMode == SFRM_Materialize);
1007 : :
1008 : : /* Use a PG_TRY block to ensure we pump libpq dry of results */
1009 [ + - ]: 23 : PG_TRY();
1010 : : {
1011 : 23 : storeInfo sinfo = {0};
1012 : : PGresult *res;
1013 : :
1014 : 23 : sinfo.fcinfo = fcinfo;
1015 : : /* Create short-lived memory context for data conversions */
1016 : 23 : sinfo.tmpcontext = AllocSetContextCreate(CurrentMemoryContext,
1017 : : "dblink temporary context",
1018 : : ALLOCSET_DEFAULT_SIZES);
1019 : :
1020 : : /* execute query, collecting any tuples into the tuplestore */
1021 : 23 : res = storeQueryResult(&sinfo, conn, sql);
1022 : :
1023 [ + - ]: 23 : if (!res ||
1024 [ + - ]: 23 : (PQresultStatus(res) != PGRES_COMMAND_OK &&
1025 [ + + ]: 23 : PQresultStatus(res) != PGRES_TUPLES_OK))
1026 : : {
1027 : 2 : dblink_res_error(conn, conname, res, fail,
1028 : : "while executing query");
1029 : : /* if fail isn't set, we'll return an empty query result */
1030 : : }
1031 [ - + ]: 21 : else if (PQresultStatus(res) == PGRES_COMMAND_OK)
1032 : : {
1033 : : /*
1034 : : * storeRow didn't get called, so we need to convert the command
1035 : : * status string to a tuple manually
1036 : : */
1037 : : TupleDesc tupdesc;
1038 : : AttInMetadata *attinmeta;
1039 : : Tuplestorestate *tupstore;
1040 : : HeapTuple tuple;
1041 : : char *values[1];
1042 : : MemoryContext oldcontext;
1043 : :
1044 : : /*
1045 : : * need a tuple descriptor representing one TEXT column to return
1046 : : * the command status string as our result tuple
1047 : : */
1048 : 0 : tupdesc = CreateTemplateTupleDesc(1);
1049 : 0 : TupleDescInitEntry(tupdesc, (AttrNumber) 1, "status",
1050 : : TEXTOID, -1, 0);
1051 : 0 : TupleDescFinalize(tupdesc);
1052 : 0 : attinmeta = TupleDescGetAttInMetadata(tupdesc);
1053 : :
1054 : 0 : oldcontext = MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory);
1055 : 0 : tupstore = tuplestore_begin_heap(true, false, work_mem);
1056 : 0 : rsinfo->setResult = tupstore;
1057 : 0 : rsinfo->setDesc = tupdesc;
1058 : 0 : MemoryContextSwitchTo(oldcontext);
1059 : :
1060 : 0 : values[0] = PQcmdStatus(res);
1061 : :
1062 : : /* build the tuple and put it into the tuplestore. */
1063 : 0 : tuple = BuildTupleFromCStrings(attinmeta, values);
1064 : 0 : tuplestore_puttuple(tupstore, tuple);
1065 : :
1066 : 0 : PQclear(res);
1067 : : }
1068 : : else
1069 : : {
1070 : : Assert(PQresultStatus(res) == PGRES_TUPLES_OK);
1071 : : /* storeRow should have created a tuplestore */
1072 : : Assert(rsinfo->setResult != NULL);
1073 : :
1074 : 21 : PQclear(res);
1075 : : }
1076 : :
1077 : : /* clean up data conversion short-lived memory context */
1078 [ + - ]: 23 : if (sinfo.tmpcontext != NULL)
1079 : 23 : MemoryContextDelete(sinfo.tmpcontext);
1080 : :
1081 : 23 : PQclear(sinfo.last_res);
1082 : 23 : PQclear(sinfo.cur_res);
1083 : : }
1084 : 0 : PG_CATCH();
1085 : : {
1086 : : PGresult *res;
1087 : :
1088 : : /* be sure to clear out any pending data in libpq */
1089 [ # # ]: 0 : while ((res = libpqsrv_get_result(conn, dblink_we_get_result)) !=
1090 : : NULL)
1091 : 0 : PQclear(res);
1092 : 0 : PG_RE_THROW();
1093 : : }
1094 [ - + ]: 23 : PG_END_TRY();
1095 : 23 : }
1096 : :
1097 : : /*
1098 : : * Execute query, and send any result rows to sinfo->tuplestore.
1099 : : */
1100 : : static PGresult *
1101 : 23 : storeQueryResult(storeInfo *sinfo, PGconn *conn, const char *sql)
1102 : : {
1103 : 23 : bool first = true;
1104 : 23 : int nestlevel = -1;
1105 : : PGresult *res;
1106 : :
1107 [ - + ]: 23 : if (!PQsendQuery(conn, sql))
1108 [ # # ]: 0 : elog(ERROR, "could not send query: %s", pchomp(PQerrorMessage(conn)));
1109 : :
1110 [ - + ]: 23 : if (!PQsetSingleRowMode(conn)) /* shouldn't fail */
1111 [ # # ]: 0 : elog(ERROR, "failed to set single-row mode for dblink query");
1112 : :
1113 : : for (;;)
1114 : : {
1115 [ - + ]: 198 : CHECK_FOR_INTERRUPTS();
1116 : :
1117 : 198 : sinfo->cur_res = libpqsrv_get_result(conn, dblink_we_get_result);
1118 [ + + ]: 198 : if (!sinfo->cur_res)
1119 : 23 : break;
1120 : :
1121 [ + + ]: 175 : if (PQresultStatus(sinfo->cur_res) == PGRES_SINGLE_TUPLE)
1122 : : {
1123 : : /* got one row from possibly-bigger resultset */
1124 : :
1125 : : /*
1126 : : * Set GUCs to ensure we read GUC-sensitive data types correctly.
1127 : : * We shouldn't do this until we have a row in hand, to ensure
1128 : : * libpq has seen any earlier ParameterStatus protocol messages.
1129 : : */
1130 [ + + + - ]: 152 : if (first && nestlevel < 0)
1131 : 21 : nestlevel = applyRemoteGucs(conn);
1132 : :
1133 : 152 : storeRow(sinfo, sinfo->cur_res, first);
1134 : :
1135 : 152 : PQclear(sinfo->cur_res);
1136 : 152 : sinfo->cur_res = NULL;
1137 : 152 : first = false;
1138 : : }
1139 : : else
1140 : : {
1141 : : /* if empty resultset, fill tuplestore header */
1142 [ + + - + ]: 23 : if (first && PQresultStatus(sinfo->cur_res) == PGRES_TUPLES_OK)
1143 : 0 : storeRow(sinfo, sinfo->cur_res, first);
1144 : :
1145 : : /* store completed result at last_res */
1146 : 23 : PQclear(sinfo->last_res);
1147 : 23 : sinfo->last_res = sinfo->cur_res;
1148 : 23 : sinfo->cur_res = NULL;
1149 : 23 : first = true;
1150 : : }
1151 : : }
1152 : :
1153 : : /* clean up GUC settings, if we changed any */
1154 : 23 : restoreLocalGucs(nestlevel);
1155 : :
1156 : : /* return last_res */
1157 : 23 : res = sinfo->last_res;
1158 : 23 : sinfo->last_res = NULL;
1159 : 23 : return res;
1160 : : }
1161 : :
1162 : : /*
1163 : : * Send single row to sinfo->tuplestore.
1164 : : *
1165 : : * If "first" is true, create the tuplestore using PGresult's metadata
1166 : : * (in this case the PGresult might contain either zero or one row).
1167 : : */
1168 : : static void
1169 : 152 : storeRow(storeInfo *sinfo, PGresult *res, bool first)
1170 : : {
1171 : 152 : int nfields = PQnfields(res);
1172 : : HeapTuple tuple;
1173 : : int i;
1174 : : MemoryContext oldcontext;
1175 : :
1176 [ + + ]: 152 : if (first)
1177 : : {
1178 : : /* Prepare for new result set */
1179 : 21 : ReturnSetInfo *rsinfo = (ReturnSetInfo *) sinfo->fcinfo->resultinfo;
1180 : : TupleDesc tupdesc;
1181 : :
1182 : : /*
1183 : : * It's possible to get more than one result set if the query string
1184 : : * contained multiple SQL commands. In that case, we follow PQexec's
1185 : : * traditional behavior of throwing away all but the last result.
1186 : : */
1187 [ - + ]: 21 : if (sinfo->tuplestore)
1188 : 0 : tuplestore_end(sinfo->tuplestore);
1189 : 21 : sinfo->tuplestore = NULL;
1190 : :
1191 : : /* get a tuple descriptor for our result type */
1192 [ + - - ]: 21 : switch (get_call_result_type(sinfo->fcinfo, NULL, &tupdesc))
1193 : : {
1194 : 21 : case TYPEFUNC_COMPOSITE:
1195 : : /* success */
1196 : 21 : break;
1197 : 0 : case TYPEFUNC_RECORD:
1198 : : /* failed to determine actual type of RECORD */
1199 [ # # ]: 0 : ereport(ERROR,
1200 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1201 : : errmsg("function returning record called in context "
1202 : : "that cannot accept type record")));
1203 : : break;
1204 : 0 : default:
1205 : : /* result type isn't composite */
1206 [ # # ]: 0 : elog(ERROR, "return type must be a row type");
1207 : : break;
1208 : : }
1209 : :
1210 : : /* make sure we have a persistent copy of the tupdesc */
1211 : 21 : tupdesc = CreateTupleDescCopy(tupdesc);
1212 : :
1213 : : /* check result and tuple descriptor have the same number of columns */
1214 [ - + ]: 21 : if (nfields != tupdesc->natts)
1215 [ # # ]: 0 : ereport(ERROR,
1216 : : (errcode(ERRCODE_DATATYPE_MISMATCH),
1217 : : errmsg("remote query result rowtype does not match "
1218 : : "the specified FROM clause rowtype")));
1219 : :
1220 : : /* Prepare attinmeta for later data conversions */
1221 : 21 : sinfo->attinmeta = TupleDescGetAttInMetadata(tupdesc);
1222 : :
1223 : : /* Create a new, empty tuplestore */
1224 : 21 : oldcontext = MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory);
1225 : 21 : sinfo->tuplestore = tuplestore_begin_heap(true, false, work_mem);
1226 : 21 : rsinfo->setResult = sinfo->tuplestore;
1227 : 21 : rsinfo->setDesc = tupdesc;
1228 : 21 : MemoryContextSwitchTo(oldcontext);
1229 : :
1230 : : /* Done if empty resultset */
1231 [ - + ]: 21 : if (PQntuples(res) == 0)
1232 : 0 : return;
1233 : :
1234 : : /*
1235 : : * Set up sufficiently-wide string pointers array; this won't change
1236 : : * in size so it's easy to preallocate.
1237 : : */
1238 [ - + ]: 21 : if (sinfo->cstrs)
1239 : 0 : pfree(sinfo->cstrs);
1240 : 21 : sinfo->cstrs = palloc_array(char *, nfields);
1241 : : }
1242 : :
1243 : : /* Should have a single-row result if we get here */
1244 : : Assert(PQntuples(res) == 1);
1245 : :
1246 : : /*
1247 : : * Do the following work in a temp context that we reset after each tuple.
1248 : : * This cleans up not only the data we have direct access to, but any
1249 : : * cruft the I/O functions might leak.
1250 : : */
1251 : 152 : oldcontext = MemoryContextSwitchTo(sinfo->tmpcontext);
1252 : :
1253 : : /*
1254 : : * Fill cstrs with null-terminated strings of column values.
1255 : : */
1256 [ + + ]: 570 : for (i = 0; i < nfields; i++)
1257 : : {
1258 [ - + ]: 418 : if (PQgetisnull(res, 0, i))
1259 : 0 : sinfo->cstrs[i] = NULL;
1260 : : else
1261 : 418 : sinfo->cstrs[i] = PQgetvalue(res, 0, i);
1262 : : }
1263 : :
1264 : : /* Convert row to a tuple, and add it to the tuplestore */
1265 : 152 : tuple = BuildTupleFromCStrings(sinfo->attinmeta, sinfo->cstrs);
1266 : :
1267 : 152 : tuplestore_puttuple(sinfo->tuplestore, tuple);
1268 : :
1269 : : /* Clean up */
1270 : 152 : MemoryContextSwitchTo(oldcontext);
1271 : 152 : MemoryContextReset(sinfo->tmpcontext);
1272 : : }
1273 : :
1274 : : /*
1275 : : * List all open dblink connections by name.
1276 : : * Returns an array of all connection names.
1277 : : * Takes no params
1278 : : */
1279 : 3 : PG_FUNCTION_INFO_V1(dblink_get_connections);
1280 : : Datum
1281 : 1 : dblink_get_connections(PG_FUNCTION_ARGS)
1282 : : {
1283 : : HASH_SEQ_STATUS status;
1284 : : remoteConnHashEnt *hentry;
1285 : 1 : ArrayBuildState *astate = NULL;
1286 : :
1287 [ + - ]: 1 : if (remoteConnHash)
1288 : : {
1289 : 1 : hash_seq_init(&status, remoteConnHash);
1290 [ + + ]: 4 : while ((hentry = (remoteConnHashEnt *) hash_seq_search(&status)) != NULL)
1291 : : {
1292 : : /* ignore it if it's not an open connection */
1293 [ - + ]: 3 : if (hentry->rconn.conn == NULL)
1294 : 0 : continue;
1295 : : /* stash away current value */
1296 : 3 : astate = accumArrayResult(astate,
1297 : 3 : CStringGetTextDatum(hentry->name),
1298 : : false, TEXTOID, CurrentMemoryContext);
1299 : : }
1300 : : }
1301 : :
1302 [ + - ]: 1 : if (astate)
1303 : 1 : PG_RETURN_DATUM(makeArrayResult(astate,
1304 : : CurrentMemoryContext));
1305 : : else
1306 : 0 : PG_RETURN_NULL();
1307 : : }
1308 : :
1309 : : /*
1310 : : * Checks if a given remote connection is busy
1311 : : *
1312 : : * Returns 1 if the connection is busy, 0 otherwise
1313 : : * Params:
1314 : : * text connection_name - name of the connection to check
1315 : : *
1316 : : */
1317 : 3 : PG_FUNCTION_INFO_V1(dblink_is_busy);
1318 : : Datum
1319 : 1 : dblink_is_busy(PG_FUNCTION_ARGS)
1320 : : {
1321 : : PGconn *conn;
1322 : :
1323 : 1 : dblink_init();
1324 : 1 : conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
1325 : :
1326 : 1 : PQconsumeInput(conn);
1327 : 1 : PG_RETURN_INT32(PQisBusy(conn));
1328 : : }
1329 : :
1330 : : /*
1331 : : * Cancels a running request on a connection
1332 : : *
1333 : : * Returns text:
1334 : : * "OK" if the cancel request has been sent correctly,
1335 : : * an error message otherwise
1336 : : *
1337 : : * Params:
1338 : : * text connection_name - name of the connection to check
1339 : : *
1340 : : */
1341 : 3 : PG_FUNCTION_INFO_V1(dblink_cancel_query);
1342 : : Datum
1343 : 1 : dblink_cancel_query(PG_FUNCTION_ARGS)
1344 : : {
1345 : : PGconn *conn;
1346 : : const char *msg;
1347 : : TimestampTz endtime;
1348 : :
1349 : 1 : dblink_init();
1350 : 1 : conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
1351 : 1 : endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(),
1352 : : 30000);
1353 : 1 : msg = libpqsrv_cancel(conn, endtime);
1354 [ + - ]: 1 : if (msg == NULL)
1355 : 1 : msg = "OK";
1356 : :
1357 : 1 : PG_RETURN_TEXT_P(cstring_to_text(msg));
1358 : : }
1359 : :
1360 : :
1361 : : /*
1362 : : * Get error message from a connection
1363 : : *
1364 : : * Returns text:
1365 : : * "OK" if no error, an error message otherwise
1366 : : *
1367 : : * Params:
1368 : : * text connection_name - name of the connection to check
1369 : : *
1370 : : */
1371 : 3 : PG_FUNCTION_INFO_V1(dblink_error_message);
1372 : : Datum
1373 : 1 : dblink_error_message(PG_FUNCTION_ARGS)
1374 : : {
1375 : : char *msg;
1376 : : PGconn *conn;
1377 : :
1378 : 1 : dblink_init();
1379 : 1 : conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
1380 : :
1381 : 1 : msg = PQerrorMessage(conn);
1382 [ + - + - ]: 1 : if (msg == NULL || msg[0] == '\0')
1383 : 1 : PG_RETURN_TEXT_P(cstring_to_text("OK"));
1384 : : else
1385 : 0 : PG_RETURN_TEXT_P(cstring_to_text(pchomp(msg)));
1386 : : }
1387 : :
1388 : : /*
1389 : : * Execute an SQL non-SELECT command
1390 : : */
1391 : 13 : PG_FUNCTION_INFO_V1(dblink_exec);
1392 : : Datum
1393 : 26 : dblink_exec(PG_FUNCTION_ARGS)
1394 : : {
1395 : 26 : text *volatile sql_cmd_status = NULL;
1396 : 26 : PGconn *volatile conn = NULL;
1397 : 26 : volatile bool freeconn = false;
1398 : :
1399 : 26 : dblink_init();
1400 : :
1401 [ + + ]: 26 : PG_TRY();
1402 : : {
1403 : 26 : PGresult *res = NULL;
1404 : 26 : char *sql = NULL;
1405 : 26 : char *conname = NULL;
1406 : 26 : bool fail = true; /* default to backward compatible behavior */
1407 : :
1408 [ - + ]: 26 : if (PG_NARGS() == 3)
1409 : : {
1410 : : /* must be text,text,bool */
1411 : 0 : conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
1412 : 0 : sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
1413 : 0 : fail = PG_GETARG_BOOL(2);
1414 : 0 : dblink_get_conn(conname, &conn, &conname, &freeconn);
1415 : : }
1416 [ + + ]: 26 : else if (PG_NARGS() == 2)
1417 : : {
1418 : : /* might be text,text or text,bool */
1419 [ + + ]: 17 : if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
1420 : : {
1421 : 1 : sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
1422 : 1 : fail = PG_GETARG_BOOL(1);
1423 : 1 : conn = pconn->conn;
1424 : : }
1425 : : else
1426 : : {
1427 : 16 : conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
1428 : 16 : sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
1429 : 16 : dblink_get_conn(conname, &conn, &conname, &freeconn);
1430 : : }
1431 : : }
1432 [ + - ]: 9 : else if (PG_NARGS() == 1)
1433 : : {
1434 : : /* must be single text argument */
1435 : 9 : conn = pconn->conn;
1436 : 9 : sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
1437 : : }
1438 : : else
1439 : : /* shouldn't happen */
1440 [ # # ]: 0 : elog(ERROR, "wrong number of arguments");
1441 : :
1442 [ - + ]: 26 : if (!conn)
1443 : 0 : dblink_conn_not_avail(conname);
1444 : :
1445 : 26 : res = libpqsrv_exec(conn, sql, dblink_we_get_result);
1446 [ + - ]: 26 : if (!res ||
1447 [ + + ]: 26 : (PQresultStatus(res) != PGRES_COMMAND_OK &&
1448 [ + - ]: 2 : PQresultStatus(res) != PGRES_TUPLES_OK))
1449 : : {
1450 : 2 : dblink_res_error(conn, conname, res, fail,
1451 : : "while executing command");
1452 : :
1453 : : /*
1454 : : * and save a copy of the command status string to return as our
1455 : : * result tuple
1456 : : */
1457 : 1 : sql_cmd_status = cstring_to_text("ERROR");
1458 : : }
1459 [ + - ]: 24 : else if (PQresultStatus(res) == PGRES_COMMAND_OK)
1460 : : {
1461 : : /*
1462 : : * and save a copy of the command status string to return as our
1463 : : * result tuple
1464 : : */
1465 : 24 : sql_cmd_status = cstring_to_text(PQcmdStatus(res));
1466 : 24 : PQclear(res);
1467 : : }
1468 : : else
1469 : : {
1470 : 0 : PQclear(res);
1471 [ # # ]: 0 : ereport(ERROR,
1472 : : (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
1473 : : errmsg("statement returning results not allowed")));
1474 : : }
1475 : : }
1476 : 1 : PG_FINALLY();
1477 : : {
1478 : : /* if needed, close the connection to the database */
1479 [ + + ]: 26 : if (freeconn)
1480 : 1 : libpqsrv_disconnect(conn);
1481 : : }
1482 [ + + ]: 26 : PG_END_TRY();
1483 : :
1484 : 25 : PG_RETURN_TEXT_P(sql_cmd_status);
1485 : : }
1486 : :
1487 : :
1488 : : /*
1489 : : * dblink_get_pkey
1490 : : *
1491 : : * Return list of primary key fields for the supplied relation,
1492 : : * or NULL if none exists.
1493 : : */
1494 : 3 : PG_FUNCTION_INFO_V1(dblink_get_pkey);
1495 : : Datum
1496 : 9 : dblink_get_pkey(PG_FUNCTION_ARGS)
1497 : : {
1498 : : int16 indnkeyatts;
1499 : : char **results;
1500 : : FuncCallContext *funcctx;
1501 : : int32 call_cntr;
1502 : : int32 max_calls;
1503 : : AttInMetadata *attinmeta;
1504 : : MemoryContext oldcontext;
1505 : :
1506 : : /* stuff done only on the first call of the function */
1507 [ + + ]: 9 : if (SRF_IS_FIRSTCALL())
1508 : : {
1509 : : Relation rel;
1510 : : TupleDesc tupdesc;
1511 : :
1512 : : /* create a function context for cross-call persistence */
1513 : 3 : funcctx = SRF_FIRSTCALL_INIT();
1514 : :
1515 : : /*
1516 : : * switch to memory context appropriate for multiple function calls
1517 : : */
1518 : 3 : oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
1519 : :
1520 : : /* open target relation */
1521 : 3 : rel = get_rel_from_relname(PG_GETARG_TEXT_PP(0), AccessShareLock, ACL_SELECT);
1522 : :
1523 : : /* get the array of attnums */
1524 : 3 : results = get_pkey_attnames(rel, &indnkeyatts);
1525 : :
1526 : 3 : relation_close(rel, AccessShareLock);
1527 : :
1528 : : /*
1529 : : * need a tuple descriptor representing one INT and one TEXT column
1530 : : */
1531 : 3 : tupdesc = CreateTemplateTupleDesc(2);
1532 : 3 : TupleDescInitEntry(tupdesc, (AttrNumber) 1, "position",
1533 : : INT4OID, -1, 0);
1534 : 3 : TupleDescInitEntry(tupdesc, (AttrNumber) 2, "colname",
1535 : : TEXTOID, -1, 0);
1536 : :
1537 : 3 : TupleDescFinalize(tupdesc);
1538 : :
1539 : : /*
1540 : : * Generate attribute metadata needed later to produce tuples from raw
1541 : : * C strings
1542 : : */
1543 : 3 : attinmeta = TupleDescGetAttInMetadata(tupdesc);
1544 : 3 : funcctx->attinmeta = attinmeta;
1545 : :
1546 [ + - + - ]: 3 : if ((results != NULL) && (indnkeyatts > 0))
1547 : : {
1548 : 3 : funcctx->max_calls = indnkeyatts;
1549 : :
1550 : : /* got results, keep track of them */
1551 : 3 : funcctx->user_fctx = results;
1552 : : }
1553 : : else
1554 : : {
1555 : : /* fast track when no results */
1556 : 0 : MemoryContextSwitchTo(oldcontext);
1557 : 0 : SRF_RETURN_DONE(funcctx);
1558 : : }
1559 : :
1560 : 3 : MemoryContextSwitchTo(oldcontext);
1561 : : }
1562 : :
1563 : : /* stuff done on every call of the function */
1564 : 9 : funcctx = SRF_PERCALL_SETUP();
1565 : :
1566 : : /*
1567 : : * initialize per-call variables
1568 : : */
1569 : 9 : call_cntr = funcctx->call_cntr;
1570 : 9 : max_calls = funcctx->max_calls;
1571 : :
1572 : 9 : results = (char **) funcctx->user_fctx;
1573 : 9 : attinmeta = funcctx->attinmeta;
1574 : :
1575 [ + + ]: 9 : if (call_cntr < max_calls) /* do when there is more left to send */
1576 : : {
1577 : : char **values;
1578 : : HeapTuple tuple;
1579 : : Datum result;
1580 : :
1581 : 6 : values = palloc_array(char *, 2);
1582 : 6 : values[0] = psprintf("%d", call_cntr + 1);
1583 : 6 : values[1] = results[call_cntr];
1584 : :
1585 : : /* build the tuple */
1586 : 6 : tuple = BuildTupleFromCStrings(attinmeta, values);
1587 : :
1588 : : /* make the tuple into a datum */
1589 : 6 : result = HeapTupleGetDatum(tuple);
1590 : :
1591 : 6 : SRF_RETURN_NEXT(funcctx, result);
1592 : : }
1593 : : else
1594 : : {
1595 : : /* do when there is no more left */
1596 : 3 : SRF_RETURN_DONE(funcctx);
1597 : : }
1598 : : }
1599 : :
1600 : :
1601 : : /*
1602 : : * dblink_build_sql_insert
1603 : : *
1604 : : * Used to generate an SQL insert statement
1605 : : * based on an existing tuple in a local relation.
1606 : : * This is useful for selectively replicating data
1607 : : * to another server via dblink.
1608 : : *
1609 : : * API:
1610 : : * <relname> - name of local table of interest
1611 : : * <pkattnums> - an int2vector of attnums which will be used
1612 : : * to identify the local tuple of interest
1613 : : * <pknumatts> - number of attnums in pkattnums
1614 : : * <src_pkattvals_arry> - text array of key values which will be used
1615 : : * to identify the local tuple of interest
1616 : : * <tgt_pkattvals_arry> - text array of key values which will be used
1617 : : * to build the string for execution remotely. These are substituted
1618 : : * for their counterparts in src_pkattvals_arry
1619 : : */
1620 : 4 : PG_FUNCTION_INFO_V1(dblink_build_sql_insert);
1621 : : Datum
1622 : 6 : dblink_build_sql_insert(PG_FUNCTION_ARGS)
1623 : : {
1624 : 6 : text *relname_text = PG_GETARG_TEXT_PP(0);
1625 : 6 : int2vector *pkattnums_arg = (int2vector *) PG_GETARG_POINTER(1);
1626 : 6 : int32 pknumatts_arg = PG_GETARG_INT32(2);
1627 : 6 : ArrayType *src_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
1628 : 6 : ArrayType *tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(4);
1629 : : Relation rel;
1630 : : int *pkattnums;
1631 : : int pknumatts;
1632 : : char **src_pkattvals;
1633 : : char **tgt_pkattvals;
1634 : : int src_nitems;
1635 : : int tgt_nitems;
1636 : : char *sql;
1637 : :
1638 : : /*
1639 : : * Open target relation.
1640 : : */
1641 : 6 : rel = get_rel_from_relname(relname_text, AccessShareLock, ACL_SELECT);
1642 : :
1643 : : /*
1644 : : * Process pkattnums argument.
1645 : : */
1646 : 6 : validate_pkattnums(rel, pkattnums_arg, pknumatts_arg,
1647 : : &pkattnums, &pknumatts);
1648 : :
1649 : : /*
1650 : : * Source array is made up of key values that will be used to locate the
1651 : : * tuple of interest from the local system.
1652 : : */
1653 : 4 : src_pkattvals = get_text_array_contents(src_pkattvals_arry, &src_nitems);
1654 : :
1655 : : /*
1656 : : * There should be one source array key value for each key attnum
1657 : : */
1658 [ - + ]: 4 : if (src_nitems != pknumatts)
1659 [ # # ]: 0 : ereport(ERROR,
1660 : : (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
1661 : : errmsg("source key array length must match number of key attributes")));
1662 : :
1663 : : /*
1664 : : * Target array is made up of key values that will be used to build the
1665 : : * SQL string for use on the remote system.
1666 : : */
1667 : 4 : tgt_pkattvals = get_text_array_contents(tgt_pkattvals_arry, &tgt_nitems);
1668 : :
1669 : : /*
1670 : : * There should be one target array key value for each key attnum
1671 : : */
1672 [ - + ]: 4 : if (tgt_nitems != pknumatts)
1673 [ # # ]: 0 : ereport(ERROR,
1674 : : (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
1675 : : errmsg("target key array length must match number of key attributes")));
1676 : :
1677 : : /*
1678 : : * Prep work is finally done. Go get the SQL string.
1679 : : */
1680 : 4 : sql = get_sql_insert(rel, pkattnums, pknumatts, src_pkattvals, tgt_pkattvals);
1681 : :
1682 : : /*
1683 : : * Now we can close the relation.
1684 : : */
1685 : 4 : relation_close(rel, AccessShareLock);
1686 : :
1687 : : /*
1688 : : * And send it
1689 : : */
1690 : 4 : PG_RETURN_TEXT_P(cstring_to_text(sql));
1691 : : }
1692 : :
1693 : :
1694 : : /*
1695 : : * dblink_build_sql_delete
1696 : : *
1697 : : * Used to generate an SQL delete statement.
1698 : : * This is useful for selectively replicating a
1699 : : * delete to another server via dblink.
1700 : : *
1701 : : * API:
1702 : : * <relname> - name of remote table of interest
1703 : : * <pkattnums> - an int2vector of attnums which will be used
1704 : : * to identify the remote tuple of interest
1705 : : * <pknumatts> - number of attnums in pkattnums
1706 : : * <tgt_pkattvals_arry> - text array of key values which will be used
1707 : : * to build the string for execution remotely.
1708 : : */
1709 : 4 : PG_FUNCTION_INFO_V1(dblink_build_sql_delete);
1710 : : Datum
1711 : 6 : dblink_build_sql_delete(PG_FUNCTION_ARGS)
1712 : : {
1713 : 6 : text *relname_text = PG_GETARG_TEXT_PP(0);
1714 : 6 : int2vector *pkattnums_arg = (int2vector *) PG_GETARG_POINTER(1);
1715 : 6 : int32 pknumatts_arg = PG_GETARG_INT32(2);
1716 : 6 : ArrayType *tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
1717 : : Relation rel;
1718 : : int *pkattnums;
1719 : : int pknumatts;
1720 : : char **tgt_pkattvals;
1721 : : int tgt_nitems;
1722 : : char *sql;
1723 : :
1724 : : /*
1725 : : * Open target relation.
1726 : : */
1727 : 6 : rel = get_rel_from_relname(relname_text, AccessShareLock, ACL_SELECT);
1728 : :
1729 : : /*
1730 : : * Process pkattnums argument.
1731 : : */
1732 : 6 : validate_pkattnums(rel, pkattnums_arg, pknumatts_arg,
1733 : : &pkattnums, &pknumatts);
1734 : :
1735 : : /*
1736 : : * Target array is made up of key values that will be used to build the
1737 : : * SQL string for use on the remote system.
1738 : : */
1739 : 4 : tgt_pkattvals = get_text_array_contents(tgt_pkattvals_arry, &tgt_nitems);
1740 : :
1741 : : /*
1742 : : * There should be one target array key value for each key attnum
1743 : : */
1744 [ - + ]: 4 : if (tgt_nitems != pknumatts)
1745 [ # # ]: 0 : ereport(ERROR,
1746 : : (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
1747 : : errmsg("target key array length must match number of key attributes")));
1748 : :
1749 : : /*
1750 : : * Prep work is finally done. Go get the SQL string.
1751 : : */
1752 : 4 : sql = get_sql_delete(rel, pkattnums, pknumatts, tgt_pkattvals);
1753 : :
1754 : : /*
1755 : : * Now we can close the relation.
1756 : : */
1757 : 4 : relation_close(rel, AccessShareLock);
1758 : :
1759 : : /*
1760 : : * And send it
1761 : : */
1762 : 4 : PG_RETURN_TEXT_P(cstring_to_text(sql));
1763 : : }
1764 : :
1765 : :
1766 : : /*
1767 : : * dblink_build_sql_update
1768 : : *
1769 : : * Used to generate an SQL update statement
1770 : : * based on an existing tuple in a local relation.
1771 : : * This is useful for selectively replicating data
1772 : : * to another server via dblink.
1773 : : *
1774 : : * API:
1775 : : * <relname> - name of local table of interest
1776 : : * <pkattnums> - an int2vector of attnums which will be used
1777 : : * to identify the local tuple of interest
1778 : : * <pknumatts> - number of attnums in pkattnums
1779 : : * <src_pkattvals_arry> - text array of key values which will be used
1780 : : * to identify the local tuple of interest
1781 : : * <tgt_pkattvals_arry> - text array of key values which will be used
1782 : : * to build the string for execution remotely. These are substituted
1783 : : * for their counterparts in src_pkattvals_arry
1784 : : */
1785 : 4 : PG_FUNCTION_INFO_V1(dblink_build_sql_update);
1786 : : Datum
1787 : 6 : dblink_build_sql_update(PG_FUNCTION_ARGS)
1788 : : {
1789 : 6 : text *relname_text = PG_GETARG_TEXT_PP(0);
1790 : 6 : int2vector *pkattnums_arg = (int2vector *) PG_GETARG_POINTER(1);
1791 : 6 : int32 pknumatts_arg = PG_GETARG_INT32(2);
1792 : 6 : ArrayType *src_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
1793 : 6 : ArrayType *tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(4);
1794 : : Relation rel;
1795 : : int *pkattnums;
1796 : : int pknumatts;
1797 : : char **src_pkattvals;
1798 : : char **tgt_pkattvals;
1799 : : int src_nitems;
1800 : : int tgt_nitems;
1801 : : char *sql;
1802 : :
1803 : : /*
1804 : : * Open target relation.
1805 : : */
1806 : 6 : rel = get_rel_from_relname(relname_text, AccessShareLock, ACL_SELECT);
1807 : :
1808 : : /*
1809 : : * Process pkattnums argument.
1810 : : */
1811 : 6 : validate_pkattnums(rel, pkattnums_arg, pknumatts_arg,
1812 : : &pkattnums, &pknumatts);
1813 : :
1814 : : /*
1815 : : * Source array is made up of key values that will be used to locate the
1816 : : * tuple of interest from the local system.
1817 : : */
1818 : 4 : src_pkattvals = get_text_array_contents(src_pkattvals_arry, &src_nitems);
1819 : :
1820 : : /*
1821 : : * There should be one source array key value for each key attnum
1822 : : */
1823 [ - + ]: 4 : if (src_nitems != pknumatts)
1824 [ # # ]: 0 : ereport(ERROR,
1825 : : (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
1826 : : errmsg("source key array length must match number of key attributes")));
1827 : :
1828 : : /*
1829 : : * Target array is made up of key values that will be used to build the
1830 : : * SQL string for use on the remote system.
1831 : : */
1832 : 4 : tgt_pkattvals = get_text_array_contents(tgt_pkattvals_arry, &tgt_nitems);
1833 : :
1834 : : /*
1835 : : * There should be one target array key value for each key attnum
1836 : : */
1837 [ - + ]: 4 : if (tgt_nitems != pknumatts)
1838 [ # # ]: 0 : ereport(ERROR,
1839 : : (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
1840 : : errmsg("target key array length must match number of key attributes")));
1841 : :
1842 : : /*
1843 : : * Prep work is finally done. Go get the SQL string.
1844 : : */
1845 : 4 : sql = get_sql_update(rel, pkattnums, pknumatts, src_pkattvals, tgt_pkattvals);
1846 : :
1847 : : /*
1848 : : * Now we can close the relation.
1849 : : */
1850 : 4 : relation_close(rel, AccessShareLock);
1851 : :
1852 : : /*
1853 : : * And send it
1854 : : */
1855 : 4 : PG_RETURN_TEXT_P(cstring_to_text(sql));
1856 : : }
1857 : :
1858 : : /*
1859 : : * dblink_current_query
1860 : : * return the current query string
1861 : : * to allow its use in (among other things)
1862 : : * rewrite rules
1863 : : */
1864 : 2 : PG_FUNCTION_INFO_V1(dblink_current_query);
1865 : : Datum
1866 : 0 : dblink_current_query(PG_FUNCTION_ARGS)
1867 : : {
1868 : : /* This is now just an alias for the built-in function current_query() */
1869 : 0 : PG_RETURN_DATUM(current_query(fcinfo));
1870 : : }
1871 : :
1872 : : /*
1873 : : * Retrieve async notifications for a connection.
1874 : : *
1875 : : * Returns a setof record of notifications, or an empty set if none received.
1876 : : * Can optionally take a named connection as parameter, but uses the unnamed
1877 : : * connection per default.
1878 : : *
1879 : : */
1880 : : #define DBLINK_NOTIFY_COLS 3
1881 : :
1882 : 5 : PG_FUNCTION_INFO_V1(dblink_get_notify);
1883 : : Datum
1884 : 2 : dblink_get_notify(PG_FUNCTION_ARGS)
1885 : : {
1886 : : PGconn *conn;
1887 : : PGnotify *notify;
1888 : 2 : ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
1889 : :
1890 : 2 : dblink_init();
1891 [ - + ]: 2 : if (PG_NARGS() == 1)
1892 : 0 : conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
1893 : : else
1894 : 2 : conn = pconn->conn;
1895 : :
1896 : 2 : InitMaterializedSRF(fcinfo, 0);
1897 : :
1898 : 2 : PQconsumeInput(conn);
1899 [ + + ]: 4 : while ((notify = PQnotifies(conn)) != NULL)
1900 : : {
1901 : : Datum values[DBLINK_NOTIFY_COLS];
1902 : : bool nulls[DBLINK_NOTIFY_COLS];
1903 : :
1904 : 2 : memset(values, 0, sizeof(values));
1905 : 2 : memset(nulls, 0, sizeof(nulls));
1906 : :
1907 [ + - ]: 2 : if (notify->relname != NULL)
1908 : 2 : values[0] = CStringGetTextDatum(notify->relname);
1909 : : else
1910 : 0 : nulls[0] = true;
1911 : :
1912 : 2 : values[1] = Int32GetDatum(notify->be_pid);
1913 : :
1914 [ + - ]: 2 : if (notify->extra != NULL)
1915 : 2 : values[2] = CStringGetTextDatum(notify->extra);
1916 : : else
1917 : 0 : nulls[2] = true;
1918 : :
1919 : 2 : tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls);
1920 : :
1921 : 2 : PQfreemem(notify);
1922 : 2 : PQconsumeInput(conn);
1923 : : }
1924 : :
1925 : 2 : return (Datum) 0;
1926 : : }
1927 : :
1928 : : /*
1929 : : * Validate the options given to a dblink foreign server or user mapping.
1930 : : * Raise an error if any option is invalid.
1931 : : *
1932 : : * We just check the names of options here, so semantic errors in options,
1933 : : * such as invalid numeric format, will be detected at the attempt to connect.
1934 : : */
1935 : 17 : PG_FUNCTION_INFO_V1(dblink_fdw_validator);
1936 : : Datum
1937 : 23 : dblink_fdw_validator(PG_FUNCTION_ARGS)
1938 : : {
1939 : 23 : List *options_list = untransformRelOptions(PG_GETARG_DATUM(0));
1940 : 23 : Oid context = PG_GETARG_OID(1);
1941 : : ListCell *cell;
1942 : :
1943 : : static const PQconninfoOption *options = NULL;
1944 : :
1945 : : /*
1946 : : * Get list of valid libpq options.
1947 : : *
1948 : : * To avoid unnecessary work, we get the list once and use it throughout
1949 : : * the lifetime of this backend process. We don't need to care about
1950 : : * memory context issues, because PQconndefaults allocates with malloc.
1951 : : */
1952 [ + + ]: 23 : if (!options)
1953 : : {
1954 : 15 : options = PQconndefaults();
1955 [ - + ]: 15 : if (!options) /* assume reason for failure is OOM */
1956 [ # # ]: 0 : ereport(ERROR,
1957 : : (errcode(ERRCODE_FDW_OUT_OF_MEMORY),
1958 : : errmsg("out of memory"),
1959 : : errdetail("Could not get libpq's default connection options.")));
1960 : : }
1961 : :
1962 : : /* Validate each supplied option. */
1963 [ + + + + : 61 : foreach(cell, options_list)
+ + ]
1964 : : {
1965 : 47 : DefElem *def = (DefElem *) lfirst(cell);
1966 : :
1967 [ + + ]: 47 : if (!is_valid_dblink_fdw_option(options, def->defname, context))
1968 : : {
1969 : : /*
1970 : : * Unknown option, or invalid option for the context specified, so
1971 : : * complain about it. Provide a hint with a valid option that
1972 : : * looks similar, if there is one.
1973 : : */
1974 : : const PQconninfoOption *opt;
1975 : : const char *closest_match;
1976 : : ClosestMatchState match_state;
1977 : 9 : bool has_valid_options = false;
1978 : :
1979 : 9 : initClosestMatch(&match_state, def->defname, 4);
1980 [ + + ]: 477 : for (opt = options; opt->keyword; opt++)
1981 : : {
1982 [ + + ]: 468 : if (is_valid_dblink_option(options, opt->keyword, context))
1983 : : {
1984 : 93 : has_valid_options = true;
1985 : 93 : updateClosestMatch(&match_state, opt->keyword);
1986 : : }
1987 : : }
1988 : :
1989 : 9 : closest_match = getClosestMatch(&match_state);
1990 [ + - + + : 9 : ereport(ERROR,
- + ]
1991 : : (errcode(ERRCODE_FDW_OPTION_NAME_NOT_FOUND),
1992 : : errmsg("invalid option \"%s\"", def->defname),
1993 : : has_valid_options ? closest_match ?
1994 : : errhint("Perhaps you meant the option \"%s\".",
1995 : : closest_match) : 0 :
1996 : : errhint("There are no valid options in this context.")));
1997 : : }
1998 : :
1999 [ + + ]: 38 : if (strcmp(def->defname, "use_scram_passthrough") == 0)
2000 : 6 : (void) defGetBoolean(def); /* accept only boolean values */
2001 : : }
2002 : :
2003 : 14 : PG_RETURN_VOID();
2004 : : }
2005 : :
2006 : :
2007 : : /*************************************************************
2008 : : * internal functions
2009 : : */
2010 : :
2011 : :
2012 : : /*
2013 : : * get_pkey_attnames
2014 : : *
2015 : : * Get the primary key attnames for the given relation.
2016 : : * Return NULL, and set indnkeyatts = 0, if no primary key exists.
2017 : : */
2018 : : static char **
2019 : 3 : get_pkey_attnames(Relation rel, int16 *indnkeyatts)
2020 : : {
2021 : : Relation indexRelation;
2022 : : ScanKeyData skey;
2023 : : SysScanDesc scan;
2024 : : HeapTuple indexTuple;
2025 : : int i;
2026 : 3 : char **result = NULL;
2027 : : TupleDesc tupdesc;
2028 : :
2029 : : /* initialize indnkeyatts to 0 in case no primary key exists */
2030 : 3 : *indnkeyatts = 0;
2031 : :
2032 : 3 : tupdesc = rel->rd_att;
2033 : :
2034 : : /* Prepare to scan pg_index for entries having indrelid = this rel. */
2035 : 3 : indexRelation = table_open(IndexRelationId, AccessShareLock);
2036 : 3 : ScanKeyInit(&skey,
2037 : : Anum_pg_index_indrelid,
2038 : : BTEqualStrategyNumber, F_OIDEQ,
2039 : : ObjectIdGetDatum(RelationGetRelid(rel)));
2040 : :
2041 : 3 : scan = systable_beginscan(indexRelation, IndexIndrelidIndexId, true,
2042 : : NULL, 1, &skey);
2043 : :
2044 [ + - ]: 3 : while (HeapTupleIsValid(indexTuple = systable_getnext(scan)))
2045 : : {
2046 : 3 : Form_pg_index index = (Form_pg_index) GETSTRUCT(indexTuple);
2047 : :
2048 : : /* we're only interested if it is the primary key */
2049 [ + - ]: 3 : if (index->indisprimary)
2050 : : {
2051 : 3 : *indnkeyatts = index->indnkeyatts;
2052 [ + - ]: 3 : if (*indnkeyatts > 0)
2053 : : {
2054 : 3 : result = palloc_array(char *, *indnkeyatts);
2055 : :
2056 [ + + ]: 9 : for (i = 0; i < *indnkeyatts; i++)
2057 : 6 : result[i] = SPI_fname(tupdesc, index->indkey.values[i]);
2058 : : }
2059 : 3 : break;
2060 : : }
2061 : : }
2062 : :
2063 : 3 : systable_endscan(scan);
2064 : 3 : table_close(indexRelation, AccessShareLock);
2065 : :
2066 : 3 : return result;
2067 : : }
2068 : :
2069 : : /*
2070 : : * Deconstruct a text[] into C-strings (note any NULL elements will be
2071 : : * returned as NULL pointers)
2072 : : */
2073 : : static char **
2074 : 20 : get_text_array_contents(ArrayType *array, int *numitems)
2075 : : {
2076 : 20 : int ndim = ARR_NDIM(array);
2077 : 20 : int *dims = ARR_DIMS(array);
2078 : : int nitems;
2079 : : int16 typlen;
2080 : : bool typbyval;
2081 : : char typalign;
2082 : : uint8 typalignby;
2083 : : char **values;
2084 : : char *ptr;
2085 : : uint8 *bitmap;
2086 : : int bitmask;
2087 : : int i;
2088 : :
2089 : : Assert(ARR_ELEMTYPE(array) == TEXTOID);
2090 : :
2091 : 20 : *numitems = nitems = ArrayGetNItems(ndim, dims);
2092 : :
2093 : 20 : get_typlenbyvalalign(ARR_ELEMTYPE(array),
2094 : : &typlen, &typbyval, &typalign);
2095 : 20 : typalignby = typalign_to_alignby(typalign);
2096 : :
2097 : 20 : values = palloc_array(char *, nitems);
2098 : :
2099 [ - + ]: 20 : ptr = ARR_DATA_PTR(array);
2100 [ - + ]: 20 : bitmap = ARR_NULLBITMAP(array);
2101 : 20 : bitmask = 1;
2102 : :
2103 [ + + ]: 55 : for (i = 0; i < nitems; i++)
2104 : : {
2105 [ - + - - ]: 35 : if (bitmap && (*bitmap & bitmask) == 0)
2106 : : {
2107 : 0 : values[i] = NULL;
2108 : : }
2109 : : else
2110 : : {
2111 : 35 : values[i] = TextDatumGetCString(PointerGetDatum(ptr));
2112 [ - + + - ]: 35 : ptr = att_addlength_pointer(ptr, typlen, ptr);
2113 : 35 : ptr = (char *) att_nominal_alignby(ptr, typalignby);
2114 : : }
2115 : :
2116 : : /* advance bitmap pointer if any */
2117 [ - + ]: 35 : if (bitmap)
2118 : : {
2119 : 0 : bitmask <<= 1;
2120 [ # # ]: 0 : if (bitmask == 0x100)
2121 : : {
2122 : 0 : bitmap++;
2123 : 0 : bitmask = 1;
2124 : : }
2125 : : }
2126 : : }
2127 : :
2128 : 20 : return values;
2129 : : }
2130 : :
2131 : : static char *
2132 : 4 : get_sql_insert(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals, char **tgt_pkattvals)
2133 : : {
2134 : : char *relname;
2135 : : HeapTuple tuple;
2136 : : TupleDesc tupdesc;
2137 : : int natts;
2138 : : StringInfoData buf;
2139 : : char *val;
2140 : : int key;
2141 : : int i;
2142 : : bool needComma;
2143 : :
2144 : 4 : initStringInfo(&buf);
2145 : :
2146 : : /* get relation name including any needed schema prefix and quoting */
2147 : 4 : relname = generate_relation_name(rel);
2148 : :
2149 : 4 : tupdesc = rel->rd_att;
2150 : 4 : natts = tupdesc->natts;
2151 : :
2152 : 4 : tuple = get_tuple_of_interest(rel, pkattnums, pknumatts, src_pkattvals);
2153 [ - + ]: 4 : if (!tuple)
2154 [ # # ]: 0 : ereport(ERROR,
2155 : : (errcode(ERRCODE_CARDINALITY_VIOLATION),
2156 : : errmsg("source row not found")));
2157 : :
2158 : 4 : appendStringInfo(&buf, "INSERT INTO %s(", relname);
2159 : :
2160 : 4 : needComma = false;
2161 [ + + ]: 19 : for (i = 0; i < natts; i++)
2162 : : {
2163 : 15 : Form_pg_attribute att = TupleDescAttr(tupdesc, i);
2164 : :
2165 [ + + ]: 15 : if (att->attisdropped)
2166 : 2 : continue;
2167 : :
2168 [ + + ]: 13 : if (needComma)
2169 : 9 : appendStringInfoChar(&buf, ',');
2170 : :
2171 : 13 : appendStringInfoString(&buf,
2172 : 13 : quote_ident_cstr(NameStr(att->attname)));
2173 : 13 : needComma = true;
2174 : : }
2175 : :
2176 : 4 : appendStringInfoString(&buf, ") VALUES(");
2177 : :
2178 : : /*
2179 : : * Note: i is physical column number (counting from 0).
2180 : : */
2181 : 4 : needComma = false;
2182 [ + + ]: 19 : for (i = 0; i < natts; i++)
2183 : : {
2184 [ + + ]: 15 : if (TupleDescAttr(tupdesc, i)->attisdropped)
2185 : 2 : continue;
2186 : :
2187 [ + + ]: 13 : if (needComma)
2188 : 9 : appendStringInfoChar(&buf, ',');
2189 : :
2190 : 13 : key = get_attnum_pk_pos(pkattnums, pknumatts, i);
2191 : :
2192 [ + + ]: 13 : if (key >= 0)
2193 [ + - ]: 7 : val = tgt_pkattvals[key] ? pstrdup(tgt_pkattvals[key]) : NULL;
2194 : : else
2195 : 6 : val = SPI_getvalue(tuple, tupdesc, i + 1);
2196 : :
2197 [ + - ]: 13 : if (val != NULL)
2198 : : {
2199 : 13 : appendStringInfoString(&buf, quote_literal_cstr(val));
2200 : 13 : pfree(val);
2201 : : }
2202 : : else
2203 : 0 : appendStringInfoString(&buf, "NULL");
2204 : 13 : needComma = true;
2205 : : }
2206 : 4 : appendStringInfoChar(&buf, ')');
2207 : :
2208 : 4 : return buf.data;
2209 : : }
2210 : :
2211 : : static char *
2212 : 4 : get_sql_delete(Relation rel, int *pkattnums, int pknumatts, char **tgt_pkattvals)
2213 : : {
2214 : : char *relname;
2215 : : TupleDesc tupdesc;
2216 : : StringInfoData buf;
2217 : : int i;
2218 : :
2219 : 4 : initStringInfo(&buf);
2220 : :
2221 : : /* get relation name including any needed schema prefix and quoting */
2222 : 4 : relname = generate_relation_name(rel);
2223 : :
2224 : 4 : tupdesc = rel->rd_att;
2225 : :
2226 : 4 : appendStringInfo(&buf, "DELETE FROM %s WHERE ", relname);
2227 [ + + ]: 11 : for (i = 0; i < pknumatts; i++)
2228 : : {
2229 : 7 : int pkattnum = pkattnums[i];
2230 : 7 : Form_pg_attribute attr = TupleDescAttr(tupdesc, pkattnum);
2231 : :
2232 [ + + ]: 7 : if (i > 0)
2233 : 3 : appendStringInfoString(&buf, " AND ");
2234 : :
2235 : 7 : appendStringInfoString(&buf,
2236 : 7 : quote_ident_cstr(NameStr(attr->attname)));
2237 : :
2238 [ + - ]: 7 : if (tgt_pkattvals[i] != NULL)
2239 : 7 : appendStringInfo(&buf, " = %s",
2240 : 7 : quote_literal_cstr(tgt_pkattvals[i]));
2241 : : else
2242 : 0 : appendStringInfoString(&buf, " IS NULL");
2243 : : }
2244 : :
2245 : 4 : return buf.data;
2246 : : }
2247 : :
2248 : : static char *
2249 : 4 : get_sql_update(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals, char **tgt_pkattvals)
2250 : : {
2251 : : char *relname;
2252 : : HeapTuple tuple;
2253 : : TupleDesc tupdesc;
2254 : : int natts;
2255 : : StringInfoData buf;
2256 : : char *val;
2257 : : int key;
2258 : : int i;
2259 : : bool needComma;
2260 : :
2261 : 4 : initStringInfo(&buf);
2262 : :
2263 : : /* get relation name including any needed schema prefix and quoting */
2264 : 4 : relname = generate_relation_name(rel);
2265 : :
2266 : 4 : tupdesc = rel->rd_att;
2267 : 4 : natts = tupdesc->natts;
2268 : :
2269 : 4 : tuple = get_tuple_of_interest(rel, pkattnums, pknumatts, src_pkattvals);
2270 [ - + ]: 4 : if (!tuple)
2271 [ # # ]: 0 : ereport(ERROR,
2272 : : (errcode(ERRCODE_CARDINALITY_VIOLATION),
2273 : : errmsg("source row not found")));
2274 : :
2275 : 4 : appendStringInfo(&buf, "UPDATE %s SET ", relname);
2276 : :
2277 : : /*
2278 : : * Note: i is physical column number (counting from 0).
2279 : : */
2280 : 4 : needComma = false;
2281 [ + + ]: 19 : for (i = 0; i < natts; i++)
2282 : : {
2283 : 15 : Form_pg_attribute attr = TupleDescAttr(tupdesc, i);
2284 : :
2285 [ + + ]: 15 : if (attr->attisdropped)
2286 : 2 : continue;
2287 : :
2288 [ + + ]: 13 : if (needComma)
2289 : 9 : appendStringInfoString(&buf, ", ");
2290 : :
2291 : 13 : appendStringInfo(&buf, "%s = ",
2292 : 13 : quote_ident_cstr(NameStr(attr->attname)));
2293 : :
2294 : 13 : key = get_attnum_pk_pos(pkattnums, pknumatts, i);
2295 : :
2296 [ + + ]: 13 : if (key >= 0)
2297 [ + - ]: 7 : val = tgt_pkattvals[key] ? pstrdup(tgt_pkattvals[key]) : NULL;
2298 : : else
2299 : 6 : val = SPI_getvalue(tuple, tupdesc, i + 1);
2300 : :
2301 [ + - ]: 13 : if (val != NULL)
2302 : : {
2303 : 13 : appendStringInfoString(&buf, quote_literal_cstr(val));
2304 : 13 : pfree(val);
2305 : : }
2306 : : else
2307 : 0 : appendStringInfoString(&buf, "NULL");
2308 : 13 : needComma = true;
2309 : : }
2310 : :
2311 : 4 : appendStringInfoString(&buf, " WHERE ");
2312 : :
2313 [ + + ]: 11 : for (i = 0; i < pknumatts; i++)
2314 : : {
2315 : 7 : int pkattnum = pkattnums[i];
2316 : 7 : Form_pg_attribute attr = TupleDescAttr(tupdesc, pkattnum);
2317 : :
2318 [ + + ]: 7 : if (i > 0)
2319 : 3 : appendStringInfoString(&buf, " AND ");
2320 : :
2321 : 7 : appendStringInfoString(&buf,
2322 : 7 : quote_ident_cstr(NameStr(attr->attname)));
2323 : :
2324 : 7 : val = tgt_pkattvals[i];
2325 : :
2326 [ + - ]: 7 : if (val != NULL)
2327 : 7 : appendStringInfo(&buf, " = %s", quote_literal_cstr(val));
2328 : : else
2329 : 0 : appendStringInfoString(&buf, " IS NULL");
2330 : : }
2331 : :
2332 : 4 : return buf.data;
2333 : : }
2334 : :
2335 : : /*
2336 : : * Return a properly quoted identifier.
2337 : : * Uses quote_ident in quote.c
2338 : : */
2339 : : static char *
2340 : 80 : quote_ident_cstr(char *rawstr)
2341 : : {
2342 : : text *rawstr_text;
2343 : : text *result_text;
2344 : : char *result;
2345 : :
2346 : 80 : rawstr_text = cstring_to_text(rawstr);
2347 : 80 : result_text = DatumGetTextPP(DirectFunctionCall1(quote_ident,
2348 : : PointerGetDatum(rawstr_text)));
2349 : 80 : result = text_to_cstring(result_text);
2350 : :
2351 : 80 : return result;
2352 : : }
2353 : :
2354 : : static int
2355 : 26 : get_attnum_pk_pos(int *pkattnums, int pknumatts, int key)
2356 : : {
2357 : : int i;
2358 : :
2359 : : /*
2360 : : * Not likely a long list anyway, so just scan for the value
2361 : : */
2362 [ + + ]: 50 : for (i = 0; i < pknumatts; i++)
2363 [ + + ]: 38 : if (key == pkattnums[i])
2364 : 14 : return i;
2365 : :
2366 : 12 : return -1;
2367 : : }
2368 : :
2369 : : static HeapTuple
2370 : 8 : get_tuple_of_interest(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals)
2371 : : {
2372 : : char *relname;
2373 : : TupleDesc tupdesc;
2374 : : int natts;
2375 : : StringInfoData buf;
2376 : : int ret;
2377 : : HeapTuple tuple;
2378 : : int i;
2379 : :
2380 : : /*
2381 : : * Connect to SPI manager
2382 : : */
2383 : 8 : SPI_connect();
2384 : :
2385 : 8 : initStringInfo(&buf);
2386 : :
2387 : : /* get relation name including any needed schema prefix and quoting */
2388 : 8 : relname = generate_relation_name(rel);
2389 : :
2390 : 8 : tupdesc = rel->rd_att;
2391 : 8 : natts = tupdesc->natts;
2392 : :
2393 : : /*
2394 : : * Build sql statement to look up tuple of interest, ie, the one matching
2395 : : * src_pkattvals. We used to use "SELECT *" here, but it's simpler to
2396 : : * generate a result tuple that matches the table's physical structure,
2397 : : * with NULLs for any dropped columns. Otherwise we have to deal with two
2398 : : * different tupdescs and everything's very confusing.
2399 : : */
2400 : 8 : appendStringInfoString(&buf, "SELECT ");
2401 : :
2402 [ + + ]: 38 : for (i = 0; i < natts; i++)
2403 : : {
2404 : 30 : Form_pg_attribute attr = TupleDescAttr(tupdesc, i);
2405 : :
2406 [ + + ]: 30 : if (i > 0)
2407 : 22 : appendStringInfoString(&buf, ", ");
2408 : :
2409 [ + + ]: 30 : if (attr->attisdropped)
2410 : 4 : appendStringInfoString(&buf, "NULL");
2411 : : else
2412 : 26 : appendStringInfoString(&buf,
2413 : 26 : quote_ident_cstr(NameStr(attr->attname)));
2414 : : }
2415 : :
2416 : 8 : appendStringInfo(&buf, " FROM %s WHERE ", relname);
2417 : :
2418 [ + + ]: 22 : for (i = 0; i < pknumatts; i++)
2419 : : {
2420 : 14 : int pkattnum = pkattnums[i];
2421 : 14 : Form_pg_attribute attr = TupleDescAttr(tupdesc, pkattnum);
2422 : :
2423 [ + + ]: 14 : if (i > 0)
2424 : 6 : appendStringInfoString(&buf, " AND ");
2425 : :
2426 : 14 : appendStringInfoString(&buf,
2427 : 14 : quote_ident_cstr(NameStr(attr->attname)));
2428 : :
2429 [ + - ]: 14 : if (src_pkattvals[i] != NULL)
2430 : 14 : appendStringInfo(&buf, " = %s",
2431 : 14 : quote_literal_cstr(src_pkattvals[i]));
2432 : : else
2433 : 0 : appendStringInfoString(&buf, " IS NULL");
2434 : : }
2435 : :
2436 : : /*
2437 : : * Retrieve the desired tuple
2438 : : */
2439 : 8 : ret = SPI_exec(buf.data, 0);
2440 : 8 : pfree(buf.data);
2441 : :
2442 : : /*
2443 : : * Only allow one qualifying tuple
2444 : : */
2445 [ + - - + ]: 8 : if ((ret == SPI_OK_SELECT) && (SPI_processed > 1))
2446 [ # # ]: 0 : ereport(ERROR,
2447 : : (errcode(ERRCODE_CARDINALITY_VIOLATION),
2448 : : errmsg("source criteria matched more than one record")));
2449 : :
2450 [ + - + - ]: 8 : else if (ret == SPI_OK_SELECT && SPI_processed == 1)
2451 : : {
2452 : 8 : SPITupleTable *tuptable = SPI_tuptable;
2453 : :
2454 : 8 : tuple = SPI_copytuple(tuptable->vals[0]);
2455 : 8 : SPI_finish();
2456 : :
2457 : 8 : return tuple;
2458 : : }
2459 : : else
2460 : : {
2461 : : /*
2462 : : * no qualifying tuples
2463 : : */
2464 : 0 : SPI_finish();
2465 : :
2466 : 0 : return NULL;
2467 : : }
2468 : :
2469 : : /*
2470 : : * never reached, but keep compiler quiet
2471 : : */
2472 : : return NULL;
2473 : : }
2474 : :
2475 : : static void
2476 : 21 : RangeVarCallbackForDblink(const RangeVar *relation,
2477 : : Oid relId, Oid oldRelId, void *arg)
2478 : : {
2479 : : AclResult aclresult;
2480 : :
2481 [ - + ]: 21 : if (!OidIsValid(relId))
2482 : 0 : return;
2483 : :
2484 : 21 : aclresult = pg_class_aclcheck(relId, GetUserId(), *((AclMode *) arg));
2485 [ - + ]: 21 : if (aclresult != ACLCHECK_OK)
2486 : 0 : aclcheck_error(aclresult, get_relkind_objtype(get_rel_relkind(relId)),
2487 : 0 : relation->relname);
2488 : : }
2489 : :
2490 : : /*
2491 : : * Open the relation named by relname_text, acquire specified type of lock,
2492 : : * verify we have specified permissions.
2493 : : * Caller must close rel when done with it.
2494 : : */
2495 : : static Relation
2496 : 21 : get_rel_from_relname(text *relname_text, LOCKMODE lockmode, AclMode aclmode)
2497 : : {
2498 : : RangeVar *relvar;
2499 : : Oid relid;
2500 : :
2501 : 21 : relvar = makeRangeVarFromNameList(textToQualifiedNameList(relname_text));
2502 : 21 : relid = RangeVarGetRelidExtended(relvar, lockmode, 0,
2503 : : RangeVarCallbackForDblink, &aclmode);
2504 : :
2505 : 21 : return table_open(relid, NoLock);
2506 : : }
2507 : :
2508 : : /*
2509 : : * generate_relation_name - copied from ruleutils.c
2510 : : * Compute the name to display for a relation
2511 : : *
2512 : : * The result includes all necessary quoting and schema-prefixing.
2513 : : */
2514 : : static char *
2515 : 20 : generate_relation_name(Relation rel)
2516 : : {
2517 : : char *nspname;
2518 : : char *result;
2519 : :
2520 : : /* Qualify the name if not visible in search path */
2521 [ + + ]: 20 : if (RelationIsVisible(RelationGetRelid(rel)))
2522 : 15 : nspname = NULL;
2523 : : else
2524 : 5 : nspname = get_namespace_name(rel->rd_rel->relnamespace);
2525 : :
2526 : 20 : result = quote_qualified_identifier(nspname, RelationGetRelationName(rel));
2527 : :
2528 : 20 : return result;
2529 : : }
2530 : :
2531 : :
2532 : : static remoteConn *
2533 : 80 : getConnectionByName(const char *name)
2534 : : {
2535 : : remoteConnHashEnt *hentry;
2536 : : char *key;
2537 : :
2538 [ + + ]: 80 : if (!remoteConnHash)
2539 : 7 : remoteConnHash = createConnHash();
2540 : :
2541 : 80 : key = pstrdup(name);
2542 : 80 : truncate_identifier(key, strlen(key), false);
2543 : 80 : hentry = (remoteConnHashEnt *) hash_search(remoteConnHash,
2544 : : key, HASH_FIND, NULL);
2545 : :
2546 [ + + + - ]: 80 : if (hentry && hentry->rconn.conn != NULL)
2547 : 68 : return &hentry->rconn;
2548 : :
2549 : 12 : return NULL;
2550 : : }
2551 : :
2552 : : static HTAB *
2553 : 8 : createConnHash(void)
2554 : : {
2555 : : HASHCTL ctl;
2556 : :
2557 : 8 : ctl.keysize = NAMEDATALEN;
2558 : 8 : ctl.entrysize = sizeof(remoteConnHashEnt);
2559 : :
2560 : 8 : return hash_create("Remote Con hash", NUMCONN, &ctl,
2561 : : HASH_ELEM | HASH_STRINGS);
2562 : : }
2563 : :
2564 : : static remoteConn *
2565 : 10 : createNewConnection(const char *name)
2566 : : {
2567 : : remoteConnHashEnt *hentry;
2568 : : bool found;
2569 : : char *key;
2570 : :
2571 [ + + ]: 10 : if (!remoteConnHash)
2572 : 1 : remoteConnHash = createConnHash();
2573 : :
2574 : 10 : key = pstrdup(name);
2575 : 10 : truncate_identifier(key, strlen(key), true);
2576 : 10 : hentry = (remoteConnHashEnt *) hash_search(remoteConnHash, key,
2577 : : HASH_ENTER, &found);
2578 : :
2579 [ + + + - ]: 10 : if (found && hentry->rconn.conn != NULL)
2580 [ + - ]: 1 : ereport(ERROR,
2581 : : (errcode(ERRCODE_DUPLICATE_OBJECT),
2582 : : errmsg("duplicate connection name")));
2583 : :
2584 : : /* New, or reusable, so initialize the rconn struct to zeroes */
2585 : 9 : memset(&hentry->rconn, 0, sizeof(remoteConn));
2586 : :
2587 : 9 : return &hentry->rconn;
2588 : : }
2589 : :
2590 : : static void
2591 : 8 : deleteConnection(const char *name)
2592 : : {
2593 : : remoteConnHashEnt *hentry;
2594 : : bool found;
2595 : : char *key;
2596 : :
2597 [ - + ]: 8 : if (!remoteConnHash)
2598 : 0 : remoteConnHash = createConnHash();
2599 : :
2600 : 8 : key = pstrdup(name);
2601 : 8 : truncate_identifier(key, strlen(key), false);
2602 : 8 : hentry = (remoteConnHashEnt *) hash_search(remoteConnHash,
2603 : : key, HASH_REMOVE, &found);
2604 : :
2605 [ - + ]: 8 : if (!hentry)
2606 [ # # ]: 0 : ereport(ERROR,
2607 : : (errcode(ERRCODE_UNDEFINED_OBJECT),
2608 : : errmsg("undefined connection name")));
2609 : 8 : }
2610 : :
2611 : : /*
2612 : : * Ensure that require_auth and SCRAM keys are correctly set on connstr.
2613 : : * SCRAM keys used to pass-through are coming from the initial connection
2614 : : * from the client with the server.
2615 : : *
2616 : : * All required SCRAM options are set by dblink, so we just need to ensure
2617 : : * that these options are not overwritten by the user.
2618 : : *
2619 : : * See appendSCRAMKeysInfo and its usage for more.
2620 : : */
2621 : : bool
2622 : 7 : dblink_connstr_has_required_scram_options(const char *connstr)
2623 : : {
2624 : : PQconninfoOption *options;
2625 : 7 : bool has_scram_server_key = false;
2626 : 7 : bool has_scram_client_key = false;
2627 : 7 : bool has_require_auth = false;
2628 : 7 : bool has_scram_keys = false;
2629 : :
2630 : 7 : options = PQconninfoParse(connstr, NULL);
2631 [ + - ]: 7 : if (options)
2632 : : {
2633 : : /*
2634 : : * Continue iterating even if we found the keys that we need to
2635 : : * validate to make sure that there is no other declaration of these
2636 : : * keys that can overwrite the first.
2637 : : */
2638 [ + + ]: 371 : for (PQconninfoOption *option = options; option->keyword != NULL; option++)
2639 : : {
2640 [ + + ]: 364 : if (strcmp(option->keyword, "require_auth") == 0)
2641 : : {
2642 [ + + + + ]: 7 : if (option->val != NULL && strcmp(option->val, "scram-sha-256") == 0)
2643 : 5 : has_require_auth = true;
2644 : : else
2645 : 2 : has_require_auth = false;
2646 : : }
2647 : :
2648 [ + + ]: 364 : if (strcmp(option->keyword, "scram_client_key") == 0)
2649 : : {
2650 [ + + + - ]: 7 : if (option->val != NULL && option->val[0] != '\0')
2651 : 6 : has_scram_client_key = true;
2652 : : else
2653 : 1 : has_scram_client_key = false;
2654 : : }
2655 : :
2656 [ + + ]: 364 : if (strcmp(option->keyword, "scram_server_key") == 0)
2657 : : {
2658 [ + + + - ]: 7 : if (option->val != NULL && option->val[0] != '\0')
2659 : 6 : has_scram_server_key = true;
2660 : : else
2661 : 1 : has_scram_server_key = false;
2662 : : }
2663 : : }
2664 : 7 : PQconninfoFree(options);
2665 : : }
2666 : :
2667 [ + + + - : 7 : has_scram_keys = has_scram_client_key && has_scram_server_key && MyProcPort != NULL && MyProcPort->has_scram_keys;
+ - + - ]
2668 : :
2669 [ + + + + ]: 7 : return (has_scram_keys && has_require_auth);
2670 : : }
2671 : :
2672 : : /*
2673 : : * We need to make sure that the connection made used credentials
2674 : : * which were provided by the user, so check what credentials were
2675 : : * used to connect and then make sure that they came from the user.
2676 : : *
2677 : : * On failure, we close "conn" and also delete the hashtable entry
2678 : : * identified by "connname" (if that's not NULL).
2679 : : */
2680 : : static void
2681 : 20 : dblink_security_check(PGconn *conn, const char *connname, const char *connstr)
2682 : : {
2683 : : /* Superuser bypasses security check */
2684 [ + + ]: 20 : if (superuser())
2685 : 18 : return;
2686 : :
2687 : : /* If password was used to connect, make sure it was one provided */
2688 [ + - - + ]: 2 : if (PQconnectionUsedPassword(conn) && dblink_connstr_has_pw(connstr))
2689 : 0 : return;
2690 : :
2691 : : /*
2692 : : * Password was not used to connect, check if SCRAM pass-through is in
2693 : : * use.
2694 : : *
2695 : : * If dblink_connstr_has_required_scram_options is true we assume that
2696 : : * UseScramPassthrough is also true because the required SCRAM keys are
2697 : : * only added if UseScramPassthrough is set, and the user is not allowed
2698 : : * to add the SCRAM keys on fdw and user mapping options.
2699 : : */
2700 [ + - + - : 2 : if (MyProcPort != NULL && MyProcPort->has_scram_keys && dblink_connstr_has_required_scram_options(connstr))
+ - ]
2701 : 2 : return;
2702 : :
2703 : : #ifdef ENABLE_GSS
2704 : : /* If GSSAPI creds used to connect, make sure it was one delegated */
2705 : : if (PQconnectionUsedGSSAPI(conn) && be_gssapi_get_delegation(MyProcPort))
2706 : : return;
2707 : : #endif
2708 : :
2709 : : /* Otherwise, fail out */
2710 : 0 : libpqsrv_disconnect(conn);
2711 [ # # ]: 0 : if (connname)
2712 : 0 : deleteConnection(connname);
2713 : :
2714 [ # # ]: 0 : ereport(ERROR,
2715 : : (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
2716 : : errmsg("password or GSSAPI delegated credentials required"),
2717 : : errdetail("Non-superusers may only connect using credentials they provide, eg: password in connection string or delegated GSSAPI credentials"),
2718 : : errhint("Ensure provided credentials match target server's authentication method.")));
2719 : : }
2720 : :
2721 : : /*
2722 : : * Function to check if the connection string includes an explicit
2723 : : * password, needed to ensure that non-superuser password-based auth
2724 : : * is using a provided password and not one picked up from the
2725 : : * environment.
2726 : : */
2727 : : static bool
2728 : 8 : dblink_connstr_has_pw(const char *connstr)
2729 : : {
2730 : : PQconninfoOption *options;
2731 : : PQconninfoOption *option;
2732 : 8 : bool connstr_gives_password = false;
2733 : :
2734 : 8 : options = PQconninfoParse(connstr, NULL);
2735 [ + - ]: 8 : if (options)
2736 : : {
2737 [ + + ]: 424 : for (option = options; option->keyword != NULL; option++)
2738 : : {
2739 [ + + ]: 416 : if (strcmp(option->keyword, "password") == 0)
2740 : : {
2741 [ - + - - ]: 8 : if (option->val != NULL && option->val[0] != '\0')
2742 : : {
2743 : 0 : connstr_gives_password = true;
2744 : 0 : break;
2745 : : }
2746 : : }
2747 : : }
2748 : 8 : PQconninfoFree(options);
2749 : : }
2750 : :
2751 : 8 : return connstr_gives_password;
2752 : : }
2753 : :
2754 : : /*
2755 : : * For non-superusers, insist that the connstr specify a password, except if
2756 : : * GSSAPI credentials have been delegated (and we check that they are used for
2757 : : * the connection in dblink_security_check later) or if SCRAM pass-through is
2758 : : * being used. This prevents a password or GSSAPI credentials from being
2759 : : * picked up from .pgpass, a service file, the environment, etc. We don't want
2760 : : * the postgres user's passwords or Kerberos credentials to be accessible to
2761 : : * non-superusers. In case of SCRAM pass-through insist that the connstr
2762 : : * has the required SCRAM pass-through options.
2763 : : */
2764 : : static void
2765 : 27 : dblink_connstr_check(const char *connstr)
2766 : : {
2767 [ + + ]: 27 : if (superuser())
2768 : 21 : return;
2769 : :
2770 [ - + ]: 6 : if (dblink_connstr_has_pw(connstr))
2771 : 0 : return;
2772 : :
2773 [ + - + + : 6 : if (MyProcPort != NULL && MyProcPort->has_scram_keys && dblink_connstr_has_required_scram_options(connstr))
+ + ]
2774 : 3 : return;
2775 : :
2776 : : #ifdef ENABLE_GSS
2777 : : if (be_gssapi_get_delegation(MyProcPort))
2778 : : return;
2779 : : #endif
2780 : :
2781 [ + - ]: 3 : ereport(ERROR,
2782 : : (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
2783 : : errmsg("password or GSSAPI delegated credentials required"),
2784 : : errdetail("Non-superusers must provide a password in the connection string or send delegated GSSAPI credentials.")));
2785 : : }
2786 : :
2787 : : /*
2788 : : * Report an error received from the remote server
2789 : : *
2790 : : * res: the received error result
2791 : : * fail: true for ERROR ereport, false for NOTICE
2792 : : * fmt and following args: sprintf-style format and values for errcontext;
2793 : : * the resulting string should be worded like "while <some action>"
2794 : : *
2795 : : * If "res" is not NULL, it'll be PQclear'ed here (unless we throw error,
2796 : : * in which case memory context cleanup will clear it eventually).
2797 : : */
2798 : : static void
2799 : 12 : dblink_res_error(PGconn *conn, const char *conname, PGresult *res,
2800 : : bool fail, const char *fmt, ...)
2801 : : {
2802 : : int level;
2803 : 12 : char *pg_diag_sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
2804 : 12 : char *message_primary = PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY);
2805 : 12 : char *message_detail = PQresultErrorField(res, PG_DIAG_MESSAGE_DETAIL);
2806 : 12 : char *message_hint = PQresultErrorField(res, PG_DIAG_MESSAGE_HINT);
2807 : 12 : char *message_context = PQresultErrorField(res, PG_DIAG_CONTEXT);
2808 : : int sqlstate;
2809 : : va_list ap;
2810 : : char dblink_context_msg[512];
2811 : :
2812 [ + + ]: 12 : if (fail)
2813 : 3 : level = ERROR;
2814 : : else
2815 : 9 : level = NOTICE;
2816 : :
2817 [ + - ]: 12 : if (pg_diag_sqlstate)
2818 : 12 : sqlstate = MAKE_SQLSTATE(pg_diag_sqlstate[0],
2819 : : pg_diag_sqlstate[1],
2820 : : pg_diag_sqlstate[2],
2821 : : pg_diag_sqlstate[3],
2822 : : pg_diag_sqlstate[4]);
2823 : : else
2824 : 0 : sqlstate = ERRCODE_CONNECTION_FAILURE;
2825 : :
2826 : : /*
2827 : : * If we don't get a message from the PGresult, try the PGconn. This is
2828 : : * needed because for connection-level failures, PQgetResult may just
2829 : : * return NULL, not a PGresult at all.
2830 : : */
2831 [ - + ]: 12 : if (message_primary == NULL)
2832 : 0 : message_primary = pchomp(PQerrorMessage(conn));
2833 : :
2834 : : /*
2835 : : * Format the basic errcontext string. Below, we'll add on something
2836 : : * about the connection name. That's a violation of the translatability
2837 : : * guidelines about constructing error messages out of parts, but since
2838 : : * there's no translation support for dblink, there's no need to worry
2839 : : * about that (yet).
2840 : : */
2841 : 12 : va_start(ap, fmt);
2842 : 12 : vsnprintf(dblink_context_msg, sizeof(dblink_context_msg), fmt, ap);
2843 : 12 : va_end(ap);
2844 : :
2845 [ + - + - : 12 : ereport(level,
+ - - + -
+ - + +
+ ]
2846 : : (errcode(sqlstate),
2847 : : (message_primary != NULL && message_primary[0] != '\0') ?
2848 : : errmsg_internal("%s", message_primary) :
2849 : : errmsg("could not obtain message string for remote error"),
2850 : : message_detail ? errdetail_internal("%s", message_detail) : 0,
2851 : : message_hint ? errhint("%s", message_hint) : 0,
2852 : : message_context ? (errcontext("%s", message_context)) : 0,
2853 : : conname ?
2854 : : (errcontext("%s on dblink connection named \"%s\"",
2855 : : dblink_context_msg, conname)) :
2856 : : (errcontext("%s on unnamed dblink connection",
2857 : : dblink_context_msg))));
2858 : 9 : PQclear(res);
2859 : 9 : }
2860 : :
2861 : : /*
2862 : : * Obtain connection string for a foreign server
2863 : : */
2864 : : static char *
2865 : 27 : get_connect_string(const char *servername)
2866 : : {
2867 : 27 : ForeignServer *foreign_server = NULL;
2868 : : UserMapping *user_mapping;
2869 : : ListCell *cell;
2870 : : StringInfoData buf;
2871 : : ForeignDataWrapper *fdw;
2872 : : AclResult aclresult;
2873 : : char *srvname;
2874 : :
2875 : : static const PQconninfoOption *options = NULL;
2876 : :
2877 : 27 : initStringInfo(&buf);
2878 : :
2879 : : /*
2880 : : * Get list of valid libpq options.
2881 : : *
2882 : : * To avoid unnecessary work, we get the list once and use it throughout
2883 : : * the lifetime of this backend process. We don't need to care about
2884 : : * memory context issues, because PQconndefaults allocates with malloc.
2885 : : */
2886 [ + + ]: 27 : if (!options)
2887 : : {
2888 : 8 : options = PQconndefaults();
2889 [ - + ]: 8 : if (!options) /* assume reason for failure is OOM */
2890 [ # # ]: 0 : ereport(ERROR,
2891 : : (errcode(ERRCODE_FDW_OUT_OF_MEMORY),
2892 : : errmsg("out of memory"),
2893 : : errdetail("Could not get libpq's default connection options.")));
2894 : : }
2895 : :
2896 : : /* first gather the server connstr options */
2897 : 27 : srvname = pstrdup(servername);
2898 : 27 : truncate_identifier(srvname, strlen(srvname), false);
2899 : 27 : foreign_server = GetForeignServerByName(srvname, true);
2900 : :
2901 [ + + ]: 27 : if (foreign_server)
2902 : : {
2903 : 7 : Oid serverid = foreign_server->serverid;
2904 : 7 : Oid fdwid = foreign_server->fdwid;
2905 : 7 : Oid userid = GetUserId();
2906 : :
2907 : 7 : user_mapping = GetUserMapping(userid, serverid);
2908 : 7 : fdw = GetForeignDataWrapper(fdwid);
2909 : :
2910 : : /* Check permissions, user must have usage on the server. */
2911 : 7 : aclresult = object_aclcheck(ForeignServerRelationId, serverid, userid, ACL_USAGE);
2912 [ - + ]: 7 : if (aclresult != ACLCHECK_OK)
2913 : 0 : aclcheck_error(aclresult, OBJECT_FOREIGN_SERVER, foreign_server->servername);
2914 : :
2915 : : /*
2916 : : * First append hardcoded options needed for SCRAM pass-through, so if
2917 : : * the user overwrites these options we can ereport on
2918 : : * dblink_connstr_check and dblink_security_check.
2919 : : */
2920 [ + - + + : 7 : if (MyProcPort != NULL && MyProcPort->has_scram_keys && UseScramPassthrough(foreign_server, user_mapping))
+ + ]
2921 : 4 : appendSCRAMKeysInfo(&buf);
2922 : :
2923 [ - + - - : 7 : foreach(cell, fdw->options)
- + ]
2924 : : {
2925 : 0 : DefElem *def = lfirst(cell);
2926 : :
2927 [ # # ]: 0 : if (is_valid_dblink_option(options, def->defname, ForeignDataWrapperRelationId))
2928 : 0 : appendStringInfo(&buf, "%s='%s' ", def->defname,
2929 : 0 : escape_param_str(strVal(def->arg)));
2930 : : }
2931 : :
2932 [ + - + + : 32 : foreach(cell, foreign_server->options)
+ + ]
2933 : : {
2934 : 25 : DefElem *def = lfirst(cell);
2935 : :
2936 [ + + ]: 25 : if (is_valid_dblink_option(options, def->defname, ForeignServerRelationId))
2937 : 20 : appendStringInfo(&buf, "%s='%s' ", def->defname,
2938 : 20 : escape_param_str(strVal(def->arg)));
2939 : : }
2940 : :
2941 [ + - + + : 15 : foreach(cell, user_mapping->options)
+ + ]
2942 : : {
2943 : :
2944 : 8 : DefElem *def = lfirst(cell);
2945 : :
2946 [ + + ]: 8 : if (is_valid_dblink_option(options, def->defname, UserMappingRelationId))
2947 : 7 : appendStringInfo(&buf, "%s='%s' ", def->defname,
2948 : 7 : escape_param_str(strVal(def->arg)));
2949 : : }
2950 : :
2951 : 7 : return buf.data;
2952 : : }
2953 : : else
2954 : 20 : return NULL;
2955 : : }
2956 : :
2957 : : /*
2958 : : * Escaping libpq connect parameter strings.
2959 : : *
2960 : : * Replaces "'" with "\'" and "\" with "\\".
2961 : : */
2962 : : static char *
2963 : 27 : escape_param_str(const char *str)
2964 : : {
2965 : : const char *cp;
2966 : : StringInfoData buf;
2967 : :
2968 : 27 : initStringInfo(&buf);
2969 : :
2970 [ + + ]: 238 : for (cp = str; *cp; cp++)
2971 : : {
2972 [ + - - + ]: 211 : if (*cp == '\\' || *cp == '\'')
2973 : 0 : appendStringInfoChar(&buf, '\\');
2974 : 211 : appendStringInfoChar(&buf, *cp);
2975 : : }
2976 : :
2977 : 27 : return buf.data;
2978 : : }
2979 : :
2980 : : /*
2981 : : * Validate the PK-attnums argument for dblink_build_sql_insert() and related
2982 : : * functions, and translate to the internal representation.
2983 : : *
2984 : : * The user supplies an int2vector of 1-based logical attnums, plus a count
2985 : : * argument (the need for the separate count argument is historical, but we
2986 : : * still check it). We check that each attnum corresponds to a valid,
2987 : : * non-dropped attribute of the rel. We do *not* prevent attnums from being
2988 : : * listed twice, though the actual use-case for such things is dubious.
2989 : : * Note that before Postgres 9.0, the user's attnums were interpreted as
2990 : : * physical not logical column numbers; this was changed for future-proofing.
2991 : : *
2992 : : * The internal representation is a palloc'd int array of 0-based physical
2993 : : * attnums.
2994 : : */
2995 : : static void
2996 : 18 : validate_pkattnums(Relation rel,
2997 : : int2vector *pkattnums_arg, int32 pknumatts_arg,
2998 : : int **pkattnums, int *pknumatts)
2999 : : {
3000 : 18 : TupleDesc tupdesc = rel->rd_att;
3001 : 18 : int natts = tupdesc->natts;
3002 : : int i;
3003 : :
3004 : : /* Don't take more array elements than there are */
3005 : 18 : pknumatts_arg = Min(pknumatts_arg, pkattnums_arg->dim1);
3006 : :
3007 : : /* Must have at least one pk attnum selected */
3008 [ - + ]: 18 : if (pknumatts_arg <= 0)
3009 [ # # ]: 0 : ereport(ERROR,
3010 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
3011 : : errmsg("number of key attributes must be > 0")));
3012 : :
3013 : : /* Allocate output array */
3014 : 18 : *pkattnums = palloc_array(int, pknumatts_arg);
3015 : 18 : *pknumatts = pknumatts_arg;
3016 : :
3017 : : /* Validate attnums and convert to internal form */
3018 [ + + ]: 57 : for (i = 0; i < pknumatts_arg; i++)
3019 : : {
3020 : 45 : int pkattnum = pkattnums_arg->values[i];
3021 : : int lnum;
3022 : : int j;
3023 : :
3024 : : /* Can throw error immediately if out of range */
3025 [ + - + + ]: 45 : if (pkattnum <= 0 || pkattnum > natts)
3026 [ + - ]: 6 : ereport(ERROR,
3027 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
3028 : : errmsg("invalid attribute number %d", pkattnum)));
3029 : :
3030 : : /* Identify which physical column has this logical number */
3031 : 39 : lnum = 0;
3032 [ + - ]: 69 : for (j = 0; j < natts; j++)
3033 : : {
3034 : : /* dropped columns don't count */
3035 [ + + ]: 69 : if (TupleDescCompactAttr(tupdesc, j)->attisdropped)
3036 : 3 : continue;
3037 : :
3038 [ + + ]: 66 : if (++lnum == pkattnum)
3039 : 39 : break;
3040 : : }
3041 : :
3042 [ + - ]: 39 : if (j < natts)
3043 : 39 : (*pkattnums)[i] = j;
3044 : : else
3045 [ # # ]: 0 : ereport(ERROR,
3046 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
3047 : : errmsg("invalid attribute number %d", pkattnum)));
3048 : : }
3049 : 12 : }
3050 : :
3051 : : /*
3052 : : * Check if the specified connection option is valid.
3053 : : *
3054 : : * We basically allow whatever libpq thinks is an option, with these
3055 : : * restrictions:
3056 : : * debug options: disallowed
3057 : : * "client_encoding": disallowed
3058 : : * "user": valid only in USER MAPPING options
3059 : : * secure options (eg password): valid only in USER MAPPING options
3060 : : * others: valid only in FOREIGN SERVER options
3061 : : *
3062 : : * We disallow client_encoding because it would be overridden anyway via
3063 : : * PQclientEncoding; allowing it to be specified would merely promote
3064 : : * confusion.
3065 : : */
3066 : : static bool
3067 : 542 : is_valid_dblink_option(const PQconninfoOption *options, const char *option,
3068 : : Oid context)
3069 : : {
3070 : : const PQconninfoOption *opt;
3071 : :
3072 : : /* Look up the option in libpq result */
3073 [ + + ]: 13652 : for (opt = options; opt->keyword; opt++)
3074 : : {
3075 [ + + ]: 13643 : if (strcmp(opt->keyword, option) == 0)
3076 : 533 : break;
3077 : : }
3078 [ + + ]: 542 : if (opt->keyword == NULL)
3079 : 9 : return false;
3080 : :
3081 : : /* Disallow debug options (particularly "replication") */
3082 [ + + ]: 533 : if (strchr(opt->dispchar, 'D'))
3083 : 38 : return false;
3084 : :
3085 : : /* Disallow "client_encoding" */
3086 [ + + ]: 495 : if (strcmp(opt->keyword, "client_encoding") == 0)
3087 : 9 : return false;
3088 : :
3089 : : /*
3090 : : * Disallow OAuth options for now, since the builtin flow communicates on
3091 : : * stderr by default and can't cache tokens yet.
3092 : : */
3093 [ + + ]: 486 : if (strncmp(opt->keyword, "oauth_", strlen("oauth_")) == 0)
3094 : 49 : return false;
3095 : :
3096 : : /*
3097 : : * If the option is "user" or marked secure, it should be specified only
3098 : : * in USER MAPPING. Others should be specified only in SERVER.
3099 : : */
3100 [ + + + + ]: 437 : if (strcmp(opt->keyword, "user") == 0 || strchr(opt->dispchar, '*'))
3101 : : {
3102 [ + + ]: 44 : if (context != UserMappingRelationId)
3103 : 12 : return false;
3104 : : }
3105 : : else
3106 : : {
3107 [ + + ]: 393 : if (context != ForeignServerRelationId)
3108 : 273 : return false;
3109 : : }
3110 : :
3111 : 152 : return true;
3112 : : }
3113 : :
3114 : : /*
3115 : : * Same as is_valid_dblink_option but also check for only dblink_fdw specific
3116 : : * options.
3117 : : */
3118 : : static bool
3119 : 47 : is_valid_dblink_fdw_option(const PQconninfoOption *options, const char *option,
3120 : : Oid context)
3121 : : {
3122 : : /*
3123 : : * These options are only valid for foreign server or user mapping
3124 : : * contexts
3125 : : */
3126 [ + + + + ]: 47 : if (context == ForeignServerRelationId || context == UserMappingRelationId)
3127 : : {
3128 [ + + ]: 45 : if (strcmp(option, "use_scram_passthrough") == 0)
3129 : 6 : return true;
3130 : : }
3131 : :
3132 : 41 : return is_valid_dblink_option(options, option, context);
3133 : : }
3134 : :
3135 : : /*
3136 : : * Copy the remote session's values of GUCs that affect datatype I/O
3137 : : * and apply them locally in a new GUC nesting level. Returns the new
3138 : : * nestlevel (which is needed by restoreLocalGucs to undo the settings),
3139 : : * or -1 if no new nestlevel was needed.
3140 : : *
3141 : : * We use the equivalent of a function SET option to allow the settings to
3142 : : * persist only until the caller calls restoreLocalGucs. If an error is
3143 : : * thrown in between, guc.c will take care of undoing the settings.
3144 : : */
3145 : : static int
3146 : 34 : applyRemoteGucs(PGconn *conn)
3147 : : {
3148 : : static const char *const GUCsAffectingIO[] = {
3149 : : "DateStyle",
3150 : : "IntervalStyle"
3151 : : };
3152 : :
3153 : 34 : int nestlevel = -1;
3154 : : int i;
3155 : :
3156 [ + + ]: 102 : for (i = 0; i < lengthof(GUCsAffectingIO); i++)
3157 : : {
3158 : 68 : const char *gucName = GUCsAffectingIO[i];
3159 : 68 : const char *remoteVal = PQparameterStatus(conn, gucName);
3160 : : const char *localVal;
3161 : :
3162 : : /*
3163 : : * If the remote server is pre-8.4, it won't have IntervalStyle, but
3164 : : * that's okay because its output format won't be ambiguous. So just
3165 : : * skip the GUC if we don't get a value for it. (We might eventually
3166 : : * need more complicated logic with remote-version checks here.)
3167 : : */
3168 [ - + ]: 68 : if (remoteVal == NULL)
3169 : 0 : continue;
3170 : :
3171 : : /*
3172 : : * Avoid GUC-setting overhead if the remote and local GUCs already
3173 : : * have the same value.
3174 : : */
3175 : 68 : localVal = GetConfigOption(gucName, false, false);
3176 : : Assert(localVal != NULL);
3177 : :
3178 [ + + ]: 68 : if (strcmp(remoteVal, localVal) == 0)
3179 : 51 : continue;
3180 : :
3181 : : /* Create new GUC nest level if we didn't already */
3182 [ + + ]: 17 : if (nestlevel < 0)
3183 : 9 : nestlevel = NewGUCNestLevel();
3184 : :
3185 : : /* Apply the option (this will throw error on failure) */
3186 : 17 : (void) set_config_option(gucName, remoteVal,
3187 : : PGC_USERSET, PGC_S_SESSION,
3188 : : GUC_ACTION_SAVE, true, 0, false);
3189 : : }
3190 : :
3191 : 34 : return nestlevel;
3192 : : }
3193 : :
3194 : : /*
3195 : : * Restore local GUCs after they have been overlaid with remote settings.
3196 : : */
3197 : : static void
3198 : 35 : restoreLocalGucs(int nestlevel)
3199 : : {
3200 : : /* Do nothing if no new nestlevel was created */
3201 [ + + ]: 35 : if (nestlevel > 0)
3202 : 8 : AtEOXact_GUC(true, nestlevel);
3203 : 35 : }
3204 : :
3205 : : /*
3206 : : * Append SCRAM client key and server key information from the global
3207 : : * MyProcPort into the given StringInfo buffer.
3208 : : */
3209 : : static void
3210 : 4 : appendSCRAMKeysInfo(StringInfo buf)
3211 : : {
3212 : : int len;
3213 : : int encoded_len;
3214 : : char *client_key;
3215 : : char *server_key;
3216 : :
3217 : 4 : len = pg_b64_enc_len(sizeof(MyProcPort->scram_ClientKey));
3218 : : /* don't forget the zero-terminator */
3219 : 4 : client_key = palloc0(len + 1);
3220 : 4 : encoded_len = pg_b64_encode(MyProcPort->scram_ClientKey,
3221 : : sizeof(MyProcPort->scram_ClientKey),
3222 : : client_key, len);
3223 [ - + ]: 4 : if (encoded_len < 0)
3224 [ # # ]: 0 : elog(ERROR, "could not encode SCRAM client key");
3225 : :
3226 : 4 : len = pg_b64_enc_len(sizeof(MyProcPort->scram_ServerKey));
3227 : : /* don't forget the zero-terminator */
3228 : 4 : server_key = palloc0(len + 1);
3229 : 4 : encoded_len = pg_b64_encode(MyProcPort->scram_ServerKey,
3230 : : sizeof(MyProcPort->scram_ServerKey),
3231 : : server_key, len);
3232 [ - + ]: 4 : if (encoded_len < 0)
3233 [ # # ]: 0 : elog(ERROR, "could not encode SCRAM server key");
3234 : :
3235 : 4 : appendStringInfo(buf, "scram_client_key='%s' ", client_key);
3236 : 4 : appendStringInfo(buf, "scram_server_key='%s' ", server_key);
3237 : 4 : appendStringInfoString(buf, "require_auth='scram-sha-256' ");
3238 : :
3239 : 4 : pfree(client_key);
3240 : 4 : pfree(server_key);
3241 : 4 : }
3242 : :
3243 : :
3244 : : /*
3245 : : * Return whether SCRAM pass-through is enabled.
3246 : : *
3247 : : * If use_scram_passthrough is specified in both the foreign server
3248 : : * and the user mapping, the user mapping setting takes precedence.
3249 : : */
3250 : : static bool
3251 : 5 : UseScramPassthrough(ForeignServer *foreign_server, UserMapping *user)
3252 : : {
3253 : : ListCell *cell;
3254 : :
3255 [ + - + + : 10 : foreach(cell, user->options)
+ + ]
3256 : : {
3257 : 6 : DefElem *def = lfirst(cell);
3258 : :
3259 [ + + ]: 6 : if (strcmp(def->defname, "use_scram_passthrough") == 0)
3260 : 1 : return defGetBoolean(def);
3261 : : }
3262 : :
3263 [ + - + - : 16 : foreach(cell, foreign_server->options)
+ - ]
3264 : : {
3265 : 16 : DefElem *def = (DefElem *) lfirst(cell);
3266 : :
3267 [ + + ]: 16 : if (strcmp(def->defname, "use_scram_passthrough") == 0)
3268 : 4 : return defGetBoolean(def);
3269 : : }
3270 : :
3271 : 0 : return false;
3272 : : }
|
