PostgreSQL Source Code  git master
connection.c File Reference
#include "postgres.h"
#include "access/htup_details.h"
#include "access/xact.h"
#include "catalog/pg_user_mapping.h"
#include "commands/defrem.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postgres_fdw.h"
#include "storage/fd.h"
#include "storage/latch.h"
#include "utils/datetime.h"
#include "utils/hsearch.h"
#include "utils/inval.h"
#include "utils/memutils.h"
#include "utils/syscache.h"
Include dependency graph for connection.c:

Go to the source code of this file.

Data Structures

struct  ConnCacheEntry
 

Typedefs

typedef Oid ConnCacheKey
 
typedef struct ConnCacheEntry ConnCacheEntry
 

Functions

static void make_new_connection (ConnCacheEntry *entry, UserMapping *user)
 
static PGconnconnect_pg_server (ForeignServer *server, UserMapping *user)
 
static void disconnect_pg_server (ConnCacheEntry *entry)
 
static void check_conn_params (const char **keywords, const char **values, UserMapping *user)
 
static void configure_remote_session (PGconn *conn)
 
static void do_sql_command (PGconn *conn, const char *sql)
 
static void begin_remote_xact (ConnCacheEntry *entry)
 
static void pgfdw_xact_callback (XactEvent event, void *arg)
 
static void pgfdw_subxact_callback (SubXactEvent event, SubTransactionId mySubid, SubTransactionId parentSubid, void *arg)
 
static void pgfdw_inval_callback (Datum arg, int cacheid, uint32 hashvalue)
 
static void pgfdw_reject_incomplete_xact_state_change (ConnCacheEntry *entry)
 
static bool pgfdw_cancel_query (PGconn *conn)
 
static bool pgfdw_exec_cleanup_query (PGconn *conn, const char *query, bool ignore_errors)
 
static bool pgfdw_get_cleanup_result (PGconn *conn, TimestampTz endtime, PGresult **result)
 
static bool UserMappingPasswordRequired (UserMapping *user)
 
PGconnGetConnection (UserMapping *user, bool will_prep_stmt)
 
void ReleaseConnection (PGconn *conn)
 
unsigned int GetCursorNumber (PGconn *conn)
 
unsigned int GetPrepStmtNumber (PGconn *conn)
 
PGresultpgfdw_exec_query (PGconn *conn, const char *query)
 
PGresultpgfdw_get_result (PGconn *conn, const char *query)
 
void pgfdw_report_error (int elevel, PGresult *res, PGconn *conn, bool clear, const char *sql)
 

Variables

static HTABConnectionHash = NULL
 
static unsigned int cursor_number = 0
 
static unsigned int prep_stmt_number = 0
 
static bool xact_got_connection = false
 

Typedef Documentation

◆ ConnCacheEntry

◆ ConnCacheKey

typedef Oid ConnCacheKey

Definition at line 47 of file connection.c.

Function Documentation

◆ begin_remote_xact()

static void begin_remote_xact ( ConnCacheEntry entry)
static

Definition at line 551 of file connection.c.

References ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, DEBUG3, do_sql_command(), elog, GetCurrentTransactionNestLevel(), IsolationIsSerializable, snprintf, and ConnCacheEntry::xact_depth.

Referenced by GetConnection().

552 {
553  int curlevel = GetCurrentTransactionNestLevel();
554 
555  /* Start main transaction if we haven't yet */
556  if (entry->xact_depth <= 0)
557  {
558  const char *sql;
559 
560  elog(DEBUG3, "starting remote transaction on connection %p",
561  entry->conn);
562 
564  sql = "START TRANSACTION ISOLATION LEVEL SERIALIZABLE";
565  else
566  sql = "START TRANSACTION ISOLATION LEVEL REPEATABLE READ";
567  entry->changing_xact_state = true;
568  do_sql_command(entry->conn, sql);
569  entry->xact_depth = 1;
570  entry->changing_xact_state = false;
571  }
572 
573  /*
574  * If we're in a subtransaction, stack up savepoints to match our level.
575  * This ensures we can rollback just the desired effects when a
576  * subtransaction aborts.
577  */
578  while (entry->xact_depth < curlevel)
579  {
580  char sql[64];
581 
582  snprintf(sql, sizeof(sql), "SAVEPOINT s%d", entry->xact_depth + 1);
583  entry->changing_xact_state = true;
584  do_sql_command(entry->conn, sql);
585  entry->xact_depth++;
586  entry->changing_xact_state = false;
587  }
588 }
#define DEBUG3
Definition: elog.h:23
bool changing_xact_state
Definition: connection.c:58
static void do_sql_command(PGconn *conn, const char *sql)
Definition: connection.c:528
int GetCurrentTransactionNestLevel(void)
Definition: xact.c:857
PGconn * conn
Definition: connection.c:52
#define IsolationIsSerializable()
Definition: xact.h:52
#define elog(elevel,...)
Definition: elog.h:214
#define snprintf
Definition: port.h:215

◆ check_conn_params()

static void check_conn_params ( const char **  keywords,
const char **  values,
UserMapping user 
)
static

Definition at line 455 of file connection.c.

References ereport, errcode(), errdetail(), errmsg(), ERROR, i, superuser_arg(), UserMapping::userid, and UserMappingPasswordRequired().

Referenced by connect_pg_server().

456 {
457  int i;
458 
459  /* no check required if superuser */
460  if (superuser_arg(user->userid))
461  return;
462 
463  /* ok if params contain a non-empty password */
464  for (i = 0; keywords[i] != NULL; i++)
465  {
466  if (strcmp(keywords[i], "password") == 0 && values[i][0] != '\0')
467  return;
468  }
469 
470  /* ok if the superuser explicitly said so at user mapping creation time */
471  if (!UserMappingPasswordRequired(user))
472  return;
473 
474  ereport(ERROR,
475  (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
476  errmsg("password is required"),
477  errdetail("Non-superusers must provide a password in the user mapping.")));
478 }
int errcode(int sqlerrcode)
Definition: elog.c:610
Oid userid
Definition: foreign.h:48
#define ERROR
Definition: elog.h:43
int errdetail(const char *fmt,...)
Definition: elog.c:954
bool superuser_arg(Oid roleid)
Definition: superuser.c:56
static bool UserMappingPasswordRequired(UserMapping *user)
Definition: connection.c:432
#define ereport(elevel,...)
Definition: elog.h:144
static Datum values[MAXATTR]
Definition: bootstrap.c:165
int errmsg(const char *fmt,...)
Definition: elog.c:821
int i

◆ configure_remote_session()

static void configure_remote_session ( PGconn conn)
static

Definition at line 492 of file connection.c.

References do_sql_command(), and PQserverVersion().

Referenced by connect_pg_server().

493 {
494  int remoteversion = PQserverVersion(conn);
495 
496  /* Force the search path to contain only pg_catalog (see deparse.c) */
497  do_sql_command(conn, "SET search_path = pg_catalog");
498 
499  /*
500  * Set remote timezone; this is basically just cosmetic, since all
501  * transmitted and returned timestamptzs should specify a zone explicitly
502  * anyway. However it makes the regression test outputs more predictable.
503  *
504  * We don't risk setting remote zone equal to ours, since the remote
505  * server might use a different timezone database. Instead, use UTC
506  * (quoted, because very old servers are picky about case).
507  */
508  do_sql_command(conn, "SET timezone = 'UTC'");
509 
510  /*
511  * Set values needed to ensure unambiguous data output from remote. (This
512  * logic should match what pg_dump does. See also set_transmission_modes
513  * in postgres_fdw.c.)
514  */
515  do_sql_command(conn, "SET datestyle = ISO");
516  if (remoteversion >= 80400)
517  do_sql_command(conn, "SET intervalstyle = postgres");
518  if (remoteversion >= 90000)
519  do_sql_command(conn, "SET extra_float_digits = 3");
520  else
521  do_sql_command(conn, "SET extra_float_digits = 2");
522 }
int PQserverVersion(const PGconn *conn)
Definition: fe-connect.c:6659
static void do_sql_command(PGconn *conn, const char *sql)
Definition: connection.c:528

◆ connect_pg_server()

static PGconn * connect_pg_server ( ForeignServer server,
UserMapping user 
)
static

