Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * connection.c
4 : * Connection management functions for postgres_fdw
5 : *
6 : * Portions Copyright (c) 2012-2018, 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 "catalog/pg_user_mapping.h"
19 : #include "access/xact.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 2660 : 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 2660 : 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 2660 : xact_got_connection = true;
141 :
142 : /* Create hash key for the entry. Assume no pad bytes in key struct */
143 2660 : key = user->umid;
144 :
145 : /*
146 : * Find or create cached entry for requested connection.
147 : */
148 2660 : entry = hash_search(ConnectionHash, &key, HASH_ENTER, &found);
149 2660 : 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 2660 : 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 2660 : 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 2660 : 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 2656 : begin_remote_xact(entry);
211 :
212 : /* Remember if caller will prepare statements */
213 2656 : entry->have_prep_stmt |= will_prep_stmt;
214 :
215 2656 : 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 1912 : do_sql_command(PGconn *conn, const char *sql)
395 : {
396 : PGresult *res;
397 :
398 1912 : if (!PQsendQuery(conn, sql))
399 0 : pgfdw_report_error(ERROR, NULL, conn, false, sql);
400 1912 : res = pgfdw_get_result(conn, sql);
401 1912 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
402 0 : pgfdw_report_error(ERROR, res, conn, true, sql);
403 1912 : PQclear(res);
404 1912 : }
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 2656 : begin_remote_xact(ConnCacheEntry *entry)
418 : {
419 2656 : int curlevel = GetCurrentTransactionNestLevel();
420 :
421 : /* Start main transaction if we haven't yet */
422 2656 : if (entry->xact_depth <= 0)
423 : {
424 : const char *sql;
425 :
426 882 : elog(DEBUG3, "starting remote transaction on connection %p",
427 : entry->conn);
428 :
429 882 : if (IsolationIsSerializable())
430 0 : sql = "START TRANSACTION ISOLATION LEVEL SERIALIZABLE";
431 : else
432 882 : sql = "START TRANSACTION ISOLATION LEVEL REPEATABLE READ";
433 882 : entry->changing_xact_state = true;
434 882 : do_sql_command(entry->conn, sql);
435 882 : entry->xact_depth = 1;
436 882 : 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 5328 : 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 2656 : }
455 :
456 : /*
457 : * Release connection reference count created by calling GetConnection.
458 : */
459 : void
460 2610 : 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 2610 : }
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 642 : GetCursorNumber(PGconn *conn)
482 : {
483 642 : 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 184 : GetPrepStmtNumber(PGconn *conn)
496 : {
497 184 : 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 5870 : 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 5870 : if (!PQsendQuery(conn, query))
515 0 : pgfdw_report_error(ERROR, NULL, conn, false, query);
516 :
517 : /* Wait for the result. */
518 5870 : 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 9776 : pgfdw_get_result(PGconn *conn, const char *query)
533 : {
534 9776 : PGresult *volatile last_res = NULL;
535 :
536 : /* In what follows, do not leak any PGresults on an error. */
537 9776 : PG_TRY();
538 : {
539 : for (;;)
540 9776 : {
541 : PGresult *res;
542 :
543 48978 : while (PQisBusy(conn))
544 : {
545 : int wc;
546 :
547 : /* Sleep until there's something to do */
548 9874 : wc = WaitLatchOrSocket(MyLatch,
549 : WL_LATCH_SET | WL_SOCKET_READABLE,
550 : PQsocket(conn),
551 : -1L, PG_WAIT_EXTENSION);
552 9874 : ResetLatch(MyLatch);
553 :
554 9874 : CHECK_FOR_INTERRUPTS();
555 :
556 : /* Data available in socket? */
557 9874 : if (wc & WL_SOCKET_READABLE)
558 : {
559 9872 : if (!PQconsumeInput(conn))
560 0 : pgfdw_report_error(ERROR, NULL, conn, false, query);
561 : }
562 : }
563 :
564 19552 : res = PQgetResult(conn);
565 19552 : if (res == NULL)
566 9776 : break; /* query is complete */
567 :
568 9776 : PQclear(last_res);
569 9776 : last_res = res;
570 : }
571 : }
572 0 : PG_CATCH();
573 : {
574 0 : PQclear(last_res);
575 0 : PG_RE_THROW();
576 : }
577 9776 : PG_END_TRY();
578 :
579 9776 : return last_res;
580 : }
581 :
582 : /*
583 : * Report an error we got from the remote server.
584 : *
585 : * elevel: error level to use (typically ERROR, but might be less)
586 : * res: PGresult containing the error
587 : * conn: connection we did the query on
588 : * clear: if true, PQclear the result (otherwise caller will handle it)
589 : * sql: NULL, or text of remote command we tried to execute
590 : *
591 : * Note: callers that choose not to throw ERROR for a remote error are
592 : * responsible for making sure that the associated ConnCacheEntry gets
593 : * marked with have_error = true.
594 : */
595 : void
596 18 : pgfdw_report_error(int elevel, PGresult *res, PGconn *conn,
597 : bool clear, const char *sql)
598 : {
599 : /* If requested, PGresult must be released before leaving this function. */
600 18 : PG_TRY();
601 : {
602 18 : char *diag_sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
603 18 : char *message_primary = PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY);
604 18 : char *message_detail = PQresultErrorField(res, PG_DIAG_MESSAGE_DETAIL);
605 18 : char *message_hint = PQresultErrorField(res, PG_DIAG_MESSAGE_HINT);
606 18 : char *message_context = PQresultErrorField(res, PG_DIAG_CONTEXT);
607 : int sqlstate;
608 :
609 18 : if (diag_sqlstate)
610 18 : sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
611 : diag_sqlstate[1],
612 : diag_sqlstate[2],
613 : diag_sqlstate[3],
614 : diag_sqlstate[4]);
615 : else
616 0 : sqlstate = ERRCODE_CONNECTION_FAILURE;
617 :
618 : /*
619 : * If we don't get a message from the PGresult, try the PGconn. This
620 : * is needed because for connection-level failures, PQexec may just
621 : * return NULL, not a PGresult at all.
622 : */
623 18 : if (message_primary == NULL)
624 0 : message_primary = pchomp(PQerrorMessage(conn));
625 :
626 18 : ereport(elevel,
627 : (errcode(sqlstate),
628 : message_primary ? errmsg_internal("%s", message_primary) :
629 : errmsg("could not obtain message string for remote error"),
630 : message_detail ? errdetail_internal("%s", message_detail) : 0,
631 : message_hint ? errhint("%s", message_hint) : 0,
632 : message_context ? errcontext("%s", message_context) : 0,
633 : sql ? errcontext("remote SQL command: %s", sql) : 0));
634 : }
635 18 : PG_CATCH();
636 : {
637 18 : if (clear)
638 16 : PQclear(res);
639 18 : PG_RE_THROW();
640 : }
641 0 : PG_END_TRY();
642 0 : if (clear)
643 0 : PQclear(res);
644 0 : }
645 :
646 : /*
647 : * pgfdw_xact_callback --- cleanup at main-transaction end.
648 : */
649 : static void
650 4648 : pgfdw_xact_callback(XactEvent event, void *arg)
651 : {
652 : HASH_SEQ_STATUS scan;
653 : ConnCacheEntry *entry;
654 :
655 : /* Quick exit if no connections were touched in this transaction. */
656 4648 : if (!xact_got_connection)
657 3768 : return;
658 :
659 : /*
660 : * Scan all connection cache entries to find open remote transactions, and
661 : * close them.
662 : */
663 880 : hash_seq_init(&scan, ConnectionHash);
664 3978 : while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
665 : {
666 : PGresult *res;
667 :
668 : /* Ignore cache entry if no open connection right now */
669 2218 : if (entry->conn == NULL)
670 4 : continue;
671 :
672 : /* If it has an open remote transaction, try to close it */
673 2214 : if (entry->xact_depth > 0)
674 : {
675 882 : bool abort_cleanup_failure = false;
676 :
677 882 : elog(DEBUG3, "closing remote transaction on connection %p",
678 : entry->conn);
679 :
680 882 : switch (event)
681 : {
682 : case XACT_EVENT_PARALLEL_PRE_COMMIT:
683 : case XACT_EVENT_PRE_COMMIT:
684 :
685 : /*
686 : * If abort cleanup previously failed for this connection,
687 : * we can't issue any more commands against it.
688 : */
689 844 : pgfdw_reject_incomplete_xact_state_change(entry);
690 :
691 : /* Commit all remote transactions during pre-commit */
692 844 : entry->changing_xact_state = true;
693 844 : do_sql_command(entry->conn, "COMMIT TRANSACTION");
694 844 : entry->changing_xact_state = false;
695 :
696 : /*
697 : * If there were any errors in subtransactions, and we
698 : * made prepared statements, do a DEALLOCATE ALL to make
699 : * sure we get rid of all prepared statements. This is
700 : * annoying and not terribly bulletproof, but it's
701 : * probably not worth trying harder.
702 : *
703 : * DEALLOCATE ALL only exists in 8.3 and later, so this
704 : * constrains how old a server postgres_fdw can
705 : * communicate with. We intentionally ignore errors in
706 : * the DEALLOCATE, so that we can hobble along to some
707 : * extent with older servers (leaking prepared statements
708 : * as we go; but we don't really support update operations
709 : * pre-8.3 anyway).
710 : */
711 844 : if (entry->have_prep_stmt && entry->have_error)
712 : {
713 0 : res = PQexec(entry->conn, "DEALLOCATE ALL");
714 0 : PQclear(res);
715 : }
716 844 : entry->have_prep_stmt = false;
717 844 : entry->have_error = false;
718 844 : break;
719 : case XACT_EVENT_PRE_PREPARE:
720 :
721 : /*
722 : * We disallow remote transactions that modified anything,
723 : * since it's not very reasonable to hold them open until
724 : * the prepared transaction is committed. For the moment,
725 : * throw error unconditionally; later we might allow
726 : * read-only cases. Note that the error will cause us to
727 : * come right back here with event == XACT_EVENT_ABORT, so
728 : * we'll clean up the connection state at that point.
729 : */
730 0 : ereport(ERROR,
731 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
732 : errmsg("cannot prepare a transaction that modified remote tables")));
733 : break;
734 : case XACT_EVENT_PARALLEL_COMMIT:
735 : case XACT_EVENT_COMMIT:
736 : case XACT_EVENT_PREPARE:
737 : /* Pre-commit should have closed the open transaction */
738 0 : elog(ERROR, "missed cleaning up connection during pre-commit");
739 : break;
740 : case XACT_EVENT_PARALLEL_ABORT:
741 : case XACT_EVENT_ABORT:
742 :
743 : /*
744 : * Don't try to clean up the connection if we're already
745 : * in error recursion trouble.
746 : */
747 38 : if (in_error_recursion_trouble())
748 0 : entry->changing_xact_state = true;
749 :
750 : /*
751 : * If connection is already unsalvageable, don't touch it
752 : * further.
753 : */
754 38 : if (entry->changing_xact_state)
755 0 : break;
756 :
757 : /*
758 : * Mark this connection as in the process of changing
759 : * transaction state.
760 : */
761 38 : entry->changing_xact_state = true;
762 :
763 : /* Assume we might have lost track of prepared statements */
764 38 : entry->have_error = true;
765 :
766 : /*
767 : * If a command has been submitted to the remote server by
768 : * using an asynchronous execution function, the command
769 : * might not have yet completed. Check to see if a
770 : * command is still being processed by the remote server,
771 : * and if so, request cancellation of the command.
772 : */
773 38 : if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE &&
774 0 : !pgfdw_cancel_query(entry->conn))
775 : {
776 : /* Unable to cancel running query. */
777 0 : abort_cleanup_failure = true;
778 : }
779 38 : else if (!pgfdw_exec_cleanup_query(entry->conn,
780 : "ABORT TRANSACTION",
781 : false))
782 : {
783 : /* Unable to abort remote transaction. */
784 0 : abort_cleanup_failure = true;
785 : }
786 56 : else if (entry->have_prep_stmt && entry->have_error &&
787 18 : !pgfdw_exec_cleanup_query(entry->conn,
788 : "DEALLOCATE ALL",
789 : true))
790 : {
791 : /* Trouble clearing prepared statements. */
792 0 : abort_cleanup_failure = true;
793 : }
794 : else
795 : {
796 38 : entry->have_prep_stmt = false;
797 38 : entry->have_error = false;
798 : }
799 :
800 : /* Disarm changing_xact_state if it all worked. */
801 38 : entry->changing_xact_state = abort_cleanup_failure;
802 38 : break;
803 : }
804 : }
805 :
806 : /* Reset state to show we're out of a transaction */
807 2214 : entry->xact_depth = 0;
808 :
809 : /*
810 : * If the connection isn't in a good idle state, discard it to
811 : * recover. Next GetConnection will open a new connection.
812 : */
813 4428 : if (PQstatus(entry->conn) != CONNECTION_OK ||
814 4428 : PQtransactionStatus(entry->conn) != PQTRANS_IDLE ||
815 2214 : entry->changing_xact_state)
816 : {
817 0 : elog(DEBUG3, "discarding connection %p", entry->conn);
818 0 : disconnect_pg_server(entry);
819 : }
820 : }
821 :
822 : /*
823 : * Regardless of the event type, we can now mark ourselves as out of the
824 : * transaction. (Note: if we are here during PRE_COMMIT or PRE_PREPARE,
825 : * this saves a useless scan of the hashtable during COMMIT or PREPARE.)
826 : */
827 880 : xact_got_connection = false;
828 :
829 : /* Also reset cursor numbering for next transaction */
830 880 : cursor_number = 0;
831 : }
832 :
833 : /*
834 : * pgfdw_subxact_callback --- cleanup at subtransaction end.
835 : */
836 : static void
837 46 : pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
838 : SubTransactionId parentSubid, void *arg)
839 : {
840 : HASH_SEQ_STATUS scan;
841 : ConnCacheEntry *entry;
842 : int curlevel;
843 :
844 : /* Nothing to do at subxact start, nor after commit. */
845 46 : if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB ||
846 : event == SUBXACT_EVENT_ABORT_SUB))
847 56 : return;
848 :
849 : /* Quick exit if no connections were touched in this transaction. */
850 18 : if (!xact_got_connection)
851 0 : return;
852 :
853 : /*
854 : * Scan all connection cache entries to find open remote subtransactions
855 : * of the current level, and close them.
856 : */
857 18 : curlevel = GetCurrentTransactionNestLevel();
858 18 : hash_seq_init(&scan, ConnectionHash);
859 90 : while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
860 : {
861 : char sql[100];
862 :
863 : /*
864 : * We only care about connections with open remote subtransactions of
865 : * the current level.
866 : */
867 54 : if (entry->conn == NULL || entry->xact_depth < curlevel)
868 38 : continue;
869 :
870 16 : if (entry->xact_depth > curlevel)
871 0 : elog(ERROR, "missed cleaning up remote subtransaction at level %d",
872 : entry->xact_depth);
873 :
874 16 : if (event == SUBXACT_EVENT_PRE_COMMIT_SUB)
875 : {
876 : /*
877 : * If abort cleanup previously failed for this connection, we
878 : * can't issue any more commands against it.
879 : */
880 10 : pgfdw_reject_incomplete_xact_state_change(entry);
881 :
882 : /* Commit all remote subtransactions during pre-commit */
883 10 : snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel);
884 10 : entry->changing_xact_state = true;
885 10 : do_sql_command(entry->conn, sql);
886 10 : entry->changing_xact_state = false;
887 : }
888 6 : else if (in_error_recursion_trouble())
889 : {
890 : /*
891 : * Don't try to clean up the connection if we're already in error
892 : * recursion trouble.
893 : */
894 0 : entry->changing_xact_state = true;
895 : }
896 6 : else if (!entry->changing_xact_state)
897 : {
898 6 : bool abort_cleanup_failure = false;
899 :
900 : /* Remember that abort cleanup is in progress. */
901 6 : entry->changing_xact_state = true;
902 :
903 : /* Assume we might have lost track of prepared statements */
904 6 : entry->have_error = true;
905 :
906 : /*
907 : * If a command has been submitted to the remote server by using
908 : * an asynchronous execution function, the command might not have
909 : * yet completed. Check to see if a command is still being
910 : * processed by the remote server, and if so, request cancellation
911 : * of the command.
912 : */
913 6 : if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE &&
914 0 : !pgfdw_cancel_query(entry->conn))
915 0 : abort_cleanup_failure = true;
916 : else
917 : {
918 : /* Rollback all remote subtransactions during abort */
919 6 : snprintf(sql, sizeof(sql),
920 : "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d",
921 : curlevel, curlevel);
922 6 : if (!pgfdw_exec_cleanup_query(entry->conn, sql, false))
923 0 : abort_cleanup_failure = true;
924 : }
925 :
926 : /* Disarm changing_xact_state if it all worked. */
927 6 : entry->changing_xact_state = abort_cleanup_failure;
928 : }
929 :
930 : /* OK, we're outta that level of subtransaction */
931 16 : entry->xact_depth--;
932 : }
933 : }
934 :
935 : /*
936 : * Connection invalidation callback function
937 : *
938 : * After a change to a pg_foreign_server or pg_user_mapping catalog entry,
939 : * mark connections depending on that entry as needing to be remade.
940 : * We can't immediately destroy them, since they might be in the midst of
941 : * a transaction, but we'll remake them at the next opportunity.
942 : *
943 : * Although most cache invalidation callbacks blow away all the related stuff
944 : * regardless of the given hashvalue, connections are expensive enough that
945 : * it's worth trying to avoid that.
946 : *
947 : * NB: We could avoid unnecessary disconnection more strictly by examining
948 : * individual option values, but it seems too much effort for the gain.
949 : */
950 : static void
951 46 : pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue)
952 : {
953 : HASH_SEQ_STATUS scan;
954 : ConnCacheEntry *entry;
955 :
956 : Assert(cacheid == FOREIGNSERVEROID || cacheid == USERMAPPINGOID);
957 :
958 : /* ConnectionHash must exist already, if we're registered */
959 46 : hash_seq_init(&scan, ConnectionHash);
960 196 : while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
961 : {
962 : /* Ignore invalid entries */
963 104 : if (entry->conn == NULL)
964 6 : continue;
965 :
966 : /* hashvalue == 0 means a cache reset, must clear all state */
967 98 : if (hashvalue == 0 ||
968 80 : (cacheid == FOREIGNSERVEROID &&
969 138 : entry->server_hashvalue == hashvalue) ||
970 18 : (cacheid == USERMAPPINGOID &&
971 18 : entry->mapping_hashvalue == hashvalue))
972 46 : entry->invalidated = true;
973 : }
974 46 : }
975 :
976 : /*
977 : * Raise an error if the given connection cache entry is marked as being
978 : * in the middle of an xact state change. This should be called at which no
979 : * such change is expected to be in progress; if one is found to be in
980 : * progress, it means that we aborted in the middle of a previous state change
981 : * and now don't know what the remote transaction state actually is.
982 : * Such connections can't safely be further used. Re-establishing the
983 : * connection would change the snapshot and roll back any writes already
984 : * performed, so that's not an option, either. Thus, we must abort.
985 : */
986 : static void
987 3514 : pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry)
988 : {
989 : HeapTuple tup;
990 : Form_pg_user_mapping umform;
991 : ForeignServer *server;
992 :
993 : /* nothing to do for inactive entries and entries of sane state */
994 3514 : if (entry->conn == NULL || !entry->changing_xact_state)
995 3514 : return;
996 :
997 : /* make sure this entry is inactive */
998 0 : disconnect_pg_server(entry);
999 :
1000 : /* find server name to be shown in the message below */
1001 0 : tup = SearchSysCache1(USERMAPPINGOID,
1002 0 : ObjectIdGetDatum(entry->key));
1003 0 : if (!HeapTupleIsValid(tup))
1004 0 : elog(ERROR, "cache lookup failed for user mapping %u", entry->key);
1005 0 : umform = (Form_pg_user_mapping) GETSTRUCT(tup);
1006 0 : server = GetForeignServer(umform->umserver);
1007 0 : ReleaseSysCache(tup);
1008 :
1009 0 : ereport(ERROR,
1010 : (errcode(ERRCODE_CONNECTION_EXCEPTION),
1011 : errmsg("connection to server \"%s\" was lost",
1012 : server->servername)));
1013 : }
1014 :
1015 : /*
1016 : * Cancel the currently-in-progress query (whose query text we do not have)
1017 : * and ignore the result. Returns true if we successfully cancel the query
1018 : * and discard any pending result, and false if not.
1019 : */
1020 : static bool
1021 0 : pgfdw_cancel_query(PGconn *conn)
1022 : {
1023 : PGcancel *cancel;
1024 : char errbuf[256];
1025 0 : PGresult *result = NULL;
1026 : TimestampTz endtime;
1027 :
1028 : /*
1029 : * If it takes too long to cancel the query and discard the result, assume
1030 : * the connection is dead.
1031 : */
1032 0 : endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);
1033 :
1034 : /*
1035 : * Issue cancel request. Unfortunately, there's no good way to limit the
1036 : * amount of time that we might block inside PQgetCancel().
1037 : */
1038 0 : if ((cancel = PQgetCancel(conn)))
1039 : {
1040 0 : if (!PQcancel(cancel, errbuf, sizeof(errbuf)))
1041 : {
1042 0 : ereport(WARNING,
1043 : (errcode(ERRCODE_CONNECTION_FAILURE),
1044 : errmsg("could not send cancel request: %s",
1045 : errbuf)));
1046 0 : PQfreeCancel(cancel);
1047 0 : return false;
1048 : }
1049 0 : PQfreeCancel(cancel);
1050 : }
1051 :
1052 : /* Get and discard the result of the query. */
1053 0 : if (pgfdw_get_cleanup_result(conn, endtime, &result))
1054 0 : return false;
1055 0 : PQclear(result);
1056 :
1057 0 : return true;
1058 : }
1059 :
1060 : /*
1061 : * Submit a query during (sub)abort cleanup and wait up to 30 seconds for the
1062 : * result. If the query is executed without error, the return value is true.
1063 : * If the query is executed successfully but returns an error, the return
1064 : * value is true if and only if ignore_errors is set. If the query can't be
1065 : * sent or times out, the return value is false.
1066 : */
1067 : static bool
1068 62 : pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors)
1069 : {
1070 62 : PGresult *result = NULL;
1071 : TimestampTz endtime;
1072 :
1073 : /*
1074 : * If it takes too long to execute a cleanup query, assume the connection
1075 : * is dead. It's fairly likely that this is why we aborted in the first
1076 : * place (e.g. statement timeout, user cancel), so the timeout shouldn't
1077 : * be too long.
1078 : */
1079 62 : endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);
1080 :
1081 : /*
1082 : * Submit a query. Since we don't use non-blocking mode, this also can
1083 : * block. But its risk is relatively small, so we ignore that for now.
1084 : */
1085 62 : if (!PQsendQuery(conn, query))
1086 : {
1087 0 : pgfdw_report_error(WARNING, NULL, conn, false, query);
1088 0 : return false;
1089 : }
1090 :
1091 : /* Get the result of the query. */
1092 62 : if (pgfdw_get_cleanup_result(conn, endtime, &result))
1093 0 : return false;
1094 :
1095 : /* Issue a warning if not successful. */
1096 62 : if (PQresultStatus(result) != PGRES_COMMAND_OK)
1097 : {
1098 0 : pgfdw_report_error(WARNING, result, conn, true, query);
1099 0 : return ignore_errors;
1100 : }
1101 62 : PQclear(result);
1102 :
1103 62 : return true;
1104 : }
1105 :
1106 : /*
1107 : * Get, during abort cleanup, the result of a query that is in progress. This
1108 : * might be a query that is being interrupted by transaction abort, or it might
1109 : * be a query that was initiated as part of transaction abort to get the remote
1110 : * side back to the appropriate state.
1111 : *
1112 : * It's not a huge problem if we throw an ERROR here, but if we get into error
1113 : * recursion trouble, we'll end up slamming the connection shut, which will
1114 : * necessitate failing the entire toplevel transaction even if subtransactions
1115 : * were used. Try to use WARNING where we can.
1116 : *
1117 : * endtime is the time at which we should give up and assume the remote
1118 : * side is dead. Returns true if the timeout expired, otherwise false.
1119 : * Sets *result except in case of a timeout.
1120 : */
1121 : static bool
1122 62 : pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result)
1123 : {
1124 62 : volatile bool timed_out = false;
1125 62 : PGresult *volatile last_res = NULL;
1126 :
1127 : /* In what follows, do not leak any PGresults on an error. */
1128 62 : PG_TRY();
1129 : {
1130 : for (;;)
1131 68 : {
1132 : PGresult *res;
1133 :
1134 322 : while (PQisBusy(conn))
1135 : {
1136 : int wc;
1137 62 : TimestampTz now = GetCurrentTimestamp();
1138 : long secs;
1139 : int microsecs;
1140 : long cur_timeout;
1141 :
1142 : /* If timeout has expired, give up, else get sleep time. */
1143 62 : if (now >= endtime)
1144 : {
1145 0 : timed_out = true;
1146 0 : goto exit;
1147 : }
1148 62 : TimestampDifference(now, endtime, &secs, µsecs);
1149 :
1150 : /* To protect against clock skew, limit sleep to one minute. */
1151 62 : cur_timeout = Min(60000, secs * USECS_PER_SEC + microsecs);
1152 :
1153 : /* Sleep until there's something to do */
1154 62 : wc = WaitLatchOrSocket(MyLatch,
1155 : WL_LATCH_SET | WL_SOCKET_READABLE | WL_TIMEOUT,
1156 : PQsocket(conn),
1157 : cur_timeout, PG_WAIT_EXTENSION);
1158 62 : ResetLatch(MyLatch);
1159 :
1160 62 : CHECK_FOR_INTERRUPTS();
1161 :
1162 : /* Data available in socket? */
1163 62 : if (wc & WL_SOCKET_READABLE)
1164 : {
1165 62 : if (!PQconsumeInput(conn))
1166 : {
1167 : /* connection trouble; treat the same as a timeout */
1168 0 : timed_out = true;
1169 0 : goto exit;
1170 : }
1171 : }
1172 : }
1173 :
1174 130 : res = PQgetResult(conn);
1175 130 : if (res == NULL)
1176 62 : break; /* query is complete */
1177 :
1178 68 : PQclear(last_res);
1179 68 : last_res = res;
1180 : }
1181 : exit: ;
1182 : }
1183 0 : PG_CATCH();
1184 : {
1185 0 : PQclear(last_res);
1186 0 : PG_RE_THROW();
1187 : }
1188 62 : PG_END_TRY();
1189 :
1190 62 : if (timed_out)
1191 0 : PQclear(last_res);
1192 : else
1193 62 : *result = last_res;
1194 62 : return timed_out;
1195 : }
|
