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