Definition at line 297 of file connection.c.

References AcquireExternalFD(), check_conn_params(), configure_remote_session(), ConnCacheEntry::conn, CONNECTION_OK, ereport, errcode(), errdetail(), errdetail_internal(), errhint(), errmsg(), ERROR, ExtractConnectionOptions(), GetDatabaseEncodingName(), list_length(), ForeignServer::options, UserMapping::options, palloc(), pchomp(), pfree(), PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, PQconnectdbParams(), PQconnectionUsedPassword(), PQerrorMessage(), PQfinish(), PQstatus(), ReleaseExternalFD(), ForeignServer::servername, superuser_arg(), UserMapping::userid, UserMappingPasswordRequired(), and values.

Referenced by make_new_connection().

298 {
299  PGconn *volatile conn = NULL;
300 
301  /*
302  * Use PG_TRY block to ensure closing connection on error.
303  */
304  PG_TRY();
305  {
306  const char **keywords;
307  const char **values;
308  int n;
309 
310  /*
311  * Construct connection params from generic options of ForeignServer
312  * and UserMapping. (Some of them might not be libpq options, in
313  * which case we'll just waste a few array slots.) Add 3 extra slots
314  * for fallback_application_name, client_encoding, end marker.
315  */
316  n = list_length(server->options) + list_length(user->options) + 3;
317  keywords = (const char **) palloc(n * sizeof(char *));
318  values = (const char **) palloc(n * sizeof(char *));
319 
320  n = 0;
321  n += ExtractConnectionOptions(server->options,
322  keywords + n, values + n);
324  keywords + n, values + n);
325 
326  /* Use "postgres_fdw" as fallback_application_name. */
327  keywords[n] = "fallback_application_name";
328  values[n] = "postgres_fdw";
329  n++;
330 
331  /* Set client_encoding so that libpq can convert encoding properly. */
332  keywords[n] = "client_encoding";
333  values[n] = GetDatabaseEncodingName();
334  n++;
335 
336  keywords[n] = values[n] = NULL;
337 
338  /* verify the set of connection parameters */
339  check_conn_params(keywords, values, user);
340 
341  /*
342  * We must obey fd.c's limit on non-virtual file descriptors. Assume
343  * that a PGconn represents one long-lived FD. (Doing this here also
344  * ensures that VFDs are closed if needed to make room.)
345  */
346  if (!AcquireExternalFD())
347  {
348 #ifndef WIN32 /* can't write #if within ereport() macro */
349  ereport(ERROR,
350  (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
351  errmsg("could not connect to server \"%s\"",
352  server->servername),
353  errdetail("There are too many open files on the local server."),
354  errhint("Raise the server's max_files_per_process and/or \"ulimit -n\" limits.")));
355 #else
356  ereport(ERROR,
357  (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
358  errmsg("could not connect to server \"%s\"",
359  server->servername),
360  errdetail("There are too many open files on the local server."),
361  errhint("Raise the server's max_files_per_process setting.")));
362 #endif
363  }
364 
365  /* OK to make connection */
366  conn = PQconnectdbParams(keywords, values, false);
367 
368  if (!conn)
369  ReleaseExternalFD(); /* because the PG_CATCH block won't */
370 
371  if (!conn || PQstatus(conn) != CONNECTION_OK)
372  ereport(ERROR,
373  (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
374  errmsg("could not connect to server \"%s\"",
375  server->servername),
376  errdetail_internal("%s", pchomp(PQerrorMessage(conn)))));
377 
378  /*
379  * Check that non-superuser has used password to establish connection;
380  * otherwise, he's piggybacking on the postgres server's user
381  * identity. See also dblink_security_check() in contrib/dblink and
382  * check_conn_params.
383  */
384  if (!superuser_arg(user->userid) && UserMappingPasswordRequired(user) &&
386  ereport(ERROR,
387  (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
388  errmsg("password is required"),
389  errdetail("Non-superuser cannot connect if the server does not request a password."),
390  errhint("Target server's authentication method must be changed or password_required=false set in the user mapping attributes.")));
391 
392  /* Prepare new session for use */
394 
395  pfree(keywords);
396  pfree(values);
397  }
398  PG_CATCH();
399  {
400  /* Release PGconn data structure if we managed to create one */
401  if (conn)
402  {
403  PQfinish(conn);
405  }
406  PG_RE_THROW();
407  }
408  PG_END_TRY();
409 
410  return conn;
411 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6669
static void configure_remote_session(PGconn *conn)
Definition: connection.c:492
int errhint(const char *fmt,...)
Definition: elog.c:1068
static void check_conn_params(const char **keywords, const char **values, UserMapping *user)
Definition: connection.c:455
int errcode(int sqlerrcode)
Definition: elog.c:610
void PQfinish(PGconn *conn)
Definition: fe-connect.c:4174
Oid userid
Definition: foreign.h:48
PGconn * PQconnectdbParams(const char *const *keywords, const char *const *values, int expand_dbname)
Definition: fe-connect.c:647
int errdetail_internal(const char *fmt,...)
Definition: elog.c:981
char * pchomp(const char *in)
Definition: mcxt.c:1215
void pfree(void *pointer)
Definition: mcxt.c:1057
int ExtractConnectionOptions(List *defelems, const char **keywords, const char **values)
Definition: option.c:333
#define ERROR
Definition: elog.h:43
PGconn * conn
Definition: streamutil.c:54
int errdetail(const char *fmt,...)
Definition: elog.c:954
List * options
Definition: foreign.h:50
bool superuser_arg(Oid roleid)
Definition: superuser.c:56
static bool UserMappingPasswordRequired(UserMapping *user)
Definition: connection.c:432
bool AcquireExternalFD(void)
Definition: fd.c:1048
#define ereport(elevel,...)
Definition: elog.h:144
#define PG_CATCH()
Definition: elog.h:305
const char * GetDatabaseEncodingName(void)
Definition: mbutils.c:1157
static int list_length(const List *l)
Definition: pg_list.h:149
#define PG_RE_THROW()
Definition: elog.h:336
static Datum values[MAXATTR]
Definition: bootstrap.c:165
int PQconnectionUsedPassword(const PGconn *conn)
Definition: fe-connect.c:6718
void ReleaseExternalFD(void)
Definition: fd.c:1101
void * palloc(Size size)
Definition: mcxt.c:950
int errmsg(const char *fmt,...)
Definition: elog.c:821
char * servername
Definition: foreign.h:39
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6616
#define PG_TRY()
Definition: elog.h:295
List * options
Definition: foreign.h:42
#define PG_END_TRY()
Definition: elog.h:320

◆ disconnect_pg_server()

static void disconnect_pg_server ( ConnCacheEntry entry)
static

Definition at line 417 of file connection.c.

References ConnCacheEntry::conn, PQfinish(), and ReleaseExternalFD().

Referenced by GetConnection(), pgfdw_reject_incomplete_xact_state_change(), and pgfdw_xact_callback().

418 {
419  if (entry->conn != NULL)
420  {
421  PQfinish(entry->conn);
422  entry->conn = NULL;
424  }
425 }
void PQfinish(PGconn *conn)
Definition: fe-connect.c:4174
PGconn * conn
Definition: connection.c:52
void ReleaseExternalFD(void)
Definition: fd.c:1101

◆ do_sql_command()

static void do_sql_command ( PGconn conn,
const char *  sql 
)
static

Definition at line 528 of file connection.c.

References ERROR, pgfdw_get_result(), pgfdw_report_error(), PGRES_COMMAND_OK, PQclear(), PQresultStatus(), and PQsendQuery().

Referenced by begin_remote_xact(), configure_remote_session(), pgfdw_subxact_callback(), and pgfdw_xact_callback().

529 {
530  PGresult *res;
531 
532  if (!PQsendQuery(conn, sql))
533  pgfdw_report_error(ERROR, NULL, conn, false, sql);
534  res = pgfdw_get_result(conn, sql);
535  if (PQresultStatus(res) != PGRES_COMMAND_OK)
536  pgfdw_report_error(ERROR, res, conn, true, sql);
537  PQclear(res);
538 }
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2692
int PQsendQuery(PGconn *conn, const char *query)
Definition: fe-exec.c:1234
#define ERROR
Definition: elog.h:43
void pgfdw_report_error(int elevel, PGresult *res, PGconn *conn, bool clear, const char *sql)
Definition: connection.c:731
void PQclear(PGresult *res)
Definition: fe-exec.c:694
PGresult * pgfdw_get_result(PGconn *conn, const char *query)
Definition: connection.c:666

◆ GetConnection()

PGconn* GetConnection ( UserMapping user,
bool  will_prep_stmt 
)

Definition at line 109 of file connection.c.

References Assert, begin_remote_xact(), CacheMemoryContext, CacheRegisterSyscacheCallback(), ConnCacheEntry::conn, CONNECTION_BAD, CopyErrorData(), CurrentMemoryContext, DEBUG3, disconnect_pg_server(), elog, HASHCTL::entrysize, ereport, errdetail_internal(), errmsg_internal(), FlushErrorState(), FOREIGNSERVEROID, FreeErrorData(), HASH_BLOBS, HASH_CONTEXT, hash_create(), HASH_ELEM, HASH_ENTER, hash_search(), ConnCacheEntry::have_prep_stmt, HASHCTL::hcxt, ConnCacheEntry::invalidated, ConnCacheEntry::key, HASHCTL::keysize, make_new_connection(), MemoryContextSwitchTo(), MemSet, pchomp(), PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, pgfdw_inval_callback(), pgfdw_reject_incomplete_xact_state_change(), pgfdw_subxact_callback(), pgfdw_xact_callback(), PQerrorMessage(), PQstatus(), RegisterSubXactCallback(), RegisterXactCallback(), ErrorData::sqlerrcode, UserMapping::umid, USERMAPPINGOID, ConnCacheEntry::xact_depth, and xact_got_connection.

Referenced by create_foreign_modify(), dumpBlobs(), dumpDatabase(), dumpDatabaseConfig(), dumpTableData_copy(), estimate_path_cost_size(), expand_foreign_server_name_patterns(), expand_schema_name_patterns(), expand_table_name_patterns(), getTables(), main(), postgresAcquireSampleRowsFunc(), postgresAnalyzeForeignTable(), postgresBeginDirectModify(), postgresBeginForeignScan(), postgresImportForeignSchema(), setup_connection(), StartLogStreamer(), StreamLog(), and StreamLogicalLog().

110 {
111  bool found;
112  bool retry = false;
113  ConnCacheEntry *entry;
116 
117  /* First time through, initialize connection cache hashtable */
118  if (ConnectionHash == NULL)
119  {
120  HASHCTL ctl;
121 
122  MemSet(&ctl, 0, sizeof(ctl));
123  ctl.keysize = sizeof(ConnCacheKey);
124  ctl.entrysize = sizeof(ConnCacheEntry);
125  /* allocate ConnectionHash in the cache context */
126  ctl.hcxt = CacheMemoryContext;
127  ConnectionHash = hash_create("postgres_fdw connections", 8,
128  &ctl,
130 
131  /*
132  * Register some callback functions that manage connection cleanup.
133  * This should be done just once in each backend.
134  */
141  }
142 
143  /* Set flag that we did GetConnection during the current transaction */
144  xact_got_connection = true;
145 
146  /* Create hash key for the entry. Assume no pad bytes in key struct */
147  key = user->umid;
148 
149  /*
150  * Find or create cached entry for requested connection.
151  */
152  entry = hash_search(ConnectionHash, &key, HASH_ENTER, &found);
153  if (!found)
154  {
155  /*
156  * We need only clear "conn" here; remaining fields will be filled
157  * later when "conn" is set.
158  */
159  entry->conn = NULL;
160  }
161 
162  /* Reject further use of connections which failed abort cleanup. */
164 
165  /*
166  * If the connection needs to be remade due to invalidation, disconnect as
167  * soon as we're out of all transactions.
168  */
169  if (entry->conn != NULL && entry->invalidated && entry->xact_depth == 0)
170  {
171  elog(DEBUG3, "closing connection %p for option changes to take effect",
172  entry->conn);
173  disconnect_pg_server(entry);
174  }
175 
176  /*
177  * If cache entry doesn't have a connection, we have to establish a new
178  * connection. (If connect_pg_server throws an error, the cache entry
179  * will remain in a valid empty state, ie conn == NULL.)
180  */
181  if (entry->conn == NULL)
182  make_new_connection(entry, user);
183 
184  /*
185  * We check the health of the cached connection here when starting a new
186  * remote transaction. If a broken connection is detected, we try to
187  * reestablish a new connection later.
188  */
189  PG_TRY();
190  {
191  /* Start a new transaction or subtransaction if needed. */
192  begin_remote_xact(entry);
193  }
194  PG_CATCH();
195  {
197  ErrorData *errdata = CopyErrorData();
198 
199  /*
200  * If connection failure is reported when starting a new remote
201  * transaction (not subtransaction), new connection will be
202  * reestablished later.
203  *
204  * After a broken connection is detected in libpq, any error other
205  * than connection failure (e.g., out-of-memory) can be thrown
206  * somewhere between return from libpq and the expected ereport() call
207  * in pgfdw_report_error(). In this case, since PQstatus() indicates
208  * CONNECTION_BAD, checking only PQstatus() causes the false detection
209  * of connection failure. To avoid this, we also verify that the
210  * error's sqlstate is ERRCODE_CONNECTION_FAILURE. Note that also
211  * checking only the sqlstate can cause another false detection
212  * because pgfdw_report_error() may report ERRCODE_CONNECTION_FAILURE
213  * for any libpq-originated error condition.
214  */
215  if (errdata->sqlerrcode != ERRCODE_CONNECTION_FAILURE ||
216  PQstatus(entry->conn) != CONNECTION_BAD ||
217  entry->xact_depth > 0)
218  {
219  MemoryContextSwitchTo(ecxt);
220  PG_RE_THROW();
221  }
222 
223  /* Clean up the error state */
224  FlushErrorState();
225  FreeErrorData(errdata);
226  errdata = NULL;
227 
228  retry = true;
229  }
230  PG_END_TRY();
231 
232  /*
233  * If a broken connection is detected, disconnect it, reestablish a new
234  * connection and retry a new remote transaction. If connection failure is
235  * reported again, we give up getting a connection.
236  */
237  if (retry)
238  {
239  Assert(entry->xact_depth == 0);
240 
241  ereport(DEBUG3,
242  (errmsg_internal("could not start remote transaction on connection %p",
243  entry->conn)),
244  errdetail_internal("%s", pchomp(PQerrorMessage(entry->conn))));
245 
246  elog(DEBUG3, "closing connection %p to reestablish a new one",
247  entry->conn);
248  disconnect_pg_server(entry);
249 
250  if (entry->conn == NULL)
251  make_new_connection(entry, user);
252 
253  begin_remote_xact(entry);
254  }
255 
256  /* Remember if caller will prepare statements */
257  entry->have_prep_stmt |= will_prep_stmt;
258 
259  return entry->conn;
260 }
Oid umid
Definition: foreign.h:47
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6669
#define HASH_CONTEXT
Definition: hsearch.h:91
#define HASH_ELEM
Definition: hsearch.h:85
MemoryContext hcxt
Definition: hsearch.h:77
int sqlerrcode
Definition: elog.h:364
#define DEBUG3
Definition: elog.h:23
struct ConnCacheEntry ConnCacheEntry
ErrorData * CopyErrorData(void)
Definition: elog.c:1471
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
Size entrysize
Definition: hsearch.h:72
#define MemSet(start, val, len)
Definition: c.h:950
bool have_prep_stmt
Definition: connection.c:56
void * hash_search(HTAB *hashp, const void *keyPtr, HASHACTION action, bool *foundPtr)
Definition: dynahash.c:919
void FlushErrorState(void)
Definition: elog.c:1565
int errdetail_internal(const char *fmt,...)
Definition: elog.c:981
char * pchomp(const char *in)
Definition: mcxt.c:1215
void FreeErrorData(ErrorData *edata)
Definition: elog.c:1527
MemoryContext CurrentMemoryContext
Definition: mcxt.c:38
bool invalidated
Definition: connection.c:59
static void pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue)
Definition: connection.c:1083
#define HASH_BLOBS
Definition: hsearch.h:86
void CacheRegisterSyscacheCallback(int cacheid, SyscacheCallbackFunction func, Datum arg)
Definition: inval.c:1434
static void pgfdw_xact_callback(XactEvent event, void *arg)
Definition: connection.c:782
static HTAB * ConnectionHash
Definition: connection.c:67
uintptr_t Datum
Definition: postgres.h:367
HTAB * hash_create(const char *tabname, long nelem, HASHCTL *info, int flags)
Definition: dynahash.c:326
void RegisterSubXactCallback(SubXactCallback callback, void *arg)
Definition: xact.c:3583
Size keysize
Definition: hsearch.h:71
#define ereport(elevel,...)
Definition: elog.h:144
int errmsg_internal(const char *fmt,...)
Definition: elog.c:908
#define PG_CATCH()
Definition: elog.h:305
PGconn * conn
Definition: connection.c:52
static void disconnect_pg_server(ConnCacheEntry *entry)
Definition: connection.c:417
#define Assert(condition)
Definition: c.h:746
void RegisterXactCallback(XactCallback callback, void *arg)
Definition: xact.c:3528
static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry)
Definition: connection.c:1119
static bool xact_got_connection
Definition: connection.c:74
#define PG_RE_THROW()
Definition: elog.h:336
static void make_new_connection(ConnCacheEntry *entry, UserMapping *user)
Definition: connection.c:267
#define elog(elevel,...)
Definition: elog.h:214
static void pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, SubTransactionId parentSubid, void *arg)
Definition: connection.c:969
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6616
#define PG_TRY()
Definition: elog.h:295
static void begin_remote_xact(ConnCacheEntry *entry)
Definition: connection.c:551
Oid ConnCacheKey
Definition: connection.c:47
#define PG_END_TRY()
Definition: elog.h:320
MemoryContext CacheMemoryContext
Definition: mcxt.c:47

