PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
connection.c File Reference
#include "postgres.h"
#include "postgres_fdw.h"
#include "access/htup_details.h"
#include "catalog/pg_user_mapping.h"
#include "access/xact.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "storage/latch.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 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)
 
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)
 
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

Definition at line 46 of file connection.c.

Function Documentation

static void begin_remote_xact ( ConnCacheEntry entry)
static

Definition at line 417 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().

418 {
419  int curlevel = GetCurrentTransactionNestLevel();
420 
421  /* Start main transaction if we haven't yet */
422  if (entry->xact_depth <= 0)
423  {
424  const char *sql;
425 
426  elog(DEBUG3, "starting remote transaction on connection %p",
427  entry->conn);
428 
430  sql = "START TRANSACTION ISOLATION LEVEL SERIALIZABLE";
431  else
432  sql = "START TRANSACTION ISOLATION LEVEL REPEATABLE READ";
433  entry->changing_xact_state = true;
434  do_sql_command(entry->conn, sql);
435  entry->xact_depth = 1;
436  entry->changing_xact_state = false;
437  }
438 
439  /*
440  * If we're in a subtransaction, stack up savepoints to match our level.
441  * This ensures we can rollback just the desired effects when a
442  * subtransaction aborts.
443  */
444  while (entry->xact_depth < curlevel)
445  {
446  char sql[64];
447 
448  snprintf(sql, sizeof(sql), "SAVEPOINT s%d", entry->xact_depth + 1);
449  entry->changing_xact_state = true;
450  do_sql_command(entry->conn, sql);
451  entry->xact_depth++;
452  entry->changing_xact_state = false;
453  }
454 }
#define DEBUG3
Definition: elog.h:23
int snprintf(char *str, size_t count, const char *fmt,...) pg_attribute_printf(3
bool changing_xact_state
Definition: connection.c:57
static void do_sql_command(PGconn *conn, const char *sql)
Definition: connection.c:394
int GetCurrentTransactionNestLevel(void)
Definition: xact.c:762
PGconn * conn
Definition: connection.c:51
#define IsolationIsSerializable()
Definition: xact.h:44
#define elog
Definition: elog.h:219
static void check_conn_params ( const char **  keywords,
const char **  values 
)
static

Definition at line 325 of file connection.c.

References ereport, errcode(), errdetail(), errmsg(), ERROR, i, and superuser().

Referenced by connect_pg_server().

326 {
327  int i;
328 
329  /* no check required if superuser */
330  if (superuser())
331  return;
332 
333  /* ok if params contain a non-empty password */
334  for (i = 0; keywords[i] != NULL; i++)
335  {
336  if (strcmp(keywords[i], "password") == 0 && values[i][0] != '\0')
337  return;
338  }
339 
340  ereport(ERROR,
341  (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
342  errmsg("password is required"),
343  errdetail("Non-superusers must provide a password in the user mapping.")));
344 }
int errcode(int sqlerrcode)
Definition: elog.c:575
bool superuser(void)
Definition: superuser.c:47
#define ERROR
Definition: elog.h:43
int errdetail(const char *fmt,...)
Definition: elog.c:873
#define ereport(elevel, rest)
Definition: elog.h:122
static Datum values[MAXATTR]
Definition: bootstrap.c:163
int errmsg(const char *fmt,...)
Definition: elog.c:797
int i
static void configure_remote_session ( PGconn conn)
static

Definition at line 358 of file connection.c.

References do_sql_command(), and PQserverVersion().

Referenced by connect_pg_server().

359 {
360  int remoteversion = PQserverVersion(conn);
361 
362  /* Force the search path to contain only pg_catalog (see deparse.c) */
363  do_sql_command(conn, "SET search_path = pg_catalog");
364 
365  /*
366  * Set remote timezone; this is basically just cosmetic, since all
367  * transmitted and returned timestamptzs should specify a zone explicitly
368  * anyway. However it makes the regression test outputs more predictable.
369  *
370  * We don't risk setting remote zone equal to ours, since the remote
371  * server might use a different timezone database. Instead, use UTC
372  * (quoted, because very old servers are picky about case).
373  */
374  do_sql_command(conn, "SET timezone = 'UTC'");
375 
376  /*
377  * Set values needed to ensure unambiguous data output from remote. (This
378  * logic should match what pg_dump does. See also set_transmission_modes
379  * in postgres_fdw.c.)
380  */
381  do_sql_command(conn, "SET datestyle = ISO");
382  if (remoteversion >= 80400)
383  do_sql_command(conn, "SET intervalstyle = postgres");
384  if (remoteversion >= 90000)
385  do_sql_command(conn, "SET extra_float_digits = 3");
386  else
387  do_sql_command(conn, "SET extra_float_digits = 2");
388 }
int PQserverVersion(const PGconn *conn)
Definition: fe-connect.c:6087
static void do_sql_command(PGconn *conn, const char *sql)
Definition: connection.c:394
static PGconn * connect_pg_server ( ForeignServer server,
UserMapping user 
)
static

Definition at line 222 of file connection.c.

References check_conn_params(), configure_remote_session(), 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(), ForeignServer::servername, superuser(), and values.

Referenced by GetConnection().

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 connection parameters and make connection */
264  check_conn_params(keywords, values);
265 
266  conn = PQconnectdbParams(keywords, values, false);
267  if (!conn || PQstatus(conn) != CONNECTION_OK)
268  ereport(ERROR,
269  (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
270  errmsg("could not connect to server \"%s\"",
271  server->servername),
272  errdetail_internal("%s", pchomp(PQerrorMessage(conn)))));
273 
274  /*
275  * Check that non-superuser has used password to establish connection;
276  * otherwise, he's piggybacking on the postgres server's user
277  * identity. See also dblink_security_check() in contrib/dblink.
278  */
279  if (!superuser() && !PQconnectionUsedPassword(conn))
280  ereport(ERROR,
281  (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
282  errmsg("password is required"),
283  errdetail("Non-superuser cannot connect if the server does not request a password."),
284  errhint("Target server's authentication method must be changed.")));
285 
286  /* Prepare new session for use */
288 
289  pfree(keywords);
290  pfree(values);
291  }
292  PG_CATCH();
293  {
294  /* Release PGconn data structure if we managed to create one */
295  if (conn)
296  PQfinish(conn);
297  PG_RE_THROW();
298  }
299  PG_END_TRY();
300 
301  return conn;
302 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6097
static void configure_remote_session(PGconn *conn)
Definition: connection.c:358
int errhint(const char *fmt,...)
Definition: elog.c:987
int errcode(int sqlerrcode)
Definition: elog.c:575
void PQfinish(PGconn *conn)
Definition: fe-connect.c:3630
bool superuser(void)
Definition: superuser.c:47
PGconn * PQconnectdbParams(const char *const *keywords, const char *const *values, int expand_dbname)
Definition: fe-connect.c:529
int errdetail_internal(const char *fmt,...)
Definition: elog.c:900
char * pchomp(const char *in)
Definition: mcxt.c:1100
void pfree(void *pointer)
Definition: mcxt.c:949
int ExtractConnectionOptions(List *defelems, const char **keywords, const char **values)
Definition: option.c:297
#define ERROR
Definition: elog.h:43
PGconn * conn
Definition: streamutil.c:43
int errdetail(const char *fmt,...)
Definition: elog.c:873
List * options
Definition: foreign.h:61
#define ereport(elevel, rest)
Definition: elog.h:122
static void check_conn_params(const char **keywords, const char **values)
Definition: connection.c:325
#define PG_CATCH()
Definition: elog.h:293
const char * GetDatabaseEncodingName(void)
Definition: mbutils.c:1021
static int list_length(const List *l)
Definition: pg_list.h:89
#define PG_RE_THROW()
Definition: elog.h:314
static Datum values[MAXATTR]
Definition: bootstrap.c:163
int PQconnectionUsedPassword(const PGconn *conn)
Definition: fe-connect.c:6146
void * palloc(Size size)
Definition: mcxt.c:848
int errmsg(const char *fmt,...)
Definition: elog.c:797
char * servername
Definition: foreign.h:50
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6044
#define PG_TRY()
Definition: elog.h:284
List * options
Definition: foreign.h:53
#define PG_END_TRY()
Definition: elog.h:300
static void disconnect_pg_server ( ConnCacheEntry entry)
static

