Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * connection.c
4 : * Connection management functions for postgres_fdw
5 : *
6 : * Portions Copyright (c) 2012-2021, PostgreSQL Global Development Group
7 : *
8 : * IDENTIFICATION
9 : * contrib/postgres_fdw/connection.c
10 : *
11 : *-------------------------------------------------------------------------
12 : */
13 : #include "postgres.h"
14 :
15 : #include "access/htup_details.h"
16 : #include "access/xact.h"
17 : #include "catalog/pg_user_mapping.h"
18 : #include "commands/defrem.h"
19 : #include "funcapi.h"
20 : #include "mb/pg_wchar.h"
21 : #include "miscadmin.h"
22 : #include "pgstat.h"
23 : #include "postgres_fdw.h"
24 : #include "storage/fd.h"
25 : #include "storage/latch.h"
26 : #include "utils/builtins.h"
27 : #include "utils/datetime.h"
28 : #include "utils/hsearch.h"
29 : #include "utils/inval.h"
30 : #include "utils/memutils.h"
31 : #include "utils/syscache.h"
32 :
33 : /*
34 : * Connection cache hash table entry
35 : *
36 : * The lookup key in this hash table is the user mapping OID. We use just one
37 : * connection per user mapping ID, which ensures that all the scans use the
38 : * same snapshot during a query. Using the user mapping OID rather than
39 : * the foreign server OID + user OID avoids creating multiple connections when
40 : * the public user mapping applies to all user OIDs.
41 : *
42 : * The "conn" pointer can be NULL if we don't currently have a live connection.
43 : * When we do have a connection, xact_depth tracks the current depth of
44 : * transactions and subtransactions open on the remote side. We need to issue
45 : * commands at the same nesting depth on the remote as we're executing at
46 : * ourselves, so that rolling back a subtransaction will kill the right
47 : * queries and not the wrong ones.
48 : */
49 : typedef Oid ConnCacheKey;
50 :
51 : typedef struct ConnCacheEntry
52 : {
53 : ConnCacheKey key; /* hash key (must be first) */
54 : PGconn *conn; /* connection to foreign server, or NULL */
55 : /* Remaining fields are invalid when conn is NULL: */
56 : int xact_depth; /* 0 = no xact open, 1 = main xact open, 2 =
57 : * one level of subxact open, etc */
58 : bool have_prep_stmt; /* have we prepared any stmts in this xact? */
59 : bool have_error; /* have any subxacts aborted in this xact? */
60 : bool changing_xact_state; /* xact state change in process */
61 : bool invalidated; /* true if reconnect is pending */
62 : bool keep_connections; /* setting value of keep_connections
63 : * server option */
64 : Oid serverid; /* foreign server OID used to get server name */
65 : uint32 server_hashvalue; /* hash value of foreign server OID */
66 : uint32 mapping_hashvalue; /* hash value of user mapping OID */
67 : PgFdwConnState state; /* extra per-connection state */
68 : } ConnCacheEntry;
69 :
70 : /*
71 : * Connection cache (initialized on first use)
72 : */
73 : static HTAB *ConnectionHash = NULL;
74 :
75 : /* for assigning cursor numbers and prepared statement numbers */
76 : static unsigned int cursor_number = 0;
77 : static unsigned int prep_stmt_number = 0;
78 :
79 : /* tracks whether any work is needed in callback functions */
80 : static bool xact_got_connection = false;
81 :
82 : /*
83 : * SQL functions
84 : */
85 4 : PG_FUNCTION_INFO_V1(postgres_fdw_get_connections);
86 4 : PG_FUNCTION_INFO_V1(postgres_fdw_disconnect);
87 4 : PG_FUNCTION_INFO_V1(postgres_fdw_disconnect_all);
88 :
89 : /* prototypes of private functions */
90 : static void make_new_connection(ConnCacheEntry *entry, UserMapping *user);
91 : static PGconn *connect_pg_server(ForeignServer *server, UserMapping *user);
92 : static void disconnect_pg_server(ConnCacheEntry *entry);
93 : static void check_conn_params(const char **keywords, const char **values, UserMapping *user);
94 : static void configure_remote_session(PGconn *conn);
95 : static void begin_remote_xact(ConnCacheEntry *entry);
96 : static void pgfdw_xact_callback(XactEvent event, void *arg);
97 : static void pgfdw_subxact_callback(SubXactEvent event,
98 : SubTransactionId mySubid,
99 : SubTransactionId parentSubid,
100 : void *arg);
101 : static void pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue);
102 : static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry);
103 : static bool pgfdw_cancel_query(PGconn *conn);
104 : static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query,
105 : bool ignore_errors);
106 : static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime,
107 : PGresult **result);
108 : static bool UserMappingPasswordRequired(UserMapping *user);
109 : static bool disconnect_cached_connections(Oid serverid);
110 :
111 : /*
112 : * Get a PGconn which can be used to execute queries on the remote PostgreSQL
113 : * server with the user's authorization. A new connection is established
114 : * if we don't already have a suitable one, and a transaction is opened at
115 : * the right subtransaction nesting depth if we didn't do that already.
116 : *
117 : * will_prep_stmt must be true if caller intends to create any prepared
118 : * statements. Since those don't go away automatically at transaction end
119 : * (not even on error), we need this flag to cue manual cleanup.
120 : *
121 : * If state is not NULL, *state receives the per-connection state associated
122 : * with the PGconn.
123 : */
124 : PGconn *
125 3102 : GetConnection(UserMapping *user, bool will_prep_stmt, PgFdwConnState **state)
126 : {
127 : bool found;
128 3102 : bool retry = false;
129 : ConnCacheEntry *entry;
130 : ConnCacheKey key;
131 3102 : MemoryContext ccxt = CurrentMemoryContext;
132 :
133 : /* First time through, initialize connection cache hashtable */
134 3102 : if (ConnectionHash == NULL)
135 : {
136 : HASHCTL ctl;
137 :
138 6 : ctl.keysize = sizeof(ConnCacheKey);
139 6 : ctl.entrysize = sizeof(ConnCacheEntry);
140 6 : ConnectionHash = hash_create("postgres_fdw connections", 8,
141 : &ctl,
142 : HASH_ELEM | HASH_BLOBS);
143 :
144 : /*
145 : * Register some callback functions that manage connection cleanup.
146 : * This should be done just once in each backend.
147 : */
148 6 : RegisterXactCallback(pgfdw_xact_callback, NULL);
149 6 : RegisterSubXactCallback(pgfdw_subxact_callback, NULL);
150 6 : CacheRegisterSyscacheCallback(FOREIGNSERVEROID,
151 : pgfdw_inval_callback, (Datum) 0);
152 6 : CacheRegisterSyscacheCallback(USERMAPPINGOID,
153 : pgfdw_inval_callback, (Datum) 0);
154 : }
155 :
156 : /* Set flag that we did GetConnection during the current transaction */
157 3102 : xact_got_connection = true;
158 :
159 : /* Create hash key for the entry. Assume no pad bytes in key struct */
160 3102 : key = user->umid;
161 :
162 : /*
163 : * Find or create cached entry for requested connection.
164 : */
165 3102 : entry = hash_search(ConnectionHash, &key, HASH_ENTER, &found);
166 3102 : if (!found)
167 : {
168 : /*
169 : * We need only clear "conn" here; remaining fields will be filled
170 : * later when "conn" is set.
171 : */
172 20 : entry->conn = NULL;
173 : }
174 :
175 : /* Reject further use of connections which failed abort cleanup. */
176 3102 : pgfdw_reject_incomplete_xact_state_change(entry);
177 :
178 : /*
179 : * If the connection needs to be remade due to invalidation, disconnect as
180 : * soon as we're out of all transactions.
181 : */
182 3102 : if (entry->conn != NULL && entry->invalidated && entry->xact_depth == 0)
183 : {
184 0 : elog(DEBUG3, "closing connection %p for option changes to take effect",
185 : entry->conn);
186 0 : disconnect_pg_server(entry);
187 : }
188 :
189 : /*
190 : * If cache entry doesn't have a connection, we have to establish a new
191 : * connection. (If connect_pg_server throws an error, the cache entry
192 : * will remain in a valid empty state, ie conn == NULL.)
193 : */
194 3102 : if (entry->conn == NULL)
195 78 : make_new_connection(entry, user);
196 :
197 : /*
198 : * We check the health of the cached connection here when starting a new
199 : * remote transaction. If a broken connection is detected, we try to
200 : * reestablish a new connection later.
201 : */
202 3090 : PG_TRY();
203 : {
204 : /* Process a pending asynchronous request if any. */
205 3090 : if (entry->state.pendingAreq)
206 0 : process_pending_request(entry->state.pendingAreq);
207 : /* Start a new transaction or subtransaction if needed. */
208 3090 : begin_remote_xact(entry);
209 : }
210 4 : PG_CATCH();
211 : {
212 4 : MemoryContext ecxt = MemoryContextSwitchTo(ccxt);
213 4 : ErrorData *errdata = CopyErrorData();
214 :
215 : /*
216 : * If connection failure is reported when starting a new remote
217 : * transaction (not subtransaction), new connection will be
218 : * reestablished later.
219 : *
220 : * After a broken connection is detected in libpq, any error other
221 : * than connection failure (e.g., out-of-memory) can be thrown
222 : * somewhere between return from libpq and the expected ereport() call
223 : * in pgfdw_report_error(). In this case, since PQstatus() indicates
224 : * CONNECTION_BAD, checking only PQstatus() causes the false detection
225 : * of connection failure. To avoid this, we also verify that the
226 : * error's sqlstate is ERRCODE_CONNECTION_FAILURE. Note that also
227 : * checking only the sqlstate can cause another false detection
228 : * because pgfdw_report_error() may report ERRCODE_CONNECTION_FAILURE
229 : * for any libpq-originated error condition.
230 : */
231 4 : if (errdata->sqlerrcode != ERRCODE_CONNECTION_FAILURE ||
232 4 : PQstatus(entry->conn) != CONNECTION_BAD ||
233 4 : entry->xact_depth > 0)
234 : {
235 2 : MemoryContextSwitchTo(ecxt);
236 2 : PG_RE_THROW();
237 : }
238 :
239 : /* Clean up the error state */
240 2 : FlushErrorState();
241 2 : FreeErrorData(errdata);
242 2 : errdata = NULL;
243 :
244 2 : retry = true;
245 : }
246 3088 : PG_END_TRY();
247 :
248 : /*
249 : * If a broken connection is detected, disconnect it, reestablish a new
250 : * connection and retry a new remote transaction. If connection failure is
251 : * reported again, we give up getting a connection.
252 : */
253 3088 : if (retry)
254 : {
255 : Assert(entry->xact_depth == 0);
256 :
257 2 : ereport(DEBUG3,
258 : (errmsg_internal("could not start remote transaction on connection %p",
259 : entry->conn)),
260 : errdetail_internal("%s", pchomp(PQerrorMessage(entry->conn))));
261 :
262 2 : elog(DEBUG3, "closing connection %p to reestablish a new one",
263 : entry->conn);
264 2 : disconnect_pg_server(entry);
265 :
266 2 : if (entry->conn == NULL)
267 2 : make_new_connection(entry, user);
268 :
269 2 : begin_remote_xact(entry);
270 : }
271 :
272 : /* Remember if caller will prepare statements */
273 3088 : entry->have_prep_stmt |= will_prep_stmt;
274 :
275 : /* If caller needs access to the per-connection state, return it. */
276 3088 : if (state)
277 1122 : *state = &entry->state;
278 :
279 3088 : return entry->conn;
280 : }
281 :
282 : /*
283 : * Reset all transient state fields in the cached connection entry and
284 : * establish new connection to the remote server.
285 : */
286 : static void
287 80 : make_new_connection(ConnCacheEntry *entry, UserMapping *user)
288 : {
289 80 : ForeignServer *server = GetForeignServer(user->serverid);
290 : ListCell *lc;
291 :
292 : Assert(entry->conn == NULL);
293 :
294 : /* Reset all transient state fields, to be sure all are clean */
295 80 : entry->xact_depth = 0;
296 80 : entry->have_prep_stmt = false;
297 80 : entry->have_error = false;
298 80 : entry->changing_xact_state = false;
299 80 : entry->invalidated = false;
300 80 : entry->serverid = server->serverid;
301 80 : entry->server_hashvalue =
302 80 : GetSysCacheHashValue1(FOREIGNSERVEROID,
303 : ObjectIdGetDatum(server->serverid));
304 80 : entry->mapping_hashvalue =
305 80 : GetSysCacheHashValue1(USERMAPPINGOID,
306 : ObjectIdGetDatum(user->umid));
307 80 : memset(&entry->state, 0, sizeof(entry->state));
308 :
309 : /*
310 : * Determine whether to keep the connection that we're about to make here
311 : * open even after the transaction using it ends, so that the subsequent
312 : * transactions can re-use it.
313 : *
314 : * It's enough to determine this only when making new connection because
315 : * all the connections to the foreign server whose keep_connections option
316 : * is changed will be closed and re-made later.
317 : *
318 : * By default, all the connections to any foreign servers are kept open.
319 : */
320 80 : entry->keep_connections = true;
321 324 : foreach(lc, server->options)
322 : {
323 244 : DefElem *def = (DefElem *) lfirst(lc);
324 :
325 244 : if (strcmp(def->defname, "keep_connections") == 0)
326 6 : entry->keep_connections = defGetBoolean(def);
327 : }
328 :
329 : /* Now try to make the connection */
330 80 : entry->conn = connect_pg_server(server, user);
331 :
332 68 : elog(DEBUG3, "new postgres_fdw connection %p for server \"%s\" (user mapping oid %u, userid %u)",
333 : entry->conn, server->servername, user->umid, user->userid);
334 68 : }
335 :
336 : /*
337 : * Connect to remote server using specified server and user mapping properties.
338 : */
339 : static PGconn *
340 80 : connect_pg_server(ForeignServer *server, UserMapping *user)
341 : {
342 80 : PGconn *volatile conn = NULL;
343 :
344 : /*
345 : * Use PG_TRY block to ensure closing connection on error.
346 : */
347 80 : PG_TRY();
348 : {
349 : const char **keywords;
350 : const char **values;
351 : int n;
352 :
353 : /*
354 : * Construct connection params from generic options of ForeignServer
355 : * and UserMapping. (Some of them might not be libpq options, in
356 : * which case we'll just waste a few array slots.) Add 3 extra slots
357 : * for fallback_application_name, client_encoding, end marker.
358 : */
359 80 : n = list_length(server->options) + list_length(user->options) + 3;
360 80 : keywords = (const char **) palloc(n * sizeof(char *));
361 80 : values = (const char **) palloc(n * sizeof(char *));
362 :
363 80 : n = 0;
364 240 : n += ExtractConnectionOptions(server->options,
365 80 : keywords + n, values + n);
366 240 : n += ExtractConnectionOptions(user->options,
367 80 : keywords + n, values + n);
368 :
369 : /* Use "postgres_fdw" as fallback_application_name. */
370 80 : keywords[n] = "fallback_application_name";
371 80 : values[n] = "postgres_fdw";
372 80 : n++;
373 :
374 : /* Set client_encoding so that libpq can convert encoding properly. */
375 80 : keywords[n] = "client_encoding";
376 80 : values[n] = GetDatabaseEncodingName();
377 80 : n++;
378 :
379 80 : keywords[n] = values[n] = NULL;
380 :
381 : /* verify the set of connection parameters */
382 80 : check_conn_params(keywords, values, user);
383 :
384 : /*
385 : * We must obey fd.c's limit on non-virtual file descriptors. Assume
386 : * that a PGconn represents one long-lived FD. (Doing this here also
387 : * ensures that VFDs are closed if needed to make room.)
388 : */
389 76 : if (!AcquireExternalFD())
390 : {
391 : #ifndef WIN32 /* can't write #if within ereport() macro */
392 0 : ereport(ERROR,
393 : (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
394 : errmsg("could not connect to server \"%s\"",
395 : server->servername),
396 : errdetail("There are too many open files on the local server."),
397 : errhint("Raise the server's max_files_per_process and/or \"ulimit -n\" limits.")));
398 : #else
399 : ereport(ERROR,
400 : (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
401 : errmsg("could not connect to server \"%s\"",
402 : server->servername),
403 : errdetail("There are too many open files on the local server."),
404 : errhint("Raise the server's max_files_per_process setting.")));
405 : #endif
406 : }
407 :
408 : /* OK to make connection */
409 76 : conn = PQconnectdbParams(keywords, values, false);
410 :
411 76 : if (!conn)
412 0 : ReleaseExternalFD(); /* because the PG_CATCH block won't */
413 :
414 76 : if (!conn || PQstatus(conn) != CONNECTION_OK)
415 4 : ereport(ERROR,
416 : (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
417 : errmsg("could not connect to server \"%s\"",
418 : server->servername),
419 : errdetail_internal("%s", pchomp(PQerrorMessage(conn)))));
420 :
421 : /*
422 : * Check that non-superuser has used password to establish connection;
423 : * otherwise, he's piggybacking on the postgres server's user
424 : * identity. See also dblink_security_check() in contrib/dblink and
425 : * check_conn_params.
426 : */
427 72 : if (!superuser_arg(user->userid) && UserMappingPasswordRequired(user) &&
428 4 : !PQconnectionUsedPassword(conn))
429 4 : ereport(ERROR,
430 : (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
431 : errmsg("password is required"),
432 : errdetail("Non-superuser cannot connect if the server does not request a password."),
433 : errhint("Target server's authentication method must be changed or password_required=false set in the user mapping attributes.")));
434 :
435 : /* Prepare new session for use */
436 68 : configure_remote_session(conn);
437 :
438 68 : pfree(keywords);
439 68 : pfree(values);
440 : }
441 12 : PG_CATCH();
442 : {
443 : /* Release PGconn data structure if we managed to create one */
444 12 : if (conn)
445 : {
446 8 : PQfinish(conn);
447 8 : ReleaseExternalFD();
448 : }
449 12 : PG_RE_THROW();
450 : }
451 68 : PG_END_TRY();
452 :
453 68 : return conn;
454 : }
455 :
456 : /*
457 : * Disconnect any open connection for a connection cache entry.
458 : */
459 : static void
460 64 : disconnect_pg_server(ConnCacheEntry *entry)
461 : {
462 64 : if (entry->conn != NULL)
463 : {
464 64 : PQfinish(entry->conn);
465 64 : entry->conn = NULL;
466 64 : ReleaseExternalFD();
467 : }
468 64 : }
469 :
470 : /*
471 : * Return true if the password_required is defined and false for this user
472 : * mapping, otherwise false. The mapping has been pre-validated.
473 : */
474 : static bool
475 10 : UserMappingPasswordRequired(UserMapping *user)
476 : {
477 : ListCell *cell;
478 :
479 16 : foreach(cell, user->options)
480 : {
481 8 : DefElem *def = (DefElem *) lfirst(cell);
482 :
483 8 : if (strcmp(def->defname, "password_required") == 0)
484 2 : return defGetBoolean(def);
485 : }
486 :
487 8 : return true;
488 : }
489 :
490 : /*
491 : * For non-superusers, insist that the connstr specify a password. This
492 : * prevents a password from being picked up from .pgpass, a service file, the
493 : * environment, etc. We don't want the postgres user's passwords,
494 : * certificates, etc to be accessible to non-superusers. (See also
495 : * dblink_connstr_check in contrib/dblink.)
496 : */
497 : static void
498 80 : check_conn_params(const char **keywords, const char **values, UserMapping *user)
499 : {
500 : int i;
501 :
502 : /* no check required if superuser */
503 80 : if (superuser_arg(user->userid))
504 70 : return;
505 :
506 : /* ok if params contain a non-empty password */
507 38 : for (i = 0; keywords[i] != NULL; i++)
508 : {
509 34 : if (strcmp(keywords[i], "password") == 0 && values[i][0] != '\0')
510 6 : return;
511 : }
512 :
513 : /* ok if the superuser explicitly said so at user mapping creation time */
514 4 : if (!UserMappingPasswordRequired(user))
515 0 : return;
516 :
517 4 : ereport(ERROR,
518 : (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
519 : errmsg("password is required"),
520 : errdetail("Non-superusers must provide a password in the user mapping.")));
521 : }
522 :
523 : /*
524 : * Issue SET commands to make sure remote session is configured properly.
525 : *
526 : * We do this just once at connection, assuming nothing will change the
527 : * values later. Since we'll never send volatile function calls to the
528 : * remote, there shouldn't be any way to break this assumption from our end.
529 : * It's possible to think of ways to break it at the remote end, eg making
530 : * a foreign table point to a view that includes a set_config call ---
531 : * but once you admit the possibility of a malicious view definition,
532 : * there are any number of ways to break things.
533 : */
534 : static void
535 68 : configure_remote_session(PGconn *conn)
536 : {
537 68 : int remoteversion = PQserverVersion(conn);
538 :
539 : /* Force the search path to contain only pg_catalog (see deparse.c) */
540 68 : do_sql_command(conn, "SET search_path = pg_catalog");
541 :
542 : /*
543 : * Set remote timezone; this is basically just cosmetic, since all
544 : * transmitted and returned timestamptzs should specify a zone explicitly
545 : * anyway. However it makes the regression test outputs more predictable.
546 : *
547 : * We don't risk setting remote zone equal to ours, since the remote
548 : * server might use a different timezone database. Instead, use UTC
549 : * (quoted, because very old servers are picky about case).
550 : */
551 68 : do_sql_command(conn, "SET timezone = 'UTC'");
552 :
553 : /*
554 : * Set values needed to ensure unambiguous data output from remote. (This
555 : * logic should match what pg_dump does. See also set_transmission_modes
556 : * in postgres_fdw.c.)
557 : */
558 68 : do_sql_command(conn, "SET datestyle = ISO");
559 68 : if (remoteversion >= 80400)
560 68 : do_sql_command(conn, "SET intervalstyle = postgres");
561 68 : if (remoteversion >= 90000)
562 68 : do_sql_command(conn, "SET extra_float_digits = 3");
563 : else
564 0 : do_sql_command(conn, "SET extra_float_digits = 2");
565 68 : }
566 :
567 : /*
568 : * Convenience subroutine to issue a non-data-returning SQL command to remote
569 : */
570 : void
571 2650 : do_sql_command(PGconn *conn, const char *sql)
572 : {
573 : PGresult *res;
574 :
575 2650 : if (!PQsendQuery(conn, sql))
576 0 : pgfdw_report_error(ERROR, NULL, conn, false, sql);
577 2650 : res = pgfdw_get_result(conn, sql);
578 2646 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
579 2 : pgfdw_report_error(ERROR, res, conn, true, sql);
580 2644 : PQclear(res);
581 2644 : }
582 :
583 : /*
584 : * Start remote transaction or subtransaction, if needed.
585 : *
586 : * Note that we always use at least REPEATABLE READ in the remote session.
587 : * This is so that, if a query initiates multiple scans of the same or
588 : * different foreign tables, we will get snapshot-consistent results from
589 : * those scans. A disadvantage is that we can't provide sane emulation of
590 : * READ COMMITTED behavior --- it would be nice if we had some other way to
591 : * control which remote queries share a snapshot.
592 : */
593 : static void
594 3092 : begin_remote_xact(ConnCacheEntry *entry)
595 : {
596 3092 : int curlevel = GetCurrentTransactionNestLevel();
597 :
598 : /* Start main transaction if we haven't yet */
599 3092 : if (entry->xact_depth <= 0)
600 : {
601 : const char *sql;
602 :
603 1160 : elog(DEBUG3, "starting remote transaction on connection %p",
604 : entry->conn);
605 :
606 1160 : if (IsolationIsSerializable())
607 0 : sql = "START TRANSACTION ISOLATION LEVEL SERIALIZABLE";
608 : else
609 1160 : sql = "START TRANSACTION ISOLATION LEVEL REPEATABLE READ";
610 1160 : entry->changing_xact_state = true;
611 1160 : do_sql_command(entry->conn, sql);
612 1158 : entry->xact_depth = 1;
613 1158 : entry->changing_xact_state = false;
614 : }
615 :
616 : /*
617 : * If we're in a subtransaction, stack up savepoints to match our level.
618 : * This ensures we can rollback just the desired effects when a
619 : * subtransaction aborts.
620 : */
621 3106 : while (entry->xact_depth < curlevel)
622 : {
623 : char sql[64];
624 :
625 18 : snprintf(sql, sizeof(sql), "SAVEPOINT s%d", entry->xact_depth + 1);
626 18 : entry->changing_xact_state = true;
627 18 : do_sql_command(entry->conn, sql);
628 16 : entry->xact_depth++;
629 16 : entry->changing_xact_state = false;
630 : }
631 3088 : }
632 :
633 : /*
634 : * Release connection reference count created by calling GetConnection.
635 : */
636 : void
637 3000 : ReleaseConnection(PGconn *conn)
638 : {
639 : /*
640 : * Currently, we don't actually track connection references because all
641 : * cleanup is managed on a transaction or subtransaction basis instead. So
642 : * there's nothing to do here.
643 : */
644 3000 : }
645 :
646 : /*
647 : * Assign a "unique" number for a cursor.
648 : *
649 : * These really only need to be unique per connection within a transaction.
650 : * For the moment we ignore the per-connection point and assign them across
651 : * all connections in the transaction, but we ask for the connection to be
652 : * supplied in case we want to refine that.
653 : *
654 : * Note that even if wraparound happens in a very long transaction, actual
655 : * collisions are highly improbable; just be sure to use %u not %d to print.
656 : */
657 : unsigned int
658 830 : GetCursorNumber(PGconn *conn)
659 : {
660 830 : return ++cursor_number;
661 : }
662 :
663 : /*
664 : * Assign a "unique" number for a prepared statement.
665 : *
666 : * This works much like GetCursorNumber, except that we never reset the counter
667 : * within a session. That's because we can't be 100% sure we've gotten rid
668 : * of all prepared statements on all connections, and it's not really worth
669 : * increasing the risk of prepared-statement name collisions by resetting.
670 : */
671 : unsigned int
672 250 : GetPrepStmtNumber(PGconn *conn)
673 : {
674 250 : return ++prep_stmt_number;
675 : }
676 :
677 : /*
678 : * Submit a query and wait for the result.
679 : *
680 : * This function is interruptible by signals.
681 : *
682 : * Caller is responsible for the error handling on the result.
683 : */
684 : PGresult *
685 6516 : pgfdw_exec_query(PGconn *conn, const char *query, PgFdwConnState *state)
686 : {
687 : /* First, process a pending asynchronous request, if any. */
688 6516 : if (state && state->pendingAreq)
689 8 : process_pending_request(state->pendingAreq);
690 :
691 : /*
692 : * Submit a query. Since we don't use non-blocking mode, this also can
693 : * block. But its risk is relatively small, so we ignore that for now.
694 : */
695 6516 : if (!PQsendQuery(conn, query))
696 0 : pgfdw_report_error(ERROR, NULL, conn, false, query);
697 :
698 : /* Wait for the result. */
699 6516 : return pgfdw_get_result(conn, query);
700 : }
701 :
702 : /*
703 : * Wait for the result from a prior asynchronous execution function call.
704 : *
705 : * This function offers quick responsiveness by checking for any interruptions.
706 : *
707 : * This function emulates PQexec()'s behavior of returning the last result
708 : * when there are many.
709 : *
710 : * Caller is responsible for the error handling on the result.
711 : */
712 : PGresult *
713 12900 : pgfdw_get_result(PGconn *conn, const char *query)
714 : {
715 12900 : PGresult *volatile last_res = NULL;
716 :
717 : /* In what follows, do not leak any PGresults on an error. */
718 12900 : PG_TRY();
719 : {
720 : for (;;)
721 12896 : {
722 : PGresult *res;
723 :
724 38684 : while (PQisBusy(conn))
725 : {
726 : int wc;
727 :
728 : /* Sleep until there's something to do */
729 12892 : wc = WaitLatchOrSocket(MyLatch,
730 : WL_LATCH_SET | WL_SOCKET_READABLE |
731 : WL_EXIT_ON_PM_DEATH,
732 : PQsocket(conn),
733 : -1L, PG_WAIT_EXTENSION);
734 12892 : ResetLatch(MyLatch);
735 :
736 12892 : CHECK_FOR_INTERRUPTS();
737 :
738 : /* Data available in socket? */
739 12892 : if (wc & WL_SOCKET_READABLE)
740 : {
741 12890 : if (!PQconsumeInput(conn))
742 4 : pgfdw_report_error(ERROR, NULL, conn, false, query);
743 : }
744 : }
745 :
746 25792 : res = PQgetResult(conn);
747 25792 : if (res == NULL)
748 12896 : break; /* query is complete */
749 :
750 12896 : PQclear(last_res);
751 12896 : last_res = res;
752 : }
753 : }
754 4 : PG_CATCH();
755 : {
756 4 : PQclear(last_res);
757 4 : PG_RE_THROW();
758 : }
759 12896 : PG_END_TRY();
760 :
761 12896 : return last_res;
762 : }
763 :
764 : /*
765 : * Report an error we got from the remote server.
766 : *
767 : * elevel: error level to use (typically ERROR, but might be less)
768 : * res: PGresult containing the error
769 : * conn: connection we did the query on
770 : * clear: if true, PQclear the result (otherwise caller will handle it)
771 : * sql: NULL, or text of remote command we tried to execute
772 : *
773 : * Note: callers that choose not to throw ERROR for a remote error are
774 : * responsible for making sure that the associated ConnCacheEntry gets
775 : * marked with have_error = true.
776 : */
777 : void
778 24 : pgfdw_report_error(int elevel, PGresult *res, PGconn *conn,
779 : bool clear, const char *sql)
780 : {
781 : /* If requested, PGresult must be released before leaving this function. */
782 24 : PG_TRY();
783 : {
784 24 : char *diag_sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
785 24 : char *message_primary = PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY);
786 24 : char *message_detail = PQresultErrorField(res, PG_DIAG_MESSAGE_DETAIL);
787 24 : char *message_hint = PQresultErrorField(res, PG_DIAG_MESSAGE_HINT);
788 24 : char *message_context = PQresultErrorField(res, PG_DIAG_CONTEXT);
789 : int sqlstate;
790 :
791 24 : if (diag_sqlstate)
792 20 : sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
793 : diag_sqlstate[1],
794 : diag_sqlstate[2],
795 : diag_sqlstate[3],
796 : diag_sqlstate[4]);
797 : else
798 4 : sqlstate = ERRCODE_CONNECTION_FAILURE;
799 :
800 : /*
801 : * If we don't get a message from the PGresult, try the PGconn. This
802 : * is needed because for connection-level failures, PQexec may just
803 : * return NULL, not a PGresult at all.
804 : */
805 24 : if (message_primary == NULL)
806 4 : message_primary = pchomp(PQerrorMessage(conn));
807 :
808 24 : ereport(elevel,
809 : (errcode(sqlstate),
810 : message_primary ? errmsg_internal("%s", message_primary) :
811 : errmsg("could not obtain message string for remote error"),
812 : message_detail ? errdetail_internal("%s", message_detail) : 0,
813 : message_hint ? errhint("%s", message_hint) : 0,
814 : message_context ? errcontext("%s", message_context) : 0,
815 : sql ? errcontext("remote SQL command: %s", sql) : 0));
816 : }
817 24 : PG_FINALLY();
818 : {
819 24 : if (clear)
820 18 : PQclear(res);
821 : }
822 24 : PG_END_TRY();
823 0 : }
824 :
825 : /*
826 : * pgfdw_xact_callback --- cleanup at main-transaction end.
827 : *
828 : * This runs just late enough that it must not enter user-defined code
829 : * locally. (Entering such code on the remote side is fine. Its remote
830 : * COMMIT TRANSACTION may run deferred triggers.)
831 : */
832 : static void
833 6038 : pgfdw_xact_callback(XactEvent event, void *arg)
834 : {
835 : HASH_SEQ_STATUS scan;
836 : ConnCacheEntry *entry;
837 :
838 : /* Quick exit if no connections were touched in this transaction. */
839 6038 : if (!xact_got_connection)
840 4912 : return;
841 :
842 : /*
843 : * Scan all connection cache entries to find open remote transactions, and
844 : * close them.
845 : */
846 1126 : hash_seq_init(&scan, ConnectionHash);
847 4570 : while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
848 : {
849 : PGresult *res;
850 :
851 : /* Ignore cache entry if no open connection right now */
852 3446 : if (entry->conn == NULL)
853 1418 : continue;
854 :
855 : /* If it has an open remote transaction, try to close it */
856 2028 : if (entry->xact_depth > 0)
857 : {
858 1160 : bool abort_cleanup_failure = false;
859 :
860 1160 : elog(DEBUG3, "closing remote transaction on connection %p",
861 : entry->conn);
862 :
863 1160 : switch (event)
864 : {
865 1100 : case XACT_EVENT_PARALLEL_PRE_COMMIT:
866 : case XACT_EVENT_PRE_COMMIT:
867 :
868 : /*
869 : * If abort cleanup previously failed for this connection,
870 : * we can't issue any more commands against it.
871 : */
872 1100 : pgfdw_reject_incomplete_xact_state_change(entry);
873 :
874 : /* Commit all remote transactions during pre-commit */
875 1100 : entry->changing_xact_state = true;
876 1100 : do_sql_command(entry->conn, "COMMIT TRANSACTION");
877 1100 : entry->changing_xact_state = false;
878 :
879 : /*
880 : * If there were any errors in subtransactions, and we
881 : * made prepared statements, do a DEALLOCATE ALL to make
882 : * sure we get rid of all prepared statements. This is
883 : * annoying and not terribly bulletproof, but it's
884 : * probably not worth trying harder.
885 : *
886 : * DEALLOCATE ALL only exists in 8.3 and later, so this
887 : * constrains how old a server postgres_fdw can
888 : * communicate with. We intentionally ignore errors in
889 : * the DEALLOCATE, so that we can hobble along to some
890 : * extent with older servers (leaking prepared statements
891 : * as we go; but we don't really support update operations
892 : * pre-8.3 anyway).
893 : */
894 1100 : if (entry->have_prep_stmt && entry->have_error)
895 : {
896 0 : res = PQexec(entry->conn, "DEALLOCATE ALL");
897 0 : PQclear(res);
898 : }
899 1100 : entry->have_prep_stmt = false;
900 1100 : entry->have_error = false;
901 1100 : break;
902 2 : case XACT_EVENT_PRE_PREPARE:
903 :
904 : /*
905 : * We disallow any remote transactions, since it's not
906 : * very reasonable to hold them open until the prepared
907 : * transaction is committed. For the moment, throw error
908 : * unconditionally; later we might allow read-only cases.
909 : * Note that the error will cause us to come right back
910 : * here with event == XACT_EVENT_ABORT, so we'll clean up
911 : * the connection state at that point.
912 : */
913 2 : ereport(ERROR,
914 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
915 : errmsg("cannot PREPARE a transaction that has operated on postgres_fdw foreign tables")));
916 : break;
917 0 : case XACT_EVENT_PARALLEL_COMMIT:
918 : case XACT_EVENT_COMMIT:
919 : case XACT_EVENT_PREPARE:
920 : /* Pre-commit should have closed the open transaction */
921 0 : elog(ERROR, "missed cleaning up connection during pre-commit");
922 : break;
923 58 : case XACT_EVENT_PARALLEL_ABORT:
924 : case XACT_EVENT_ABORT:
925 :
926 : /*
927 : * Don't try to clean up the connection if we're already
928 : * in error recursion trouble.
929 : */
930 58 : if (in_error_recursion_trouble())
931 0 : entry->changing_xact_state = true;
932 :
933 : /*
934 : * If connection is already unsalvageable, don't touch it
935 : * further.
936 : */
937 58 : if (entry->changing_xact_state)
938 2 : break;
939 :
940 : /*
941 : * Mark this connection as in the process of changing
942 : * transaction state.
943 : */
944 56 : entry->changing_xact_state = true;
945 :
946 : /* Assume we might have lost track of prepared statements */
947 56 : entry->have_error = true;
948 :
949 : /*
950 : * If a command has been submitted to the remote server by
951 : * using an asynchronous execution function, the command
952 : * might not have yet completed. Check to see if a
953 : * command is still being processed by the remote server,
954 : * and if so, request cancellation of the command.
955 : */
956 56 : if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE &&
957 0 : !pgfdw_cancel_query(entry->conn))
958 : {
959 : /* Unable to cancel running query. */
960 0 : abort_cleanup_failure = true;
961 : }
962 56 : else if (!pgfdw_exec_cleanup_query(entry->conn,
963 : "ABORT TRANSACTION",
964 : false))
965 : {
966 : /* Unable to abort remote transaction. */
967 0 : abort_cleanup_failure = true;
968 : }
969 56 : else if (entry->have_prep_stmt && entry->have_error &&
970 24 : !pgfdw_exec_cleanup_query(entry->conn,
971 : "DEALLOCATE ALL",
972 : true))
973 : {
974 : /* Trouble clearing prepared statements. */
975 0 : abort_cleanup_failure = true;
976 : }
977 : else
978 : {
979 56 : entry->have_prep_stmt = false;
980 56 : entry->have_error = false;
981 : /* Also reset per-connection state */
982 56 : memset(&entry->state, 0, sizeof(entry->state));
983 : }
984 :
985 : /* Disarm changing_xact_state if it all worked. */
986 56 : entry->changing_xact_state = abort_cleanup_failure;
987 56 : break;
988 : }
989 868 : }
990 :
991 : /* Reset state to show we're out of a transaction */
992 2026 : entry->xact_depth = 0;
993 :
994 : /*
995 : * If the connection isn't in a good idle state, it is marked as
996 : * invalid or keep_connections option of its server is disabled, then
997 : * discard it to recover. Next GetConnection will open a new
998 : * connection.
999 : */
1000 4050 : if (PQstatus(entry->conn) != CONNECTION_OK ||
1001 2024 : PQtransactionStatus(entry->conn) != PQTRANS_IDLE ||
1002 2024 : entry->changing_xact_state ||
1003 2024 : entry->invalidated ||
1004 2020 : !entry->keep_connections)
1005 : {
1006 8 : elog(DEBUG3, "discarding connection %p", entry->conn);
1007 8 : disconnect_pg_server(entry);
1008 : }
1009 : }
1010 :
1011 : /*
1012 : * Regardless of the event type, we can now mark ourselves as out of the
1013 : * transaction. (Note: if we are here during PRE_COMMIT or PRE_PREPARE,
1014 : * this saves a useless scan of the hashtable during COMMIT or PREPARE.)
1015 : */
1016 1124 : xact_got_connection = false;
1017 :
1018 : /* Also reset cursor numbering for next transaction */
1019 1124 : cursor_number = 0;
1020 : }
1021 :
1022 : /*
1023 : * pgfdw_subxact_callback --- cleanup at subtransaction end.
1024 : */
1025 : static void
1026 50 : pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
1027 : SubTransactionId parentSubid, void *arg)
1028 : {
1029 : HASH_SEQ_STATUS scan;
1030 : ConnCacheEntry *entry;
1031 : int curlevel;
1032 :
1033 : /* Nothing to do at subxact start, nor after commit. */
1034 50 : if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB ||
1035 : event == SUBXACT_EVENT_ABORT_SUB))
1036 30 : return;
1037 :
1038 : /* Quick exit if no connections were touched in this transaction. */
1039 20 : if (!xact_got_connection)
1040 0 : return;
1041 :
1042 : /*
1043 : * Scan all connection cache entries to find open remote subtransactions
1044 : * of the current level, and close them.
1045 : */
1046 20 : curlevel = GetCurrentTransactionNestLevel();
1047 20 : hash_seq_init(&scan, ConnectionHash);
1048 84 : while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
1049 : {
1050 : char sql[100];
1051 :
1052 : /*
1053 : * We only care about connections with open remote subtransactions of
1054 : * the current level.
1055 : */
1056 64 : if (entry->conn == NULL || entry->xact_depth < curlevel)
1057 48 : continue;
1058 :
1059 16 : if (entry->xact_depth > curlevel)
1060 0 : elog(ERROR, "missed cleaning up remote subtransaction at level %d",
1061 : entry->xact_depth);
1062 :
1063 16 : if (event == SUBXACT_EVENT_PRE_COMMIT_SUB)
1064 : {
1065 : /*
1066 : * If abort cleanup previously failed for this connection, we
1067 : * can't issue any more commands against it.
1068 : */
1069 10 : pgfdw_reject_incomplete_xact_state_change(entry);
1070 :
1071 : /* Commit all remote subtransactions during pre-commit */
1072 10 : snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel);
1073 10 : entry->changing_xact_state = true;
1074 10 : do_sql_command(entry->conn, sql);
1075 10 : entry->changing_xact_state = false;
1076 : }
1077 6 : else if (in_error_recursion_trouble())
1078 : {
1079 : /*
1080 : * Don't try to clean up the connection if we're already in error
1081 : * recursion trouble.
1082 : */
1083 0 : entry->changing_xact_state = true;
1084 : }
1085 6 : else if (!entry->changing_xact_state)
1086 : {
1087 6 : bool abort_cleanup_failure = false;
1088 :
1089 : /* Remember that abort cleanup is in progress. */
1090 6 : entry->changing_xact_state = true;
1091 :
1092 : /* Assume we might have lost track of prepared statements */
1093 6 : entry->have_error = true;
1094 :
1095 : /*
1096 : * If a command has been submitted to the remote server by using
1097 : * an asynchronous execution function, the command might not have
1098 : * yet completed. Check to see if a command is still being
1099 : * processed by the remote server, and if so, request cancellation
1100 : * of the command.
1101 : */
1102 6 : if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE &&
1103 0 : !pgfdw_cancel_query(entry->conn))
1104 0 : abort_cleanup_failure = true;
1105 : else
1106 : {
1107 : /* Rollback all remote subtransactions during abort */
1108 6 : snprintf(sql, sizeof(sql),
1109 : "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d",
1110 : curlevel, curlevel);
1111 6 : if (!pgfdw_exec_cleanup_query(entry->conn, sql, false))
1112 0 : abort_cleanup_failure = true;
1113 : }
1114 :
1115 : /* Disarm changing_xact_state if it all worked. */
1116 6 : entry->changing_xact_state = abort_cleanup_failure;
1117 : }
1118 :
1119 : /* OK, we're outta that level of subtransaction */
1120 16 : entry->xact_depth--;
1121 : }
1122 : }
1123 :
1124 : /*
1125 : * Connection invalidation callback function
1126 : *
1127 : * After a change to a pg_foreign_server or pg_user_mapping catalog entry,
1128 : * close connections depending on that entry immediately if current transaction
1129 : * has not used those connections yet. Otherwise, mark those connections as
1130 : * invalid and then make pgfdw_xact_callback() close them at the end of current
1131 : * transaction, since they cannot be closed in the midst of the transaction
1132 : * using them. Closed connections will be remade at the next opportunity if
1133 : * necessary.
1134 : *
1135 : * Although most cache invalidation callbacks blow away all the related stuff
1136 : * regardless of the given hashvalue, connections are expensive enough that
1137 : * it's worth trying to avoid that.
1138 : *
1139 : * NB: We could avoid unnecessary disconnection more strictly by examining
1140 : * individual option values, but it seems too much effort for the gain.
1141 : */
1142 : static void
1143 186 : pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue)
1144 : {
1145 : HASH_SEQ_STATUS scan;
1146 : ConnCacheEntry *entry;
1147 :
1148 : Assert(cacheid == FOREIGNSERVEROID || cacheid == USERMAPPINGOID);
1149 :
1150 : /* ConnectionHash must exist already, if we're registered */
1151 186 : hash_seq_init(&scan, ConnectionHash);
1152 994 : while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
1153 : {
1154 : /* Ignore invalid entries */
1155 808 : if (entry->conn == NULL)
1156 600 : continue;
1157 :
1158 : /* hashvalue == 0 means a cache reset, must clear all state */
1159 208 : if (hashvalue == 0 ||
1160 124 : (cacheid == FOREIGNSERVEROID &&
1161 208 : entry->server_hashvalue == hashvalue) ||
1162 84 : (cacheid == USERMAPPINGOID &&
1163 84 : entry->mapping_hashvalue == hashvalue))
1164 : {
1165 : /*
1166 : * Close the connection immediately if it's not used yet in this
1167 : * transaction. Otherwise mark it as invalid so that
1168 : * pgfdw_xact_callback() can close it at the end of this
1169 : * transaction.
1170 : */
1171 50 : if (entry->xact_depth == 0)
1172 : {
1173 44 : elog(DEBUG3, "discarding connection %p", entry->conn);
1174 44 : disconnect_pg_server(entry);
1175 : }
1176 : else
1177 6 : entry->invalidated = true;
1178 : }
1179 : }
1180 186 : }
1181 :
1182 : /*
1183 : * Raise an error if the given connection cache entry is marked as being
1184 : * in the middle of an xact state change. This should be called at which no
1185 : * such change is expected to be in progress; if one is found to be in
1186 : * progress, it means that we aborted in the middle of a previous state change
1187 : * and now don't know what the remote transaction state actually is.
1188 : * Such connections can't safely be further used. Re-establishing the
1189 : * connection would change the snapshot and roll back any writes already
1190 : * performed, so that's not an option, either. Thus, we must abort.
1191 : */
1192 : static void
1193 4212 : pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry)
1194 : {
1195 : ForeignServer *server;
1196 :
1197 : /* nothing to do for inactive entries and entries of sane state */
1198 4212 : if (entry->conn == NULL || !entry->changing_xact_state)
1199 4212 : return;
1200 :
1201 : /* make sure this entry is inactive */
1202 0 : disconnect_pg_server(entry);
1203 :
1204 : /* find server name to be shown in the message below */
1205 0 : server = GetForeignServer(entry->serverid);
1206 :
1207 0 : ereport(ERROR,
1208 : (errcode(ERRCODE_CONNECTION_EXCEPTION),
1209 : errmsg("connection to server \"%s\" was lost",
1210 : server->servername)));
1211 : }
1212 :
1213 : /*
1214 : * Cancel the currently-in-progress query (whose query text we do not have)
1215 : * and ignore the result. Returns true if we successfully cancel the query
1216 : * and discard any pending result, and false if not.
1217 : *
1218 : * XXX: if the query was one sent by fetch_more_data_begin(), we could get the
1219 : * query text from the pendingAreq saved in the per-connection state, then
1220 : * report the query using it.
1221 : */
1222 : static bool
1223 0 : pgfdw_cancel_query(PGconn *conn)
1224 : {
1225 : PGcancel *cancel;
1226 : char errbuf[256];
1227 0 : PGresult *result = NULL;
1228 : TimestampTz endtime;
1229 :
1230 : /*
1231 : * If it takes too long to cancel the query and discard the result, assume
1232 : * the connection is dead.
1233 : */
1234 0 : endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);
1235 :
1236 : /*
1237 : * Issue cancel request. Unfortunately, there's no good way to limit the
1238 : * amount of time that we might block inside PQgetCancel().
1239 : */
1240 0 : if ((cancel = PQgetCancel(conn)))
1241 : {
1242 0 : if (!PQcancel(cancel, errbuf, sizeof(errbuf)))
1243 : {
1244 0 : ereport(WARNING,
1245 : (errcode(ERRCODE_CONNECTION_FAILURE),
1246 : errmsg("could not send cancel request: %s",
1247 : errbuf)));
1248 0 : PQfreeCancel(cancel);
1249 0 : return false;
1250 : }
1251 0 : PQfreeCancel(cancel);
1252 : }
1253 :
1254 : /* Get and discard the result of the query. */
1255 0 : if (pgfdw_get_cleanup_result(conn, endtime, &result))
1256 0 : return false;
1257 0 : PQclear(result);
1258 :
1259 0 : return true;
1260 : }
1261 :
1262 : /*
1263 : * Submit a query during (sub)abort cleanup and wait up to 30 seconds for the
1264 : * result. If the query is executed without error, the return value is true.
1265 : * If the query is executed successfully but returns an error, the return
1266 : * value is true if and only if ignore_errors is set. If the query can't be
1267 : * sent or times out, the return value is false.
1268 : */
1269 : static bool
1270 86 : pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors)
1271 : {
1272 86 : PGresult *result = NULL;
1273 : TimestampTz endtime;
1274 :
1275 : /*
1276 : * If it takes too long to execute a cleanup query, assume the connection
1277 : * is dead. It's fairly likely that this is why we aborted in the first
1278 : * place (e.g. statement timeout, user cancel), so the timeout shouldn't
1279 : * be too long.
1280 : */
1281 86 : endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);
1282 :
1283 : /*
1284 : * Submit a query. Since we don't use non-blocking mode, this also can
1285 : * block. But its risk is relatively small, so we ignore that for now.
1286 : */
1287 86 : if (!PQsendQuery(conn, query))
1288 : {
1289 0 : pgfdw_report_error(WARNING, NULL, conn, false, query);
1290 0 : return false;
1291 : }
1292 :
1293 : /* Get the result of the query. */
1294 86 : if (pgfdw_get_cleanup_result(conn, endtime, &result))
1295 0 : return false;
1296 :
1297 : /* Issue a warning if not successful. */
1298 86 : if (PQresultStatus(result) != PGRES_COMMAND_OK)
1299 : {
1300 0 : pgfdw_report_error(WARNING, result, conn, true, query);
1301 0 : return ignore_errors;
1302 : }
1303 86 : PQclear(result);
1304 :
1305 86 : return true;
1306 : }
1307 :
1308 : /*
1309 : * Get, during abort cleanup, the result of a query that is in progress. This
1310 : * might be a query that is being interrupted by transaction abort, or it might
1311 : * be a query that was initiated as part of transaction abort to get the remote
1312 : * side back to the appropriate state.
1313 : *
1314 : * It's not a huge problem if we throw an ERROR here, but if we get into error
1315 : * recursion trouble, we'll end up slamming the connection shut, which will
1316 : * necessitate failing the entire toplevel transaction even if subtransactions
1317 : * were used. Try to use WARNING where we can.
1318 : *
1319 : * endtime is the time at which we should give up and assume the remote
1320 : * side is dead. Returns true if the timeout expired, otherwise false.
1321 : * Sets *result except in case of a timeout.
1322 : */
1323 : static bool
1324 86 : pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result)
1325 : {
1326 86 : volatile bool timed_out = false;
1327 86 : PGresult *volatile last_res = NULL;
1328 :
1329 : /* In what follows, do not leak any PGresults on an error. */
1330 86 : PG_TRY();
1331 : {
1332 : for (;;)
1333 92 : {
1334 : PGresult *res;
1335 :
1336 264 : while (PQisBusy(conn))
1337 : {
1338 : int wc;
1339 86 : TimestampTz now = GetCurrentTimestamp();
1340 : long cur_timeout;
1341 :
1342 : /* If timeout has expired, give up, else get sleep time. */
1343 86 : cur_timeout = TimestampDifferenceMilliseconds(now, endtime);
1344 86 : if (cur_timeout <= 0)
1345 : {
1346 0 : timed_out = true;
1347 0 : goto exit;
1348 : }
1349 :
1350 : /* Sleep until there's something to do */
1351 86 : wc = WaitLatchOrSocket(MyLatch,
1352 : WL_LATCH_SET | WL_SOCKET_READABLE |
1353 : WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
1354 : PQsocket(conn),
1355 : cur_timeout, PG_WAIT_EXTENSION);
1356 86 : ResetLatch(MyLatch);
1357 :
1358 86 : CHECK_FOR_INTERRUPTS();
1359 :
1360 : /* Data available in socket? */
1361 86 : if (wc & WL_SOCKET_READABLE)
1362 : {
1363 86 : if (!PQconsumeInput(conn))
1364 : {
1365 : /* connection trouble; treat the same as a timeout */
1366 0 : timed_out = true;
1367 0 : goto exit;
1368 : }
1369 : }
1370 : }
1371 :
1372 178 : res = PQgetResult(conn);
1373 178 : if (res == NULL)
1374 86 : break; /* query is complete */
1375 :
1376 92 : PQclear(last_res);
1377 92 : last_res = res;
1378 : }
1379 86 : exit: ;
1380 : }
1381 0 : PG_CATCH();
1382 : {
1383 0 : PQclear(last_res);
1384 0 : PG_RE_THROW();
1385 : }
1386 86 : PG_END_TRY();
1387 :
1388 86 : if (timed_out)
1389 0 : PQclear(last_res);
1390 : else
1391 86 : *result = last_res;
1392 86 : return timed_out;
1393 : }
1394 :
1395 : /*
1396 : * List active foreign server connections.
1397 : *
1398 : * This function takes no input parameter and returns setof record made of
1399 : * following values:
1400 : * - server_name - server name of active connection. In case the foreign server
1401 : * is dropped but still the connection is active, then the server name will
1402 : * be NULL in output.
1403 : * - valid - true/false representing whether the connection is valid or not.
1404 : * Note that the connections can get invalidated in pgfdw_inval_callback.
1405 : *
1406 : * No records are returned when there are no cached connections at all.
1407 : */
1408 : Datum
1409 22 : postgres_fdw_get_connections(PG_FUNCTION_ARGS)
1410 : {
1411 : #define POSTGRES_FDW_GET_CONNECTIONS_COLS 2
1412 22 : ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
1413 : TupleDesc tupdesc;
1414 : Tuplestorestate *tupstore;
1415 : MemoryContext per_query_ctx;
1416 : MemoryContext oldcontext;
1417 : HASH_SEQ_STATUS scan;
1418 : ConnCacheEntry *entry;
1419 :
1420 : /* check to see if caller supports us returning a tuplestore */
1421 22 : if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
1422 0 : ereport(ERROR,
1423 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1424 : errmsg("set-valued function called in context that cannot accept a set")));
1425 22 : if (!(rsinfo->allowedModes & SFRM_Materialize))
1426 0 : ereport(ERROR,
1427 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1428 : errmsg("materialize mode required, but it is not allowed in this context")));
1429 :
1430 : /* Build a tuple descriptor for our result type */
1431 22 : if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
1432 0 : elog(ERROR, "return type must be a row type");
1433 :
1434 : /* Build tuplestore to hold the result rows */
1435 22 : per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
1436 22 : oldcontext = MemoryContextSwitchTo(per_query_ctx);
1437 :
1438 22 : tupstore = tuplestore_begin_heap(true, false, work_mem);
1439 22 : rsinfo->returnMode = SFRM_Materialize;
1440 22 : rsinfo->setResult = tupstore;
1441 22 : rsinfo->setDesc = tupdesc;
1442 :
1443 22 : MemoryContextSwitchTo(oldcontext);
1444 :
1445 : /* If cache doesn't exist, we return no records */
1446 22 : if (!ConnectionHash)
1447 : {
1448 : /* clean up and return the tuplestore */
1449 : tuplestore_donestoring(tupstore);
1450 :
1451 0 : PG_RETURN_VOID();
1452 : }
1453 :
1454 22 : hash_seq_init(&scan, ConnectionHash);
1455 164 : while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
1456 : {
1457 : ForeignServer *server;
1458 : Datum values[POSTGRES_FDW_GET_CONNECTIONS_COLS];
1459 : bool nulls[POSTGRES_FDW_GET_CONNECTIONS_COLS];
1460 :
1461 : /* We only look for open remote connections */
1462 142 : if (!entry->conn)
1463 120 : continue;
1464 :
1465 22 : server = GetForeignServerExtended(entry->serverid, FSV_MISSING_OK);
1466 :
1467 66 : MemSet(values, 0, sizeof(values));
1468 22 : MemSet(nulls, 0, sizeof(nulls));
1469 :
1470 : /*
1471 : * The foreign server may have been dropped in current explicit
1472 : * transaction. It is not possible to drop the server from another
1473 : * session when the connection associated with it is in use in the
1474 : * current transaction, if tried so, the drop query in another session
1475 : * blocks until the current transaction finishes.
1476 : *
1477 : * Even though the server is dropped in the current transaction, the
1478 : * cache can still have associated active connection entry, say we
1479 : * call such connections dangling. Since we can not fetch the server
1480 : * name from system catalogs for dangling connections, instead we show
1481 : * NULL value for server name in output.
1482 : *
1483 : * We could have done better by storing the server name in the cache
1484 : * entry instead of server oid so that it could be used in the output.
1485 : * But the server name in each cache entry requires 64 bytes of
1486 : * memory, which is huge, when there are many cached connections and
1487 : * the use case i.e. dropping the foreign server within the explicit
1488 : * current transaction seems rare. So, we chose to show NULL value for
1489 : * server name in output.
1490 : *
1491 : * Such dangling connections get closed either in next use or at the
1492 : * end of current explicit transaction in pgfdw_xact_callback.
1493 : */
1494 22 : if (!server)
1495 : {
1496 : /*
1497 : * If the server has been dropped in the current explicit
1498 : * transaction, then this entry would have been invalidated in
1499 : * pgfdw_inval_callback at the end of drop server command. Note
1500 : * that this connection would not have been closed in
1501 : * pgfdw_inval_callback because it is still being used in the
1502 : * current explicit transaction. So, assert that here.
1503 : */
1504 : Assert(entry->conn && entry->xact_depth > 0 && entry->invalidated);
1505 :
1506 : /* Show null, if no server name was found */
1507 2 : nulls[0] = true;
1508 : }
1509 : else
1510 20 : values[0] = CStringGetTextDatum(server->servername);
1511 :
1512 22 : values[1] = BoolGetDatum(!entry->invalidated);
1513 :
1514 22 : tuplestore_putvalues(tupstore, tupdesc, values, nulls);
1515 : }
1516 :
1517 : /* clean up and return the tuplestore */
1518 : tuplestore_donestoring(tupstore);
1519 :
1520 22 : PG_RETURN_VOID();
1521 : }
1522 :
1523 : /*
1524 : * Disconnect the specified cached connections.
1525 : *
1526 : * This function discards the open connections that are established by
1527 : * postgres_fdw from the local session to the foreign server with
1528 : * the given name. Note that there can be multiple connections to
1529 : * the given server using different user mappings. If the connections
1530 : * are used in the current local transaction, they are not disconnected
1531 : * and warning messages are reported. This function returns true
1532 : * if it disconnects at least one connection, otherwise false. If no
1533 : * foreign server with the given name is found, an error is reported.
1534 : */
1535 : Datum
1536 8 : postgres_fdw_disconnect(PG_FUNCTION_ARGS)
1537 : {
1538 : ForeignServer *server;
1539 : char *servername;
1540 :
1541 8 : servername = text_to_cstring(PG_GETARG_TEXT_PP(0));
1542 8 : server = GetForeignServerByName(servername, false);
1543 :
1544 6 : PG_RETURN_BOOL(disconnect_cached_connections(server->serverid));
1545 : }
1546 :
1547 : /*
1548 : * Disconnect all the cached connections.
1549 : *
1550 : * This function discards all the open connections that are established by
1551 : * postgres_fdw from the local session to the foreign servers.
1552 : * If the connections are used in the current local transaction, they are
1553 : * not disconnected and warning messages are reported. This function
1554 : * returns true if it disconnects at least one connection, otherwise false.
1555 : */
1556 : Datum
1557 8 : postgres_fdw_disconnect_all(PG_FUNCTION_ARGS)
1558 : {
1559 8 : PG_RETURN_BOOL(disconnect_cached_connections(InvalidOid));
1560 : }
1561 :
1562 : /*
1563 : * Workhorse to disconnect cached connections.
1564 : *
1565 : * This function scans all the connection cache entries and disconnects
1566 : * the open connections whose foreign server OID matches with
1567 : * the specified one. If InvalidOid is specified, it disconnects all
1568 : * the cached connections.
1569 : *
1570 : * This function emits a warning for each connection that's used in
1571 : * the current transaction and doesn't close it. It returns true if
1572 : * it disconnects at least one connection, otherwise false.
1573 : *
1574 : * Note that this function disconnects even the connections that are
1575 : * established by other users in the same local session using different
1576 : * user mappings. This leads even non-superuser to be able to close
1577 : * the connections established by superusers in the same local session.
1578 : *
1579 : * XXX As of now we don't see any security risk doing this. But we should
1580 : * set some restrictions on that, for example, prevent non-superuser
1581 : * from closing the connections established by superusers even
1582 : * in the same session?
1583 : */
1584 : static bool
1585 14 : disconnect_cached_connections(Oid serverid)
1586 : {
1587 : HASH_SEQ_STATUS scan;
1588 : ConnCacheEntry *entry;
1589 14 : bool all = !OidIsValid(serverid);
1590 14 : bool result = false;
1591 :
1592 : /*
1593 : * Connection cache hashtable has not been initialized yet in this
1594 : * session, so return false.
1595 : */
1596 14 : if (!ConnectionHash)
1597 0 : return false;
1598 :
1599 14 : hash_seq_init(&scan, ConnectionHash);
1600 100 : while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
1601 : {
1602 : /* Ignore cache entry if no open connection right now. */
1603 86 : if (!entry->conn)
1604 64 : continue;
1605 :
1606 22 : if (all || entry->serverid == serverid)
1607 : {
1608 : /*
1609 : * Emit a warning because the connection to close is used in the
1610 : * current transaction and cannot be disconnected right now.
1611 : */
1612 16 : if (entry->xact_depth > 0)
1613 : {
1614 : ForeignServer *server;
1615 :
1616 6 : server = GetForeignServerExtended(entry->serverid,
1617 : FSV_MISSING_OK);
1618 :
1619 6 : if (!server)
1620 : {
1621 : /*
1622 : * If the foreign server was dropped while its connection
1623 : * was used in the current transaction, the connection
1624 : * must have been marked as invalid by
1625 : * pgfdw_inval_callback at the end of DROP SERVER command.
1626 : */
1627 : Assert(entry->invalidated);
1628 :
1629 0 : ereport(WARNING,
1630 : (errmsg("cannot close dropped server connection because it is still in use")));
1631 : }
1632 : else
1633 6 : ereport(WARNING,
1634 : (errmsg("cannot close connection for server \"%s\" because it is still in use",
1635 : server->servername)));
1636 : }
1637 : else
1638 : {
1639 10 : elog(DEBUG3, "discarding connection %p", entry->conn);
1640 10 : disconnect_pg_server(entry);
1641 10 : result = true;
1642 : }
1643 : }
1644 : }
1645 :
1646 14 : return result;
1647 : }
|