◆ GetCursorNumber()

unsigned int GetCursorNumber ( PGconn conn)

Definition at line 615 of file connection.c.

References cursor_number.

Referenced by postgresAcquireSampleRowsFunc(), and postgresBeginForeignScan().

616 {
617  return ++cursor_number;
618 }
static unsigned int cursor_number
Definition: connection.c:70

◆ GetPrepStmtNumber()

unsigned int GetPrepStmtNumber ( PGconn conn)

Definition at line 629 of file connection.c.

References prep_stmt_number.

Referenced by prepare_foreign_modify().

630 {
631  return ++prep_stmt_number;
632 }
static unsigned int prep_stmt_number
Definition: connection.c:71

◆ make_new_connection()

static void make_new_connection ( ConnCacheEntry entry,
UserMapping user 
)
static

Definition at line 267 of file connection.c.

References Assert, ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, connect_pg_server(), DEBUG3, elog, FOREIGNSERVEROID, GetForeignServer(), GetSysCacheHashValue1, ConnCacheEntry::have_error, ConnCacheEntry::have_prep_stmt, ConnCacheEntry::invalidated, ConnCacheEntry::mapping_hashvalue, ObjectIdGetDatum, ConnCacheEntry::server_hashvalue, ForeignServer::serverid, UserMapping::serverid, ForeignServer::servername, UserMapping::umid, UserMapping::userid, USERMAPPINGOID, and ConnCacheEntry::xact_depth.