Definition at line 308 of file connection.c.

References ConnCacheEntry::conn, and PQfinish().

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

309 {
310  if (entry->conn != NULL)
311  {
312  PQfinish(entry->conn);
313  entry->conn = NULL;
314  }
315 }
void PQfinish(PGconn *conn)
Definition: fe-connect.c:3630
PGconn * conn
Definition: connection.c:51
static void do_sql_command ( PGconn conn,
const char *  sql 
)
static

Definition at line 394 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().

395 {
396  PGresult *res;
397 
398  if (!PQsendQuery(conn, sql))
399  pgfdw_report_error(ERROR, NULL, conn, false, sql);
400  res = pgfdw_get_result(conn, sql);
401  if (PQresultStatus(res) != PGRES_COMMAND_OK)
402  pgfdw_report_error(ERROR, res, conn, true, sql);
403  PQclear(res);
404 }
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
int PQsendQuery(PGconn *conn, const char *query)
Definition: fe-exec.c:1183
#define ERROR
Definition: elog.h:43
void pgfdw_report_error(int elevel, PGresult *res, PGconn *conn, bool clear, const char *sql)
Definition: connection.c:596
void PQclear(PGresult *res)
Definition: fe-exec.c:671
PGresult * pgfdw_get_result(PGconn *conn, const char *query)
Definition: connection.c:532
PGconn* GetConnection ( UserMapping user,
bool  will_prep_stmt 
)

Definition at line 107 of file connection.c.

References begin_remote_xact(), CacheMemoryContext, CacheRegisterSyscacheCallback(), ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, connect_pg_server(), DEBUG3, disconnect_pg_server(), elog, HASHCTL::entrysize, FOREIGNSERVEROID, GetForeignServer(), GetSysCacheHashValue1, HASH_BLOBS, HASH_CONTEXT, hash_create(), HASH_ELEM, HASH_ENTER, hash_search(), ConnCacheEntry::have_error, ConnCacheEntry::have_prep_stmt, HASHCTL::hcxt, ConnCacheEntry::invalidated, HASHCTL::keysize, ConnCacheEntry::mapping_hashvalue, MemSet, ObjectIdGetDatum, pgfdw_inval_callback(), pgfdw_reject_incomplete_xact_state_change(), pgfdw_subxact_callback(), pgfdw_xact_callback(), RegisterSubXactCallback(), RegisterXactCallback(), ConnCacheEntry::server_hashvalue, ForeignServer::serverid, UserMapping::serverid, ForeignServer::servername, UserMapping::umid, UserMapping::userid, USERMAPPINGOID, ConnCacheEntry::xact_depth, and xact_got_connection.

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

