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