Referenced by GetConnection().

268 {
269  ForeignServer *server = GetForeignServer(user->serverid);
270 
271  Assert(entry->conn == NULL);
272 
273  /* Reset all transient state fields, to be sure all are clean */
274  entry->xact_depth = 0;
275  entry->have_prep_stmt = false;
276  entry->have_error = false;
277  entry->changing_xact_state = false;
278  entry->invalidated = false;
279  entry->server_hashvalue =
281  ObjectIdGetDatum(server->serverid));
282  entry->mapping_hashvalue =
284  ObjectIdGetDatum(user->umid));
285 
286  /* Now try to make the connection */
287  entry->conn = connect_pg_server(server, user);
288 
289  elog(DEBUG3, "new postgres_fdw connection %p for server \"%s\" (user mapping oid %u, userid %u)",
290  entry->conn, server->servername, user->umid, user->userid);
291 }
Oid umid
Definition: foreign.h:47
#define DEBUG3
Definition: elog.h:23
bool have_prep_stmt
Definition: connection.c:56
#define GetSysCacheHashValue1(cacheId, key1)
Definition: syscache.h:201
uint32 server_hashvalue
Definition: connection.c:60
Oid userid
Definition: foreign.h:48
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
bool changing_xact_state
Definition: connection.c:58
bool invalidated
Definition: connection.c:59
ForeignServer * GetForeignServer(Oid serverid)
Definition: foreign.c:109
uint32 mapping_hashvalue
Definition: connection.c:61
PGconn * conn
Definition: connection.c:52
#define Assert(condition)
Definition: c.h:746
Oid serverid
Definition: foreign.h:49
char * servername
Definition: foreign.h:39
#define elog(elevel,...)
Definition: elog.h:214
Oid serverid
Definition: foreign.h:36
static PGconn * connect_pg_server(ForeignServer *server, UserMapping *user)
Definition: connection.c:297

◆ pgfdw_cancel_query()

static bool pgfdw_cancel_query ( PGconn conn)
static

Definition at line 1153 of file connection.c.

References ereport, errcode(), errmsg(), GetCurrentTimestamp(), pgfdw_get_cleanup_result(), PQcancel(), PQclear(), PQfreeCancel(), PQgetCancel(), TimestampTzPlusMilliseconds, and WARNING.

Referenced by pgfdw_subxact_callback(), and pgfdw_xact_callback().

1154 {
1155  PGcancel *cancel;
1156  char errbuf[256];
1157  PGresult *result = NULL;
1158  TimestampTz endtime;
1159 
1160  /*
1161  * If it takes too long to cancel the query and discard the result, assume
1162  * the connection is dead.
1163  */
1165 
1166  /*
1167  * Issue cancel request. Unfortunately, there's no good way to limit the
1168  * amount of time that we might block inside PQgetCancel().
1169  */
1170  if ((cancel = PQgetCancel(conn)))
1171  {
1172  if (!PQcancel(cancel, errbuf, sizeof(errbuf)))
1173  {
1174  ereport(WARNING,
1175  (errcode(ERRCODE_CONNECTION_FAILURE),
1176  errmsg("could not send cancel request: %s",
1177  errbuf)));
1178  PQfreeCancel(cancel);
1179  return false;
1180  }
1181  PQfreeCancel(cancel);
1182  }
1183 
1184  /* Get and discard the result of the query. */
1185  if (pgfdw_get_cleanup_result(conn, endtime, &result))
1186  return false;
1187  PQclear(result);
1188 
1189  return true;
1190 }
static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result)
Definition: connection.c:1254
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1578
int64 TimestampTz
Definition: timestamp.h:39
void PQfreeCancel(PGcancel *cancel)
Definition: fe-connect.c:4318
int errcode(int sqlerrcode)
Definition: elog.c:610
PGcancel * PQgetCancel(PGconn *conn)
Definition: fe-connect.c:4295
#define WARNING
Definition: elog.h:40
#define TimestampTzPlusMilliseconds(tz, ms)
Definition: timestamp.h:56
#define ereport(elevel,...)
Definition: elog.h:144
void PQclear(PGresult *res)
Definition: fe-exec.c:694
int PQcancel(PGcancel *cancel, char *errbuf, int errbufsize)
Definition: fe-connect.c:4450
int errmsg(const char *fmt,...)
Definition: elog.c:821

◆ pgfdw_exec_cleanup_query()

static bool pgfdw_exec_cleanup_query ( PGconn conn,
const char *  query,
bool  ignore_errors 
)
static

Definition at line 1200 of file connection.c.

References GetCurrentTimestamp(), pgfdw_get_cleanup_result(), pgfdw_report_error(), PGRES_COMMAND_OK, PQclear(), PQresultStatus(), PQsendQuery(), TimestampTzPlusMilliseconds, and WARNING.

Referenced by pgfdw_subxact_callback(), and pgfdw_xact_callback().