108 {
109  bool found;
110  ConnCacheEntry *entry;
111  ConnCacheKey key;
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 }
Oid umid
Definition: foreign.h:58
#define HASH_CONTEXT
Definition: hsearch.h:93
#define HASH_ELEM
Definition: hsearch.h:87
MemoryContext hcxt
Definition: hsearch.h:78
#define DEBUG3
Definition: elog.h:23
struct ConnCacheEntry ConnCacheEntry
Size entrysize
Definition: hsearch.h:73
#define MemSet(start, val, len)
Definition: c.h:846
bool have_prep_stmt
Definition: connection.c:55
#define GetSysCacheHashValue1(cacheId, key1)
Definition: syscache.h:195
uint32 server_hashvalue
Definition: connection.c:59
void * hash_search(HTAB *hashp, const void *keyPtr, HASHACTION action, bool *foundPtr)
Definition: dynahash.c:902
Oid userid
Definition: foreign.h:59
#define ObjectIdGetDatum(X)
Definition: postgres.h:513
bool changing_xact_state
Definition: connection.c:57
bool invalidated
Definition: connection.c:58
static void pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue)
Definition: connection.c:951
#define HASH_BLOBS
Definition: hsearch.h:88
void CacheRegisterSyscacheCallback(int cacheid, SyscacheCallbackFunction func, Datum arg)
Definition: inval.c:1389
static void pgfdw_xact_callback(XactEvent event, void *arg)
Definition: connection.c:650
static HTAB * ConnectionHash
Definition: connection.c:66
ForeignServer * GetForeignServer(Oid serverid)
Definition: foreign.c:93
uintptr_t Datum
Definition: postgres.h:372
HTAB * hash_create(const char *tabname, long nelem, HASHCTL *info, int flags)
Definition: dynahash.c:316
void RegisterSubXactCallback(SubXactCallback callback, void *arg)
Definition: xact.c:3366
Size keysize
Definition: hsearch.h:72
uint32 mapping_hashvalue
Definition: connection.c:60
PGconn * conn
Definition: connection.c:51
static void disconnect_pg_server(ConnCacheEntry *entry)
Definition: connection.c:308
void RegisterXactCallback(XactCallback callback, void *arg)
Definition: xact.c:3311
static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry)
Definition: connection.c:987
static bool xact_got_connection
Definition: connection.c:73
Oid serverid
Definition: foreign.h:60
char * servername
Definition: foreign.h:50
#define elog
Definition: elog.h:219
static void pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, SubTransactionId parentSubid, void *arg)
Definition: connection.c:837
static void begin_remote_xact(ConnCacheEntry *entry)
Definition: connection.c:417
Oid ConnCacheKey
Definition: connection.c:46
Oid serverid
Definition: foreign.h:47
MemoryContext CacheMemoryContext
Definition: mcxt.c:46
static PGconn * connect_pg_server(ForeignServer *server, UserMapping *user)
Definition: connection.c:222
unsigned int GetCursorNumber ( PGconn conn)

Definition at line 481 of file connection.c.

References cursor_number.

Referenced by postgresAcquireSampleRowsFunc(), and postgresBeginForeignScan().

482 {
483  return ++cursor_number;
484 }
static unsigned int cursor_number
Definition: connection.c:69
unsigned int GetPrepStmtNumber ( PGconn conn)

Definition at line 495 of file connection.c.

References prep_stmt_number.

Referenced by prepare_foreign_modify().

496 {
497  return ++prep_stmt_number;
498 }
static unsigned int prep_stmt_number
Definition: connection.c:70
static bool pgfdw_cancel_query ( PGconn conn)
static

Definition at line 1021 of file connection.c.

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

Referenced by pgfdw_subxact_callback(), and pgfdw_xact_callback().

1022 {
1023  PGcancel *cancel;
1024  char errbuf[256];
1025  PGresult *result = NULL;
1026  TimestampTz endtime;
1027 
1028  /*
1029  * If it takes too long to cancel the query and discard the result, assume
1030  * the connection is dead.
1031  */
1033 
1034  /*
1035  * Issue cancel request. Unfortunately, there's no good way to limit the
1036  * amount of time that we might block inside PQgetCancel().
1037  */
1038  if ((cancel = PQgetCancel(conn)))
1039  {
1040  if (!PQcancel(cancel, errbuf, sizeof(errbuf)))
1041  {
1042  ereport(WARNING,
1043  (errcode(ERRCODE_CONNECTION_FAILURE),
1044  errmsg("could not send cancel request: %s",
1045  errbuf)));
1046  PQfreeCancel(cancel);
1047  return false;
1048  }
1049  PQfreeCancel(cancel);
1050  }
1051 
1052  /* Get and discard the result of the query. */
1053  if (pgfdw_get_cleanup_result(conn, endtime, &result))
1054  return false;
1055  PQclear(result);
1056 
1057  return true;
1058 }
static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result)
Definition: connection.c:1122
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1570
int64 TimestampTz
Definition: timestamp.h:39
void PQfreeCancel(PGcancel *cancel)
Definition: fe-connect.c:3774
int errcode(int sqlerrcode)
Definition: elog.c:575
return result
Definition: formatting.c:1633
PGcancel * PQgetCancel(PGconn *conn)
Definition: fe-connect.c:3751
#define ereport(elevel, rest)
Definition: elog.h:122
#define WARNING
Definition: elog.h:40
#define TimestampTzPlusMilliseconds(tz, ms)
Definition: timestamp.h:56
void PQclear(PGresult *res)
Definition: fe-exec.c:671
int PQcancel(PGcancel *cancel, char *errbuf, int errbufsize)
Definition: fe-connect.c:3906
int errmsg(const char *fmt,...)
Definition: elog.c:797
static bool pgfdw_exec_cleanup_query ( PGconn conn,
const char *  query,
bool  ignore_errors 
)
static

