4 * Functions returning results from a remote database
6 * Joe Conway <mail@joeconway.com>
8 * Darko Prenosil <Darko.Prenosil@finteh.hr>
9 * Shridhar Daithankar <shridhar_daithankar@persistent.co.in>
11 * contrib/dblink/dblink.c
12 * Copyright (c) 2001-2024, PostgreSQL Global Development Group
13 * ALL RIGHTS RESERVED;
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.
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.
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.
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 "executor/spi.h"
47 #include "foreign/foreign.h"
49 #include "lib/stringinfo.h"
51 #include "libpq/libpq-be.h"
52 #include "libpq/libpq-be-fe-helpers.h"
53 #include "mb/pg_wchar.h"
54 #include "miscadmin.h"
55 #include "parser/scansup.h"
56 #include "utils/acl.h"
57 #include "utils/builtins.h"
58 #include "utils/fmgroids.h"
59 #include "utils/guc.h"
60 #include "utils/lsyscache.h"
61 #include "utils/memutils.h"
62 #include "utils/rel.h"
63 #include "utils/varlena.h"
64 #include "utils/wait_event.h"
68 typedef struct remoteConn
70 PGconn
*conn
; /* Hold the remote connection */
71 int openCursorCount
; /* The number of open cursors */
72 bool newXactForCursor
; /* Opened a transaction for a cursor */
75 typedef struct storeInfo
77 FunctionCallInfo fcinfo
;
78 Tuplestorestate
*tuplestore
;
79 AttInMetadata
*attinmeta
;
80 MemoryContext tmpcontext
;
82 /* temp storage for results to avoid leaks on exception */
88 * Internal declarations
90 static Datum
dblink_record_internal(FunctionCallInfo fcinfo
, bool is_async
);
91 static void prepTuplestoreResult(FunctionCallInfo fcinfo
);
92 static void materializeResult(FunctionCallInfo fcinfo
, PGconn
*conn
,
94 static void materializeQueryResult(FunctionCallInfo fcinfo
,
99 static PGresult
*storeQueryResult(volatile storeInfo
*sinfo
, PGconn
*conn
, const char *sql
);
100 static void storeRow(volatile storeInfo
*sinfo
, PGresult
*res
, bool first
);
101 static remoteConn
*getConnectionByName(const char *name
);
102 static HTAB
*createConnHash(void);
103 static void createNewConnection(const char *name
, remoteConn
*rconn
);
104 static void deleteConnection(const char *name
);
105 static char **get_pkey_attnames(Relation rel
, int16
*indnkeyatts
);
106 static char **get_text_array_contents(ArrayType
*array
, int *numitems
);
107 static char *get_sql_insert(Relation rel
, int *pkattnums
, int pknumatts
, char **src_pkattvals
, char **tgt_pkattvals
);
108 static char *get_sql_delete(Relation rel
, int *pkattnums
, int pknumatts
, char **tgt_pkattvals
);
109 static char *get_sql_update(Relation rel
, int *pkattnums
, int pknumatts
, char **src_pkattvals
, char **tgt_pkattvals
);
110 static char *quote_ident_cstr(char *rawstr
);
111 static int get_attnum_pk_pos(int *pkattnums
, int pknumatts
, int key
);
112 static HeapTuple
get_tuple_of_interest(Relation rel
, int *pkattnums
, int pknumatts
, char **src_pkattvals
);
113 static Relation
get_rel_from_relname(text
*relname_text
, LOCKMODE lockmode
, AclMode aclmode
);
114 static char *generate_relation_name(Relation rel
);
115 static void dblink_connstr_check(const char *connstr
);
116 static bool dblink_connstr_has_pw(const char *connstr
);
117 static void dblink_security_check(PGconn
*conn
, remoteConn
*rconn
, const char *connstr
);
118 static void dblink_res_error(PGconn
*conn
, const char *conname
, PGresult
*res
,
119 bool fail
, const char *fmt
,...) pg_attribute_printf(5, 6);
120 static char *get_connect_string(const char *servername
);
121 static char *escape_param_str(const char *str
);
122 static void validate_pkattnums(Relation rel
,
123 int2vector
*pkattnums_arg
, int32 pknumatts_arg
,
124 int **pkattnums
, int *pknumatts
);
125 static bool is_valid_dblink_option(const PQconninfoOption
*options
,
126 const char *option
, Oid context
);
127 static int applyRemoteGucs(PGconn
*conn
);
128 static void restoreLocalGucs(int nestlevel
);
131 static remoteConn
*pconn
= NULL
;
132 static HTAB
*remoteConnHash
= NULL
;
134 /* custom wait event values, retrieved from shared memory */
135 static uint32 dblink_we_connect
= 0;
136 static uint32 dblink_we_get_conn
= 0;
137 static uint32 dblink_we_get_result
= 0;
140 * Following is list that holds multiple remote connections.
141 * Calling convention of each dblink function changes to accept
142 * connection name as the first parameter. The connection list is
143 * much like ecpg e.g. a mapping between a name and a PGconn object.
146 typedef struct remoteConnHashEnt
148 char name
[NAMEDATALEN
];
152 /* initial number of connection hashes */
156 xpstrdup(const char *in
)
164 pg_attribute_noreturn()
165 dblink_res_internalerror(PGconn
*conn
, PGresult
*res
, const char *p2
)
167 char *msg
= pchomp(PQerrorMessage(conn
));
170 elog(ERROR
, "%s: %s", p2
, msg
);
174 pg_attribute_noreturn()
175 dblink_conn_not_avail(const char *conname
)
179 (errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST
),
180 errmsg("connection \"%s\" not available", conname
)));
183 (errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST
),
184 errmsg("connection not available")));
188 dblink_get_conn(char *conname_or_str
,
189 PGconn
*volatile *conn_p
, char **conname_p
, volatile bool *freeconn_p
)
191 remoteConn
*rconn
= getConnectionByName(conname_or_str
);
199 conname
= conname_or_str
;
206 connstr
= get_connect_string(conname_or_str
);
208 connstr
= conname_or_str
;
209 dblink_connstr_check(connstr
);
211 /* first time, allocate or get the custom wait event */
212 if (dblink_we_get_conn
== 0)
213 dblink_we_get_conn
= WaitEventExtensionNew("DblinkGetConnect");
215 /* OK to make connection */
216 conn
= libpqsrv_connect(connstr
, dblink_we_get_conn
);
218 if (PQstatus(conn
) == CONNECTION_BAD
)
220 char *msg
= pchomp(PQerrorMessage(conn
));
222 libpqsrv_disconnect(conn
);
224 (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION
),
225 errmsg("could not establish connection"),
226 errdetail_internal("%s", msg
)));
228 dblink_security_check(conn
, rconn
, connstr
);
229 if (PQclientEncoding(conn
) != GetDatabaseEncoding())
230 PQsetClientEncoding(conn
, GetDatabaseEncodingName());
236 *conname_p
= conname
;
237 *freeconn_p
= freeconn
;
241 dblink_get_named_conn(const char *conname
)
243 remoteConn
*rconn
= getConnectionByName(conname
);
248 dblink_conn_not_avail(conname
);
249 return NULL
; /* keep compiler quiet */
257 if (dblink_we_get_result
== 0)
258 dblink_we_get_result
= WaitEventExtensionNew("DblinkGetResult");
260 pconn
= (remoteConn
*) MemoryContextAlloc(TopMemoryContext
, sizeof(remoteConn
));
262 pconn
->openCursorCount
= 0;
263 pconn
->newXactForCursor
= false;
268 * Create a persistent connection to another database
270 PG_FUNCTION_INFO_V1(dblink_connect
);
272 dblink_connect(PG_FUNCTION_ARGS
)
274 char *conname_or_str
= NULL
;
275 char *connstr
= NULL
;
276 char *connname
= NULL
;
279 remoteConn
*rconn
= NULL
;
285 conname_or_str
= text_to_cstring(PG_GETARG_TEXT_PP(1));
286 connname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
288 else if (PG_NARGS() == 1)
289 conname_or_str
= text_to_cstring(PG_GETARG_TEXT_PP(0));
293 rconn
= (remoteConn
*) MemoryContextAlloc(TopMemoryContext
,
296 rconn
->openCursorCount
= 0;
297 rconn
->newXactForCursor
= false;
300 /* first check for valid foreign data server */
301 connstr
= get_connect_string(conname_or_str
);
303 connstr
= conname_or_str
;
305 /* check password in connection string if not superuser */
306 dblink_connstr_check(connstr
);
308 /* first time, allocate or get the custom wait event */
309 if (dblink_we_connect
== 0)
310 dblink_we_connect
= WaitEventExtensionNew("DblinkConnect");
312 /* OK to make connection */
313 conn
= libpqsrv_connect(connstr
, dblink_we_connect
);
315 if (PQstatus(conn
) == CONNECTION_BAD
)
317 msg
= pchomp(PQerrorMessage(conn
));
318 libpqsrv_disconnect(conn
);
323 (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION
),
324 errmsg("could not establish connection"),
325 errdetail_internal("%s", msg
)));
328 /* check password actually used if not superuser */
329 dblink_security_check(conn
, rconn
, connstr
);
331 /* attempt to set client encoding to match server encoding, if needed */
332 if (PQclientEncoding(conn
) != GetDatabaseEncoding())
333 PQsetClientEncoding(conn
, GetDatabaseEncodingName());
338 createNewConnection(connname
, rconn
);
343 libpqsrv_disconnect(pconn
->conn
);
347 PG_RETURN_TEXT_P(cstring_to_text("OK"));
351 * Clear a persistent connection to another database
353 PG_FUNCTION_INFO_V1(dblink_disconnect
);
355 dblink_disconnect(PG_FUNCTION_ARGS
)
357 char *conname
= NULL
;
358 remoteConn
*rconn
= NULL
;
365 conname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
366 rconn
= getConnectionByName(conname
);
374 dblink_conn_not_avail(conname
);
376 libpqsrv_disconnect(conn
);
379 deleteConnection(conname
);
385 PG_RETURN_TEXT_P(cstring_to_text("OK"));
389 * opens a cursor using a persistent connection
391 PG_FUNCTION_INFO_V1(dblink_open
);
393 dblink_open(PG_FUNCTION_ARGS
)
395 PGresult
*res
= NULL
;
397 char *curname
= NULL
;
399 char *conname
= NULL
;
401 remoteConn
*rconn
= NULL
;
402 bool fail
= true; /* default to backward compatible behavior */
405 initStringInfo(&buf
);
410 curname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
411 sql
= text_to_cstring(PG_GETARG_TEXT_PP(1));
414 else if (PG_NARGS() == 3)
416 /* might be text,text,text or text,text,bool */
417 if (get_fn_expr_argtype(fcinfo
->flinfo
, 2) == BOOLOID
)
419 curname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
420 sql
= text_to_cstring(PG_GETARG_TEXT_PP(1));
421 fail
= PG_GETARG_BOOL(2);
426 conname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
427 curname
= text_to_cstring(PG_GETARG_TEXT_PP(1));
428 sql
= text_to_cstring(PG_GETARG_TEXT_PP(2));
429 rconn
= getConnectionByName(conname
);
432 else if (PG_NARGS() == 4)
434 /* text,text,text,bool */
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 fail
= PG_GETARG_BOOL(3);
439 rconn
= getConnectionByName(conname
);
442 if (!rconn
|| !rconn
->conn
)
443 dblink_conn_not_avail(conname
);
447 /* If we are not in a transaction, start one */
448 if (PQtransactionStatus(conn
) == PQTRANS_IDLE
)
450 res
= libpqsrv_exec(conn
, "BEGIN", dblink_we_get_result
);
451 if (PQresultStatus(res
) != PGRES_COMMAND_OK
)
452 dblink_res_internalerror(conn
, res
, "begin error");
454 rconn
->newXactForCursor
= true;
457 * Since transaction state was IDLE, we force cursor count to
458 * initially be 0. This is needed as a previous ABORT might have wiped
459 * out our transaction without maintaining the cursor count for us.
461 rconn
->openCursorCount
= 0;
464 /* if we started a transaction, increment cursor count */
465 if (rconn
->newXactForCursor
)
466 (rconn
->openCursorCount
)++;
468 appendStringInfo(&buf
, "DECLARE %s CURSOR FOR %s", curname
, sql
);
469 res
= libpqsrv_exec(conn
, buf
.data
, dblink_we_get_result
);
470 if (!res
|| PQresultStatus(res
) != PGRES_COMMAND_OK
)
472 dblink_res_error(conn
, conname
, res
, fail
,
473 "while opening cursor \"%s\"", curname
);
474 PG_RETURN_TEXT_P(cstring_to_text("ERROR"));
478 PG_RETURN_TEXT_P(cstring_to_text("OK"));
484 PG_FUNCTION_INFO_V1(dblink_close
);
486 dblink_close(PG_FUNCTION_ARGS
)
489 PGresult
*res
= NULL
;
490 char *curname
= NULL
;
491 char *conname
= NULL
;
493 remoteConn
*rconn
= NULL
;
494 bool fail
= true; /* default to backward compatible behavior */
497 initStringInfo(&buf
);
502 curname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
505 else if (PG_NARGS() == 2)
507 /* might be text,text or text,bool */
508 if (get_fn_expr_argtype(fcinfo
->flinfo
, 1) == BOOLOID
)
510 curname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
511 fail
= PG_GETARG_BOOL(1);
516 conname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
517 curname
= text_to_cstring(PG_GETARG_TEXT_PP(1));
518 rconn
= getConnectionByName(conname
);
524 conname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
525 curname
= text_to_cstring(PG_GETARG_TEXT_PP(1));
526 fail
= PG_GETARG_BOOL(2);
527 rconn
= getConnectionByName(conname
);
530 if (!rconn
|| !rconn
->conn
)
531 dblink_conn_not_avail(conname
);
535 appendStringInfo(&buf
, "CLOSE %s", curname
);
537 /* close the cursor */
538 res
= libpqsrv_exec(conn
, buf
.data
, dblink_we_get_result
);
539 if (!res
|| PQresultStatus(res
) != PGRES_COMMAND_OK
)
541 dblink_res_error(conn
, conname
, res
, fail
,
542 "while closing cursor \"%s\"", curname
);
543 PG_RETURN_TEXT_P(cstring_to_text("ERROR"));
548 /* if we started a transaction, decrement cursor count */
549 if (rconn
->newXactForCursor
)
551 (rconn
->openCursorCount
)--;
553 /* if count is zero, commit the transaction */
554 if (rconn
->openCursorCount
== 0)
556 rconn
->newXactForCursor
= false;
558 res
= libpqsrv_exec(conn
, "COMMIT", dblink_we_get_result
);
559 if (PQresultStatus(res
) != PGRES_COMMAND_OK
)
560 dblink_res_internalerror(conn
, res
, "commit error");
565 PG_RETURN_TEXT_P(cstring_to_text("OK"));
569 * Fetch results from an open cursor
571 PG_FUNCTION_INFO_V1(dblink_fetch
);
573 dblink_fetch(PG_FUNCTION_ARGS
)
575 PGresult
*res
= NULL
;
576 char *conname
= NULL
;
577 remoteConn
*rconn
= NULL
;
580 char *curname
= NULL
;
582 bool fail
= true; /* default to backward compatible */
584 prepTuplestoreResult(fcinfo
);
590 /* text,text,int,bool */
591 conname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
592 curname
= text_to_cstring(PG_GETARG_TEXT_PP(1));
593 howmany
= PG_GETARG_INT32(2);
594 fail
= PG_GETARG_BOOL(3);
596 rconn
= getConnectionByName(conname
);
600 else if (PG_NARGS() == 3)
602 /* text,text,int or text,int,bool */
603 if (get_fn_expr_argtype(fcinfo
->flinfo
, 2) == BOOLOID
)
605 curname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
606 howmany
= PG_GETARG_INT32(1);
607 fail
= PG_GETARG_BOOL(2);
612 conname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
613 curname
= text_to_cstring(PG_GETARG_TEXT_PP(1));
614 howmany
= PG_GETARG_INT32(2);
616 rconn
= getConnectionByName(conname
);
621 else if (PG_NARGS() == 2)
624 curname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
625 howmany
= PG_GETARG_INT32(1);
630 dblink_conn_not_avail(conname
);
632 initStringInfo(&buf
);
633 appendStringInfo(&buf
, "FETCH %d FROM %s", howmany
, curname
);
636 * Try to execute the query. Note that since libpq uses malloc, the
637 * PGresult will be long-lived even though we are still in a short-lived
640 res
= libpqsrv_exec(conn
, buf
.data
, dblink_we_get_result
);
642 (PQresultStatus(res
) != PGRES_COMMAND_OK
&&
643 PQresultStatus(res
) != PGRES_TUPLES_OK
))
645 dblink_res_error(conn
, conname
, res
, fail
,
646 "while fetching from cursor \"%s\"", curname
);
649 else if (PQresultStatus(res
) == PGRES_COMMAND_OK
)
651 /* cursor does not exist - closed already or bad name */
654 (errcode(ERRCODE_INVALID_CURSOR_NAME
),
655 errmsg("cursor \"%s\" does not exist", curname
)));
658 materializeResult(fcinfo
, conn
, res
);
663 * Note: this is the new preferred version of dblink
665 PG_FUNCTION_INFO_V1(dblink_record
);
667 dblink_record(PG_FUNCTION_ARGS
)
669 return dblink_record_internal(fcinfo
, false);
672 PG_FUNCTION_INFO_V1(dblink_send_query
);
674 dblink_send_query(PG_FUNCTION_ARGS
)
682 conn
= dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
683 sql
= text_to_cstring(PG_GETARG_TEXT_PP(1));
686 /* shouldn't happen */
687 elog(ERROR
, "wrong number of arguments");
689 /* async query send */
690 retval
= PQsendQuery(conn
, sql
);
692 elog(NOTICE
, "could not send query: %s", pchomp(PQerrorMessage(conn
)));
694 PG_RETURN_INT32(retval
);
697 PG_FUNCTION_INFO_V1(dblink_get_result
);
699 dblink_get_result(PG_FUNCTION_ARGS
)
701 return dblink_record_internal(fcinfo
, true);
705 dblink_record_internal(FunctionCallInfo fcinfo
, bool is_async
)
707 PGconn
*volatile conn
= NULL
;
708 volatile bool freeconn
= false;
710 prepTuplestoreResult(fcinfo
);
717 char *conname
= NULL
;
718 bool fail
= true; /* default to backward compatible */
725 conname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
726 sql
= text_to_cstring(PG_GETARG_TEXT_PP(1));
727 fail
= PG_GETARG_BOOL(2);
728 dblink_get_conn(conname
, &conn
, &conname
, &freeconn
);
730 else if (PG_NARGS() == 2)
732 /* text,text or text,bool */
733 if (get_fn_expr_argtype(fcinfo
->flinfo
, 1) == BOOLOID
)
735 sql
= text_to_cstring(PG_GETARG_TEXT_PP(0));
736 fail
= PG_GETARG_BOOL(1);
741 conname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
742 sql
= text_to_cstring(PG_GETARG_TEXT_PP(1));
743 dblink_get_conn(conname
, &conn
, &conname
, &freeconn
);
746 else if (PG_NARGS() == 1)
750 sql
= text_to_cstring(PG_GETARG_TEXT_PP(0));
753 /* shouldn't happen */
754 elog(ERROR
, "wrong number of arguments");
758 /* get async result */
759 conname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
764 fail
= PG_GETARG_BOOL(1);
765 conn
= dblink_get_named_conn(conname
);
767 else if (PG_NARGS() == 1)
770 conn
= dblink_get_named_conn(conname
);
773 /* shouldn't happen */
774 elog(ERROR
, "wrong number of arguments");
778 dblink_conn_not_avail(conname
);
782 /* synchronous query, use efficient tuple collection method */
783 materializeQueryResult(fcinfo
, conn
, conname
, sql
, fail
);
787 /* async result retrieval, do it the old way */
788 PGresult
*res
= libpqsrv_get_result(conn
, dblink_we_get_result
);
790 /* NULL means we're all done with the async results */
793 if (PQresultStatus(res
) != PGRES_COMMAND_OK
&&
794 PQresultStatus(res
) != PGRES_TUPLES_OK
)
796 dblink_res_error(conn
, conname
, res
, fail
,
797 "while executing query");
798 /* if fail isn't set, we'll return an empty query result */
802 materializeResult(fcinfo
, conn
, res
);
809 /* if needed, close the connection to the database */
811 libpqsrv_disconnect(conn
);
819 * Verify function caller can handle a tuplestore result, and set up for that.
821 * Note: if the caller returns without actually creating a tuplestore, the
822 * executor will treat the function result as an empty set.
825 prepTuplestoreResult(FunctionCallInfo fcinfo
)
827 ReturnSetInfo
*rsinfo
= (ReturnSetInfo
*) fcinfo
->resultinfo
;
829 /* check to see if query supports us returning a tuplestore */
830 if (rsinfo
== NULL
|| !IsA(rsinfo
, ReturnSetInfo
))
832 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED
),
833 errmsg("set-valued function called in context that cannot accept a set")));
834 if (!(rsinfo
->allowedModes
& SFRM_Materialize
))
836 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED
),
837 errmsg("materialize mode required, but it is not allowed in this context")));
839 /* let the executor know we're sending back a tuplestore */
840 rsinfo
->returnMode
= SFRM_Materialize
;
842 /* caller must fill these to return a non-empty result */
843 rsinfo
->setResult
= NULL
;
844 rsinfo
->setDesc
= NULL
;
848 * Copy the contents of the PGresult into a tuplestore to be returned
849 * as the result of the current function.
850 * The PGresult will be released in this function.
853 materializeResult(FunctionCallInfo fcinfo
, PGconn
*conn
, PGresult
*res
)
855 ReturnSetInfo
*rsinfo
= (ReturnSetInfo
*) fcinfo
->resultinfo
;
857 /* prepTuplestoreResult must have been called previously */
858 Assert(rsinfo
->returnMode
== SFRM_Materialize
);
867 if (PQresultStatus(res
) == PGRES_COMMAND_OK
)
872 * need a tuple descriptor representing one TEXT column to return
873 * the command status string as our result tuple
875 tupdesc
= CreateTemplateTupleDesc(1);
876 TupleDescInitEntry(tupdesc
, (AttrNumber
) 1, "status",
883 Assert(PQresultStatus(res
) == PGRES_TUPLES_OK
);
887 /* get a tuple descriptor for our result type */
888 switch (get_call_result_type(fcinfo
, NULL
, &tupdesc
))
890 case TYPEFUNC_COMPOSITE
:
893 case TYPEFUNC_RECORD
:
894 /* failed to determine actual type of RECORD */
896 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED
),
897 errmsg("function returning record called in context "
898 "that cannot accept type record")));
901 /* result type isn't composite */
902 elog(ERROR
, "return type must be a row type");
906 /* make sure we have a persistent copy of the tupdesc */
907 tupdesc
= CreateTupleDescCopy(tupdesc
);
908 ntuples
= PQntuples(res
);
909 nfields
= PQnfields(res
);
913 * check result and tuple descriptor have the same number of columns
915 if (nfields
!= tupdesc
->natts
)
917 (errcode(ERRCODE_DATATYPE_MISMATCH
),
918 errmsg("remote query result rowtype does not match "
919 "the specified FROM clause rowtype")));
923 AttInMetadata
*attinmeta
;
925 Tuplestorestate
*tupstore
;
926 MemoryContext oldcontext
;
930 attinmeta
= TupleDescGetAttInMetadata(tupdesc
);
932 /* Set GUCs to ensure we read GUC-sensitive data types correctly */
934 nestlevel
= applyRemoteGucs(conn
);
936 oldcontext
= MemoryContextSwitchTo(rsinfo
->econtext
->ecxt_per_query_memory
);
937 tupstore
= tuplestore_begin_heap(true, false, work_mem
);
938 rsinfo
->setResult
= tupstore
;
939 rsinfo
->setDesc
= tupdesc
;
940 MemoryContextSwitchTo(oldcontext
);
942 values
= palloc_array(char *, nfields
);
944 /* put all tuples into the tuplestore */
945 for (row
= 0; row
< ntuples
; row
++)
953 for (i
= 0; i
< nfields
; i
++)
955 if (PQgetisnull(res
, row
, i
))
958 values
[i
] = PQgetvalue(res
, row
, i
);
963 values
[0] = PQcmdStatus(res
);
966 /* build the tuple and put it into the tuplestore. */
967 tuple
= BuildTupleFromCStrings(attinmeta
, values
);
968 tuplestore_puttuple(tupstore
, tuple
);
971 /* clean up GUC settings, if we changed any */
972 restoreLocalGucs(nestlevel
);
977 /* be sure to release the libpq result */
984 * Execute the given SQL command and store its results into a tuplestore
985 * to be returned as the result of the current function.
987 * This is equivalent to PQexec followed by materializeResult, but we make
988 * use of libpq's single-row mode to avoid accumulating the whole result
989 * inside libpq before it gets transferred to the tuplestore.
992 materializeQueryResult(FunctionCallInfo fcinfo
,
998 ReturnSetInfo
*rsinfo
= (ReturnSetInfo
*) fcinfo
->resultinfo
;
999 PGresult
*volatile res
= NULL
;
1000 volatile storeInfo sinfo
= {0};
1002 /* prepTuplestoreResult must have been called previously */
1003 Assert(rsinfo
->returnMode
== SFRM_Materialize
);
1005 sinfo
.fcinfo
= fcinfo
;
1009 /* Create short-lived memory context for data conversions */
1010 sinfo
.tmpcontext
= AllocSetContextCreate(CurrentMemoryContext
,
1011 "dblink temporary context",
1012 ALLOCSET_DEFAULT_SIZES
);
1014 /* execute query, collecting any tuples into the tuplestore */
1015 res
= storeQueryResult(&sinfo
, conn
, sql
);
1018 (PQresultStatus(res
) != PGRES_COMMAND_OK
&&
1019 PQresultStatus(res
) != PGRES_TUPLES_OK
))
1022 * dblink_res_error will clear the passed PGresult, so we need
1023 * this ugly dance to avoid doing so twice during error exit
1025 PGresult
*res1
= res
;
1028 dblink_res_error(conn
, conname
, res1
, fail
,
1029 "while executing query");
1030 /* if fail isn't set, we'll return an empty query result */
1032 else if (PQresultStatus(res
) == PGRES_COMMAND_OK
)
1035 * storeRow didn't get called, so we need to convert the command
1036 * status string to a tuple manually
1039 AttInMetadata
*attinmeta
;
1040 Tuplestorestate
*tupstore
;
1043 MemoryContext oldcontext
;
1046 * need a tuple descriptor representing one TEXT column to return
1047 * the command status string as our result tuple
1049 tupdesc
= CreateTemplateTupleDesc(1);
1050 TupleDescInitEntry(tupdesc
, (AttrNumber
) 1, "status",
1052 attinmeta
= TupleDescGetAttInMetadata(tupdesc
);
1054 oldcontext
= MemoryContextSwitchTo(rsinfo
->econtext
->ecxt_per_query_memory
);
1055 tupstore
= tuplestore_begin_heap(true, false, work_mem
);
1056 rsinfo
->setResult
= tupstore
;
1057 rsinfo
->setDesc
= tupdesc
;
1058 MemoryContextSwitchTo(oldcontext
);
1060 values
[0] = PQcmdStatus(res
);
1062 /* build the tuple and put it into the tuplestore. */
1063 tuple
= BuildTupleFromCStrings(attinmeta
, values
);
1064 tuplestore_puttuple(tupstore
, tuple
);
1071 Assert(PQresultStatus(res
) == PGRES_TUPLES_OK
);
1072 /* storeRow should have created a tuplestore */
1073 Assert(rsinfo
->setResult
!= NULL
);
1079 /* clean up data conversion short-lived memory context */
1080 if (sinfo
.tmpcontext
!= NULL
)
1081 MemoryContextDelete(sinfo
.tmpcontext
);
1082 sinfo
.tmpcontext
= NULL
;
1084 PQclear(sinfo
.last_res
);
1085 sinfo
.last_res
= NULL
;
1086 PQclear(sinfo
.cur_res
);
1087 sinfo
.cur_res
= NULL
;
1091 /* be sure to release any libpq result we collected */
1093 PQclear(sinfo
.last_res
);
1094 PQclear(sinfo
.cur_res
);
1095 /* and clear out any pending data in libpq */
1096 while ((res
= libpqsrv_get_result(conn
, dblink_we_get_result
)) !=
1105 * Execute query, and send any result rows to sinfo->tuplestore.
1108 storeQueryResult(volatile storeInfo
*sinfo
, PGconn
*conn
, const char *sql
)
1114 if (!PQsendQuery(conn
, sql
))
1115 elog(ERROR
, "could not send query: %s", pchomp(PQerrorMessage(conn
)));
1117 if (!PQsetSingleRowMode(conn
)) /* shouldn't fail */
1118 elog(ERROR
, "failed to set single-row mode for dblink query");
1122 CHECK_FOR_INTERRUPTS();
1124 sinfo
->cur_res
= libpqsrv_get_result(conn
, dblink_we_get_result
);
1125 if (!sinfo
->cur_res
)
1128 if (PQresultStatus(sinfo
->cur_res
) == PGRES_SINGLE_TUPLE
)
1130 /* got one row from possibly-bigger resultset */
1133 * Set GUCs to ensure we read GUC-sensitive data types correctly.
1134 * We shouldn't do this until we have a row in hand, to ensure
1135 * libpq has seen any earlier ParameterStatus protocol messages.
1137 if (first
&& nestlevel
< 0)
1138 nestlevel
= applyRemoteGucs(conn
);
1140 storeRow(sinfo
, sinfo
->cur_res
, first
);
1142 PQclear(sinfo
->cur_res
);
1143 sinfo
->cur_res
= NULL
;
1148 /* if empty resultset, fill tuplestore header */
1149 if (first
&& PQresultStatus(sinfo
->cur_res
) == PGRES_TUPLES_OK
)
1150 storeRow(sinfo
, sinfo
->cur_res
, first
);
1152 /* store completed result at last_res */
1153 PQclear(sinfo
->last_res
);
1154 sinfo
->last_res
= sinfo
->cur_res
;
1155 sinfo
->cur_res
= NULL
;
1160 /* clean up GUC settings, if we changed any */
1161 restoreLocalGucs(nestlevel
);
1163 /* return last_res */
1164 res
= sinfo
->last_res
;
1165 sinfo
->last_res
= NULL
;
1170 * Send single row to sinfo->tuplestore.
1172 * If "first" is true, create the tuplestore using PGresult's metadata
1173 * (in this case the PGresult might contain either zero or one row).
1176 storeRow(volatile storeInfo
*sinfo
, PGresult
*res
, bool first
)
1178 int nfields
= PQnfields(res
);
1181 MemoryContext oldcontext
;
1185 /* Prepare for new result set */
1186 ReturnSetInfo
*rsinfo
= (ReturnSetInfo
*) sinfo
->fcinfo
->resultinfo
;
1190 * It's possible to get more than one result set if the query string
1191 * contained multiple SQL commands. In that case, we follow PQexec's
1192 * traditional behavior of throwing away all but the last result.
1194 if (sinfo
->tuplestore
)
1195 tuplestore_end(sinfo
->tuplestore
);
1196 sinfo
->tuplestore
= NULL
;
1198 /* get a tuple descriptor for our result type */
1199 switch (get_call_result_type(sinfo
->fcinfo
, NULL
, &tupdesc
))
1201 case TYPEFUNC_COMPOSITE
:
1204 case TYPEFUNC_RECORD
:
1205 /* failed to determine actual type of RECORD */
1207 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED
),
1208 errmsg("function returning record called in context "
1209 "that cannot accept type record")));
1212 /* result type isn't composite */
1213 elog(ERROR
, "return type must be a row type");
1217 /* make sure we have a persistent copy of the tupdesc */
1218 tupdesc
= CreateTupleDescCopy(tupdesc
);
1220 /* check result and tuple descriptor have the same number of columns */
1221 if (nfields
!= tupdesc
->natts
)
1223 (errcode(ERRCODE_DATATYPE_MISMATCH
),
1224 errmsg("remote query result rowtype does not match "
1225 "the specified FROM clause rowtype")));
1227 /* Prepare attinmeta for later data conversions */
1228 sinfo
->attinmeta
= TupleDescGetAttInMetadata(tupdesc
);
1230 /* Create a new, empty tuplestore */
1231 oldcontext
= MemoryContextSwitchTo(rsinfo
->econtext
->ecxt_per_query_memory
);
1232 sinfo
->tuplestore
= tuplestore_begin_heap(true, false, work_mem
);
1233 rsinfo
->setResult
= sinfo
->tuplestore
;
1234 rsinfo
->setDesc
= tupdesc
;
1235 MemoryContextSwitchTo(oldcontext
);
1237 /* Done if empty resultset */
1238 if (PQntuples(res
) == 0)
1242 * Set up sufficiently-wide string pointers array; this won't change
1243 * in size so it's easy to preallocate.
1246 pfree(sinfo
->cstrs
);
1247 sinfo
->cstrs
= palloc_array(char *, nfields
);
1250 /* Should have a single-row result if we get here */
1251 Assert(PQntuples(res
) == 1);
1254 * Do the following work in a temp context that we reset after each tuple.
1255 * This cleans up not only the data we have direct access to, but any
1256 * cruft the I/O functions might leak.
1258 oldcontext
= MemoryContextSwitchTo(sinfo
->tmpcontext
);
1261 * Fill cstrs with null-terminated strings of column values.
1263 for (i
= 0; i
< nfields
; i
++)
1265 if (PQgetisnull(res
, 0, i
))
1266 sinfo
->cstrs
[i
] = NULL
;
1268 sinfo
->cstrs
[i
] = PQgetvalue(res
, 0, i
);
1271 /* Convert row to a tuple, and add it to the tuplestore */
1272 tuple
= BuildTupleFromCStrings(sinfo
->attinmeta
, sinfo
->cstrs
);
1274 tuplestore_puttuple(sinfo
->tuplestore
, tuple
);
1277 MemoryContextSwitchTo(oldcontext
);
1278 MemoryContextReset(sinfo
->tmpcontext
);
1282 * List all open dblink connections by name.
1283 * Returns an array of all connection names.
1286 PG_FUNCTION_INFO_V1(dblink_get_connections
);
1288 dblink_get_connections(PG_FUNCTION_ARGS
)
1290 HASH_SEQ_STATUS status
;
1291 remoteConnHashEnt
*hentry
;
1292 ArrayBuildState
*astate
= NULL
;
1296 hash_seq_init(&status
, remoteConnHash
);
1297 while ((hentry
= (remoteConnHashEnt
*) hash_seq_search(&status
)) != NULL
)
1299 /* stash away current value */
1300 astate
= accumArrayResult(astate
,
1301 CStringGetTextDatum(hentry
->name
),
1302 false, TEXTOID
, CurrentMemoryContext
);
1307 PG_RETURN_DATUM(makeArrayResult(astate
,
1308 CurrentMemoryContext
));
1314 * Checks if a given remote connection is busy
1316 * Returns 1 if the connection is busy, 0 otherwise
1318 * text connection_name - name of the connection to check
1321 PG_FUNCTION_INFO_V1(dblink_is_busy
);
1323 dblink_is_busy(PG_FUNCTION_ARGS
)
1328 conn
= dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
1330 PQconsumeInput(conn
);
1331 PG_RETURN_INT32(PQisBusy(conn
));
1335 * Cancels a running request on a connection
1338 * "OK" if the cancel request has been sent correctly,
1339 * an error message otherwise
1342 * text connection_name - name of the connection to check
1345 PG_FUNCTION_INFO_V1(dblink_cancel_query
);
1347 dblink_cancel_query(PG_FUNCTION_ARGS
)
1351 TimestampTz endtime
;
1354 conn
= dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
1355 endtime
= TimestampTzPlusMilliseconds(GetCurrentTimestamp(),
1357 msg
= libpqsrv_cancel(conn
, endtime
);
1361 PG_RETURN_TEXT_P(cstring_to_text(msg
));
1366 * Get error message from a connection
1369 * "OK" if no error, an error message otherwise
1372 * text connection_name - name of the connection to check
1375 PG_FUNCTION_INFO_V1(dblink_error_message
);
1377 dblink_error_message(PG_FUNCTION_ARGS
)
1383 conn
= dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
1385 msg
= PQerrorMessage(conn
);
1386 if (msg
== NULL
|| msg
[0] == '\0')
1387 PG_RETURN_TEXT_P(cstring_to_text("OK"));
1389 PG_RETURN_TEXT_P(cstring_to_text(pchomp(msg
)));
1393 * Execute an SQL non-SELECT command
1395 PG_FUNCTION_INFO_V1(dblink_exec
);
1397 dblink_exec(PG_FUNCTION_ARGS
)
1399 text
*volatile sql_cmd_status
= NULL
;
1400 PGconn
*volatile conn
= NULL
;
1401 volatile bool freeconn
= false;
1407 PGresult
*res
= NULL
;
1409 char *conname
= NULL
;
1410 bool fail
= true; /* default to backward compatible behavior */
1412 if (PG_NARGS() == 3)
1414 /* must be text,text,bool */
1415 conname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
1416 sql
= text_to_cstring(PG_GETARG_TEXT_PP(1));
1417 fail
= PG_GETARG_BOOL(2);
1418 dblink_get_conn(conname
, &conn
, &conname
, &freeconn
);
1420 else if (PG_NARGS() == 2)
1422 /* might be text,text or text,bool */
1423 if (get_fn_expr_argtype(fcinfo
->flinfo
, 1) == BOOLOID
)
1425 sql
= text_to_cstring(PG_GETARG_TEXT_PP(0));
1426 fail
= PG_GETARG_BOOL(1);
1431 conname
= text_to_cstring(PG_GETARG_TEXT_PP(0));
1432 sql
= text_to_cstring(PG_GETARG_TEXT_PP(1));
1433 dblink_get_conn(conname
, &conn
, &conname
, &freeconn
);
1436 else if (PG_NARGS() == 1)
1438 /* must be single text argument */
1440 sql
= text_to_cstring(PG_GETARG_TEXT_PP(0));
1443 /* shouldn't happen */
1444 elog(ERROR
, "wrong number of arguments");
1447 dblink_conn_not_avail(conname
);
1449 res
= libpqsrv_exec(conn
, sql
, dblink_we_get_result
);
1451 (PQresultStatus(res
) != PGRES_COMMAND_OK
&&
1452 PQresultStatus(res
) != PGRES_TUPLES_OK
))
1454 dblink_res_error(conn
, conname
, res
, fail
,
1455 "while executing command");
1458 * and save a copy of the command status string to return as our
1461 sql_cmd_status
= cstring_to_text("ERROR");
1463 else if (PQresultStatus(res
) == PGRES_COMMAND_OK
)
1466 * and save a copy of the command status string to return as our
1469 sql_cmd_status
= cstring_to_text(PQcmdStatus(res
));
1476 (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED
),
1477 errmsg("statement returning results not allowed")));
1482 /* if needed, close the connection to the database */
1484 libpqsrv_disconnect(conn
);
1488 PG_RETURN_TEXT_P(sql_cmd_status
);
1495 * Return list of primary key fields for the supplied relation,
1496 * or NULL if none exists.
1498 PG_FUNCTION_INFO_V1(dblink_get_pkey
);
1500 dblink_get_pkey(PG_FUNCTION_ARGS
)
1504 FuncCallContext
*funcctx
;
1507 AttInMetadata
*attinmeta
;
1508 MemoryContext oldcontext
;
1510 /* stuff done only on the first call of the function */
1511 if (SRF_IS_FIRSTCALL())
1516 /* create a function context for cross-call persistence */
1517 funcctx
= SRF_FIRSTCALL_INIT();
1520 * switch to memory context appropriate for multiple function calls
1522 oldcontext
= MemoryContextSwitchTo(funcctx
->multi_call_memory_ctx
);
1524 /* open target relation */
1525 rel
= get_rel_from_relname(PG_GETARG_TEXT_PP(0), AccessShareLock
, ACL_SELECT
);
1527 /* get the array of attnums */
1528 results
= get_pkey_attnames(rel
, &indnkeyatts
);
1530 relation_close(rel
, AccessShareLock
);
1533 * need a tuple descriptor representing one INT and one TEXT column
1535 tupdesc
= CreateTemplateTupleDesc(2);
1536 TupleDescInitEntry(tupdesc
, (AttrNumber
) 1, "position",
1538 TupleDescInitEntry(tupdesc
, (AttrNumber
) 2, "colname",
1542 * Generate attribute metadata needed later to produce tuples from raw
1545 attinmeta
= TupleDescGetAttInMetadata(tupdesc
);
1546 funcctx
->attinmeta
= attinmeta
;
1548 if ((results
!= NULL
) && (indnkeyatts
> 0))
1550 funcctx
->max_calls
= indnkeyatts
;
1552 /* got results, keep track of them */
1553 funcctx
->user_fctx
= results
;
1557 /* fast track when no results */
1558 MemoryContextSwitchTo(oldcontext
);
1559 SRF_RETURN_DONE(funcctx
);
1562 MemoryContextSwitchTo(oldcontext
);
1565 /* stuff done on every call of the function */
1566 funcctx
= SRF_PERCALL_SETUP();
1569 * initialize per-call variables
1571 call_cntr
= funcctx
->call_cntr
;
1572 max_calls
= funcctx
->max_calls
;
1574 results
= (char **) funcctx
->user_fctx
;
1575 attinmeta
= funcctx
->attinmeta
;
1577 if (call_cntr
< max_calls
) /* do when there is more left to send */
1583 values
= palloc_array(char *, 2);
1584 values
[0] = psprintf("%d", call_cntr
+ 1);
1585 values
[1] = results
[call_cntr
];
1587 /* build the tuple */
1588 tuple
= BuildTupleFromCStrings(attinmeta
, values
);
1590 /* make the tuple into a datum */
1591 result
= HeapTupleGetDatum(tuple
);
1593 SRF_RETURN_NEXT(funcctx
, result
);
1597 /* do when there is no more left */
1598 SRF_RETURN_DONE(funcctx
);
1604 * dblink_build_sql_insert
1606 * Used to generate an SQL insert statement
1607 * based on an existing tuple in a local relation.
1608 * This is useful for selectively replicating data
1609 * to another server via dblink.
1612 * <relname> - name of local table of interest
1613 * <pkattnums> - an int2vector of attnums which will be used
1614 * to identify the local tuple of interest
1615 * <pknumatts> - number of attnums in pkattnums
1616 * <src_pkattvals_arry> - text array of key values which will be used
1617 * to identify the local tuple of interest
1618 * <tgt_pkattvals_arry> - text array of key values which will be used
1619 * to build the string for execution remotely. These are substituted
1620 * for their counterparts in src_pkattvals_arry
1622 PG_FUNCTION_INFO_V1(dblink_build_sql_insert
);
1624 dblink_build_sql_insert(PG_FUNCTION_ARGS
)
1626 text
*relname_text
= PG_GETARG_TEXT_PP(0);
1627 int2vector
*pkattnums_arg
= (int2vector
*) PG_GETARG_POINTER(1);
1628 int32 pknumatts_arg
= PG_GETARG_INT32(2);
1629 ArrayType
*src_pkattvals_arry
= PG_GETARG_ARRAYTYPE_P(3);
1630 ArrayType
*tgt_pkattvals_arry
= PG_GETARG_ARRAYTYPE_P(4);
1634 char **src_pkattvals
;
1635 char **tgt_pkattvals
;
1641 * Open target relation.
1643 rel
= get_rel_from_relname(relname_text
, AccessShareLock
, ACL_SELECT
);
1646 * Process pkattnums argument.
1648 validate_pkattnums(rel
, pkattnums_arg
, pknumatts_arg
,
1649 &pkattnums
, &pknumatts
);
1652 * Source array is made up of key values that will be used to locate the
1653 * tuple of interest from the local system.
1655 src_pkattvals
= get_text_array_contents(src_pkattvals_arry
, &src_nitems
);
1658 * There should be one source array key value for each key attnum
1660 if (src_nitems
!= pknumatts
)
1662 (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR
),
1663 errmsg("source key array length must match number of key attributes")));
1666 * Target array is made up of key values that will be used to build the
1667 * SQL string for use on the remote system.
1669 tgt_pkattvals
= get_text_array_contents(tgt_pkattvals_arry
, &tgt_nitems
);
1672 * There should be one target array key value for each key attnum
1674 if (tgt_nitems
!= pknumatts
)
1676 (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR
),
1677 errmsg("target key array length must match number of key attributes")));
1680 * Prep work is finally done. Go get the SQL string.
1682 sql
= get_sql_insert(rel
, pkattnums
, pknumatts
, src_pkattvals
, tgt_pkattvals
);
1685 * Now we can close the relation.
1687 relation_close(rel
, AccessShareLock
);
1692 PG_RETURN_TEXT_P(cstring_to_text(sql
));
1697 * dblink_build_sql_delete
1699 * Used to generate an SQL delete statement.
1700 * This is useful for selectively replicating a
1701 * delete to another server via dblink.
1704 * <relname> - name of remote table of interest
1705 * <pkattnums> - an int2vector of attnums which will be used
1706 * to identify the remote tuple of interest
1707 * <pknumatts> - number of attnums in pkattnums
1708 * <tgt_pkattvals_arry> - text array of key values which will be used
1709 * to build the string for execution remotely.
1711 PG_FUNCTION_INFO_V1(dblink_build_sql_delete
);
1713 dblink_build_sql_delete(PG_FUNCTION_ARGS
)
1715 text
*relname_text
= PG_GETARG_TEXT_PP(0);
1716 int2vector
*pkattnums_arg
= (int2vector
*) PG_GETARG_POINTER(1);
1717 int32 pknumatts_arg
= PG_GETARG_INT32(2);
1718 ArrayType
*tgt_pkattvals_arry
= PG_GETARG_ARRAYTYPE_P(3);
1722 char **tgt_pkattvals
;
1727 * Open target relation.
1729 rel
= get_rel_from_relname(relname_text
, AccessShareLock
, ACL_SELECT
);
1732 * Process pkattnums argument.
1734 validate_pkattnums(rel
, pkattnums_arg
, pknumatts_arg
,
1735 &pkattnums
, &pknumatts
);
1738 * Target array is made up of key values that will be used to build the
1739 * SQL string for use on the remote system.
1741 tgt_pkattvals
= get_text_array_contents(tgt_pkattvals_arry
, &tgt_nitems
);
1744 * There should be one target array key value for each key attnum
1746 if (tgt_nitems
!= pknumatts
)
1748 (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR
),
1749 errmsg("target key array length must match number of key attributes")));
1752 * Prep work is finally done. Go get the SQL string.
1754 sql
= get_sql_delete(rel
, pkattnums
, pknumatts
, tgt_pkattvals
);
1757 * Now we can close the relation.
1759 relation_close(rel
, AccessShareLock
);
1764 PG_RETURN_TEXT_P(cstring_to_text(sql
));
1769 * dblink_build_sql_update
1771 * Used to generate an SQL update statement
1772 * based on an existing tuple in a local relation.
1773 * This is useful for selectively replicating data
1774 * to another server via dblink.
1777 * <relname> - name of local table of interest
1778 * <pkattnums> - an int2vector of attnums which will be used
1779 * to identify the local tuple of interest
1780 * <pknumatts> - number of attnums in pkattnums
1781 * <src_pkattvals_arry> - text array of key values which will be used
1782 * to identify the local tuple of interest
1783 * <tgt_pkattvals_arry> - text array of key values which will be used
1784 * to build the string for execution remotely. These are substituted
1785 * for their counterparts in src_pkattvals_arry
1787 PG_FUNCTION_INFO_V1(dblink_build_sql_update
);
1789 dblink_build_sql_update(PG_FUNCTION_ARGS
)
1791 text
*relname_text
= PG_GETARG_TEXT_PP(0);
1792 int2vector
*pkattnums_arg
= (int2vector
*) PG_GETARG_POINTER(1);
1793 int32 pknumatts_arg
= PG_GETARG_INT32(2);
1794 ArrayType
*src_pkattvals_arry
= PG_GETARG_ARRAYTYPE_P(3);
1795 ArrayType
*tgt_pkattvals_arry
= PG_GETARG_ARRAYTYPE_P(4);
1799 char **src_pkattvals
;
1800 char **tgt_pkattvals
;
1806 * Open target relation.
1808 rel
= get_rel_from_relname(relname_text
, AccessShareLock
, ACL_SELECT
);
1811 * Process pkattnums argument.
1813 validate_pkattnums(rel
, pkattnums_arg
, pknumatts_arg
,
1814 &pkattnums
, &pknumatts
);
1817 * Source array is made up of key values that will be used to locate the
1818 * tuple of interest from the local system.
1820 src_pkattvals
= get_text_array_contents(src_pkattvals_arry
, &src_nitems
);
1823 * There should be one source array key value for each key attnum
1825 if (src_nitems
!= pknumatts
)
1827 (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR
),
1828 errmsg("source key array length must match number of key attributes")));
1831 * Target array is made up of key values that will be used to build the
1832 * SQL string for use on the remote system.
1834 tgt_pkattvals
= get_text_array_contents(tgt_pkattvals_arry
, &tgt_nitems
);
1837 * There should be one target array key value for each key attnum
1839 if (tgt_nitems
!= pknumatts
)
1841 (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR
),
1842 errmsg("target key array length must match number of key attributes")));
1845 * Prep work is finally done. Go get the SQL string.
1847 sql
= get_sql_update(rel
, pkattnums
, pknumatts
, src_pkattvals
, tgt_pkattvals
);
1850 * Now we can close the relation.
1852 relation_close(rel
, AccessShareLock
);
1857 PG_RETURN_TEXT_P(cstring_to_text(sql
));
1861 * dblink_current_query
1862 * return the current query string
1863 * to allow its use in (among other things)
1866 PG_FUNCTION_INFO_V1(dblink_current_query
);
1868 dblink_current_query(PG_FUNCTION_ARGS
)
1870 /* This is now just an alias for the built-in function current_query() */
1871 PG_RETURN_DATUM(current_query(fcinfo
));
1875 * Retrieve async notifications for a connection.
1877 * Returns a setof record of notifications, or an empty set if none received.
1878 * Can optionally take a named connection as parameter, but uses the unnamed
1879 * connection per default.
1882 #define DBLINK_NOTIFY_COLS 3
1884 PG_FUNCTION_INFO_V1(dblink_get_notify
);
1886 dblink_get_notify(PG_FUNCTION_ARGS
)
1890 ReturnSetInfo
*rsinfo
= (ReturnSetInfo
*) fcinfo
->resultinfo
;
1893 if (PG_NARGS() == 1)
1894 conn
= dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
1898 InitMaterializedSRF(fcinfo
, 0);
1900 PQconsumeInput(conn
);
1901 while ((notify
= PQnotifies(conn
)) != NULL
)
1903 Datum values
[DBLINK_NOTIFY_COLS
];
1904 bool nulls
[DBLINK_NOTIFY_COLS
];
1906 memset(values
, 0, sizeof(values
));
1907 memset(nulls
, 0, sizeof(nulls
));
1909 if (notify
->relname
!= NULL
)
1910 values
[0] = CStringGetTextDatum(notify
->relname
);
1914 values
[1] = Int32GetDatum(notify
->be_pid
);
1916 if (notify
->extra
!= NULL
)
1917 values
[2] = CStringGetTextDatum(notify
->extra
);
1921 tuplestore_putvalues(rsinfo
->setResult
, rsinfo
->setDesc
, values
, nulls
);
1924 PQconsumeInput(conn
);
1931 * Validate the options given to a dblink foreign server or user mapping.
1932 * Raise an error if any option is invalid.
1934 * We just check the names of options here, so semantic errors in options,
1935 * such as invalid numeric format, will be detected at the attempt to connect.
1937 PG_FUNCTION_INFO_V1(dblink_fdw_validator
);
1939 dblink_fdw_validator(PG_FUNCTION_ARGS
)
1941 List
*options_list
= untransformRelOptions(PG_GETARG_DATUM(0));
1942 Oid context
= PG_GETARG_OID(1);
1945 static const PQconninfoOption
*options
= NULL
;
1948 * Get list of valid libpq options.
1950 * To avoid unnecessary work, we get the list once and use it throughout
1951 * the lifetime of this backend process. We don't need to care about
1952 * memory context issues, because PQconndefaults allocates with malloc.
1956 options
= PQconndefaults();
1957 if (!options
) /* assume reason for failure is OOM */
1959 (errcode(ERRCODE_FDW_OUT_OF_MEMORY
),
1960 errmsg("out of memory"),
1961 errdetail("Could not get libpq's default connection options.")));
1964 /* Validate each supplied option. */
1965 foreach(cell
, options_list
)
1967 DefElem
*def
= (DefElem
*) lfirst(cell
);
1969 if (!is_valid_dblink_option(options
, def
->defname
, context
))
1972 * Unknown option, or invalid option for the context specified, so
1973 * complain about it. Provide a hint with a valid option that
1974 * looks similar, if there is one.
1976 const PQconninfoOption
*opt
;
1977 const char *closest_match
;
1978 ClosestMatchState match_state
;
1979 bool has_valid_options
= false;
1981 initClosestMatch(&match_state
, def
->defname
, 4);
1982 for (opt
= options
; opt
->keyword
; opt
++)
1984 if (is_valid_dblink_option(options
, opt
->keyword
, context
))
1986 has_valid_options
= true;
1987 updateClosestMatch(&match_state
, opt
->keyword
);
1991 closest_match
= getClosestMatch(&match_state
);
1993 (errcode(ERRCODE_FDW_OPTION_NAME_NOT_FOUND
),
1994 errmsg("invalid option \"%s\"", def
->defname
),
1995 has_valid_options
? closest_match
?
1996 errhint("Perhaps you meant the option \"%s\".",
1997 closest_match
) : 0 :
1998 errhint("There are no valid options in this context.")));
2006 /*************************************************************
2007 * internal functions
2014 * Get the primary key attnames for the given relation.
2015 * Return NULL, and set indnkeyatts = 0, if no primary key exists.
2018 get_pkey_attnames(Relation rel
, int16
*indnkeyatts
)
2020 Relation indexRelation
;
2023 HeapTuple indexTuple
;
2025 char **result
= NULL
;
2028 /* initialize indnkeyatts to 0 in case no primary key exists */
2031 tupdesc
= rel
->rd_att
;
2033 /* Prepare to scan pg_index for entries having indrelid = this rel. */
2034 indexRelation
= table_open(IndexRelationId
, AccessShareLock
);
2036 Anum_pg_index_indrelid
,
2037 BTEqualStrategyNumber
, F_OIDEQ
,
2038 ObjectIdGetDatum(RelationGetRelid(rel
)));
2040 scan
= systable_beginscan(indexRelation
, IndexIndrelidIndexId
, true,
2043 while (HeapTupleIsValid(indexTuple
= systable_getnext(scan
)))
2045 Form_pg_index index
= (Form_pg_index
) GETSTRUCT(indexTuple
);
2047 /* we're only interested if it is the primary key */
2048 if (index
->indisprimary
)
2050 *indnkeyatts
= index
->indnkeyatts
;
2051 if (*indnkeyatts
> 0)
2053 result
= palloc_array(char *, *indnkeyatts
);
2055 for (i
= 0; i
< *indnkeyatts
; i
++)
2056 result
[i
] = SPI_fname(tupdesc
, index
->indkey
.values
[i
]);
2062 systable_endscan(scan
);
2063 table_close(indexRelation
, AccessShareLock
);
2069 * Deconstruct a text[] into C-strings (note any NULL elements will be
2070 * returned as NULL pointers)
2073 get_text_array_contents(ArrayType
*array
, int *numitems
)
2075 int ndim
= ARR_NDIM(array
);
2076 int *dims
= ARR_DIMS(array
);
2087 Assert(ARR_ELEMTYPE(array
) == TEXTOID
);
2089 *numitems
= nitems
= ArrayGetNItems(ndim
, dims
);
2091 get_typlenbyvalalign(ARR_ELEMTYPE(array
),
2092 &typlen
, &typbyval
, &typalign
);
2094 values
= palloc_array(char *, nitems
);
2096 ptr
= ARR_DATA_PTR(array
);
2097 bitmap
= ARR_NULLBITMAP(array
);
2100 for (i
= 0; i
< nitems
; i
++)
2102 if (bitmap
&& (*bitmap
& bitmask
) == 0)
2108 values
[i
] = TextDatumGetCString(PointerGetDatum(ptr
));
2109 ptr
= att_addlength_pointer(ptr
, typlen
, ptr
);
2110 ptr
= (char *) att_align_nominal(ptr
, typalign
);
2113 /* advance bitmap pointer if any */
2117 if (bitmask
== 0x100)
2129 get_sql_insert(Relation rel
, int *pkattnums
, int pknumatts
, char **src_pkattvals
, char **tgt_pkattvals
)
2141 initStringInfo(&buf
);
2143 /* get relation name including any needed schema prefix and quoting */
2144 relname
= generate_relation_name(rel
);
2146 tupdesc
= rel
->rd_att
;
2147 natts
= tupdesc
->natts
;
2149 tuple
= get_tuple_of_interest(rel
, pkattnums
, pknumatts
, src_pkattvals
);
2152 (errcode(ERRCODE_CARDINALITY_VIOLATION
),
2153 errmsg("source row not found")));
2155 appendStringInfo(&buf
, "INSERT INTO %s(", relname
);
2158 for (i
= 0; i
< natts
; i
++)
2160 Form_pg_attribute att
= TupleDescAttr(tupdesc
, i
);
2162 if (att
->attisdropped
)
2166 appendStringInfoChar(&buf
, ',');
2168 appendStringInfoString(&buf
,
2169 quote_ident_cstr(NameStr(att
->attname
)));
2173 appendStringInfoString(&buf
, ") VALUES(");
2176 * Note: i is physical column number (counting from 0).
2179 for (i
= 0; i
< natts
; i
++)
2181 if (TupleDescAttr(tupdesc
, i
)->attisdropped
)
2185 appendStringInfoChar(&buf
, ',');
2187 key
= get_attnum_pk_pos(pkattnums
, pknumatts
, i
);
2190 val
= tgt_pkattvals
[key
] ? pstrdup(tgt_pkattvals
[key
]) : NULL
;
2192 val
= SPI_getvalue(tuple
, tupdesc
, i
+ 1);
2196 appendStringInfoString(&buf
, quote_literal_cstr(val
));
2200 appendStringInfoString(&buf
, "NULL");
2203 appendStringInfoChar(&buf
, ')');
2209 get_sql_delete(Relation rel
, int *pkattnums
, int pknumatts
, char **tgt_pkattvals
)
2216 initStringInfo(&buf
);
2218 /* get relation name including any needed schema prefix and quoting */
2219 relname
= generate_relation_name(rel
);
2221 tupdesc
= rel
->rd_att
;
2223 appendStringInfo(&buf
, "DELETE FROM %s WHERE ", relname
);
2224 for (i
= 0; i
< pknumatts
; i
++)
2226 int pkattnum
= pkattnums
[i
];
2227 Form_pg_attribute attr
= TupleDescAttr(tupdesc
, pkattnum
);
2230 appendStringInfoString(&buf
, " AND ");
2232 appendStringInfoString(&buf
,
2233 quote_ident_cstr(NameStr(attr
->attname
)));
2235 if (tgt_pkattvals
[i
] != NULL
)
2236 appendStringInfo(&buf
, " = %s",
2237 quote_literal_cstr(tgt_pkattvals
[i
]));
2239 appendStringInfoString(&buf
, " IS NULL");
2246 get_sql_update(Relation rel
, int *pkattnums
, int pknumatts
, char **src_pkattvals
, char **tgt_pkattvals
)
2258 initStringInfo(&buf
);
2260 /* get relation name including any needed schema prefix and quoting */
2261 relname
= generate_relation_name(rel
);
2263 tupdesc
= rel
->rd_att
;
2264 natts
= tupdesc
->natts
;
2266 tuple
= get_tuple_of_interest(rel
, pkattnums
, pknumatts
, src_pkattvals
);
2269 (errcode(ERRCODE_CARDINALITY_VIOLATION
),
2270 errmsg("source row not found")));
2272 appendStringInfo(&buf
, "UPDATE %s SET ", relname
);
2275 * Note: i is physical column number (counting from 0).
2278 for (i
= 0; i
< natts
; i
++)
2280 Form_pg_attribute attr
= TupleDescAttr(tupdesc
, i
);
2282 if (attr
->attisdropped
)
2286 appendStringInfoString(&buf
, ", ");
2288 appendStringInfo(&buf
, "%s = ",
2289 quote_ident_cstr(NameStr(attr
->attname
)));
2291 key
= get_attnum_pk_pos(pkattnums
, pknumatts
, i
);
2294 val
= tgt_pkattvals
[key
] ? pstrdup(tgt_pkattvals
[key
]) : NULL
;
2296 val
= SPI_getvalue(tuple
, tupdesc
, i
+ 1);
2300 appendStringInfoString(&buf
, quote_literal_cstr(val
));
2304 appendStringInfoString(&buf
, "NULL");
2308 appendStringInfoString(&buf
, " WHERE ");
2310 for (i
= 0; i
< pknumatts
; i
++)
2312 int pkattnum
= pkattnums
[i
];
2313 Form_pg_attribute attr
= TupleDescAttr(tupdesc
, pkattnum
);
2316 appendStringInfoString(&buf
, " AND ");
2318 appendStringInfoString(&buf
,
2319 quote_ident_cstr(NameStr(attr
->attname
)));
2321 val
= tgt_pkattvals
[i
];
2324 appendStringInfo(&buf
, " = %s", quote_literal_cstr(val
));
2326 appendStringInfoString(&buf
, " IS NULL");
2333 * Return a properly quoted identifier.
2334 * Uses quote_ident in quote.c
2337 quote_ident_cstr(char *rawstr
)
2343 rawstr_text
= cstring_to_text(rawstr
);
2344 result_text
= DatumGetTextPP(DirectFunctionCall1(quote_ident
,
2345 PointerGetDatum(rawstr_text
)));
2346 result
= text_to_cstring(result_text
);
2352 get_attnum_pk_pos(int *pkattnums
, int pknumatts
, int key
)
2357 * Not likely a long list anyway, so just scan for the value
2359 for (i
= 0; i
< pknumatts
; i
++)
2360 if (key
== pkattnums
[i
])
2367 get_tuple_of_interest(Relation rel
, int *pkattnums
, int pknumatts
, char **src_pkattvals
)
2378 * Connect to SPI manager
2380 if ((ret
= SPI_connect()) < 0)
2381 /* internal error */
2382 elog(ERROR
, "SPI connect failure - returned %d", ret
);
2384 initStringInfo(&buf
);
2386 /* get relation name including any needed schema prefix and quoting */
2387 relname
= generate_relation_name(rel
);
2389 tupdesc
= rel
->rd_att
;
2390 natts
= tupdesc
->natts
;
2393 * Build sql statement to look up tuple of interest, ie, the one matching
2394 * src_pkattvals. We used to use "SELECT *" here, but it's simpler to
2395 * generate a result tuple that matches the table's physical structure,
2396 * with NULLs for any dropped columns. Otherwise we have to deal with two
2397 * different tupdescs and everything's very confusing.
2399 appendStringInfoString(&buf
, "SELECT ");
2401 for (i
= 0; i
< natts
; i
++)
2403 Form_pg_attribute attr
= TupleDescAttr(tupdesc
, i
);
2406 appendStringInfoString(&buf
, ", ");
2408 if (attr
->attisdropped
)
2409 appendStringInfoString(&buf
, "NULL");
2411 appendStringInfoString(&buf
,
2412 quote_ident_cstr(NameStr(attr
->attname
)));
2415 appendStringInfo(&buf
, " FROM %s WHERE ", relname
);
2417 for (i
= 0; i
< pknumatts
; i
++)
2419 int pkattnum
= pkattnums
[i
];
2420 Form_pg_attribute attr
= TupleDescAttr(tupdesc
, pkattnum
);
2423 appendStringInfoString(&buf
, " AND ");
2425 appendStringInfoString(&buf
,
2426 quote_ident_cstr(NameStr(attr
->attname
)));
2428 if (src_pkattvals
[i
] != NULL
)
2429 appendStringInfo(&buf
, " = %s",
2430 quote_literal_cstr(src_pkattvals
[i
]));
2432 appendStringInfoString(&buf
, " IS NULL");
2436 * Retrieve the desired tuple
2438 ret
= SPI_exec(buf
.data
, 0);
2442 * Only allow one qualifying tuple
2444 if ((ret
== SPI_OK_SELECT
) && (SPI_processed
> 1))
2446 (errcode(ERRCODE_CARDINALITY_VIOLATION
),
2447 errmsg("source criteria matched more than one record")));
2449 else if (ret
== SPI_OK_SELECT
&& SPI_processed
== 1)
2451 SPITupleTable
*tuptable
= SPI_tuptable
;
2453 tuple
= SPI_copytuple(tuptable
->vals
[0]);
2461 * no qualifying tuples
2469 * never reached, but keep compiler quiet
2475 * Open the relation named by relname_text, acquire specified type of lock,
2476 * verify we have specified permissions.
2477 * Caller must close rel when done with it.
2480 get_rel_from_relname(text
*relname_text
, LOCKMODE lockmode
, AclMode aclmode
)
2484 AclResult aclresult
;
2486 relvar
= makeRangeVarFromNameList(textToQualifiedNameList(relname_text
));
2487 rel
= table_openrv(relvar
, lockmode
);
2489 aclresult
= pg_class_aclcheck(RelationGetRelid(rel
), GetUserId(),
2491 if (aclresult
!= ACLCHECK_OK
)
2492 aclcheck_error(aclresult
, get_relkind_objtype(rel
->rd_rel
->relkind
),
2493 RelationGetRelationName(rel
));
2499 * generate_relation_name - copied from ruleutils.c
2500 * Compute the name to display for a relation
2502 * The result includes all necessary quoting and schema-prefixing.
2505 generate_relation_name(Relation rel
)
2510 /* Qualify the name if not visible in search path */
2511 if (RelationIsVisible(RelationGetRelid(rel
)))
2514 nspname
= get_namespace_name(rel
->rd_rel
->relnamespace
);
2516 result
= quote_qualified_identifier(nspname
, RelationGetRelationName(rel
));
2523 getConnectionByName(const char *name
)
2525 remoteConnHashEnt
*hentry
;
2528 if (!remoteConnHash
)
2529 remoteConnHash
= createConnHash();
2531 key
= pstrdup(name
);
2532 truncate_identifier(key
, strlen(key
), false);
2533 hentry
= (remoteConnHashEnt
*) hash_search(remoteConnHash
,
2534 key
, HASH_FIND
, NULL
);
2537 return hentry
->rconn
;
2543 createConnHash(void)
2547 ctl
.keysize
= NAMEDATALEN
;
2548 ctl
.entrysize
= sizeof(remoteConnHashEnt
);
2550 return hash_create("Remote Con hash", NUMCONN
, &ctl
,
2551 HASH_ELEM
| HASH_STRINGS
);
2555 createNewConnection(const char *name
, remoteConn
*rconn
)
2557 remoteConnHashEnt
*hentry
;
2561 if (!remoteConnHash
)
2562 remoteConnHash
= createConnHash();
2564 key
= pstrdup(name
);
2565 truncate_identifier(key
, strlen(key
), true);
2566 hentry
= (remoteConnHashEnt
*) hash_search(remoteConnHash
, key
,
2567 HASH_ENTER
, &found
);
2571 libpqsrv_disconnect(rconn
->conn
);
2575 (errcode(ERRCODE_DUPLICATE_OBJECT
),
2576 errmsg("duplicate connection name")));
2579 hentry
->rconn
= rconn
;
2583 deleteConnection(const char *name
)
2585 remoteConnHashEnt
*hentry
;
2589 if (!remoteConnHash
)
2590 remoteConnHash
= createConnHash();
2592 key
= pstrdup(name
);
2593 truncate_identifier(key
, strlen(key
), false);
2594 hentry
= (remoteConnHashEnt
*) hash_search(remoteConnHash
,
2595 key
, HASH_REMOVE
, &found
);
2599 (errcode(ERRCODE_UNDEFINED_OBJECT
),
2600 errmsg("undefined connection name")));
2604 * We need to make sure that the connection made used credentials
2605 * which were provided by the user, so check what credentials were
2606 * used to connect and then make sure that they came from the user.
2609 dblink_security_check(PGconn
*conn
, remoteConn
*rconn
, const char *connstr
)
2611 /* Superuser bypasses security check */
2615 /* If password was used to connect, make sure it was one provided */
2616 if (PQconnectionUsedPassword(conn
) && dblink_connstr_has_pw(connstr
))
2620 /* If GSSAPI creds used to connect, make sure it was one delegated */
2621 if (PQconnectionUsedGSSAPI(conn
) && be_gssapi_get_delegation(MyProcPort
))
2625 /* Otherwise, fail out */
2626 libpqsrv_disconnect(conn
);
2631 (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED
),
2632 errmsg("password or GSSAPI delegated credentials required"),
2633 errdetail("Non-superusers may only connect using credentials they provide, eg: password in connection string or delegated GSSAPI credentials"),
2634 errhint("Ensure provided credentials match target server's authentication method.")));
2638 * Function to check if the connection string includes an explicit
2639 * password, needed to ensure that non-superuser password-based auth
2640 * is using a provided password and not one picked up from the
2644 dblink_connstr_has_pw(const char *connstr
)
2646 PQconninfoOption
*options
;
2647 PQconninfoOption
*option
;
2648 bool connstr_gives_password
= false;
2650 options
= PQconninfoParse(connstr
, NULL
);
2653 for (option
= options
; option
->keyword
!= NULL
; option
++)
2655 if (strcmp(option
->keyword
, "password") == 0)
2657 if (option
->val
!= NULL
&& option
->val
[0] != '\0')
2659 connstr_gives_password
= true;
2664 PQconninfoFree(options
);
2667 return connstr_gives_password
;
2671 * For non-superusers, insist that the connstr specify a password, except
2672 * if GSSAPI credentials have been delegated (and we check that they are used
2673 * for the connection in dblink_security_check later). This prevents a
2674 * password or GSSAPI credentials from being picked up from .pgpass, a
2675 * service file, the environment, etc. We don't want the postgres user's
2676 * passwords or Kerberos credentials to be accessible to non-superusers.
2679 dblink_connstr_check(const char *connstr
)
2684 if (dblink_connstr_has_pw(connstr
))
2688 if (be_gssapi_get_delegation(MyProcPort
))
2693 (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED
),
2694 errmsg("password or GSSAPI delegated credentials required"),
2695 errdetail("Non-superusers must provide a password in the connection string or send delegated GSSAPI credentials.")));
2699 * Report an error received from the remote server
2701 * res: the received error result (will be freed)
2702 * fail: true for ERROR ereport, false for NOTICE
2703 * fmt and following args: sprintf-style format and values for errcontext;
2704 * the resulting string should be worded like "while <some action>"
2707 dblink_res_error(PGconn
*conn
, const char *conname
, PGresult
*res
,
2708 bool fail
, const char *fmt
,...)
2711 char *pg_diag_sqlstate
= PQresultErrorField(res
, PG_DIAG_SQLSTATE
);
2712 char *pg_diag_message_primary
= PQresultErrorField(res
, PG_DIAG_MESSAGE_PRIMARY
);
2713 char *pg_diag_message_detail
= PQresultErrorField(res
, PG_DIAG_MESSAGE_DETAIL
);
2714 char *pg_diag_message_hint
= PQresultErrorField(res
, PG_DIAG_MESSAGE_HINT
);
2715 char *pg_diag_context
= PQresultErrorField(res
, PG_DIAG_CONTEXT
);
2717 char *message_primary
;
2718 char *message_detail
;
2720 char *message_context
;
2722 char dblink_context_msg
[512];
2729 if (pg_diag_sqlstate
)
2730 sqlstate
= MAKE_SQLSTATE(pg_diag_sqlstate
[0],
2731 pg_diag_sqlstate
[1],
2732 pg_diag_sqlstate
[2],
2733 pg_diag_sqlstate
[3],
2734 pg_diag_sqlstate
[4]);
2736 sqlstate
= ERRCODE_CONNECTION_FAILURE
;
2738 message_primary
= xpstrdup(pg_diag_message_primary
);
2739 message_detail
= xpstrdup(pg_diag_message_detail
);
2740 message_hint
= xpstrdup(pg_diag_message_hint
);
2741 message_context
= xpstrdup(pg_diag_context
);
2744 * If we don't get a message from the PGresult, try the PGconn. This is
2745 * needed because for connection-level failures, PQgetResult may just
2746 * return NULL, not a PGresult at all.
2748 if (message_primary
== NULL
)
2749 message_primary
= pchomp(PQerrorMessage(conn
));
2752 * Now that we've copied all the data we need out of the PGresult, it's
2753 * safe to free it. We must do this to avoid PGresult leakage. We're
2754 * leaking all the strings too, but those are in palloc'd memory that will
2755 * get cleaned up eventually.
2760 * Format the basic errcontext string. Below, we'll add on something
2761 * about the connection name. That's a violation of the translatability
2762 * guidelines about constructing error messages out of parts, but since
2763 * there's no translation support for dblink, there's no need to worry
2767 vsnprintf(dblink_context_msg
, sizeof(dblink_context_msg
), fmt
, ap
);
2772 (message_primary
!= NULL
&& message_primary
[0] != '\0') ?
2773 errmsg_internal("%s", message_primary
) :
2774 errmsg("could not obtain message string for remote error"),
2775 message_detail
? errdetail_internal("%s", message_detail
) : 0,
2776 message_hint
? errhint("%s", message_hint
) : 0,
2777 message_context
? (errcontext("%s", message_context
)) : 0,
2779 (errcontext("%s on dblink connection named \"%s\"",
2780 dblink_context_msg
, conname
)) :
2781 (errcontext("%s on unnamed dblink connection",
2782 dblink_context_msg
))));
2786 * Obtain connection string for a foreign server
2789 get_connect_string(const char *servername
)
2791 ForeignServer
*foreign_server
= NULL
;
2792 UserMapping
*user_mapping
;
2795 ForeignDataWrapper
*fdw
;
2796 AclResult aclresult
;
2799 static const PQconninfoOption
*options
= NULL
;
2801 initStringInfo(&buf
);
2804 * Get list of valid libpq options.
2806 * To avoid unnecessary work, we get the list once and use it throughout
2807 * the lifetime of this backend process. We don't need to care about
2808 * memory context issues, because PQconndefaults allocates with malloc.
2812 options
= PQconndefaults();
2813 if (!options
) /* assume reason for failure is OOM */
2815 (errcode(ERRCODE_FDW_OUT_OF_MEMORY
),
2816 errmsg("out of memory"),
2817 errdetail("Could not get libpq's default connection options.")));
2820 /* first gather the server connstr options */
2821 srvname
= pstrdup(servername
);
2822 truncate_identifier(srvname
, strlen(srvname
), false);
2823 foreign_server
= GetForeignServerByName(srvname
, true);
2827 Oid serverid
= foreign_server
->serverid
;
2828 Oid fdwid
= foreign_server
->fdwid
;
2829 Oid userid
= GetUserId();
2831 user_mapping
= GetUserMapping(userid
, serverid
);
2832 fdw
= GetForeignDataWrapper(fdwid
);
2834 /* Check permissions, user must have usage on the server. */
2835 aclresult
= object_aclcheck(ForeignServerRelationId
, serverid
, userid
, ACL_USAGE
);
2836 if (aclresult
!= ACLCHECK_OK
)
2837 aclcheck_error(aclresult
, OBJECT_FOREIGN_SERVER
, foreign_server
->servername
);
2839 foreach(cell
, fdw
->options
)
2841 DefElem
*def
= lfirst(cell
);
2843 if (is_valid_dblink_option(options
, def
->defname
, ForeignDataWrapperRelationId
))
2844 appendStringInfo(&buf
, "%s='%s' ", def
->defname
,
2845 escape_param_str(strVal(def
->arg
)));
2848 foreach(cell
, foreign_server
->options
)
2850 DefElem
*def
= lfirst(cell
);
2852 if (is_valid_dblink_option(options
, def
->defname
, ForeignServerRelationId
))
2853 appendStringInfo(&buf
, "%s='%s' ", def
->defname
,
2854 escape_param_str(strVal(def
->arg
)));
2857 foreach(cell
, user_mapping
->options
)
2860 DefElem
*def
= lfirst(cell
);
2862 if (is_valid_dblink_option(options
, def
->defname
, UserMappingRelationId
))
2863 appendStringInfo(&buf
, "%s='%s' ", def
->defname
,
2864 escape_param_str(strVal(def
->arg
)));
2874 * Escaping libpq connect parameter strings.
2876 * Replaces "'" with "\'" and "\" with "\\".
2879 escape_param_str(const char *str
)
2884 initStringInfo(&buf
);
2886 for (cp
= str
; *cp
; cp
++)
2888 if (*cp
== '\\' || *cp
== '\'')
2889 appendStringInfoChar(&buf
, '\\');
2890 appendStringInfoChar(&buf
, *cp
);
2897 * Validate the PK-attnums argument for dblink_build_sql_insert() and related
2898 * functions, and translate to the internal representation.
2900 * The user supplies an int2vector of 1-based logical attnums, plus a count
2901 * argument (the need for the separate count argument is historical, but we
2902 * still check it). We check that each attnum corresponds to a valid,
2903 * non-dropped attribute of the rel. We do *not* prevent attnums from being
2904 * listed twice, though the actual use-case for such things is dubious.
2905 * Note that before Postgres 9.0, the user's attnums were interpreted as
2906 * physical not logical column numbers; this was changed for future-proofing.
2908 * The internal representation is a palloc'd int array of 0-based physical
2912 validate_pkattnums(Relation rel
,
2913 int2vector
*pkattnums_arg
, int32 pknumatts_arg
,
2914 int **pkattnums
, int *pknumatts
)
2916 TupleDesc tupdesc
= rel
->rd_att
;
2917 int natts
= tupdesc
->natts
;
2920 /* Don't take more array elements than there are */
2921 pknumatts_arg
= Min(pknumatts_arg
, pkattnums_arg
->dim1
);
2923 /* Must have at least one pk attnum selected */
2924 if (pknumatts_arg
<= 0)
2926 (errcode(ERRCODE_INVALID_PARAMETER_VALUE
),
2927 errmsg("number of key attributes must be > 0")));
2929 /* Allocate output array */
2930 *pkattnums
= palloc_array(int, pknumatts_arg
);
2931 *pknumatts
= pknumatts_arg
;
2933 /* Validate attnums and convert to internal form */
2934 for (i
= 0; i
< pknumatts_arg
; i
++)
2936 int pkattnum
= pkattnums_arg
->values
[i
];
2940 /* Can throw error immediately if out of range */
2941 if (pkattnum
<= 0 || pkattnum
> natts
)
2943 (errcode(ERRCODE_INVALID_PARAMETER_VALUE
),
2944 errmsg("invalid attribute number %d", pkattnum
)));
2946 /* Identify which physical column has this logical number */
2948 for (j
= 0; j
< natts
; j
++)
2950 /* dropped columns don't count */
2951 if (TupleDescAttr(tupdesc
, j
)->attisdropped
)
2954 if (++lnum
== pkattnum
)
2959 (*pkattnums
)[i
] = j
;
2962 (errcode(ERRCODE_INVALID_PARAMETER_VALUE
),
2963 errmsg("invalid attribute number %d", pkattnum
)));
2968 * Check if the specified connection option is valid.
2970 * We basically allow whatever libpq thinks is an option, with these
2972 * debug options: disallowed
2973 * "client_encoding": disallowed
2974 * "user": valid only in USER MAPPING options
2975 * secure options (eg password): valid only in USER MAPPING options
2976 * others: valid only in FOREIGN SERVER options
2978 * We disallow client_encoding because it would be overridden anyway via
2979 * PQclientEncoding; allowing it to be specified would merely promote
2983 is_valid_dblink_option(const PQconninfoOption
*options
, const char *option
,
2986 const PQconninfoOption
*opt
;
2988 /* Look up the option in libpq result */
2989 for (opt
= options
; opt
->keyword
; opt
++)
2991 if (strcmp(opt
->keyword
, option
) == 0)
2994 if (opt
->keyword
== NULL
)
2997 /* Disallow debug options (particularly "replication") */
2998 if (strchr(opt
->dispchar
, 'D'))
3001 /* Disallow "client_encoding" */
3002 if (strcmp(opt
->keyword
, "client_encoding") == 0)
3006 * If the option is "user" or marked secure, it should be specified only
3007 * in USER MAPPING. Others should be specified only in SERVER.
3009 if (strcmp(opt
->keyword
, "user") == 0 || strchr(opt
->dispchar
, '*'))
3011 if (context
!= UserMappingRelationId
)
3016 if (context
!= ForeignServerRelationId
)
3024 * Copy the remote session's values of GUCs that affect datatype I/O
3025 * and apply them locally in a new GUC nesting level. Returns the new
3026 * nestlevel (which is needed by restoreLocalGucs to undo the settings),
3027 * or -1 if no new nestlevel was needed.
3029 * We use the equivalent of a function SET option to allow the settings to
3030 * persist only until the caller calls restoreLocalGucs. If an error is
3031 * thrown in between, guc.c will take care of undoing the settings.
3034 applyRemoteGucs(PGconn
*conn
)
3036 static const char *const GUCsAffectingIO
[] = {
3044 for (i
= 0; i
< lengthof(GUCsAffectingIO
); i
++)
3046 const char *gucName
= GUCsAffectingIO
[i
];
3047 const char *remoteVal
= PQparameterStatus(conn
, gucName
);
3048 const char *localVal
;
3051 * If the remote server is pre-8.4, it won't have IntervalStyle, but
3052 * that's okay because its output format won't be ambiguous. So just
3053 * skip the GUC if we don't get a value for it. (We might eventually
3054 * need more complicated logic with remote-version checks here.)
3056 if (remoteVal
== NULL
)
3060 * Avoid GUC-setting overhead if the remote and local GUCs already
3061 * have the same value.
3063 localVal
= GetConfigOption(gucName
, false, false);
3064 Assert(localVal
!= NULL
);
3066 if (strcmp(remoteVal
, localVal
) == 0)
3069 /* Create new GUC nest level if we didn't already */
3071 nestlevel
= NewGUCNestLevel();
3073 /* Apply the option (this will throw error on failure) */
3074 (void) set_config_option(gucName
, remoteVal
,
3075 PGC_USERSET
, PGC_S_SESSION
,
3076 GUC_ACTION_SAVE
, true, 0, false);
3083 * Restore local GUCs after they have been overlaid with remote settings.
3086 restoreLocalGucs(int nestlevel
)
3088 /* Do nothing if no new nestlevel was created */
3090 AtEOXact_GUC(true, nestlevel
);