1201 {
1202  PGresult *result = NULL;
1203  TimestampTz endtime;
1204 
1205  /*
1206  * If it takes too long to execute a cleanup query, assume the connection
1207  * is dead. It's fairly likely that this is why we aborted in the first
1208  * place (e.g. statement timeout, user cancel), so the timeout shouldn't
1209  * be too long.
1210  */
1212 
1213  /*
1214  * Submit a query. Since we don't use non-blocking mode, this also can
1215  * block. But its risk is relatively small, so we ignore that for now.
1216  */
1217  if (!PQsendQuery(conn, query))
1218  {
1219  pgfdw_report_error(WARNING, NULL, conn, false, query);
1220  return false;
1221  }
1222 
1223  /* Get the result of the query. */
1224  if (pgfdw_get_cleanup_result(conn, endtime, &result))
1225  return false;
1226 
1227  /* Issue a warning if not successful. */
1228  if (PQresultStatus(result) != PGRES_COMMAND_OK)
1229  {
1230  pgfdw_report_error(WARNING, result, conn, true, query);
1231  return ignore_errors;
1232  }
1233  PQclear(result);
1234 
1235  return true;
1236 }
static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result)
Definition: connection.c:1254
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1578
int64 TimestampTz
Definition: timestamp.h:39
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2692
int PQsendQuery(PGconn *conn, const char *query)
Definition: fe-exec.c:1234
void pgfdw_report_error(int elevel, PGresult *res, PGconn *conn, bool clear, const char *sql)
Definition: connection.c:731
#define WARNING
Definition: elog.h:40
#define TimestampTzPlusMilliseconds(tz, ms)
Definition: timestamp.h:56
void PQclear(PGresult *res)
Definition: fe-exec.c:694

◆ pgfdw_exec_query()

PGresult* pgfdw_exec_query ( PGconn conn,
const char *  query 
)

Definition at line 642 of file connection.c.

References ERROR, pgfdw_get_result(), pgfdw_report_error(), and PQsendQuery().

Referenced by close_cursor(), fetch_more_data(), finish_foreign_modify(), get_remote_estimate(), postgresAcquireSampleRowsFunc(), postgresAnalyzeForeignTable(), postgresImportForeignSchema(), and postgresReScanForeignScan().

643 {
644  /*
645  * Submit a query. Since we don't use non-blocking mode, this also can
646  * block. But its risk is relatively small, so we ignore that for now.
647  */
648  if (!PQsendQuery(conn, query))
649  pgfdw_report_error(ERROR, NULL, conn, false, query);
650 
651  /* Wait for the result. */
652  return pgfdw_get_result(conn, query);
653 }
int PQsendQuery(PGconn *conn, const char *query)
Definition: fe-exec.c:1234
#define ERROR
Definition: elog.h:43
void pgfdw_report_error(int elevel, PGresult *res, PGconn *conn, bool clear, const char *sql)
Definition: connection.c:731
PGresult * pgfdw_get_result(PGconn *conn, const char *query)
Definition: connection.c:666

◆ pgfdw_get_cleanup_result()

static bool pgfdw_get_cleanup_result ( PGconn conn,
TimestampTz  endtime,
PGresult **  result 
)
static

Definition at line 1254 of file connection.c.

References CHECK_FOR_INTERRUPTS, GetCurrentTimestamp(), Min, MyLatch, now(), PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, PG_WAIT_EXTENSION, PQclear(), PQconsumeInput(), PQgetResult(), PQisBusy(), PQsocket(), ResetLatch(), TimestampDifference(), USECS_PER_SEC, WaitLatchOrSocket(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, WL_SOCKET_READABLE, and WL_TIMEOUT.

Referenced by pgfdw_cancel_query(), and pgfdw_exec_cleanup_query().

1255 {
1256  volatile bool timed_out = false;
1257  PGresult *volatile last_res = NULL;
1258 
1259  /* In what follows, do not leak any PGresults on an error. */
1260  PG_TRY();
1261  {
1262  for (;;)
1263  {
1264  PGresult *res;
1265 
1266  while (PQisBusy(conn))
1267  {
1268  int wc;
1270  long secs;
1271  int microsecs;
1272  long cur_timeout;
1273 
1274  /* If timeout has expired, give up, else get sleep time. */
1275  if (now >= endtime)
1276  {
1277  timed_out = true;
1278  goto exit;
1279  }
1280  TimestampDifference(now, endtime, &secs, &microsecs);
1281 
1282  /* To protect against clock skew, limit sleep to one minute. */
1283  cur_timeout = Min(60000, secs * USECS_PER_SEC + microsecs);
1284 
1285  /* Sleep until there's something to do */
1289  PQsocket(conn),
1290  cur_timeout, PG_WAIT_EXTENSION);
1292 
1294 
1295  /* Data available in socket? */
1296  if (wc & WL_SOCKET_READABLE)
1297  {
1298  if (!PQconsumeInput(conn))
1299  {
1300  /* connection trouble; treat the same as a timeout */
1301  timed_out = true;
1302  goto exit;
1303  }
1304  }
1305  }
1306 
1307  res = PQgetResult(conn);
1308  if (res == NULL)
1309  break; /* query is complete */
1310 
1311  PQclear(last_res);
1312  last_res = res;
1313  }
1314 exit: ;
1315  }
1316  PG_CATCH();
1317  {
1318  PQclear(last_res);
1319  PG_RE_THROW();
1320  }
1321  PG_END_TRY();
1322 
1323  if (timed_out)
1324  PQclear(last_res);
1325  else
1326  *result = last_res;
1327  return timed_out;
1328 }
#define WL_TIMEOUT
Definition: latch.h:127
#define USECS_PER_SEC
Definition: timestamp.h:94
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1578
int64 TimestampTz
Definition: timestamp.h:39
#define Min(x, y)
Definition: c.h:928
#define WL_SOCKET_READABLE
Definition: latch.h:125
void ResetLatch(Latch *latch)
Definition: latch.c:588
int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:438
#define PG_WAIT_EXTENSION
Definition: pgstat.h:854
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1704
void PQclear(PGresult *res)
Definition: fe-exec.c:694
#define PG_CATCH()
Definition: elog.h:305
int PQisBusy(PGconn *conn)
Definition: fe-exec.c:1754
#define PG_RE_THROW()
Definition: elog.h:336
struct Latch * MyLatch
Definition: globals.c:54
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:99
void TimestampDifference(TimestampTz start_time, TimestampTz stop_time, long *secs, int *microsecs)
Definition: timestamp.c:1652
#define PG_TRY()
Definition: elog.h:295
#define WL_LATCH_SET
Definition: latch.h:124
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6687
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1542
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1778
#define PG_END_TRY()
Definition: elog.h:320
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:129

◆ pgfdw_get_result()

PGresult* pgfdw_get_result ( PGconn conn,
const char *  query 
)

Definition at line 666 of file connection.c.

References CHECK_FOR_INTERRUPTS, ERROR, MyLatch, PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, PG_WAIT_EXTENSION, pgfdw_report_error(), PQclear(), PQconsumeInput(), PQgetResult(), PQisBusy(), PQsocket(), ResetLatch(), WaitLatchOrSocket(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_SOCKET_READABLE.

Referenced by create_cursor(), do_sql_command(), execute_dml_stmt(), execute_foreign_modify(), pgfdw_exec_query(), and prepare_foreign_modify().

667 {
668  PGresult *volatile last_res = NULL;
669 
670  /* In what follows, do not leak any PGresults on an error. */
671  PG_TRY();
672  {
673  for (;;)
674  {
675  PGresult *res;
676 
677  while (PQisBusy(conn))
678  {
679  int wc;
680 
681  /* Sleep until there's something to do */
685  PQsocket(conn),
686  -1L, PG_WAIT_EXTENSION);
688 
690 
691  /* Data available in socket? */
692  if (wc & WL_SOCKET_READABLE)
693  {
694  if (!PQconsumeInput(conn))
695  pgfdw_report_error(ERROR, NULL, conn, false, query);
696  }
697  }
698 
699  res = PQgetResult(conn);
700  if (res == NULL)
701  break; /* query is complete */
702 
703  PQclear(last_res);
704  last_res = res;
705  }
706  }
707  PG_CATCH();
708  {
709  PQclear(last_res);
710  PG_RE_THROW();
711  }
712  PG_END_TRY();
713 
714  return last_res;
715 }
#define WL_SOCKET_READABLE
Definition: latch.h:125
void ResetLatch(Latch *latch)
Definition: latch.c:588
#define ERROR
Definition: elog.h:43
int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:438
void pgfdw_report_error(int elevel, PGresult *res, PGconn *conn, bool clear, const char *sql)
Definition: connection.c:731
#define PG_WAIT_EXTENSION
Definition: pgstat.h:854
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1704
void PQclear(PGresult *res)
Definition: fe-exec.c:694
#define PG_CATCH()
Definition: elog.h:305
int PQisBusy(PGconn *conn)
Definition: fe-exec.c:1754
#define PG_RE_THROW()
Definition: elog.h:336
struct Latch * MyLatch
Definition: globals.c:54
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:99
#define PG_TRY()
Definition: elog.h:295
#define WL_LATCH_SET
Definition: latch.h:124
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6687
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1778
#define PG_END_TRY()
Definition: elog.h:320
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:129