Definition at line 1068 of file connection.c.

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

Referenced by pgfdw_subxact_callback(), and pgfdw_xact_callback().

1069 {
1070  PGresult *result = NULL;
1071  TimestampTz endtime;
1072 
1073  /*
1074  * If it takes too long to execute a cleanup query, assume the connection
1075  * is dead. It's fairly likely that this is why we aborted in the first
1076  * place (e.g. statement timeout, user cancel), so the timeout shouldn't
1077  * be too long.
1078  */
1080 
1081  /*
1082  * Submit a query. Since we don't use non-blocking mode, this also can
1083  * block. But its risk is relatively small, so we ignore that for now.
1084  */
1085  if (!PQsendQuery(conn, query))
1086  {
1087  pgfdw_report_error(WARNING, NULL, conn, false, query);
1088  return false;
1089  }
1090 
1091  /* Get the result of the query. */
1092  if (pgfdw_get_cleanup_result(conn, endtime, &result))
1093  return false;
1094 
1095  /* Issue a warning if not successful. */
1096  if (PQresultStatus(result) != PGRES_COMMAND_OK)
1097  {
1098  pgfdw_report_error(WARNING, result, conn, true, query);
1099  return ignore_errors;
1100  }
1101  PQclear(result);
1102 
1103  return true;
1104 }
static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result)
Definition: connection.c:1122
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1570
int64 TimestampTz
Definition: timestamp.h:39
return result
Definition: formatting.c:1633
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
int PQsendQuery(PGconn *conn, const char *query)
Definition: fe-exec.c:1183
void pgfdw_report_error(int elevel, PGresult *res, PGconn *conn, bool clear, const char *sql)
Definition: connection.c:596
#define WARNING
Definition: elog.h:40
#define TimestampTzPlusMilliseconds(tz, ms)
Definition: timestamp.h:56
void PQclear(PGresult *res)
Definition: fe-exec.c:671
PGresult* pgfdw_exec_query ( PGconn conn,
const char *  query 
)

Definition at line 508 of file connection.c.

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

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

509 {
510  /*
511  * Submit a query. Since we don't use non-blocking mode, this also can
512  * block. But its risk is relatively small, so we ignore that for now.
513  */
514  if (!PQsendQuery(conn, query))
515  pgfdw_report_error(ERROR, NULL, conn, false, query);
516 
517  /* Wait for the result. */
518  return pgfdw_get_result(conn, query);
519 }
int PQsendQuery(PGconn *conn, const char *query)
Definition: fe-exec.c:1183
#define ERROR
Definition: elog.h:43
void pgfdw_report_error(int elevel, PGresult *res, PGconn *conn, bool clear, const char *sql)
Definition: connection.c:596
PGresult * pgfdw_get_result(PGconn *conn, const char *query)
Definition: connection.c:532
static bool pgfdw_get_cleanup_result ( PGconn conn,
TimestampTz  endtime,
PGresult **  result 
)
static

Definition at line 1122 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_LATCH_SET, WL_SOCKET_READABLE, and WL_TIMEOUT.

Referenced by pgfdw_cancel_query(), and pgfdw_exec_cleanup_query().

