Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * connection.c
4 : * Connection management functions for postgres_fdw
5 : *
6 : * Portions Copyright (c) 2012-2020, 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 "mb/pg_wchar.h"
20 : #include "miscadmin.h"
21 : #include "pgstat.h"
22 : #include "postgres_fdw.h"
23 : #include "storage/fd.h"
24 : #include "storage/latch.h"
25 : #include "utils/hsearch.h"
26 : #include "utils/inval.h"
27 : #include "utils/memutils.h"
28 : #include "utils/syscache.h"
29 :
30 : /*
31 : * Connection cache hash table entry
32 : *
33 : * The lookup key in this hash table is the user mapping OID. We use just one
34 : * connection per user mapping ID, which ensures that all the scans use the
35 : * same snapshot during a query. Using the user mapping OID rather than
36 : * the foreign server OID + user OID avoids creating multiple connections when
37 : * the public user mapping applies to all user OIDs.
38 : *
39 : * The "conn" pointer can be NULL if we don't currently have a live connection.
40 : * When we do have a connection, xact_depth tracks the current depth of
41 : * transactions and subtransactions open on the remote side. We need to issue
42 : * commands at the same nesting depth on the remote as we're executing at
43 : * ourselves, so that rolling back a subtransaction will kill the right
44 : * queries and not the wrong ones.
45 : */
46 : typedef Oid ConnCacheKey;
47 :
48 : typedef struct ConnCacheEntry
49 : {
50 : ConnCacheKey key; /* hash key (must be first) */
51 : PGconn *conn; /* connection to foreign server, or NULL */
52 : /* Remaining fields are invalid when conn is NULL: */
53 : int xact_depth; /* 0 = no xact open, 1 = main xact open, 2 =
54 : * one level of subxact open, etc */
55 : bool have_prep_stmt; /* have we prepared any stmts in this xact? */
56 : bool have_error; /* have any subxacts aborted in this xact? */
57 : bool changing_xact_state; /* xact state change in process */
58 : bool invalidated; /* true if reconnect is pending */
59 : uint32 server_hashvalue; /* hash value of foreign server OID */
60 : uint32 mapping_hashvalue; /* hash value of user mapping OID */
61 : } ConnCacheEntry;
62 :
63 : /*
64 : * Connection cache (initialized on first use)
65 : */
66 : static HTAB *ConnectionHash = NULL;
67 :
68 : /* for assigning cursor numbers and prepared statement numbers */
69 : static unsigned int cursor_number = 0;
70 : static unsigned int prep_stmt_number = 0;
71 :
72 : /* tracks whether any work is needed in callback functions */
73 : static bool xact_got_connection = false;
74 :
75 : /* prototypes of private functions */
76 : static PGconn *connect_pg_server(ForeignServer *server, UserMapping *user);
77 : static void disconnect_pg_server(ConnCacheEntry *entry);
78 : static void check_conn_params(const char **keywords, const char **values, UserMapping *user);
79 : static void configure_remote_session(PGconn *conn);
80 : static void do_sql_command(PGconn *conn, const char *sql);
81 : static void begin_remote_xact(ConnCacheEntry *entry);
82 : static void pgfdw_xact_callback(XactEvent event, void *arg);
83 : static void pgfdw_subxact_callback(SubXactEvent event,
84 : SubTransactionId mySubid,
85 : SubTransactionId parentSubid,
86 : void *arg);
87 : static void pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue);
88 : static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry);
89 : static bool pgfdw_cancel_query(PGconn *conn);
90 : static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query,
91 : bool ignore_errors);
92 : static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime,
93 : PGresult **result);
94 : static bool UserMappingPasswordRequired(UserMapping *user);
95 :
96 : /*
97 : * Get a PGconn which can be used to execute queries on the remote PostgreSQL
98 : * server with the user's authorization. A new connection is established
99 : * if we don't already have a suitable one, and a transaction is opened at
100 : * the right subtransaction nesting depth if we didn't do that already.
101 : *
102 : * will_prep_stmt must be true if caller intends to create any prepared
103 : * statements. Since those don't go away automatically at transaction end
104 : * (not even on error), we need this flag to cue manual cleanup.
105 : */
106 : PGconn *
107 2826 : GetConnection(UserMapping *user, bool will_prep_stmt)
108 : {
109 : bool found;
110 : ConnCacheEntry *entry;
111 : ConnCacheKey key;
112 :
113 : /* First time through, initialize connection cache hashtable */
114 2826 : if (ConnectionHash == NULL)
115 : {
116 : HASHCTL ctl;
117 :
118 84 : MemSet(&ctl, 0, sizeof(ctl));
119 6 : ctl.keysize = sizeof(ConnCacheKey);
120 6 : ctl.entrysize = sizeof(ConnCacheEntry);
121 : /* allocate ConnectionHash in the cache context */
122 6 : ctl.hcxt = CacheMemoryContext;
123 6 : ConnectionHash = hash_create("postgres_fdw connections", 8,
124 : &ctl,
125 : HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
126 :
127 : /*
128 : * Register some callback functions that manage connection cleanup.
129 : * This should be done just once in each backend.
130 : */
131 6 : RegisterXactCallback(pgfdw_xact_callback, NULL);
132 6 : RegisterSubXactCallback(pgfdw_subxact_callback, NULL);
133 6 : CacheRegisterSyscacheCallback(FOREIGNSERVEROID,
134 : pgfdw_inval_callback, (Datum) 0);
135 6 : CacheRegisterSyscacheCallback(USERMAPPINGOID,
136 : pgfdw_inval_callback, (Datum) 0);
137 : }
138 :
139 : /* Set flag that we did GetConnection during the current transaction */
140 2826 : xact_got_connection = true;
141 :
142 : /* Create hash key for the entry. Assume no pad bytes in key struct */
143 2826 : key = user->umid;
144 :
145 : /*
146 : * Find or create cached entry for requested connection.
147 : */
148 2826 : entry = hash_search(ConnectionHash, &key, HASH_ENTER, &found);
149 2826 : if (!found)
150 : {
151 : /*
152 : * We need only clear "conn" here; remaining fields will be filled
153 : * later when "conn" is set.
154 : */
155 14 : entry->conn = NULL;
156 : }
157 :
158 : /* Reject further use of connections which failed abort cleanup. */
159 2826 : pgfdw_reject_incomplete_xact_state_change(entry);
160 :
161 : /*
162 : * If the connection needs to be remade due to invalidation, disconnect as
163 : * soon as we're out of all transactions.
164 : */
165 2826 : if (entry->conn != NULL && entry->invalidated && entry->xact_depth == 0)
166 : {
167 26 : elog(DEBUG3, "closing connection %p for option changes to take effect",
168 : entry->conn);
169 26 : disconnect_pg_server(entry);
170 : }
171 :
172 : /*
173 : * We don't check the health of cached connection here, because it would
174 : * require some overhead. Broken connection will be detected when the
175 : * connection is actually used.
176 : */
177 :
178 : /*
179 : * If cache entry doesn't have a connection, we have to establish a new
180 : * connection. (If connect_pg_server throws an error, the cache entry
181 : * will remain in a valid empty state, ie conn == NULL.)
182 : */
183 2826 : if (entry->conn == NULL)
184 : {
185 52 : ForeignServer *server = GetForeignServer(user->serverid);
186 :
187 : /* Reset all transient state fields, to be sure all are clean */
188 52 : entry->xact_depth = 0;
189 52 : entry->have_prep_stmt = false;
190 52 : entry->have_error = false;
191 52 : entry->changing_xact_state = false;
192 52 : entry->invalidated = false;
193 52 : entry->server_hashvalue =
194 52 : GetSysCacheHashValue1(FOREIGNSERVEROID,
195 : ObjectIdGetDatum(server->serverid));
196 52 : entry->mapping_hashvalue =
197 52 : GetSysCacheHashValue1(USERMAPPINGOID,
198 : ObjectIdGetDatum(user->umid));
199 :
200 : /* Now try to make the connection */
201 52 : entry->conn = connect_pg_server(server, user);
202 :
203 40 : elog(DEBUG3, "new postgres_fdw connection %p for server \"%s\" (user mapping oid %u, userid %u)",
204 : entry->conn, server->servername, user->umid, user->userid);
205 : }
206 :
207 : /*
208 : * Start a new transaction or subtransaction if needed.
209 : */
210 2814 : begin_remote_xact(entry);
211 :
212 : /* Remember if caller will prepare statements */
213 2814 : entry->have_prep_stmt |= will_prep_stmt;
214 :
215 2814 : return entry->conn;
216 : }
217 :
218 : /*
219 : * Connect to remote server using specified server and user mapping properties.
220 : */
221 : static PGconn *
222 52 : connect_pg_server(ForeignServer *server, UserMapping *user)
223 : {
224 52 : PGconn *volatile conn = NULL;
225 :
226 : /*
227 : * Use PG_TRY block to ensure closing connection on error.
228 : */
229 52 : PG_TRY();
230 : {
231 : const char **keywords;
232 : const char **values;
233 : int n;
234 :
235 : /*
236 : * Construct connection params from generic options of ForeignServer
237 : * and UserMapping. (Some of them might not be libpq options, in
238 : * which case we'll just waste a few array slots.) Add 3 extra slots
239 : * for fallback_application_name, client_encoding, end marker.
240 : */
241 52 : n = list_length(server->options) + list_length(user->options) + 3;
242 52 : keywords = (const char **) palloc(n * sizeof(char *));
243 52 : values = (const char **) palloc(n * sizeof(char *));
244 :
245 52 : n = 0;
246 156 : n += ExtractConnectionOptions(server->options,
247 52 : keywords + n, values + n);
248 156 : n += ExtractConnectionOptions(user->options,
249 52 : keywords + n, values + n);
250 :
251 : /* Use "postgres_fdw" as fallback_application_name. */
252 52 : keywords[n] = "fallback_application_name";
253 52 : values[n] = "postgres_fdw";
254 52 : n++;
255 :
256 : /* Set client_encoding so that libpq can convert encoding properly. */
257 52 : keywords[n] = "client_encoding";
258 52 : values[n] = GetDatabaseEncodingName();
259 52 : n++;
260 :
261 52 : keywords[n] = values[n] = NULL;
262 :
263 : /* verify the set of connection parameters */
264 52 : check_conn_params(keywords, values, user);
265 :
266 : /*
267 : * We must obey fd.c's limit on non-virtual file descriptors. Assume
268 : * that a PGconn represents one long-lived FD. (Doing this here also
269 : * ensures that VFDs are closed if needed to make room.)
270 : */
271 48 : if (!AcquireExternalFD())
272 : {
273 : #ifndef WIN32 /* can't write #if within ereport() macro */
274 0 : ereport(ERROR,
275 : (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
276 : errmsg("could not connect to server \"%s\"",
277 : server->servername),
278 : errdetail("There are too many open files on the local server."),
279 : errhint("Raise the server's max_files_per_process and/or \"ulimit -n\" limits.")));
280 : #else
281 : ereport(ERROR,
282 : (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
283 : errmsg("could not connect to server \"%s\"",
284 : server->servername),
285 : errdetail("There are too many open files on the local server."),
286 : errhint("Raise the server's max_files_per_process setting.")));
287 : #endif
288 : }
289 :
290 : /* OK to make connection */
291 48 : conn = PQconnectdbParams(keywords, values, false);
292 :
293 48 : if (!conn)
294 0 : ReleaseExternalFD(); /* because the PG_CATCH block won't */
295 :
296 48 : if (!conn || PQstatus(conn) != CONNECTION_OK)
297 4 : ereport(ERROR,
298 : (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
299 : errmsg("could not connect to server \"%s\"",
300 : server->servername),
301 : errdetail_internal("%s", pchomp(PQerrorMessage(conn)))));
302 :
303 : /*
304 : * Check that non-superuser has used password to establish connection;
305 : * otherwise, he's piggybacking on the postgres server's user
306 : * identity. See also dblink_security_check() in contrib/dblink and
307 : * check_conn_params.
308 : */
309 44 : if (!superuser_arg(user->userid) && UserMappingPasswordRequired(user) &&
310 4 : !PQconnectionUsedPassword(conn))
311 4 : ereport(ERROR,
312 : (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
313 : errmsg("password is required"),
314 : errdetail("Non-superuser cannot connect if the server does not request a password."),
315 : errhint("Target server's authentication method must be changed or password_required=false set in the user mapping attributes.")));
316 :
317 : /* Prepare new session for use */
318 40 : configure_remote_session(conn);
319 :
320 40 : pfree(keywords);
321 40 : pfree(values);
322 : }
323 12 : PG_CATCH();
324 : {
325 : /* Release PGconn data structure if we managed to create one */
326 12 : if (conn)
327 : {
328 8 : PQfinish(conn);
329 8 : ReleaseExternalFD();
330 : }
331 12 : PG_RE_THROW();
332 : }
333 40 : PG_END_TRY();
334 :
335 40 : return conn;
336 : }
337 :
338 : /*
339 : * Disconnect any open connection for a connection cache entry.
340 : */
341 : static void
342 26 : disconnect_pg_server(ConnCacheEntry *entry)
343 : {
344 26 : if (entry->conn != NULL)
345 : {
346 26 : PQfinish(entry->conn);
347 26 : entry->conn = NULL;
348 26 : ReleaseExternalFD();
349 : }
350 26 : }
351 :
352 : /*
353 : * Return true if the password_required is defined and false for this user
354 : * mapping, otherwise false. The mapping has been pre-validated.
355 : */
356 : static bool
357 10 : UserMappingPasswordRequired(UserMapping *user)
358 : {
359 : ListCell *cell;
360 :
361 16 : foreach(cell, user->options)
362 : {
363 8 : DefElem *def = (DefElem *) lfirst(cell);
364 :
365 8 : if (strcmp(def->defname, "password_required") == 0)
366 2 : return defGetBoolean(def);
367 : }
368 :
369 8 : return true;
370 : }
371 :
372 : /*
373 : * For non-superusers, insist that the connstr specify a password. This
374 : * prevents a password from being picked up from .pgpass, a service file, the
375 : * environment, etc. We don't want the postgres user's passwords,
376 : * certificates, etc to be accessible to non-superusers. (See also
377 : * dblink_connstr_check in contrib/dblink.)
378 : */
379 : static void
380 52 : check_conn_params(const char **keywords, const char **values, UserMapping *user)
381 : {
382 : int i;
383 :
384 : /* no check required if superuser */
385 52 : if (superuser_arg(user->userid))
386 42 : return;
387 :
388 : /* ok if params contain a non-empty password */
389 38 : for (i = 0; keywords[i] != NULL; i++)
390 : {
391 34 : if (strcmp(keywords[i], "password") == 0 && values[i][0] != '\0')
392 6 : return;
393 : }
394 :
395 : /* ok if the superuser explicitly said so at user mapping creation time */
396 4 : if (!UserMappingPasswordRequired(user))
397 0 : return;
398 :
399 4 : ereport(ERROR,
400 : (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
401 : errmsg("password is required"),
402 : errdetail("Non-superusers must provide a password in the user mapping.")));
403 : }
404 :
405 : /*
406 : * Issue SET commands to make sure remote session is configured properly.
407 : *
408 : * We do this just once at connection, assuming nothing will change the
409 : * values later. Since we'll never send volatile function calls to the
410 : * remote, there shouldn't be any way to break this assumption from our end.
411 : * It's possible to think of ways to break it at the remote end, eg making
412 : * a foreign table point to a view that includes a set_config call ---
413 : * but once you admit the possibility of a malicious view definition,
414 : * there are any number of ways to break things.
415 : */
416 : static void
417 40 : configure_remote_session(PGconn *conn)
418 : {
419 40 : int remoteversion = PQserverVersion(conn);
420 :
421 : /* Force the search path to contain only pg_catalog (see deparse.c) */
422 40 : do_sql_command(conn, "SET search_path = pg_catalog");
423 :
424 : /*
425 : * Set remote timezone; this is basically just cosmetic, since all
426 : * transmitted and returned timestamptzs should specify a zone explicitly
427 : * anyway. However it makes the regression test outputs more predictable.
428 : *
429 : * We don't risk setting remote zone equal to ours, since the remote
430 : * server might use a different timezone database. Instead, use UTC
431 : * (quoted, because very old servers are picky about case).
432 : */
433 40 : do_sql_command(conn, "SET timezone = 'UTC'");
434 :
435 : /*
436 : * Set values needed to ensure unambiguous data output from remote. (This
437 : * logic should match what pg_dump does. See also set_transmission_modes
438 : * in postgres_fdw.c.)
439 : */
440 40 : do_sql_command(conn, "SET datestyle = ISO");
441 40 : if (remoteversion >= 80400)
442 40 : do_sql_command(conn, "SET intervalstyle = postgres");
443 40 : if (remoteversion >= 90000)
444 40 : do_sql_command(conn, "SET extra_float_digits = 3");
445 : else
446 0 : do_sql_command(conn, "SET extra_float_digits = 2");
447 40 : }
448 :
449 : /*
450 : * Convenience subroutine to issue a non-data-returning SQL command to remote
451 : */
452 : static void
453 2086 : do_sql_command(PGconn *conn, const char *sql)
454 : {
455 : PGresult *res;
456 :
457 2086 : if (!PQsendQuery(conn, sql))
458 0 : pgfdw_report_error(ERROR, NULL, conn, false, sql);
459 2086 : res = pgfdw_get_result(conn, sql);
460 2086 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
461 0 : pgfdw_report_error(ERROR, res, conn, true, sql);
462 2086 : PQclear(res);
463 2086 : }
464 :
465 : /*
466 : * Start remote transaction or subtransaction, if needed.
467 : *
468 : * Note that we always use at least REPEATABLE READ in the remote session.
469 : * This is so that, if a query initiates multiple scans of the same or
470 : * different foreign tables, we will get snapshot-consistent results from
471 : * those scans. A disadvantage is that we can't provide sane emulation of
472 : * READ COMMITTED behavior --- it would be nice if we had some other way to
473 : * control which remote queries share a snapshot.
474 : */
475 : static void
476 2814 : begin_remote_xact(ConnCacheEntry *entry)
477 : {
478 2814 : int curlevel = GetCurrentTransactionNestLevel();
479 :
480 : /* Start main transaction if we haven't yet */
481 2814 : if (entry->xact_depth <= 0)
482 : {
483 : const char *sql;
484 :
485 952 : elog(DEBUG3, "starting remote transaction on connection %p",
486 : entry->conn);
487 :
488 952 : if (IsolationIsSerializable())
489 0 : sql = "START TRANSACTION ISOLATION LEVEL SERIALIZABLE";
490 : else
491 952 : sql = "START TRANSACTION ISOLATION LEVEL REPEATABLE READ";
492 952 : entry->changing_xact_state = true;
493 952 : do_sql_command(entry->conn, sql);
494 952 : entry->xact_depth = 1;
495 952 : entry->changing_xact_state = false;
496 : }
497 :
498 : /*
499 : * If we're in a subtransaction, stack up savepoints to match our level.
500 : * This ensures we can rollback just the desired effects when a
501 : * subtransaction aborts.
502 : */
503 2830 : while (entry->xact_depth < curlevel)
504 : {
505 : char sql[64];
506 :
507 16 : snprintf(sql, sizeof(sql), "SAVEPOINT s%d", entry->xact_depth + 1);
508 16 : entry->changing_xact_state = true;
509 16 : do_sql_command(entry->conn, sql);
510 16 : entry->xact_depth++;
511 16 : entry->changing_xact_state = false;
512 : }
513 2814 : }
514 :
515 : /*
516 : * Release connection reference count created by calling GetConnection.
517 : */
518 : void
519 2762 : ReleaseConnection(PGconn *conn)
520 : {
521 : /*
522 : * Currently, we don't actually track connection references because all
523 : * cleanup is managed on a transaction or subtransaction basis instead. So
524 : * there's nothing to do here.
525 : */
526 2762 : }
527 :
528 : /*
529 : * Assign a "unique" number for a cursor.
530 : *
531 : * These really only need to be unique per connection within a transaction.
532 : * For the moment we ignore the per-connection point and assign them across
533 : * all connections in the transaction, but we ask for the connection to be
534 : * supplied in case we want to refine that.
535 : *
536 : * Note that even if wraparound happens in a very long transaction, actual
537 : * collisions are highly improbable; just be sure to use %u not %d to print.
538 : */
539 : unsigned int
540 672 : GetCursorNumber(PGconn *conn)
541 : {
542 672 : return ++cursor_number;
543 : }
544 :
545 : /*
546 : * Assign a "unique" number for a prepared statement.
547 : *
548 : * This works much like GetCursorNumber, except that we never reset the counter
549 : * within a session. That's because we can't be 100% sure we've gotten rid
550 : * of all prepared statements on all connections, and it's not really worth
551 : * increasing the risk of prepared-statement name collisions by resetting.
552 : */
553 : unsigned int
554 216 : GetPrepStmtNumber(PGconn *conn)
555 : {
556 216 : return ++prep_stmt_number;
557 : }
558 :
559 : /*
560 : * Submit a query and wait for the result.
561 : *
562 : * This function is interruptible by signals.
563 : *
564 : * Caller is responsible for the error handling on the result.
565 : */
566 : PGresult *
567 6076 : pgfdw_exec_query(PGconn *conn, const char *query)
568 : {
569 : /*
570 : * Submit a query. Since we don't use non-blocking mode, this also can
571 : * block. But its risk is relatively small, so we ignore that for now.
572 : */
573 6076 : if (!PQsendQuery(conn, query))
574 0 : pgfdw_report_error(ERROR, NULL, conn, false, query);
575 :
576 : /* Wait for the result. */
577 6076 : return pgfdw_get_result(conn, query);
578 : }
579 :
580 : /*
581 : * Wait for the result from a prior asynchronous execution function call.
582 : *
583 : * This function offers quick responsiveness by checking for any interruptions.
584 : *
585 : * This function emulates PQexec()'s behavior of returning the last result
586 : * when there are many.
587 : *
588 : * Caller is responsible for the error handling on the result.
589 : */
590 : PGresult *
591 10308 : pgfdw_get_result(PGconn *conn, const char *query)
592 : {
593 10308 : PGresult *volatile last_res = NULL;
594 :
595 : /* In what follows, do not leak any PGresults on an error. */
596 10308 : PG_TRY();
597 : {
598 : for (;;)
599 10308 : {
600 : PGresult *res;
601 :
602 31014 : while (PQisBusy(conn))
603 : {
604 : int wc;
605 :
606 : /* Sleep until there's something to do */
607 10398 : wc = WaitLatchOrSocket(MyLatch,
608 : WL_LATCH_SET | WL_SOCKET_READABLE |
609 : WL_EXIT_ON_PM_DEATH,
610 : PQsocket(conn),
611 : -1L, PG_WAIT_EXTENSION);
612 10398 : ResetLatch(MyLatch);
613 :
614 10398 : CHECK_FOR_INTERRUPTS();
615 :
616 : /* Data available in socket? */
617 10398 : if (wc & WL_SOCKET_READABLE)
618 : {
619 10394 : if (!PQconsumeInput(conn))
620 0 : pgfdw_report_error(ERROR, NULL, conn, false, query);
621 : }
622 : }
623 :
624 20616 : res = PQgetResult(conn);
625 20616 : if (res == NULL)
626 10308 : break; /* query is complete */
627 :
628 10308 : PQclear(last_res);
629 10308 : last_res = res;
630 : }
631 : }
632 0 : PG_CATCH();
633 : {
634 0 : PQclear(last_res);
635 0 : PG_RE_THROW();
636 : }
637 10308 : PG_END_TRY();
638 :
639 10308 : return last_res;
640 : }
641 :
642 : /*
643 : * Report an error we got from the remote server.
644 : *
645 : * elevel: error level to use (typically ERROR, but might be less)
646 : * res: PGresult containing the error
647 : * conn: connection we did the query on
648 : * clear: if true, PQclear the result (otherwise caller will handle it)
649 : * sql: NULL, or text of remote command we tried to execute
650 : *
651 : * Note: callers that choose not to throw ERROR for a remote error are
652 : * responsible for making sure that the associated ConnCacheEntry gets
653 : * marked with have_error = true.
654 : */
655 : void
656 18 : pgfdw_report_error(int elevel, PGresult *res, PGconn *conn,
657 : bool clear, const char *sql)
658 : {
659 : /* If requested, PGresult must be released before leaving this function. */
660 18 : PG_TRY();
661 : {
662 18 : char *diag_sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
663 18 : char *message_primary = PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY);
664 18 : char *message_detail = PQresultErrorField(res, PG_DIAG_MESSAGE_DETAIL);
665 18 : char *message_hint = PQresultErrorField(res, PG_DIAG_MESSAGE_HINT);
666 18 : char *message_context = PQresultErrorField(res, PG_DIAG_CONTEXT);
667 : int sqlstate;
668 :
669 18 : if (diag_sqlstate)
670 18 : sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
671 : diag_sqlstate[1],
672 : diag_sqlstate[2],
673 : diag_sqlstate[3],
674 : diag_sqlstate[4]);
675 : else
676 0 : sqlstate = ERRCODE_CONNECTION_FAILURE;
677 :
678 : /*
679 : * If we don't get a message from the PGresult, try the PGconn. This
680 : * is needed because for connection-level failures, PQexec may just
681 : * return NULL, not a PGresult at all.
682 : */
683 18 : if (message_primary == NULL)
684 0 : message_primary = pchomp(PQerrorMessage(conn));
685 :
686 18 : ereport(elevel,
687 : (errcode(sqlstate),
688 : message_primary ? errmsg_internal("%s", message_primary) :
689 : errmsg("could not obtain message string for remote error"),
690 : message_detail ? errdetail_internal("%s", message_detail) : 0,
691 : message_hint ? errhint("%s", message_hint) : 0,
692 : message_context ? errcontext("%s", message_context) : 0,
693 : sql ? errcontext("remote SQL command: %s", sql) : 0));
694 : }
695 18 : PG_FINALLY();
696 : {
697 18 : if (clear)
698 16 : PQclear(res);
699 : }
700 18 : PG_END_TRY();
701 0 : }
702 :
703 : /*
704 : * pgfdw_xact_callback --- cleanup at main-transaction end.
705 : */
706 : static void
707 4930 : pgfdw_xact_callback(XactEvent event, void *arg)
708 : {
709 : HASH_SEQ_STATUS scan;
710 : ConnCacheEntry *entry;
711 :
712 : /* Quick exit if no connections were touched in this transaction. */
713 4930 : if (!xact_got_connection)
714 3970 : return;
715 :
716 : /*
717 : * Scan all connection cache entries to find open remote transactions, and
718 : * close them.
719 : */
720 960 : hash_seq_init(&scan, ConnectionHash);
721 3416 : while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
722 : {
723 : PGresult *res;
724 :
725 : /* Ignore cache entry if no open connection right now */
726 2458 : if (entry->conn == NULL)
727 12 : continue;
728 :
729 : /* If it has an open remote transaction, try to close it */
730 2446 : if (entry->xact_depth > 0)
731 : {
732 954 : bool abort_cleanup_failure = false;
733 :
734 954 : elog(DEBUG3, "closing remote transaction on connection %p",
735 : entry->conn);
736 :
737 954 : switch (event)
738 : {
739 908 : case XACT_EVENT_PARALLEL_PRE_COMMIT:
740 : case XACT_EVENT_PRE_COMMIT:
741 :
742 : /*
743 : * If abort cleanup previously failed for this connection,
744 : * we can't issue any more commands against it.
745 : */
746 908 : pgfdw_reject_incomplete_xact_state_change(entry);
747 :
748 : /* Commit all remote transactions during pre-commit */
749 908 : entry->changing_xact_state = true;
750 908 : do_sql_command(entry->conn, "COMMIT TRANSACTION");
751 908 : entry->changing_xact_state = false;
752 :
753 : /*
754 : * If there were any errors in subtransactions, and we
755 : * made prepared statements, do a DEALLOCATE ALL to make
756 : * sure we get rid of all prepared statements. This is
757 : * annoying and not terribly bulletproof, but it's
758 : * probably not worth trying harder.
759 : *
760 : * DEALLOCATE ALL only exists in 8.3 and later, so this
761 : * constrains how old a server postgres_fdw can
762 : * communicate with. We intentionally ignore errors in
763 : * the DEALLOCATE, so that we can hobble along to some
764 : * extent with older servers (leaking prepared statements
765 : * as we go; but we don't really support update operations
766 : * pre-8.3 anyway).
767 : */
768 908 : if (entry->have_prep_stmt && entry->have_error)
769 : {
770 0 : res = PQexec(entry->conn, "DEALLOCATE ALL");
771 0 : PQclear(res);
772 : }
773 908 : entry->have_prep_stmt = false;
774 908 : entry->have_error = false;
775 908 : break;
776 2 : case XACT_EVENT_PRE_PREPARE:
777 :
778 : /*
779 : * We disallow any remote transactions, since it's not
780 : * very reasonable to hold them open until the prepared
781 : * transaction is committed. For the moment, throw error
782 : * unconditionally; later we might allow read-only cases.
783 : * Note that the error will cause us to come right back
784 : * here with event == XACT_EVENT_ABORT, so we'll clean up
785 : * the connection state at that point.
786 : */
787 2 : ereport(ERROR,
788 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
789 : errmsg("cannot PREPARE a transaction that has operated on postgres_fdw foreign tables")));
790 : break;
791 0 : case XACT_EVENT_PARALLEL_COMMIT:
792 : case XACT_EVENT_COMMIT:
793 : case XACT_EVENT_PREPARE:
794 : /* Pre-commit should have closed the open transaction */
795 0 : elog(ERROR, "missed cleaning up connection during pre-commit");
796 : break;
797 44 : case XACT_EVENT_PARALLEL_ABORT:
798 : case XACT_EVENT_ABORT:
799 :
800 : /*
801 : * Don't try to clean up the connection if we're already
802 : * in error recursion trouble.
803 : */
804 44 : if (in_error_recursion_trouble())
805 0 : entry->changing_xact_state = true;
806 :
807 : /*
808 : * If connection is already unsalvageable, don't touch it
809 : * further.
810 : */
811 44 : if (entry->changing_xact_state)
812 0 : break;
813 :
814 : /*
815 : * Mark this connection as in the process of changing
816 : * transaction state.
817 : */
818 44 : entry->changing_xact_state = true;
819 :
820 : /* Assume we might have lost track of prepared statements */
821 44 : entry->have_error = true;
822 :
823 : /*
824 : * If a command has been submitted to the remote server by
825 : * using an asynchronous execution function, the command
826 : * might not have yet completed. Check to see if a
827 : * command is still being processed by the remote server,
828 : * and if so, request cancellation of the command.
829 : */
830 44 : if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE &&
831 0 : !pgfdw_cancel_query(entry->conn))
832 : {
833 : /* Unable to cancel running query. */
834 0 : abort_cleanup_failure = true;
835 : }
836 44 : else if (!pgfdw_exec_cleanup_query(entry->conn,
837 : "ABORT TRANSACTION",
838 : false))
839 : {
840 : /* Unable to abort remote transaction. */
841 0 : abort_cleanup_failure = true;
842 : }
843 44 : else if (entry->have_prep_stmt && entry->have_error &&
844 20 : !pgfdw_exec_cleanup_query(entry->conn,
845 : "DEALLOCATE ALL",
846 : true))
847 : {
848 : /* Trouble clearing prepared statements. */
849 0 : abort_cleanup_failure = true;
850 : }
851 : else
852 : {
853 44 : entry->have_prep_stmt = false;
854 44 : entry->have_error = false;
855 : }
856 :
857 : /* Disarm changing_xact_state if it all worked. */
858 44 : entry->changing_xact_state = abort_cleanup_failure;
859 44 : break;
860 : }
861 1492 : }
862 :
863 : /* Reset state to show we're out of a transaction */
864 2444 : entry->xact_depth = 0;
865 :
866 : /*
867 : * If the connection isn't in a good idle state, discard it to
868 : * recover. Next GetConnection will open a new connection.
869 : */
870 4888 : if (PQstatus(entry->conn) != CONNECTION_OK ||
871 2444 : PQtransactionStatus(entry->conn) != PQTRANS_IDLE ||
872 2444 : entry->changing_xact_state)
873 : {
874 0 : elog(DEBUG3, "discarding connection %p", entry->conn);
875 0 : disconnect_pg_server(entry);
876 : }
877 : }
878 :
879 : /*
880 : * Regardless of the event type, we can now mark ourselves as out of the
881 : * transaction. (Note: if we are here during PRE_COMMIT or PRE_PREPARE,
882 : * this saves a useless scan of the hashtable during COMMIT or PREPARE.)
883 : */
884 958 : xact_got_connection = false;
885 :
886 : /* Also reset cursor numbering for next transaction */
887 958 : cursor_number = 0;
888 : }
889 :
890 : /*
891 : * pgfdw_subxact_callback --- cleanup at subtransaction end.
892 : */
893 : static void
894 46 : pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
895 : SubTransactionId parentSubid, void *arg)
896 : {
897 : HASH_SEQ_STATUS scan;
898 : ConnCacheEntry *entry;
899 : int curlevel;
900 :
901 : /* Nothing to do at subxact start, nor after commit. */
902 46 : if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB ||
903 : event == SUBXACT_EVENT_ABORT_SUB))
904 28 : return;
905 :
906 : /* Quick exit if no connections were touched in this transaction. */
907 18 : if (!xact_got_connection)
908 0 : return;
909 :
910 : /*
911 : * Scan all connection cache entries to find open remote subtransactions
912 : * of the current level, and close them.
913 : */
914 18 : curlevel = GetCurrentTransactionNestLevel();
915 18 : hash_seq_init(&scan, ConnectionHash);
916 72 : while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
917 : {
918 : char sql[100];
919 :
920 : /*
921 : * We only care about connections with open remote subtransactions of
922 : * the current level.
923 : */
924 54 : if (entry->conn == NULL || entry->xact_depth < curlevel)
925 38 : continue;
926 :
927 16 : if (entry->xact_depth > curlevel)
928 0 : elog(ERROR, "missed cleaning up remote subtransaction at level %d",
929 : entry->xact_depth);
930 :
931 16 : if (event == SUBXACT_EVENT_PRE_COMMIT_SUB)
932 : {
933 : /*
934 : * If abort cleanup previously failed for this connection, we
935 : * can't issue any more commands against it.
936 : */
937 10 : pgfdw_reject_incomplete_xact_state_change(entry);
938 :
939 : /* Commit all remote subtransactions during pre-commit */
940 10 : snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel);
941 10 : entry->changing_xact_state = true;
942 10 : do_sql_command(entry->conn, sql);
943 10 : entry->changing_xact_state = false;
944 : }
945 6 : else if (in_error_recursion_trouble())
946 : {
947 : /*
948 : * Don't try to clean up the connection if we're already in error
949 : * recursion trouble.
950 : */
951 0 : entry->changing_xact_state = true;
952 : }
953 6 : else if (!entry->changing_xact_state)
954 : {
955 6 : bool abort_cleanup_failure = false;
956 :
957 : /* Remember that abort cleanup is in progress. */
958 6 : entry->changing_xact_state = true;
959 :
960 : /* Assume we might have lost track of prepared statements */
961 6 : entry->have_error = true;
962 :
963 : /*
964 : * If a command has been submitted to the remote server by using
965 : * an asynchronous execution function, the command might not have
966 : * yet completed. Check to see if a command is still being
967 : * processed by the remote server, and if so, request cancellation
968 : * of the command.
969 : */
970 6 : if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE &&
971 0 : !pgfdw_cancel_query(entry->conn))
972 0 : abort_cleanup_failure = true;
973 : else
974 : {
975 : /* Rollback all remote subtransactions during abort */
976 6 : snprintf(sql, sizeof(sql),
977 : "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d",
978 : curlevel, curlevel);
979 6 : if (!pgfdw_exec_cleanup_query(entry->conn, sql, false))
980 0 : abort_cleanup_failure = true;
981 : }
982 :
983 : /* Disarm changing_xact_state if it all worked. */
984 6 : entry->changing_xact_state = abort_cleanup_failure;
985 : }
986 :
987 : /* OK, we're outta that level of subtransaction */
988 16 : entry->xact_depth--;
989 : }
990 : }
991 :
992 : /*
993 : * Connection invalidation callback function
994 : *
995 : * After a change to a pg_foreign_server or pg_user_mapping catalog entry,
996 : * mark connections depending on that entry as needing to be remade.
997 : * We can't immediately destroy them, since they might be in the midst of
998 : * a transaction, but we'll remake them at the next opportunity.
999 : *
1000 : * Although most cache invalidation callbacks blow away all the related stuff
1001 : * regardless of the given hashvalue, connections are expensive enough that
1002 : * it's worth trying to avoid that.
1003 : *
1004 : * NB: We could avoid unnecessary disconnection more strictly by examining
1005 : * individual option values, but it seems too much effort for the gain.
1006 : */
1007 : static void
1008 112 : pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue)
1009 : {
1010 : HASH_SEQ_STATUS scan;
1011 : ConnCacheEntry *entry;
1012 :
1013 : Assert(cacheid == FOREIGNSERVEROID || cacheid == USERMAPPINGOID);
1014 :
1015 : /* ConnectionHash must exist already, if we're registered */
1016 112 : hash_seq_init(&scan, ConnectionHash);
1017 420 : while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
1018 : {
1019 : /* Ignore invalid entries */
1020 308 : if (entry->conn == NULL)
1021 16 : continue;
1022 :
1023 : /* hashvalue == 0 means a cache reset, must clear all state */
1024 292 : if (hashvalue == 0 ||
1025 168 : (cacheid == FOREIGNSERVEROID &&
1026 292 : entry->server_hashvalue == hashvalue) ||
1027 124 : (cacheid == USERMAPPINGOID &&
1028 124 : entry->mapping_hashvalue == hashvalue))
1029 108 : entry->invalidated = true;
1030 : }
1031 112 : }
1032 :
1033 : /*
1034 : * Raise an error if the given connection cache entry is marked as being
1035 : * in the middle of an xact state change. This should be called at which no
1036 : * such change is expected to be in progress; if one is found to be in
1037 : * progress, it means that we aborted in the middle of a previous state change
1038 : * and now don't know what the remote transaction state actually is.
1039 : * Such connections can't safely be further used. Re-establishing the
1040 : * connection would change the snapshot and roll back any writes already
1041 : * performed, so that's not an option, either. Thus, we must abort.
1042 : */
1043 : static void
1044 3744 : pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry)
1045 : {
1046 : HeapTuple tup;
1047 : Form_pg_user_mapping umform;
1048 : ForeignServer *server;
1049 :
1050 : /* nothing to do for inactive entries and entries of sane state */
1051 3744 : if (entry->conn == NULL || !entry->changing_xact_state)
1052 3744 : return;
1053 :
1054 : /* make sure this entry is inactive */
1055 0 : disconnect_pg_server(entry);
1056 :
1057 : /* find server name to be shown in the message below */
1058 0 : tup = SearchSysCache1(USERMAPPINGOID,
1059 0 : ObjectIdGetDatum(entry->key));
1060 0 : if (!HeapTupleIsValid(tup))
1061 0 : elog(ERROR, "cache lookup failed for user mapping %u", entry->key);
1062 0 : umform = (Form_pg_user_mapping) GETSTRUCT(tup);
1063 0 : server = GetForeignServer(umform->umserver);
1064 0 : ReleaseSysCache(tup);
1065 :
1066 0 : ereport(ERROR,
1067 : (errcode(ERRCODE_CONNECTION_EXCEPTION),
1068 : errmsg("connection to server \"%s\" was lost",
1069 : server->servername)));
1070 : }
1071 :
1072 : /*
1073 : * Cancel the currently-in-progress query (whose query text we do not have)
1074 : * and ignore the result. Returns true if we successfully cancel the query
1075 : * and discard any pending result, and false if not.
1076 : */
1077 : static bool
1078 0 : pgfdw_cancel_query(PGconn *conn)
1079 : {
1080 : PGcancel *cancel;
1081 : char errbuf[256];
1082 0 : PGresult *result = NULL;
1083 : TimestampTz endtime;
1084 :
1085 : /*
1086 : * If it takes too long to cancel the query and discard the result, assume
1087 : * the connection is dead.
1088 : */
1089 0 : endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);
1090 :
1091 : /*
1092 : * Issue cancel request. Unfortunately, there's no good way to limit the
1093 : * amount of time that we might block inside PQgetCancel().
1094 : */
1095 0 : if ((cancel = PQgetCancel(conn)))
1096 : {
1097 0 : if (!PQcancel(cancel, errbuf, sizeof(errbuf)))
1098 : {
1099 0 : ereport(WARNING,
1100 : (errcode(ERRCODE_CONNECTION_FAILURE),
1101 : errmsg("could not send cancel request: %s",
1102 : errbuf)));
1103 0 : PQfreeCancel(cancel);
1104 0 : return false;
1105 : }
1106 0 : PQfreeCancel(cancel);
1107 : }
1108 :
1109 : /* Get and discard the result of the query. */
1110 0 : if (pgfdw_get_cleanup_result(conn, endtime, &result))
1111 0 : return false;
1112 0 : PQclear(result);
1113 :
1114 0 : return true;
1115 : }
1116 :
1117 : /*
1118 : * Submit a query during (sub)abort cleanup and wait up to 30 seconds for the
1119 : * result. If the query is executed without error, the return value is true.
1120 : * If the query is executed successfully but returns an error, the return
1121 : * value is true if and only if ignore_errors is set. If the query can't be
1122 : * sent or times out, the return value is false.
1123 : */
1124 : static bool
1125 70 : pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors)
1126 : {
1127 70 : PGresult *result = NULL;
1128 : TimestampTz endtime;
1129 :
1130 : /*
1131 : * If it takes too long to execute a cleanup query, assume the connection
1132 : * is dead. It's fairly likely that this is why we aborted in the first
1133 : * place (e.g. statement timeout, user cancel), so the timeout shouldn't
1134 : * be too long.
1135 : */
1136 70 : endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);
1137 :
1138 : /*
1139 : * Submit a query. Since we don't use non-blocking mode, this also can
1140 : * block. But its risk is relatively small, so we ignore that for now.
1141 : */
1142 70 : if (!PQsendQuery(conn, query))
1143 : {
1144 0 : pgfdw_report_error(WARNING, NULL, conn, false, query);
1145 0 : return false;
1146 : }
1147 :
1148 : /* Get the result of the query. */
1149 70 : if (pgfdw_get_cleanup_result(conn, endtime, &result))
1150 0 : return false;
1151 :
1152 : /* Issue a warning if not successful. */
1153 70 : if (PQresultStatus(result) != PGRES_COMMAND_OK)
1154 : {
1155 0 : pgfdw_report_error(WARNING, result, conn, true, query);
1156 0 : return ignore_errors;
1157 : }
1158 70 : PQclear(result);
1159 :
1160 70 : return true;
1161 : }
1162 :
1163 : /*
1164 : * Get, during abort cleanup, the result of a query that is in progress. This
1165 : * might be a query that is being interrupted by transaction abort, or it might
1166 : * be a query that was initiated as part of transaction abort to get the remote
1167 : * side back to the appropriate state.
1168 : *
1169 : * It's not a huge problem if we throw an ERROR here, but if we get into error
1170 : * recursion trouble, we'll end up slamming the connection shut, which will
1171 : * necessitate failing the entire toplevel transaction even if subtransactions
1172 : * were used. Try to use WARNING where we can.
1173 : *
1174 : * endtime is the time at which we should give up and assume the remote
1175 : * side is dead. Returns true if the timeout expired, otherwise false.
1176 : * Sets *result except in case of a timeout.
1177 : */
1178 : static bool
1179 70 : pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result)
1180 : {
1181 70 : volatile bool timed_out = false;
1182 70 : PGresult *volatile last_res = NULL;
1183 :
1184 : /* In what follows, do not leak any PGresults on an error. */
1185 70 : PG_TRY();
1186 : {
1187 : for (;;)
1188 76 : {
1189 : PGresult *res;
1190 :
1191 216 : while (PQisBusy(conn))
1192 : {
1193 : int wc;
1194 70 : TimestampTz now = GetCurrentTimestamp();
1195 : long secs;
1196 : int microsecs;
1197 : long cur_timeout;
1198 :
1199 : /* If timeout has expired, give up, else get sleep time. */
1200 70 : if (now >= endtime)
1201 : {
1202 0 : timed_out = true;
1203 0 : goto exit;
1204 : }
1205 70 : TimestampDifference(now, endtime, &secs, µsecs);
1206 :
1207 : /* To protect against clock skew, limit sleep to one minute. */
1208 70 : cur_timeout = Min(60000, secs * USECS_PER_SEC + microsecs);
1209 :
1210 : /* Sleep until there's something to do */
1211 70 : wc = WaitLatchOrSocket(MyLatch,
1212 : WL_LATCH_SET | WL_SOCKET_READABLE |
1213 : WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
1214 : PQsocket(conn),
1215 : cur_timeout, PG_WAIT_EXTENSION);
1216 70 : ResetLatch(MyLatch);
1217 :
1218 70 : CHECK_FOR_INTERRUPTS();
1219 :
1220 : /* Data available in socket? */
1221 70 : if (wc & WL_SOCKET_READABLE)
1222 : {
1223 70 : if (!PQconsumeInput(conn))
1224 : {
1225 : /* connection trouble; treat the same as a timeout */
1226 0 : timed_out = true;
1227 0 : goto exit;
1228 : }
1229 : }
1230 : }
1231 :
1232 146 : res = PQgetResult(conn);
1233 146 : if (res == NULL)
1234 70 : break; /* query is complete */
1235 :
1236 76 : PQclear(last_res);
1237 76 : last_res = res;
1238 : }
1239 70 : exit: ;
1240 : }
1241 0 : PG_CATCH();
1242 : {
1243 0 : PQclear(last_res);
1244 0 : PG_RE_THROW();
1245 : }
1246 70 : PG_END_TRY();
1247 :
1248 70 : if (timed_out)
1249 0 : PQclear(last_res);
1250 : else
1251 70 : *result = last_res;
1252 70 : return timed_out;
1253 : }
|
