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