1123 {
1124  volatile bool timed_out = false;
1125  PGresult *volatile last_res = NULL;
1126 
1127  /* In what follows, do not leak any PGresults on an error. */
1128  PG_TRY();
1129  {
1130  for (;;)
1131  {
1132  PGresult *res;
1133 
1134  while (PQisBusy(conn))
1135  {
1136  int wc;
1138  long secs;
1139  int microsecs;
1140  long cur_timeout;
1141 
1142  /* If timeout has expired, give up, else get sleep time. */
1143  if (now >= endtime)
1144  {
1145  timed_out = true;
1146  goto exit;
1147  }
1148  TimestampDifference(now, endtime, &secs, &microsecs);
1149 
1150  /* To protect against clock skew, limit sleep to one minute. */
1151  cur_timeout = Min(60000, secs * USECS_PER_SEC + microsecs);
1152 
1153  /* Sleep until there's something to do */
1156  PQsocket(conn),
1157  cur_timeout, PG_WAIT_EXTENSION);
1159 
1161 
1162  /* Data available in socket? */
1163  if (wc & WL_SOCKET_READABLE)
1164  {
1165  if (!PQconsumeInput(conn))
1166  {
1167  /* connection trouble; treat the same as a timeout */
1168  timed_out = true;
1169  goto exit;
1170  }
1171  }
1172  }
1173 
1174  res = PQgetResult(conn);
1175  if (res == NULL)
1176  break; /* query is complete */
1177 
1178  PQclear(last_res);
1179  last_res = res;
1180  }
1181 exit: ;
1182  }
1183  PG_CATCH();
1184  {
1185  PQclear(last_res);
1186  PG_RE_THROW();
1187  }
1188  PG_END_TRY();
1189 
1190  if (timed_out)
1191  PQclear(last_res);
1192  else
1193  *result = last_res;
1194  return timed_out;
1195 }
#define WL_TIMEOUT
Definition: latch.h:127
#define USECS_PER_SEC
Definition: timestamp.h:94
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1570
int64 TimestampTz
Definition: timestamp.h:39
#define Min(x, y)
Definition: c.h:795
#define WL_SOCKET_READABLE
Definition: latch.h:125
void ResetLatch(volatile Latch *latch)
Definition: latch.c:497
#define PG_WAIT_EXTENSION
Definition: pgstat.h:742
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1682
void PQclear(PGresult *res)
Definition: fe-exec.c:671
#define PG_CATCH()
Definition: elog.h:293
int WaitLatchOrSocket(volatile Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:356
int PQisBusy(PGconn *conn)
Definition: fe-exec.c:1732
#define PG_RE_THROW()
Definition: elog.h:314
struct Latch * MyLatch
Definition: globals.c:52
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
void TimestampDifference(TimestampTz start_time, TimestampTz stop_time, long *secs, int *microsecs)
Definition: timestamp.c:1624
#define PG_TRY()
Definition: elog.h:284
#define WL_LATCH_SET
Definition: latch.h:124
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6115
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1534
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1753
#define PG_END_TRY()
Definition: elog.h:300
PGresult* pgfdw_get_result ( PGconn conn,
const char *  query 
)

Definition at line 532 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_LATCH_SET, and WL_SOCKET_READABLE.

Referenced by create_cursor(), do_sql_command(), execute_dml_stmt(), pgfdw_exec_query(), postgresExecForeignDelete(), postgresExecForeignInsert(), postgresExecForeignUpdate(), and prepare_foreign_modify().

533 {
534  PGresult *volatile last_res = NULL;
535 
536  /* In what follows, do not leak any PGresults on an error. */
537  PG_TRY();
538  {
539  for (;;)
540  {
541  PGresult *res;
542 
543  while (PQisBusy(conn))
544  {
545  int wc;
546 
547  /* Sleep until there's something to do */
550  PQsocket(conn),
551  -1L, PG_WAIT_EXTENSION);
553 
555 
556  /* Data available in socket? */
557  if (wc & WL_SOCKET_READABLE)
558  {
559  if (!PQconsumeInput(conn))
560  pgfdw_report_error(ERROR, NULL, conn, false, query);
561  }
562  }
563 
564  res = PQgetResult(conn);
565  if (res == NULL)
566  break; /* query is complete */
567 
568  PQclear(last_res);
569  last_res = res;
570  }
571  }
572  PG_CATCH();
573  {
574  PQclear(last_res);
575  PG_RE_THROW();
576  }
577  PG_END_TRY();
578 
579  return last_res;
580 }
#define WL_SOCKET_READABLE
Definition: latch.h:125
void ResetLatch(volatile Latch *latch)
Definition: latch.c:497
#define ERROR
Definition: elog.h:43
void pgfdw_report_error(int elevel, PGresult *res, PGconn *conn, bool clear, const char *sql)
Definition: connection.c:596
#define PG_WAIT_EXTENSION
Definition: pgstat.h:742
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1682
void PQclear(PGresult *res)
Definition: fe-exec.c:671
#define PG_CATCH()
Definition: elog.h:293
int WaitLatchOrSocket(volatile Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:356
int PQisBusy(PGconn *conn)
Definition: fe-exec.c:1732
#define PG_RE_THROW()
Definition: elog.h:314
struct Latch * MyLatch
Definition: globals.c:52
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
#define PG_TRY()
Definition: elog.h:284
#define WL_LATCH_SET
Definition: latch.h:124
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6115
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1753
#define PG_END_TRY()
Definition: elog.h:300
static void pgfdw_inval_callback ( Datum  arg,
int  cacheid,
uint32  hashvalue 
)
static

Definition at line 951 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().

952 {
953  HASH_SEQ_STATUS scan;
954  ConnCacheEntry *entry;
955 
956  Assert(cacheid == FOREIGNSERVEROID || cacheid == USERMAPPINGOID);
957 
958  /* ConnectionHash must exist already, if we're registered */
960  while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
961  {
962  /* Ignore invalid entries */
963  if (entry->conn == NULL)
964  continue;
965 
966  /* hashvalue == 0 means a cache reset, must clear all state */
967  if (hashvalue == 0 ||
968  (cacheid == FOREIGNSERVEROID &&
969  entry->server_hashvalue == hashvalue) ||
970  (cacheid == USERMAPPINGOID &&
971  entry->mapping_hashvalue == hashvalue))
972  entry->invalidated = true;
973  }
974 }
uint32 server_hashvalue
Definition: connection.c:59
bool invalidated
Definition: connection.c:58
static HTAB * ConnectionHash
Definition: connection.c:66
uint32 mapping_hashvalue
Definition: connection.c:60
PGconn * conn
Definition: connection.c:51
#define Assert(condition)
Definition: c.h:664
void * hash_seq_search(HASH_SEQ_STATUS *status)
Definition: dynahash.c:1385
void hash_seq_init(HASH_SEQ_STATUS *status, HTAB *hashp)
Definition: dynahash.c:1375
static void pgfdw_reject_incomplete_xact_state_change ( ConnCacheEntry entry)
static

Definition at line 987 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().

988 {
989  HeapTuple tup;
990  Form_pg_user_mapping umform;
991  ForeignServer *server;
992 
993  /* nothing to do for inactive entries and entries of sane state */
994  if (entry->conn == NULL || !entry->changing_xact_state)
995  return;
996 
997  /* make sure this entry is inactive */
998  disconnect_pg_server(entry);
999 
1000  /* find server name to be shown in the message below */
1002  ObjectIdGetDatum(entry->key));
1003  if (!HeapTupleIsValid(tup))
1004  elog(ERROR, "cache lookup failed for user mapping %u", entry->key);
1005  umform = (Form_pg_user_mapping) GETSTRUCT(tup);
1006  server = GetForeignServer(umform->umserver);
1007  ReleaseSysCache(tup);
1008 
1009  ereport(ERROR,
1010  (errcode(ERRCODE_CONNECTION_EXCEPTION),
1011  errmsg("connection to server \"%s\" was lost",
1012  server->servername)));
1013 }
#define GETSTRUCT(TUP)
Definition: htup_details.h:656
ConnCacheKey key
Definition: connection.c:50
int errcode(int sqlerrcode)
Definition: elog.c:575
#define SearchSysCache1(cacheId, key1)
Definition: syscache.h:159
#define ObjectIdGetDatum(X)
Definition: postgres.h:513
#define ERROR
Definition: elog.h:43
bool changing_xact_state
Definition: connection.c:57
FormData_pg_user_mapping * Form_pg_user_mapping
#define ereport(elevel, rest)
Definition: elog.h:122
ForeignServer * GetForeignServer(Oid serverid)
Definition: foreign.c:93
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1117
PGconn * conn
Definition: connection.c:51
#define HeapTupleIsValid(tuple)
Definition: htup.h:77
static void disconnect_pg_server(ConnCacheEntry *entry)
Definition: connection.c:308
int errmsg(const char *fmt,...)
Definition: elog.c:797
char * servername
Definition: foreign.h:50
#define elog
Definition: elog.h:219
void pgfdw_report_error ( int  elevel,
PGresult res,
PGconn conn,
bool  clear,
const char *  sql 
)