◆ pgfdw_inval_callback()

static void pgfdw_inval_callback ( Datum  arg,
int  cacheid,
uint32  hashvalue 
)
static

Definition at line 1083 of file connection.c.

References Assert, ConnCacheEntry::conn, FOREIGNSERVEROID, hash_seq_init(), hash_seq_search(), ConnCacheEntry::invalidated, ConnCacheEntry::mapping_hashvalue, ConnCacheEntry::server_hashvalue, and USERMAPPINGOID.

Referenced by GetConnection().

1084 {
1085  HASH_SEQ_STATUS scan;
1086  ConnCacheEntry *entry;
1087 
1088  Assert(cacheid == FOREIGNSERVEROID || cacheid == USERMAPPINGOID);
1089 
1090  /* ConnectionHash must exist already, if we're registered */
1091  hash_seq_init(&scan, ConnectionHash);
1092  while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
1093  {
1094  /* Ignore invalid entries */
1095  if (entry->conn == NULL)
1096  continue;
1097 
1098  /* hashvalue == 0 means a cache reset, must clear all state */
1099  if (hashvalue == 0 ||
1100  (cacheid == FOREIGNSERVEROID &&
1101  entry->server_hashvalue == hashvalue) ||
1102  (cacheid == USERMAPPINGOID &&
1103  entry->mapping_hashvalue == hashvalue))
1104  entry->invalidated = true;
1105  }
1106 }
uint32 server_hashvalue
Definition: connection.c:60
bool invalidated
Definition: connection.c:59
static HTAB * ConnectionHash
Definition: connection.c:67
uint32 mapping_hashvalue
Definition: connection.c:61
PGconn * conn
Definition: connection.c:52
#define Assert(condition)
Definition: c.h:746
void * hash_seq_search(HASH_SEQ_STATUS *status)
Definition: dynahash.c:1401
void hash_seq_init(HASH_SEQ_STATUS *status, HTAB *hashp)
Definition: dynahash.c:1391

◆ pgfdw_reject_incomplete_xact_state_change()

static void pgfdw_reject_incomplete_xact_state_change ( ConnCacheEntry entry)
static

Definition at line 1119 of file connection.c.

References ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, disconnect_pg_server(), elog, ereport, errcode(), errmsg(), ERROR, GetForeignServer(), GETSTRUCT, HeapTupleIsValid, ConnCacheEntry::key, ObjectIdGetDatum, ReleaseSysCache(), SearchSysCache1(), ForeignServer::servername, and USERMAPPINGOID.

Referenced by GetConnection(), pgfdw_subxact_callback(), and pgfdw_xact_callback().

1120 {
1121  HeapTuple tup;
1122  Form_pg_user_mapping umform;
1123  ForeignServer *server;
1124 
1125  /* nothing to do for inactive entries and entries of sane state */
1126  if (entry->conn == NULL || !entry->changing_xact_state)
1127  return;
1128 
1129  /* make sure this entry is inactive */
1130  disconnect_pg_server(entry);
1131 
1132  /* find server name to be shown in the message below */
1134  ObjectIdGetDatum(entry->key));
1135  if (!HeapTupleIsValid(tup))
1136  elog(ERROR, "cache lookup failed for user mapping %u", entry->key);
1137  umform = (Form_pg_user_mapping) GETSTRUCT(tup);
1138  server = GetForeignServer(umform->umserver);
1139  ReleaseSysCache(tup);
1140 
1141  ereport(ERROR,
1142  (errcode(ERRCODE_CONNECTION_EXCEPTION),
1143  errmsg("connection to server \"%s\" was lost",
1144  server->servername)));
1145 }
#define GETSTRUCT(TUP)
Definition: htup_details.h:655
ConnCacheKey key
Definition: connection.c:51
int errcode(int sqlerrcode)
Definition: elog.c:610
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
#define ERROR
Definition: elog.h:43
bool changing_xact_state
Definition: connection.c:58
FormData_pg_user_mapping * Form_pg_user_mapping
HeapTuple SearchSysCache1(int cacheId, Datum key1)
Definition: syscache.c:1116
ForeignServer * GetForeignServer(Oid serverid)
Definition: foreign.c:109
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1164
#define ereport(elevel,...)
Definition: elog.h:144
PGconn * conn
Definition: connection.c:52
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
static void disconnect_pg_server(ConnCacheEntry *entry)
Definition: connection.c:417
int errmsg(const char *fmt,...)
Definition: elog.c:821
char * servername
Definition: foreign.h:39
#define elog(elevel,...)
Definition: elog.h:214

◆ pgfdw_report_error()

void pgfdw_report_error ( int  elevel,
PGresult res,
PGconn conn,
bool  clear,
const char *  sql 
)

Definition at line 731 of file connection.c.

References ereport, errcode(), errcontext, errdetail_internal(), errhint(), errmsg(), errmsg_internal(), MAKE_SQLSTATE, pchomp(), PG_DIAG_CONTEXT, PG_DIAG_MESSAGE_DETAIL, PG_DIAG_MESSAGE_HINT, PG_DIAG_MESSAGE_PRIMARY, PG_DIAG_SQLSTATE, PG_END_TRY, PG_FINALLY, PG_TRY, PQclear(), PQerrorMessage(), and PQresultErrorField().

Referenced by close_cursor(), create_cursor(), do_sql_command(), execute_dml_stmt(), execute_foreign_modify(), fetch_more_data(), finish_foreign_modify(), get_remote_estimate(), pgfdw_exec_cleanup_query(), pgfdw_exec_query(), pgfdw_get_result(), postgresAcquireSampleRowsFunc(), postgresAnalyzeForeignTable(), postgresImportForeignSchema(), postgresReScanForeignScan(), and prepare_foreign_modify().

733 {
734  /* If requested, PGresult must be released before leaving this function. */
735  PG_TRY();
736  {
737  char *diag_sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
738  char *message_primary = PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY);
739  char *message_detail = PQresultErrorField(res, PG_DIAG_MESSAGE_DETAIL);
740  char *message_hint = PQresultErrorField(res, PG_DIAG_MESSAGE_HINT);
741  char *message_context = PQresultErrorField(res, PG_DIAG_CONTEXT);
742  int sqlstate;
743 
744  if (diag_sqlstate)
745  sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
746  diag_sqlstate[1],
747  diag_sqlstate[2],
748  diag_sqlstate[3],
749  diag_sqlstate[4]);
750  else
751  sqlstate = ERRCODE_CONNECTION_FAILURE;
752 
753  /*
754  * If we don't get a message from the PGresult, try the PGconn. This
755  * is needed because for connection-level failures, PQexec may just
756  * return NULL, not a PGresult at all.
757  */
758  if (message_primary == NULL)
759  message_primary = pchomp(PQerrorMessage(conn));
760 
761  ereport(elevel,
762  (errcode(sqlstate),
763  message_primary ? errmsg_internal("%s", message_primary) :
764  errmsg("could not obtain message string for remote error"),
765  message_detail ? errdetail_internal("%s", message_detail) : 0,
766  message_hint ? errhint("%s", message_hint) : 0,
767  message_context ? errcontext("%s", message_context) : 0,
768  sql ? errcontext("remote SQL command: %s", sql) : 0));
769  }
770  PG_FINALLY();
771  {
772  if (clear)
773  PQclear(res);
774  }
775  PG_END_TRY();
776 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6669
int errhint(const char *fmt,...)
Definition: elog.c:1068
#define PG_DIAG_MESSAGE_PRIMARY
Definition: postgres_ext.h:58
#define PG_DIAG_MESSAGE_DETAIL
Definition: postgres_ext.h:59
#define MAKE_SQLSTATE(ch1, ch2, ch3, ch4, ch5)
Definition: elog.h:62
int errcode(int sqlerrcode)
Definition: elog.c:610
#define PG_DIAG_SQLSTATE
Definition: postgres_ext.h:57
int errdetail_internal(const char *fmt,...)
Definition: elog.c:981
char * pchomp(const char *in)
Definition: mcxt.c:1215
static int elevel
Definition: vacuumlazy.c:333
#define PG_FINALLY()
Definition: elog.h:312
#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
int errmsg_internal(const char *fmt,...)
Definition: elog.c:908
char * PQresultErrorField(const PGresult *res, int fieldcode)
Definition: fe-exec.c:2754
int errmsg(const char *fmt,...)
Definition: elog.c:821
#define errcontext
Definition: elog.h:185
#define PG_TRY()
Definition: elog.h:295
#define PG_END_TRY()
Definition: elog.h:320
#define PG_DIAG_CONTEXT
Definition: postgres_ext.h:64

