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