Definition at line 596 of file connection.c.

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

Referenced by close_cursor(), create_cursor(), do_sql_command(), execute_dml_stmt(), fetch_more_data(), get_remote_estimate(), pgfdw_exec_cleanup_query(), pgfdw_exec_query(), pgfdw_get_result(), postgresAcquireSampleRowsFunc(), postgresAnalyzeForeignTable(), postgresEndForeignModify(), postgresExecForeignDelete(), postgresExecForeignInsert(), postgresExecForeignUpdate(), postgresImportForeignSchema(), postgresReScanForeignScan(), and prepare_foreign_modify().

598 {
599  /* If requested, PGresult must be released before leaving this function. */
600  PG_TRY();
601  {
602  char *diag_sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
603  char *message_primary = PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY);
604  char *message_detail = PQresultErrorField(res, PG_DIAG_MESSAGE_DETAIL);
605  char *message_hint = PQresultErrorField(res, PG_DIAG_MESSAGE_HINT);
606  char *message_context = PQresultErrorField(res, PG_DIAG_CONTEXT);
607  int sqlstate;
608 
609  if (diag_sqlstate)
610  sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
611  diag_sqlstate[1],
612  diag_sqlstate[2],
613  diag_sqlstate[3],
614  diag_sqlstate[4]);
615  else
616  sqlstate = ERRCODE_CONNECTION_FAILURE;
617 
618  /*
619  * If we don't get a message from the PGresult, try the PGconn. This
620  * is needed because for connection-level failures, PQexec may just
621  * return NULL, not a PGresult at all.
622  */
623  if (message_primary == NULL)
624  message_primary = pchomp(PQerrorMessage(conn));
625 
626  ereport(elevel,
627  (errcode(sqlstate),
628  message_primary ? errmsg_internal("%s", message_primary) :
629  errmsg("could not obtain message string for remote error"),
630  message_detail ? errdetail_internal("%s", message_detail) : 0,
631  message_hint ? errhint("%s", message_hint) : 0,
632  message_context ? errcontext("%s", message_context) : 0,
633  sql ? errcontext("Remote SQL command: %s", sql) : 0));
634  }
635  PG_CATCH();
636  {
637  if (clear)
638  PQclear(res);
639  PG_RE_THROW();
640  }
641  PG_END_TRY();
642  if (clear)
643  PQclear(res);
644 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6097
int errhint(const char *fmt,...)
Definition: elog.c:987
#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:575
#define PG_DIAG_SQLSTATE
Definition: postgres_ext.h:57
int errdetail_internal(const char *fmt,...)
Definition: elog.c:900
char * pchomp(const char *in)
Definition: mcxt.c:1100
#define ereport(elevel, rest)
Definition: elog.h:122
static int elevel
Definition: vacuumlazy.c:136
#define PG_DIAG_MESSAGE_HINT
Definition: postgres_ext.h:60
void PQclear(PGresult *res)
Definition: fe-exec.c:671
int errmsg_internal(const char *fmt,...)
Definition: elog.c:827
#define PG_CATCH()
Definition: elog.h:293
char * PQresultErrorField(const PGresult *res, int fieldcode)
Definition: fe-exec.c:2709
#define PG_RE_THROW()
Definition: elog.h:314
int errmsg(const char *fmt,...)
Definition: elog.c:797
#define errcontext
Definition: elog.h:164
#define PG_TRY()
Definition: elog.h:284
#define PG_END_TRY()
Definition: elog.h:300
#define PG_DIAG_CONTEXT
Definition: postgres_ext.h:64
static void pgfdw_subxact_callback ( SubXactEvent  event,
SubTransactionId  mySubid,
SubTransactionId  parentSubid,
void *  arg 
)
static

Definition at line 837 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().