◆ pgfdw_subxact_callback()

static void pgfdw_subxact_callback ( SubXactEvent  event,
SubTransactionId  mySubid,
SubTransactionId  parentSubid,
void *  arg 
)
static

Definition at line 969 of file connection.c.

References ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, do_sql_command(), elog, ERROR, GetCurrentTransactionNestLevel(), hash_seq_init(), hash_seq_search(), ConnCacheEntry::have_error, in_error_recursion_trouble(), pgfdw_cancel_query(), pgfdw_exec_cleanup_query(), pgfdw_reject_incomplete_xact_state_change(), PQTRANS_ACTIVE, PQtransactionStatus(), snprintf, SUBXACT_EVENT_ABORT_SUB, SUBXACT_EVENT_PRE_COMMIT_SUB, ConnCacheEntry::xact_depth, and xact_got_connection.

Referenced by GetConnection().

971 {
972  HASH_SEQ_STATUS scan;
973  ConnCacheEntry *entry;
974  int curlevel;
975 
976  /* Nothing to do at subxact start, nor after commit. */
977  if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB ||
978  event == SUBXACT_EVENT_ABORT_SUB))
979  return;
980 
981  /* Quick exit if no connections were touched in this transaction. */
982  if (!xact_got_connection)
983  return;
984 
985  /*
986  * Scan all connection cache entries to find open remote subtransactions
987  * of the current level, and close them.
988  */
989  curlevel = GetCurrentTransactionNestLevel();
991  while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
992  {
993  char sql[100];
994 
995  /*
996  * We only care about connections with open remote subtransactions of
997  * the current level.
998  */
999  if (entry->conn == NULL || entry->xact_depth < curlevel)
1000  continue;
1001 
1002  if (entry->xact_depth > curlevel)
1003  elog(ERROR, "missed cleaning up remote subtransaction at level %d",
1004  entry->xact_depth);
1005 
1006  if (event == SUBXACT_EVENT_PRE_COMMIT_SUB)
1007  {
1008  /*
1009  * If abort cleanup previously failed for this connection, we
1010  * can't issue any more commands against it.
1011  */
1013 
1014  /* Commit all remote subtransactions during pre-commit */
1015  snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel);
1016  entry->changing_xact_state = true;
1017  do_sql_command(entry->conn, sql);
1018  entry->changing_xact_state = false;
1019  }
1020  else if (in_error_recursion_trouble())
1021  {
1022  /*
1023  * Don't try to clean up the connection if we're already in error
1024  * recursion trouble.
1025  */
1026  entry->changing_xact_state = true;
1027  }
1028  else if (!entry->changing_xact_state)
1029  {
1030  bool abort_cleanup_failure = false;
1031 
1032  /* Remember that abort cleanup is in progress. */
1033  entry->changing_xact_state = true;
1034 
1035  /* Assume we might have lost track of prepared statements */
1036  entry->have_error = true;
1037 
1038  /*
1039  * If a command has been submitted to the remote server by using
1040  * an asynchronous execution function, the command might not have
1041  * yet completed. Check to see if a command is still being
1042  * processed by the remote server, and if so, request cancellation
1043  * of the command.
1044  */
1045  if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE &&
1046  !pgfdw_cancel_query(entry->conn))
1047  abort_cleanup_failure = true;
1048  else
1049  {
1050  /* Rollback all remote subtransactions during abort */
1051  snprintf(sql, sizeof(sql),
1052  "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d",
1053  curlevel, curlevel);
1054  if (!pgfdw_exec_cleanup_query(entry->conn, sql, false))
1055  abort_cleanup_failure = true;
1056  }
1057 
1058  /* Disarm changing_xact_state if it all worked. */
1059  entry->changing_xact_state = abort_cleanup_failure;
1060  }
1061 
1062  /* OK, we're outta that level of subtransaction */
1063  entry->xact_depth--;
1064  }
1065 }
#define ERROR
Definition: elog.h:43
bool changing_xact_state
Definition: connection.c:58
static void do_sql_command(PGconn *conn, const char *sql)
Definition: connection.c:528
PGTransactionStatusType PQtransactionStatus(const PGconn *conn)
Definition: fe-connect.c:6624
static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors)
Definition: connection.c:1200
static HTAB * ConnectionHash
Definition: connection.c:67
static bool pgfdw_cancel_query(PGconn *conn)
Definition: connection.c:1153
int GetCurrentTransactionNestLevel(void)
Definition: xact.c:857
bool in_error_recursion_trouble(void)
Definition: elog.c:198
PGconn * conn
Definition: connection.c:52
static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry)
Definition: connection.c:1119
static bool xact_got_connection
Definition: connection.c:74
void * hash_seq_search(HASH_SEQ_STATUS *status)
Definition: dynahash.c:1401
void hash_seq_init(HASH_SEQ_STATUS *status, HTAB *hashp)
Definition: dynahash.c:1391
#define elog(elevel,...)
Definition: elog.h:214
#define snprintf
Definition: port.h:215

◆ pgfdw_xact_callback()

static void pgfdw_xact_callback ( XactEvent  event,
void *  arg 
)
static

Definition at line 782 of file connection.c.

References ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, CONNECTION_OK, cursor_number, DEBUG3, disconnect_pg_server(), do_sql_command(), elog, ereport, errcode(), errmsg(), ERROR, hash_seq_init(), hash_seq_search(), ConnCacheEntry::have_error, ConnCacheEntry::have_prep_stmt, in_error_recursion_trouble(), pgfdw_cancel_query(), pgfdw_exec_cleanup_query(), pgfdw_reject_incomplete_xact_state_change(), PQclear(), PQexec(), PQstatus(), PQTRANS_ACTIVE, PQTRANS_IDLE, PQtransactionStatus(), ConnCacheEntry::xact_depth, XACT_EVENT_ABORT, XACT_EVENT_COMMIT, XACT_EVENT_PARALLEL_ABORT, XACT_EVENT_PARALLEL_COMMIT, XACT_EVENT_PARALLEL_PRE_COMMIT, XACT_EVENT_PRE_COMMIT, XACT_EVENT_PRE_PREPARE, XACT_EVENT_PREPARE, and xact_got_connection.

Referenced by GetConnection().