839 {
840  HASH_SEQ_STATUS scan;
841  ConnCacheEntry *entry;
842  int curlevel;
843 
844  /* Nothing to do at subxact start, nor after commit. */
845  if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB ||
846  event == SUBXACT_EVENT_ABORT_SUB))
847  return;
848 
849  /* Quick exit if no connections were touched in this transaction. */
850  if (!xact_got_connection)
851  return;
852 
853  /*
854  * Scan all connection cache entries to find open remote subtransactions
855  * of the current level, and close them.
856  */
857  curlevel = GetCurrentTransactionNestLevel();
859  while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
860  {
861  char sql[100];
862 
863  /*
864  * We only care about connections with open remote subtransactions of
865  * the current level.
866  */
867  if (entry->conn == NULL || entry->xact_depth < curlevel)
868  continue;
869 
870  if (entry->xact_depth > curlevel)
871  elog(ERROR, "missed cleaning up remote subtransaction at level %d",
872  entry->xact_depth);
873 
874  if (event == SUBXACT_EVENT_PRE_COMMIT_SUB)
875  {
876  /*
877  * If abort cleanup previously failed for this connection, we
878  * can't issue any more commands against it.
879  */
881 
882  /* Commit all remote subtransactions during pre-commit */
883  snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel);
884  entry->changing_xact_state = true;
885  do_sql_command(entry->conn, sql);
886  entry->changing_xact_state = false;
887  }
888  else if (in_error_recursion_trouble())
889  {
890  /*
891  * Don't try to clean up the connection if we're already in error
892  * recursion trouble.
893  */
894  entry->changing_xact_state = true;
895  }
896  else if (!entry->changing_xact_state)
897  {
898  bool abort_cleanup_failure = false;
899 
900  /* Remember that abort cleanup is in progress. */
901  entry->changing_xact_state = true;
902 
903  /* Assume we might have lost track of prepared statements */
904  entry->have_error = true;
905 
906  /*
907  * If a command has been submitted to the remote server by using
908  * an asynchronous execution function, the command might not have
909  * yet completed. Check to see if a command is still being
910  * processed by the remote server, and if so, request cancellation
911  * of the command.
912  */
913  if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE &&
914  !pgfdw_cancel_query(entry->conn))
915  abort_cleanup_failure = true;
916  else
917  {
918  /* Rollback all remote subtransactions during abort */
919  snprintf(sql, sizeof(sql),
920  "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d",
921  curlevel, curlevel);
922  if (!pgfdw_exec_cleanup_query(entry->conn, sql, false))
923  abort_cleanup_failure = true;
924  }
925 
926  /* Disarm changing_xact_state if it all worked. */
927  entry->changing_xact_state = abort_cleanup_failure;
928  }
929 
930  /* OK, we're outta that level of subtransaction */
931  entry->xact_depth--;
932  }
933 }
int snprintf(char *str, size_t count, const char *fmt,...) pg_attribute_printf(3
#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:394
PGTransactionStatusType PQtransactionStatus(const PGconn *conn)
Definition: fe-connect.c:6052
static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors)
Definition: connection.c:1068
static HTAB * ConnectionHash
Definition: connection.c:66
static bool pgfdw_cancel_query(PGconn *conn)
Definition: connection.c:1021
int GetCurrentTransactionNestLevel(void)
Definition: xact.c:762
bool in_error_recursion_trouble(void)
Definition: elog.c:196
PGconn * conn
Definition: connection.c:51
static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry)
Definition: connection.c:987
static bool xact_got_connection
Definition: connection.c:73
void * hash_seq_search(HASH_SEQ_STATUS *status)
Definition: dynahash.c:1385
void hash_seq_init(HASH_SEQ_STATUS *status, HTAB *hashp)
Definition: dynahash.c:1375
#define elog
Definition: elog.h:219
static void pgfdw_xact_callback ( XactEvent  event,
void *  arg 
)
static

Definition at line 650 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().

651 {
652  HASH_SEQ_STATUS scan;
653  ConnCacheEntry *entry;
654 
655  /* Quick exit if no connections were touched in this transaction. */
656  if (!xact_got_connection)
657  return;
658 
659  /*
660  * Scan all connection cache entries to find open remote transactions, and
661  * close them.
662  */
664  while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
665  {
666  PGresult *res;
667 
668  /* Ignore cache entry if no open connection right now */
669  if (entry->conn == NULL)
670  continue;
671 
672  /* If it has an open remote transaction, try to close it */
673  if (entry->xact_depth > 0)
674  {
675  bool abort_cleanup_failure = false;
676 
677  elog(DEBUG3, "closing remote transaction on connection %p",
678  entry->conn);
679 
680  switch (event)
681  {
684 
685  /*
686  * If abort cleanup previously failed for this connection,
687  * we can't issue any more commands against it.
688  */
690 
691  /* Commit all remote transactions during pre-commit */
692  entry->changing_xact_state = true;
693  do_sql_command(entry->conn, "COMMIT TRANSACTION");
694  entry->changing_xact_state = false;
695 
696  /*
697  * If there were any errors in subtransactions, and we
698  * made prepared statements, do a DEALLOCATE ALL to make
699  * sure we get rid of all prepared statements. This is
700  * annoying and not terribly bulletproof, but it's
701  * probably not worth trying harder.
702  *
703  * DEALLOCATE ALL only exists in 8.3 and later, so this
704  * constrains how old a server postgres_fdw can
705  * communicate with. We intentionally ignore errors in
706  * the DEALLOCATE, so that we can hobble along to some
707  * extent with older servers (leaking prepared statements
708  * as we go; but we don't really support update operations
709  * pre-8.3 anyway).
710  */
711  if (entry->have_prep_stmt && entry->have_error)
712  {
713  res = PQexec(entry->conn, "DEALLOCATE ALL");
714  PQclear(res);
715  }
716  entry->have_prep_stmt = false;
717  entry->have_error = false;
718  break;
720 
721  /*
722  * We disallow remote transactions that modified anything,
723  * since it's not very reasonable to hold them open until
724  * the prepared transaction is committed. For the moment,
725  * throw error unconditionally; later we might allow
726  * read-only cases. Note that the error will cause us to
727  * come right back here with event == XACT_EVENT_ABORT, so
728  * we'll clean up the connection state at that point.
729  */
730  ereport(ERROR,
731  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
732  errmsg("cannot prepare a transaction that modified remote tables")));
733  break;
735  case XACT_EVENT_COMMIT:
736  case XACT_EVENT_PREPARE:
737  /* Pre-commit should have closed the open transaction */
738  elog(ERROR, "missed cleaning up connection during pre-commit");
739  break;
741  case XACT_EVENT_ABORT:
742 
743  /*
744  * Don't try to clean up the connection if we're already
745  * in error recursion trouble.
746  */
748  entry->changing_xact_state = true;
749 
750  /*
751  * If connection is already unsalvageable, don't touch it
752  * further.
753  */
754  if (entry->changing_xact_state)
755  break;
756 
757  /*
758  * Mark this connection as in the process of changing
759  * transaction state.
760  */
761  entry->changing_xact_state = true;
762 
763  /* Assume we might have lost track of prepared statements */
764  entry->have_error = true;
765 
766  /*
767  * If a command has been submitted to the remote server by
768  * using an asynchronous execution function, the command
769  * might not have yet completed. Check to see if a
770  * command is still being processed by the remote server,
771  * and if so, request cancellation of the command.
772  */
773  if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE &&
774  !pgfdw_cancel_query(entry->conn))
775  {
776  /* Unable to cancel running query. */
777  abort_cleanup_failure = true;
778  }
779  else if (!pgfdw_exec_cleanup_query(entry->conn,
780  "ABORT TRANSACTION",
781  false))
782  {
783  /* Unable to abort remote transaction. */
784  abort_cleanup_failure = true;
785  }
786  else if (entry->have_prep_stmt && entry->have_error &&
788  "DEALLOCATE ALL",
789  true))
790  {
791  /* Trouble clearing prepared statements. */
792  abort_cleanup_failure = true;
793  }
794  else
795  {
796  entry->have_prep_stmt = false;
797  entry->have_error = false;
798  }
799 
800  /* Disarm changing_xact_state if it all worked. */
801  entry->changing_xact_state = abort_cleanup_failure;
802  break;
803  }
804  }
805 
806  /* Reset state to show we're out of a transaction */
807  entry->xact_depth = 0;
808 
809  /*
810  * If the connection isn't in a good idle state, discard it to
811  * recover. Next GetConnection will open a new connection.
812  */
813  if (PQstatus(entry->conn) != CONNECTION_OK ||
815  entry->changing_xact_state)
816  {
817  elog(DEBUG3, "discarding connection %p", entry->conn);
818  disconnect_pg_server(entry);
819  }
820  }
821 
822  /*
823  * Regardless of the event type, we can now mark ourselves as out of the
824  * transaction. (Note: if we are here during PRE_COMMIT or PRE_PREPARE,
825  * this saves a useless scan of the hashtable during COMMIT or PREPARE.)
826  */
827  xact_got_connection = false;
828 
829  /* Also reset cursor numbering for next transaction */
830  cursor_number = 0;
831 }
#define DEBUG3
Definition: elog.h:23
int errcode(int sqlerrcode)
Definition: elog.c:575
bool have_prep_stmt
Definition: connection.c:55
#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:394
static unsigned int cursor_number
Definition: connection.c:69
#define ereport(elevel, rest)
Definition: elog.h:122
PGTransactionStatusType PQtransactionStatus(const PGconn *conn)
Definition: fe-connect.c:6052
static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors)
Definition: connection.c:1068
static HTAB * ConnectionHash
Definition: connection.c:66
static bool pgfdw_cancel_query(PGconn *conn)
Definition: connection.c:1021
void PQclear(PGresult *res)
Definition: fe-exec.c:671
bool in_error_recursion_trouble(void)
Definition: elog.c:196
PGconn * conn
Definition: connection.c:51
static void disconnect_pg_server(ConnCacheEntry *entry)
Definition: connection.c:308
static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry)
Definition: connection.c:987
static bool xact_got_connection
Definition: connection.c:73
void * hash_seq_search(HASH_SEQ_STATUS *status)
Definition: dynahash.c:1385
void hash_seq_init(HASH_SEQ_STATUS *status, HTAB *hashp)
Definition: dynahash.c:1375
int errmsg(const char *fmt,...)
Definition: elog.c:797
PGresult * PQexec(PGconn *conn, const char *query)
Definition: fe-exec.c:1897
#define elog
Definition: elog.h:219
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6044
void ReleaseConnection ( PGconn conn)

Definition at line 460 of file connection.c.

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

461 {
462  /*
463  * Currently, we don't actually track connection references because all
464  * cleanup is managed on a transaction or subtransaction basis instead. So
465  * there's nothing to do here.
466  */
467 }

Variable Documentation

HTAB* ConnectionHash = NULL
static

Definition at line 66 of file connection.c.

unsigned int cursor_number = 0
static
unsigned int prep_stmt_number = 0
static

Definition at line 70 of file connection.c.

Referenced by GetPrepStmtNumber().

bool xact_got_connection = false
static

Definition at line 73 of file connection.c.

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