783 {
784  HASH_SEQ_STATUS scan;
785  ConnCacheEntry *entry;
786 
787  /* Quick exit if no connections were touched in this transaction. */
788  if (!xact_got_connection)
789  return;
790 
791  /*
792  * Scan all connection cache entries to find open remote transactions, and
793  * close them.
794  */
796  while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
797  {
798  PGresult *res;
799 
800  /* Ignore cache entry if no open connection right now */
801  if (entry->conn == NULL)
802  continue;
803 
804  /* If it has an open remote transaction, try to close it */
805  if (entry->xact_depth > 0)
806  {
807  bool abort_cleanup_failure = false;
808 
809  elog(DEBUG3, "closing remote transaction on connection %p",
810  entry->conn);
811 
812  switch (event)
813  {
816 
817  /*
818  * If abort cleanup previously failed for this connection,
819  * we can't issue any more commands against it.
820  */
822 
823  /* Commit all remote transactions during pre-commit */
824  entry->changing_xact_state = true;
825  do_sql_command(entry->conn, "COMMIT TRANSACTION");
826  entry->changing_xact_state = false;
827 
828  /*
829  * If there were any errors in subtransactions, and we
830  * made prepared statements, do a DEALLOCATE ALL to make
831  * sure we get rid of all prepared statements. This is
832  * annoying and not terribly bulletproof, but it's
833  * probably not worth trying harder.
834  *
835  * DEALLOCATE ALL only exists in 8.3 and later, so this
836  * constrains how old a server postgres_fdw can
837  * communicate with. We intentionally ignore errors in
838  * the DEALLOCATE, so that we can hobble along to some
839  * extent with older servers (leaking prepared statements
840  * as we go; but we don't really support update operations
841  * pre-8.3 anyway).
842  */
843  if (entry->have_prep_stmt && entry->have_error)
844  {
845  res = PQexec(entry->conn, "DEALLOCATE ALL");
846  PQclear(res);
847  }
848  entry->have_prep_stmt = false;
849  entry->have_error = false;
850  break;
852 
853  /*
854  * We disallow any remote transactions, since it's not
855  * very reasonable to hold them open until the prepared
856  * transaction is committed. For the moment, throw error
857  * unconditionally; later we might allow read-only cases.
858  * Note that the error will cause us to come right back
859  * here with event == XACT_EVENT_ABORT, so we'll clean up
860  * the connection state at that point.
861  */
862  ereport(ERROR,
863  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
864  errmsg("cannot PREPARE a transaction that has operated on postgres_fdw foreign tables")));
865  break;
867  case XACT_EVENT_COMMIT:
868  case XACT_EVENT_PREPARE:
869  /* Pre-commit should have closed the open transaction */
870  elog(ERROR, "missed cleaning up connection during pre-commit");
871  break;
873  case XACT_EVENT_ABORT:
874 
875  /*
876  * Don't try to clean up the connection if we're already
877  * in error recursion trouble.
878  */
880  entry->changing_xact_state = true;
881 
882  /*
883  * If connection is already unsalvageable, don't touch it
884  * further.
885  */
886  if (entry->changing_xact_state)
887  break;
888 
889  /*
890  * Mark this connection as in the process of changing
891  * transaction state.
892  */
893  entry->changing_xact_state = true;
894 
895  /* Assume we might have lost track of prepared statements */
896  entry->have_error = true;
897 
898  /*
899  * If a command has been submitted to the remote server by
900  * using an asynchronous execution function, the command
901  * might not have yet completed. Check to see if a
902  * command is still being processed by the remote server,
903  * and if so, request cancellation of the command.
904  */
905  if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE &&
906  !pgfdw_cancel_query(entry->conn))
907  {
908  /* Unable to cancel running query. */
909  abort_cleanup_failure = true;
910  }
911  else if (!pgfdw_exec_cleanup_query(entry->conn,
912  "ABORT TRANSACTION",
913  false))
914  {
915  /* Unable to abort remote transaction. */
916  abort_cleanup_failure = true;
917  }
918  else if (entry->have_prep_stmt && entry->have_error &&
920  "DEALLOCATE ALL",
921  true))
922  {
923  /* Trouble clearing prepared statements. */
924  abort_cleanup_failure = true;
925  }
926  else
927  {
928  entry->have_prep_stmt = false;
929  entry->have_error = false;
930  }
931 
932  /* Disarm changing_xact_state if it all worked. */
933  entry->changing_xact_state = abort_cleanup_failure;
934  break;
935  }
936  }
937 
938  /* Reset state to show we're out of a transaction */
939  entry->xact_depth = 0;
940 
941  /*
942  * If the connection isn't in a good idle state, discard it to
943  * recover. Next GetConnection will open a new connection.
944  */
945  if (PQstatus(entry->conn) != CONNECTION_OK ||
947  entry->changing_xact_state)
948  {
949  elog(DEBUG3, "discarding connection %p", entry->conn);
950  disconnect_pg_server(entry);
951  }
952  }
953 
954  /*
955  * Regardless of the event type, we can now mark ourselves as out of the
956  * transaction. (Note: if we are here during PRE_COMMIT or PRE_PREPARE,
957  * this saves a useless scan of the hashtable during COMMIT or PREPARE.)
958  */
959  xact_got_connection = false;
960 
961  /* Also reset cursor numbering for next transaction */
962  cursor_number = 0;
963 }
#define DEBUG3
Definition: elog.h:23
int errcode(int sqlerrcode)
Definition: elog.c:610
bool have_prep_stmt
Definition: connection.c:56
#define ERROR
Definition: elog.h:43
bool changing_xact_state
Definition: connection.c:58
static void do_sql_command(PGconn *conn, const char *sql)
Definition: connection.c:528
static unsigned int cursor_number
Definition: connection.c:70
PGTransactionStatusType PQtransactionStatus(const PGconn *conn)
Definition: fe-connect.c:6624
static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors)
Definition: connection.c:1200
static HTAB * ConnectionHash
Definition: connection.c:67
static bool pgfdw_cancel_query(PGconn *conn)
Definition: connection.c:1153
#define ereport(elevel,...)
Definition: elog.h:144
void PQclear(PGresult *res)
Definition: fe-exec.c:694
bool in_error_recursion_trouble(void)
Definition: elog.c:198
PGconn * conn
Definition: connection.c:52
static void disconnect_pg_server(ConnCacheEntry *entry)
Definition: connection.c:417
static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry)
Definition: connection.c:1119
static bool xact_got_connection
Definition: connection.c:74
void * hash_seq_search(HASH_SEQ_STATUS *status)
Definition: dynahash.c:1401
void hash_seq_init(HASH_SEQ_STATUS *status, HTAB *hashp)
Definition: dynahash.c:1391
int errmsg(const char *fmt,...)
Definition: elog.c:821
#define elog(elevel,...)
Definition: elog.h:214
PGresult * PQexec(PGconn *conn, const char *query)
Definition: fe-exec.c:1939
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6616

◆ ReleaseConnection()

void ReleaseConnection ( PGconn conn)

Definition at line 594 of file connection.c.

Referenced by estimate_path_cost_size(), finish_foreign_modify(), postgresAcquireSampleRowsFunc(), postgresAnalyzeForeignTable(), postgresEndDirectModify(), postgresEndForeignScan(), and postgresImportForeignSchema().

595 {
596  /*
597  * Currently, we don't actually track connection references because all
598  * cleanup is managed on a transaction or subtransaction basis instead. So
599  * there's nothing to do here.
600  */
601 }

◆ UserMappingPasswordRequired()

static bool UserMappingPasswordRequired ( UserMapping user)
static

Definition at line 432 of file connection.c.

References defGetBoolean(), DefElem::defname, lfirst, and UserMapping::options.

Referenced by check_conn_params(), and connect_pg_server().

433 {
434  ListCell *cell;
435 
436  foreach(cell, user->options)
437  {
438  DefElem *def = (DefElem *) lfirst(cell);
439 
440  if (strcmp(def->defname, "password_required") == 0)
441  return defGetBoolean(def);
442  }
443 
444  return true;
445 }
bool defGetBoolean(DefElem *def)
Definition: define.c:111
List * options
Definition: foreign.h:50
#define lfirst(lc)
Definition: pg_list.h:169
char * defname
Definition: parsenodes.h:733

Variable Documentation

◆ ConnectionHash

HTAB* ConnectionHash = NULL
static

Definition at line 67 of file connection.c.

◆ cursor_number

unsigned int cursor_number = 0
static

Definition at line 70 of file connection.c.

Referenced by GetCursorNumber(), and pgfdw_xact_callback().

◆ prep_stmt_number

unsigned int prep_stmt_number = 0
static

Definition at line 71 of file connection.c.

Referenced by GetPrepStmtNumber().

◆ xact_got_connection

bool xact_got_connection = false
static

Definition at line 74 of file connection.c.

Referenced by GetConnection(), pgfdw_subxact_callback(), and pgfdw_xact_callback().