#include "postgres.h"
#include <poll.h>
#include "access/htup_details.h"
#include "access/xact.h"
#include "catalog/pg_user_mapping.h"
#include "commands/defrem.h"
#include "funcapi.h"
#include "libpq/libpq-be.h"
#include "libpq/libpq-be-fe-helpers.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/builtins.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:

Data Structures
struct	ConnCacheEntry

Macros
#define	CONNECTION_CLEANUP_TIMEOUT 30000

#define	CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel)

#define	POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_1 2

#define	POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_2 5

#define	POSTGRES_FDW_GET_CONNECTIONS_COLS 5 /* maximum of above */

Typedefs
typedef Oid	ConnCacheKey

typedef struct ConnCacheEntry	ConnCacheEntry

Enumerations
enum	pgfdwVersion { PGFDW_V1_1 = 0 , PGFDW_V1_2 }

Functions
	PG_FUNCTION_INFO_V1 (postgres_fdw_get_connections)

	PG_FUNCTION_INFO_V1 (postgres_fdw_get_connections_1_2)

	PG_FUNCTION_INFO_V1 (postgres_fdw_disconnect)

	PG_FUNCTION_INFO_V1 (postgres_fdw_disconnect_all)

static void	make_new_connection (ConnCacheEntry entry, UserMapping user)

static PGconn *	connect_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_begin (PGconn conn, const char sql)

static void	do_sql_command_end (PGconn conn, const char sql, bool consume_input)

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 void	pgfdw_reset_xact_state (ConnCacheEntry *entry, bool toplevel)

static bool	pgfdw_cancel_query (PGconn *conn)

static bool	pgfdw_cancel_query_begin (PGconn *conn, TimestampTz endtime)

static bool	pgfdw_cancel_query_end (PGconn *conn, TimestampTz endtime, bool consume_input)

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

static bool	pgfdw_exec_cleanup_query_begin (PGconn conn, const char query)

static bool	pgfdw_exec_cleanup_query_end (PGconn conn, const char query, TimestampTz endtime, bool consume_input, bool ignore_errors)

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

static void	pgfdw_abort_cleanup (ConnCacheEntry *entry, bool toplevel)

static bool	pgfdw_abort_cleanup_begin (ConnCacheEntry entry, bool toplevel, List pending_entries, List *cancel_requested)

static void	pgfdw_finish_pre_commit_cleanup (List *pending_entries)

static void	pgfdw_finish_pre_subcommit_cleanup (List *pending_entries, int curlevel)

static void	pgfdw_finish_abort_cleanup (List pending_entries, List cancel_requested, bool toplevel)

static void	pgfdw_security_check (const char keywords, const char values, UserMapping user, PGconn conn)

static bool	UserMappingPasswordRequired (UserMapping *user)

static bool	disconnect_cached_connections (Oid serverid)

static void	postgres_fdw_get_connections_internal (FunctionCallInfo fcinfo, enum pgfdwVersion api_version)

static int	pgfdw_conn_check (PGconn *conn)

static bool	pgfdw_conn_checkable (void)

PGconn *	GetConnection (UserMapping user, bool will_prep_stmt, PgFdwConnState *state)

void	do_sql_command (PGconn conn, const char sql)

void	ReleaseConnection (PGconn *conn)

unsigned int	GetCursorNumber (PGconn *conn)

unsigned int	GetPrepStmtNumber (PGconn *conn)

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

PGresult *	pgfdw_get_result (PGconn *conn)

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

Datum	postgres_fdw_get_connections_1_2 (PG_FUNCTION_ARGS)

Datum	postgres_fdw_get_connections (PG_FUNCTION_ARGS)

Datum	postgres_fdw_disconnect (PG_FUNCTION_ARGS)

Datum	postgres_fdw_disconnect_all (PG_FUNCTION_ARGS)

Variables
static HTAB *	ConnectionHash = NULL

static unsigned int	cursor_number = 0

static unsigned int	prep_stmt_number = 0

static bool	xact_got_connection = false

static uint32	pgfdw_we_cleanup_result = 0

static uint32	pgfdw_we_connect = 0

static uint32	pgfdw_we_get_result = 0

Macro Definition Documentation

◆ CONNECTION_CLEANUP_TIMEOUT

#define CONNECTION_CLEANUP_TIMEOUT 30000

Definition at line 100 of file connection.c.

◆ CONSTRUCT_ABORT_COMMAND

#define CONSTRUCT_ABORT_COMMAND	(	sql,
		entry,
		toplevel
	)

Value:

    do { \
        if (toplevel) \
            snprintf((sql), sizeof(sql), \
                     "ABORT TRANSACTION"); \
		else \
			snprintf((sql), sizeof(sql), \
                     "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", \
                     (entry)->xact_depth, (entry)->xact_depth); \
    } while(0)

Definition at line 103 of file connection.c.

◆ POSTGRES_FDW_GET_CONNECTIONS_COLS

#define POSTGRES_FDW_GET_CONNECTIONS_COLS 5 /* maximum of above */

Definition at line 2001 of file connection.c.

◆ POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_1

#define POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_1 2

Definition at line 1999 of file connection.c.

◆ POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_2

#define POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_2 5

Definition at line 2000 of file connection.c.

Typedef Documentation

◆ ConnCacheEntry

typedef struct ConnCacheEntry ConnCacheEntry

◆ ConnCacheKey

typedef Oid ConnCacheKey

Definition at line 55 of file connection.c.

Enumeration Type Documentation

◆ pgfdwVersion

enum pgfdwVersion

Enumerator
PGFDW_V1_1
PGFDW_V1_2

Definition at line 117 of file connection.c.

 {
     PGFDW_V1_1 = 0,
     PGFDW_V1_2,
 };

Function Documentation

◆ begin_remote_xact()

static void begin_remote_xact ( ConnCacheEntry * entry )

static

Definition at line 760 of file connection.c.

 {
     int         curlevel = GetCurrentTransactionNestLevel();
  
     /* Start main transaction if we haven't yet */
     if (entry->xact_depth <= 0)
     {
         const char *sql;
  
         elog(DEBUG3, "starting remote transaction on connection %p",
              entry->conn);
  
         if (IsolationIsSerializable())
             sql = "START TRANSACTION ISOLATION LEVEL SERIALIZABLE";
         else
             sql = "START TRANSACTION ISOLATION LEVEL REPEATABLE READ";
         entry->changing_xact_state = true;
         do_sql_command(entry->conn, sql);
         entry->xact_depth = 1;
         entry->changing_xact_state = false;
     }
  
     /*
      * If we're in a subtransaction, stack up savepoints to match our level.
      * This ensures we can rollback just the desired effects when a
      * subtransaction aborts.
      */
     while (entry->xact_depth < curlevel)
     {
         char        sql[64];
  
         snprintf(sql, sizeof(sql), "SAVEPOINT s%d", entry->xact_depth + 1);
         entry->changing_xact_state = true;
         do_sql_command(entry->conn, sql);
         entry->xact_depth++;
         entry->changing_xact_state = false;
     }
 }

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

Referenced by GetConnection().

◆ check_conn_params()

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

static

Definition at line 636 of file connection.c.

 {
     int         i;
  
     /* no check required if superuser */
     if (superuser_arg(user->userid))
         return;
  
 #ifdef ENABLE_GSS
     /* ok if the user provided their own delegated credentials */
     if (be_gssapi_get_delegation(MyProcPort))
         return;
 #endif
  
     /* ok if params contain a non-empty password */
     for (i = 0; keywords[i] != NULL; i++)
     {
         if (strcmp(keywords[i], "password") == 0 && values[i][0] != '\0')
             return;
     }
  
     /* ok if the superuser explicitly said so at user mapping creation time */
     if (!UserMappingPasswordRequired(user))
         return;
  
     ereport(ERROR,
             (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
              errmsg("password or GSSAPI delegated credentials required"),
              errdetail("Non-superusers must delegate GSSAPI credentials or provide a password in the user mapping.")));
 }

References be_gssapi_get_delegation(), ereport, errcode(), errdetail(), errmsg(), ERROR, i, MyProcPort, superuser_arg(), user, UserMappingPasswordRequired(), and values.

Referenced by connect_pg_server().

◆ configure_remote_session()

static void configure_remote_session ( PGconn * conn )

static

Definition at line 679 of file connection.c.

 {
     int         remoteversion = PQserverVersion(conn);
  
     /* Force the search path to contain only pg_catalog (see deparse.c) */
     do_sql_command(conn, "SET search_path = pg_catalog");
  
     /*
      * Set remote timezone; this is basically just cosmetic, since all
      * transmitted and returned timestamptzs should specify a zone explicitly
      * anyway.  However it makes the regression test outputs more predictable.
      *
      * We don't risk setting remote zone equal to ours, since the remote
      * server might use a different timezone database.  Instead, use GMT
      * (quoted, because very old servers are picky about case).  That's
      * guaranteed to work regardless of the remote's timezone database,
      * because pg_tzset() hard-wires it (at least in PG 9.2 and later).
      */
     do_sql_command(conn, "SET timezone = 'GMT'");
  
     /*
      * Set values needed to ensure unambiguous data output from remote.  (This
      * logic should match what pg_dump does.  See also set_transmission_modes
      * in postgres_fdw.c.)
      */
     do_sql_command(conn, "SET datestyle = ISO");
     if (remoteversion >= 80400)
         do_sql_command(conn, "SET intervalstyle = postgres");
     if (remoteversion >= 90000)
         do_sql_command(conn, "SET extra_float_digits = 3");
     else
         do_sql_command(conn, "SET extra_float_digits = 2");
 }

References conn, do_sql_command(), and PQserverVersion().

Referenced by connect_pg_server().

◆ connect_pg_server()

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

static

Definition at line 462 of file connection.c.

 {
     PGconn     *volatile conn = NULL;
  
     /*
      * Use PG_TRY block to ensure closing connection on error.
      */
     PG_TRY();
     {
         const char **keywords;
         const char **values;
         char       *appname = NULL;
         int         n;
  
         /*
          * Construct connection params from generic options of ForeignServer
          * and UserMapping.  (Some of them might not be libpq options, in
          * which case we'll just waste a few array slots.)  Add 4 extra slots
          * for application_name, fallback_application_name, client_encoding,
          * end marker.
          */
         n = list_length(server->options) + list_length(user->options) + 4;
         keywords = (const char **) palloc(n * sizeof(char *));
         values = (const char **) palloc(n * sizeof(char *));
  
         n = 0;
         n += ExtractConnectionOptions(server->options,
                                       keywords + n, values + n);
         n += ExtractConnectionOptions(user->options,
                                       keywords + n, values + n);
  
         /*
          * Use pgfdw_application_name as application_name if set.
          *
          * PQconnectdbParams() processes the parameter arrays from start to
          * end. If any key word is repeated, the last value is used. Therefore
          * note that pgfdw_application_name must be added to the arrays after
          * options of ForeignServer are, so that it can override
          * application_name set in ForeignServer.
          */
         if (pgfdw_application_name && *pgfdw_application_name != '\0')
         {
             keywords[n] = "application_name";
             values[n] = pgfdw_application_name;
             n++;
         }
  
         /*
          * Search the parameter arrays to find application_name setting, and
          * replace escape sequences in it with status information if found.
          * The arrays are searched backwards because the last value is used if
          * application_name is repeatedly set.
          */
         for (int i = n - 1; i >= 0; i--)
         {
             if (strcmp(keywords[i], "application_name") == 0 &&
                 *(values[i]) != '\0')
             {
                 /*
                  * Use this application_name setting if it's not empty string
                  * even after any escape sequences in it are replaced.
                  */
                 appname = process_pgfdw_appname(values[i]);
                 if (appname[0] != '\0')
                 {
                     values[i] = appname;
                     break;
                 }
  
                 /*
                  * This empty application_name is not used, so we set
                  * values[i] to NULL and keep searching the array to find the
                  * next one.
                  */
                 values[i] = NULL;
                 pfree(appname);
                 appname = NULL;
             }
         }
  
         /* Use "postgres_fdw" as fallback_application_name */
         keywords[n] = "fallback_application_name";
         values[n] = "postgres_fdw";
         n++;
  
         /* Set client_encoding so that libpq can convert encoding properly. */
         keywords[n] = "client_encoding";
         values[n] = GetDatabaseEncodingName();
         n++;
  
         keywords[n] = values[n] = NULL;
  
         /* verify the set of connection parameters */
         check_conn_params(keywords, values, user);
  
         /* first time, allocate or get the custom wait event */
         if (pgfdw_we_connect == 0)
             pgfdw_we_connect = WaitEventExtensionNew("PostgresFdwConnect");
  
         /* OK to make connection */
         conn = libpqsrv_connect_params(keywords, values,
                                        false,   /* expand_dbname */
                                        pgfdw_we_connect);
  
         if (!conn || PQstatus(conn) != CONNECTION_OK)
             ereport(ERROR,
                     (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
                      errmsg("could not connect to server \"%s\"",
                             server->servername),
                      errdetail_internal("%s", pchomp(PQerrorMessage(conn)))));
  
         /* Perform post-connection security checks */
         pgfdw_security_check(keywords, values, user, conn);
  
         /* Prepare new session for use */
         configure_remote_session(conn);
  
         if (appname != NULL)
             pfree(appname);
         pfree(keywords);
         pfree(values);
     }
     PG_CATCH();
     {
         libpqsrv_disconnect(conn);
         PG_RE_THROW();
     }
     PG_END_TRY();
  
     return conn;
 }

References check_conn_params(), configure_remote_session(), conn, CONNECTION_OK, ereport, errcode(), errdetail_internal(), errmsg(), ERROR, ExtractConnectionOptions(), GetDatabaseEncodingName(), i, libpqsrv_connect_params(), libpqsrv_disconnect(), list_length(), ForeignServer::options, palloc(), pchomp(), pfree(), PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, pgfdw_application_name, pgfdw_security_check(), pgfdw_we_connect, PQerrorMessage(), PQstatus(), process_pgfdw_appname(), ForeignServer::servername, user, values, and WaitEventExtensionNew().

Referenced by make_new_connection().

◆ disconnect_cached_connections()

static bool disconnect_cached_connections ( Oid serverid )

static

Definition at line 2254 of file connection.c.

 {
     HASH_SEQ_STATUS scan;
     ConnCacheEntry *entry;
     bool        all = !OidIsValid(serverid);
     bool        result = false;
  
     /*
      * Connection cache hashtable has not been initialized yet in this
      * session, so return false.
      */
     if (!ConnectionHash)
         return false;
  
     hash_seq_init(&scan, ConnectionHash);
     while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
     {
         /* Ignore cache entry if no open connection right now. */
         if (!entry->conn)
             continue;
  
         if (all || entry->serverid == serverid)
         {
             /*
              * Emit a warning because the connection to close is used in the
              * current transaction and cannot be disconnected right now.
              */
             if (entry->xact_depth > 0)
             {
                 ForeignServer *server;
  
                 server = GetForeignServerExtended(entry->serverid,
                                                   FSV_MISSING_OK);
  
                 if (!server)
                 {
                     /*
                      * If the foreign server was dropped while its connection
                      * was used in the current transaction, the connection
                      * must have been marked as invalid by
                      * pgfdw_inval_callback at the end of DROP SERVER command.
                      */
                     Assert(entry->invalidated);
  
                     ereport(WARNING,
                             (errmsg("cannot close dropped server connection because it is still in use")));
                 }
                 else
                     ereport(WARNING,
                             (errmsg("cannot close connection for server \"%s\" because it is still in use",
                                     server->servername)));
             }
             else
             {
                 elog(DEBUG3, "discarding connection %p", entry->conn);
                 disconnect_pg_server(entry);
                 result = true;
             }
         }
     }
  
     return result;
 }

References Assert, ConnCacheEntry::conn, ConnectionHash, DEBUG3, disconnect_pg_server(), elog, ereport, errmsg(), FSV_MISSING_OK, GetForeignServerExtended(), hash_seq_init(), hash_seq_search(), ConnCacheEntry::invalidated, OidIsValid, ConnCacheEntry::serverid, ForeignServer::servername, WARNING, and ConnCacheEntry::xact_depth.

Referenced by postgres_fdw_disconnect(), and postgres_fdw_disconnect_all().

◆ disconnect_pg_server()

static void disconnect_pg_server ( ConnCacheEntry * entry )

static

Definition at line 598 of file connection.c.

 {
     if (entry->conn != NULL)
     {
         libpqsrv_disconnect(entry->conn);
         entry->conn = NULL;
     }
 }

References ConnCacheEntry::conn, and libpqsrv_disconnect().

Referenced by disconnect_cached_connections(), GetConnection(), pgfdw_inval_callback(), pgfdw_reject_incomplete_xact_state_change(), and pgfdw_reset_xact_state().

◆ do_sql_command()

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

Definition at line 717 of file connection.c.

 {
     do_sql_command_begin(conn, sql);
     do_sql_command_end(conn, sql, false);
 }

References conn, do_sql_command_begin(), and do_sql_command_end().

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

◆ do_sql_command_begin()

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

static

Definition at line 724 of file connection.c.

 {
     if (!PQsendQuery(conn, sql))
         pgfdw_report_error(ERROR, NULL, conn, false, sql);
 }

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

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

◆ do_sql_command_end()

static void do_sql_command_end	(	PGconn *	conn,
		const char *	sql,
		bool	consume_input
	)

static

Definition at line 731 of file connection.c.

 {
     PGresult   *res;
  
     /*
      * If requested, consume whatever data is available from the socket. (Note
      * that if all data is available, this allows pgfdw_get_result to call
      * PQgetResult without forcing the overhead of WaitLatchOrSocket, which
      * would be large compared to the overhead of PQconsumeInput.)
      */
     if (consume_input && !PQconsumeInput(conn))
         pgfdw_report_error(ERROR, NULL, conn, false, sql);
     res = pgfdw_get_result(conn);
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pgfdw_report_error(ERROR, res, conn, true, sql);
     PQclear(res);
 }

References conn, ERROR, pgfdw_get_result(), pgfdw_report_error(), PGRES_COMMAND_OK, PQclear(), PQconsumeInput(), PQresultStatus(), and res.

Referenced by do_sql_command(), pgfdw_finish_pre_commit_cleanup(), and pgfdw_finish_pre_subcommit_cleanup().

◆ GetConnection()

PGconn* GetConnection	(	UserMapping *	user,
		bool	will_prep_stmt,
		PgFdwConnState **	state
	)

Definition at line 195 of file connection.c.

 {
     bool        found;
     bool        retry = false;
     ConnCacheEntry *entry;
     ConnCacheKey key;
     MemoryContext ccxt = CurrentMemoryContext;
  
     /* First time through, initialize connection cache hashtable */
     if (ConnectionHash == NULL)
     {
         HASHCTL     ctl;
  
         if (pgfdw_we_get_result == 0)
             pgfdw_we_get_result =
                 WaitEventExtensionNew("PostgresFdwGetResult");
  
         ctl.keysize = sizeof(ConnCacheKey);
         ctl.entrysize = sizeof(ConnCacheEntry);
         ConnectionHash = hash_create("postgres_fdw connections", 8,
                                      &ctl,
                                      HASH_ELEM | HASH_BLOBS);
  
         /*
          * Register some callback functions that manage connection cleanup.
          * This should be done just once in each backend.
          */
         RegisterXactCallback(pgfdw_xact_callback, NULL);
         RegisterSubXactCallback(pgfdw_subxact_callback, NULL);
         CacheRegisterSyscacheCallback(FOREIGNSERVEROID,
                                       pgfdw_inval_callback, (Datum) 0);
         CacheRegisterSyscacheCallback(USERMAPPINGOID,
                                       pgfdw_inval_callback, (Datum) 0);
     }
  
     /* Set flag that we did GetConnection during the current transaction */
     xact_got_connection = true;
  
     /* Create hash key for the entry.  Assume no pad bytes in key struct */
     key = user->umid;
  
     /*
      * Find or create cached entry for requested connection.
      */
     entry = hash_search(ConnectionHash, &key, HASH_ENTER, &found);
     if (!found)
     {
         /*
          * We need only clear "conn" here; remaining fields will be filled
          * later when "conn" is set.
          */
         entry->conn = NULL;
     }
  
     /* Reject further use of connections which failed abort cleanup. */
     pgfdw_reject_incomplete_xact_state_change(entry);
  
     /*
      * If the connection needs to be remade due to invalidation, disconnect as
      * soon as we're out of all transactions.
      */
     if (entry->conn != NULL && entry->invalidated && entry->xact_depth == 0)
     {
         elog(DEBUG3, "closing connection %p for option changes to take effect",
              entry->conn);
         disconnect_pg_server(entry);
     }
  
     /*
      * If cache entry doesn't have a connection, we have to establish a new
      * connection.  (If connect_pg_server throws an error, the cache entry
      * will remain in a valid empty state, ie conn == NULL.)
      */
     if (entry->conn == NULL)
         make_new_connection(entry, user);
  
     /*
      * We check the health of the cached connection here when using it.  In
      * cases where we're out of all transactions, if a broken connection is
      * detected, we try to reestablish a new connection later.
      */
     PG_TRY();
     {
         /* Process a pending asynchronous request if any. */
         if (entry->state.pendingAreq)
             process_pending_request(entry->state.pendingAreq);
         /* Start a new transaction or subtransaction if needed. */
         begin_remote_xact(entry);
     }
     PG_CATCH();
     {
         MemoryContext ecxt = MemoryContextSwitchTo(ccxt);
         ErrorData  *errdata = CopyErrorData();
  
         /*
          * Determine whether to try to reestablish the connection.
          *
          * After a broken connection is detected in libpq, any error other
          * than connection failure (e.g., out-of-memory) can be thrown
          * somewhere between return from libpq and the expected ereport() call
          * in pgfdw_report_error(). In this case, since PQstatus() indicates
          * CONNECTION_BAD, checking only PQstatus() causes the false detection
          * of connection failure. To avoid this, we also verify that the
          * error's sqlstate is ERRCODE_CONNECTION_FAILURE. Note that also
          * checking only the sqlstate can cause another false detection
          * because pgfdw_report_error() may report ERRCODE_CONNECTION_FAILURE
          * for any libpq-originated error condition.
          */
         if (errdata->sqlerrcode != ERRCODE_CONNECTION_FAILURE ||
             PQstatus(entry->conn) != CONNECTION_BAD ||
             entry->xact_depth > 0)
         {
             MemoryContextSwitchTo(ecxt);
             PG_RE_THROW();
         }
  
         /* Clean up the error state */
         FlushErrorState();
         FreeErrorData(errdata);
         errdata = NULL;
  
         retry = true;
     }
     PG_END_TRY();
  
     /*
      * If a broken connection is detected, disconnect it, reestablish a new
      * connection and retry a new remote transaction. If connection failure is
      * reported again, we give up getting a connection.
      */
     if (retry)
     {
         Assert(entry->xact_depth == 0);
  
         ereport(DEBUG3,
                 (errmsg_internal("could not start remote transaction on connection %p",
                                  entry->conn)),
                 errdetail_internal("%s", pchomp(PQerrorMessage(entry->conn))));
  
         elog(DEBUG3, "closing connection %p to reestablish a new one",
              entry->conn);
         disconnect_pg_server(entry);
  
         make_new_connection(entry, user);
  
         begin_remote_xact(entry);
     }
  
     /* Remember if caller will prepare statements */
     entry->have_prep_stmt |= will_prep_stmt;
  
     /* If caller needs access to the per-connection state, return it. */
     if (state)
         *state = &entry->state;
  
     return entry->conn;
 }

References Assert, begin_remote_xact(), CacheRegisterSyscacheCallback(), ConnCacheEntry::conn, CONNECTION_BAD, ConnectionHash, CopyErrorData(), ctl, CurrentMemoryContext, DEBUG3, disconnect_pg_server(), elog, ereport, errdetail_internal(), errmsg_internal(), FlushErrorState(), FreeErrorData(), HASH_BLOBS, hash_create(), HASH_ELEM, HASH_ENTER, hash_search(), ConnCacheEntry::have_prep_stmt, ConnCacheEntry::invalidated, sort-test::key, make_new_connection(), MemoryContextSwitchTo(), pchomp(), PgFdwConnState::pendingAreq, PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, pgfdw_inval_callback(), pgfdw_reject_incomplete_xact_state_change(), pgfdw_subxact_callback(), pgfdw_we_get_result, pgfdw_xact_callback(), PQerrorMessage(), PQstatus(), process_pending_request(), RegisterSubXactCallback(), RegisterXactCallback(), ErrorData::sqlerrcode, ConnCacheEntry::state, user, WaitEventExtensionNew(), ConnCacheEntry::xact_depth, and xact_got_connection.

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

◆ GetCursorNumber()

unsigned int GetCursorNumber ( PGconn * conn )

Definition at line 824 of file connection.c.

 {
     return ++cursor_number;
 }

References cursor_number.

Referenced by postgresAcquireSampleRowsFunc(), and postgresBeginForeignScan().

◆ GetPrepStmtNumber()

unsigned int GetPrepStmtNumber ( PGconn * conn )

Definition at line 838 of file connection.c.

 {
     return ++prep_stmt_number;
 }

References prep_stmt_number.

Referenced by prepare_foreign_modify().

◆ make_new_connection()

static void make_new_connection	(	ConnCacheEntry *	entry,
		UserMapping *	user
	)

static

Definition at line 358 of file connection.c.

 {
     ForeignServer *server = GetForeignServer(user->serverid);
     ListCell   *lc;
  
     Assert(entry->conn == NULL);
  
     /* Reset all transient state fields, to be sure all are clean */
     entry->xact_depth = 0;
     entry->have_prep_stmt = false;
     entry->have_error = false;
     entry->changing_xact_state = false;
     entry->invalidated = false;
     entry->serverid = server->serverid;
     entry->server_hashvalue =
         GetSysCacheHashValue1(FOREIGNSERVEROID,
                               ObjectIdGetDatum(server->serverid));
     entry->mapping_hashvalue =
         GetSysCacheHashValue1(USERMAPPINGOID,
                               ObjectIdGetDatum(user->umid));
     memset(&entry->state, 0, sizeof(entry->state));
  
     /*
      * Determine whether to keep the connection that we're about to make here
      * open even after the transaction using it ends, so that the subsequent
      * transactions can re-use it.
      *
      * By default, all the connections to any foreign servers are kept open.
      *
      * Also determine whether to commit/abort (sub)transactions opened on the
      * remote server in parallel at (sub)transaction end, which is disabled by
      * default.
      *
      * Note: it's enough to determine these only when making a new connection
      * because if these settings for it are changed, it will be closed and
      * re-made later.
      */
     entry->keep_connections = true;
     entry->parallel_commit = false;
     entry->parallel_abort = false;
     foreach(lc, server->options)
     {
         DefElem    *def = (DefElem *) lfirst(lc);
  
         if (strcmp(def->defname, "keep_connections") == 0)
             entry->keep_connections = defGetBoolean(def);
         else if (strcmp(def->defname, "parallel_commit") == 0)
             entry->parallel_commit = defGetBoolean(def);
         else if (strcmp(def->defname, "parallel_abort") == 0)
             entry->parallel_abort = defGetBoolean(def);
     }
  
     /* Now try to make the connection */
     entry->conn = connect_pg_server(server, user);
  
     elog(DEBUG3, "new postgres_fdw connection %p for server \"%s\" (user mapping oid %u, userid %u)",
          entry->conn, server->servername, user->umid, user->userid);
 }

References Assert, ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, connect_pg_server(), DEBUG3, defGetBoolean(), DefElem::defname, elog, GetForeignServer(), GetSysCacheHashValue1, ConnCacheEntry::have_error, ConnCacheEntry::have_prep_stmt, ConnCacheEntry::invalidated, ConnCacheEntry::keep_connections, lfirst, ConnCacheEntry::mapping_hashvalue, ObjectIdGetDatum(), ForeignServer::options, ConnCacheEntry::parallel_abort, ConnCacheEntry::parallel_commit, ConnCacheEntry::server_hashvalue, ConnCacheEntry::serverid, ForeignServer::serverid, ForeignServer::servername, ConnCacheEntry::state, user, and ConnCacheEntry::xact_depth.

Referenced by GetConnection().

◆ PG_FUNCTION_INFO_V1() [1/4]

PG_FUNCTION_INFO_V1 ( postgres_fdw_disconnect )

◆ PG_FUNCTION_INFO_V1() [2/4]

PG_FUNCTION_INFO_V1 ( postgres_fdw_disconnect_all )

◆ PG_FUNCTION_INFO_V1() [3/4]

PG_FUNCTION_INFO_V1 ( postgres_fdw_get_connections )

◆ PG_FUNCTION_INFO_V1() [4/4]

PG_FUNCTION_INFO_V1 ( postgres_fdw_get_connections_1_2 )

◆ pgfdw_abort_cleanup()

static void pgfdw_abort_cleanup	(	ConnCacheEntry *	entry,
		bool	toplevel
	)

static

Definition at line 1601 of file connection.c.

 {
     char        sql[100];
  
     /*
      * Don't try to clean up the connection if we're already in error
      * recursion trouble.
      */
     if (in_error_recursion_trouble())
         entry->changing_xact_state = true;
  
     /*
      * If connection is already unsalvageable, don't touch it further.
      */
     if (entry->changing_xact_state)
         return;
  
     /*
      * Mark this connection as in the process of changing transaction state.
      */
     entry->changing_xact_state = true;
  
     /* Assume we might have lost track of prepared statements */
     entry->have_error = true;
  
     /*
      * If a command has been submitted to the remote server by using an
      * asynchronous execution function, the command might not have yet
      * completed.  Check to see if a command is still being processed by the
      * remote server, and if so, request cancellation of the command.
      */
     if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE &&
         !pgfdw_cancel_query(entry->conn))
         return;                 /* Unable to cancel running query */
  
     CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel);
     if (!pgfdw_exec_cleanup_query(entry->conn, sql, false))
         return;                 /* Unable to abort remote (sub)transaction */
  
     if (toplevel)
     {
         if (entry->have_prep_stmt && entry->have_error &&
             !pgfdw_exec_cleanup_query(entry->conn,
                                       "DEALLOCATE ALL",
                                       true))
             return;             /* Trouble clearing prepared statements */
  
         entry->have_prep_stmt = false;
         entry->have_error = false;
     }
  
     /*
      * If pendingAreq of the per-connection state is not NULL, it means that
      * an asynchronous fetch begun by fetch_more_data_begin() was not done
      * successfully and thus the per-connection state was not reset in
      * fetch_more_data(); in that case reset the per-connection state here.
      */
     if (entry->state.pendingAreq)
         memset(&entry->state, 0, sizeof(entry->state));
  
     /* Disarm changing_xact_state if it all worked */
     entry->changing_xact_state = false;
 }

References ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, CONSTRUCT_ABORT_COMMAND, ConnCacheEntry::have_error, ConnCacheEntry::have_prep_stmt, in_error_recursion_trouble(), PgFdwConnState::pendingAreq, pgfdw_cancel_query(), pgfdw_exec_cleanup_query(), PQTRANS_ACTIVE, PQtransactionStatus(), and ConnCacheEntry::state.

Referenced by pgfdw_subxact_callback(), and pgfdw_xact_callback().

◆ pgfdw_abort_cleanup_begin()

static bool pgfdw_abort_cleanup_begin	(	ConnCacheEntry *	entry,
		bool	toplevel,
		List **	pending_entries,
		List **	cancel_requested
	)

static

Definition at line 1675 of file connection.c.

 {
     /*
      * Don't try to clean up the connection if we're already in error
      * recursion trouble.
      */
     if (in_error_recursion_trouble())
         entry->changing_xact_state = true;
  
     /*
      * If connection is already unsalvageable, don't touch it further.
      */
     if (entry->changing_xact_state)
         return false;
  
     /*
      * Mark this connection as in the process of changing transaction state.
      */
     entry->changing_xact_state = true;
  
     /* Assume we might have lost track of prepared statements */
     entry->have_error = true;
  
     /*
      * If a command has been submitted to the remote server by using an
      * asynchronous execution function, the command might not have yet
      * completed.  Check to see if a command is still being processed by the
      * remote server, and if so, request cancellation of the command.
      */
     if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE)
     {
         TimestampTz endtime;
  
         endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(),
                                               CONNECTION_CLEANUP_TIMEOUT);
         if (!pgfdw_cancel_query_begin(entry->conn, endtime))
             return false;       /* Unable to cancel running query */
         *cancel_requested = lappend(*cancel_requested, entry);
     }
     else
     {
         char        sql[100];
  
         CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel);
         if (!pgfdw_exec_cleanup_query_begin(entry->conn, sql))
             return false;       /* Unable to abort remote transaction */
         *pending_entries = lappend(*pending_entries, entry);
     }
  
     return true;
 }

References ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, CONNECTION_CLEANUP_TIMEOUT, CONSTRUCT_ABORT_COMMAND, GetCurrentTimestamp(), ConnCacheEntry::have_error, in_error_recursion_trouble(), lappend(), pgfdw_cancel_query_begin(), pgfdw_exec_cleanup_query_begin(), PQTRANS_ACTIVE, PQtransactionStatus(), and TimestampTzPlusMilliseconds.

Referenced by pgfdw_subxact_callback(), and pgfdw_xact_callback().

◆ pgfdw_cancel_query()

static bool pgfdw_cancel_query ( PGconn * conn )

static

Definition at line 1327 of file connection.c.

 {
     TimestampTz endtime;
  
     /*
      * If it takes too long to cancel the query and discard the result, assume
      * the connection is dead.
      */
     endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(),
                                           CONNECTION_CLEANUP_TIMEOUT);
  
     if (!pgfdw_cancel_query_begin(conn, endtime))
         return false;
     return pgfdw_cancel_query_end(conn, endtime, false);
 }

References conn, CONNECTION_CLEANUP_TIMEOUT, GetCurrentTimestamp(), pgfdw_cancel_query_begin(), pgfdw_cancel_query_end(), and TimestampTzPlusMilliseconds.

Referenced by pgfdw_abort_cleanup().

◆ pgfdw_cancel_query_begin()

static bool pgfdw_cancel_query_begin	(	PGconn *	conn,
		TimestampTz	endtime
	)

static

Definition at line 1353 of file connection.c.

 {
     const char *errormsg = libpqsrv_cancel(conn, endtime);
  
     if (errormsg != NULL)
         ereport(WARNING,
                 errcode(ERRCODE_CONNECTION_FAILURE),
                 errmsg("could not send cancel request: %s", errormsg));
  
     return errormsg == NULL;
 }

References conn, ereport, errcode(), errmsg(), libpqsrv_cancel(), and WARNING.

Referenced by pgfdw_abort_cleanup_begin(), and pgfdw_cancel_query().

◆ pgfdw_cancel_query_end()

static bool pgfdw_cancel_query_end	(	PGconn *	conn,
		TimestampTz	endtime,
		bool	consume_input
	)

static

Definition at line 1366 of file connection.c.

 {
     PGresult   *result = NULL;
     bool        timed_out;
  
     /*
      * If requested, consume whatever data is available from the socket. (Note
      * that if all data is available, this allows pgfdw_get_cleanup_result to
      * call PQgetResult without forcing the overhead of WaitLatchOrSocket,
      * which would be large compared to the overhead of PQconsumeInput.)
      */
     if (consume_input && !PQconsumeInput(conn))
     {
         ereport(WARNING,
                 (errcode(ERRCODE_CONNECTION_FAILURE),
                  errmsg("could not get result of cancel request: %s",
                         pchomp(PQerrorMessage(conn)))));
         return false;
     }
  
     /* Get and discard the result of the query. */
     if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out))
     {
         if (timed_out)
             ereport(WARNING,
                     (errmsg("could not get result of cancel request due to timeout")));
         else
             ereport(WARNING,
                     (errcode(ERRCODE_CONNECTION_FAILURE),
                      errmsg("could not get result of cancel request: %s",
                             pchomp(PQerrorMessage(conn)))));
  
         return false;
     }
     PQclear(result);
  
     return true;
 }

References conn, ereport, errcode(), errmsg(), pchomp(), pgfdw_get_cleanup_result(), PQclear(), PQconsumeInput(), PQerrorMessage(), and WARNING.

Referenced by pgfdw_cancel_query(), and pgfdw_finish_abort_cleanup().

◆ pgfdw_conn_check()

static int pgfdw_conn_check ( PGconn * conn )

static

Definition at line 2326 of file connection.c.

 {
     int         sock = PQsocket(conn);
  
     if (PQstatus(conn) != CONNECTION_OK || sock == -1)
         return -1;
  
 #if (defined(HAVE_POLL) && defined(POLLRDHUP))
     {
         struct pollfd input_fd;
         int         result;
  
         input_fd.fd = sock;
         input_fd.events = POLLRDHUP;
         input_fd.revents = 0;
  
         do
             result = poll(&input_fd, 1, 0);
         while (result < 0 && errno == EINTR);
  
         if (result < 0)
             return -1;
  
         return (input_fd.revents &
                 (POLLRDHUP | POLLHUP | POLLERR | POLLNVAL)) ? 1 : 0;
     }
 #else
     return 0;
 #endif
 }

References conn, CONNECTION_OK, EINTR, PQsocket(), and PQstatus().

Referenced by postgres_fdw_get_connections_internal().

◆ pgfdw_conn_checkable()

static bool pgfdw_conn_checkable ( void )

static

Definition at line 2363 of file connection.c.

 {
 #if (defined(HAVE_POLL) && defined(POLLRDHUP))
     return true;
 #else
     return false;
 #endif
 }

Referenced by postgres_fdw_get_connections_internal().

◆ pgfdw_exec_cleanup_query()

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

static

Definition at line 1418 of file connection.c.

 {
     TimestampTz endtime;
  
     /*
      * If it takes too long to execute a cleanup query, assume the connection
      * is dead.  It's fairly likely that this is why we aborted in the first
      * place (e.g. statement timeout, user cancel), so the timeout shouldn't
      * be too long.
      */
     endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(),
                                           CONNECTION_CLEANUP_TIMEOUT);
  
     if (!pgfdw_exec_cleanup_query_begin(conn, query))
         return false;
     return pgfdw_exec_cleanup_query_end(conn, query, endtime,
                                         false, ignore_errors);
 }

References conn, CONNECTION_CLEANUP_TIMEOUT, GetCurrentTimestamp(), pgfdw_exec_cleanup_query_begin(), pgfdw_exec_cleanup_query_end(), and TimestampTzPlusMilliseconds.

Referenced by pgfdw_abort_cleanup().

◆ pgfdw_exec_cleanup_query_begin()

static bool pgfdw_exec_cleanup_query_begin	(	PGconn *	conn,
		const char *	query
	)

static

Definition at line 1438 of file connection.c.

 {
     Assert(query != NULL);
  
     /*
      * Submit a query.  Since we don't use non-blocking mode, this also can
      * block.  But its risk is relatively small, so we ignore that for now.
      */
     if (!PQsendQuery(conn, query))
     {
         pgfdw_report_error(WARNING, NULL, conn, false, query);
         return false;
     }
  
     return true;
 }

References Assert, conn, pgfdw_report_error(), PQsendQuery(), and WARNING.

Referenced by pgfdw_abort_cleanup_begin(), pgfdw_exec_cleanup_query(), and pgfdw_finish_abort_cleanup().

◆ pgfdw_exec_cleanup_query_end()

static bool pgfdw_exec_cleanup_query_end	(	PGconn *	conn,
		const char *	query,
		TimestampTz	endtime,
		bool	consume_input,
		bool	ignore_errors
	)

static

Definition at line 1456 of file connection.c.

 {
     PGresult   *result = NULL;
     bool        timed_out;
  
     Assert(query != NULL);
  
     /*
      * If requested, consume whatever data is available from the socket. (Note
      * that if all data is available, this allows pgfdw_get_cleanup_result to
      * call PQgetResult without forcing the overhead of WaitLatchOrSocket,
      * which would be large compared to the overhead of PQconsumeInput.)
      */
     if (consume_input && !PQconsumeInput(conn))
     {
         pgfdw_report_error(WARNING, NULL, conn, false, query);
         return false;
     }
  
     /* Get the result of the query. */
     if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out))
     {
         if (timed_out)
             ereport(WARNING,
                     (errmsg("could not get query result due to timeout"),
                      errcontext("remote SQL command: %s", query)));
         else
             pgfdw_report_error(WARNING, NULL, conn, false, query);
  
         return false;
     }
  
     /* Issue a warning if not successful. */
     if (PQresultStatus(result) != PGRES_COMMAND_OK)
     {
         pgfdw_report_error(WARNING, result, conn, true, query);
         return ignore_errors;
     }
     PQclear(result);
  
     return true;
 }

References Assert, conn, ereport, errcontext, errmsg(), pgfdw_get_cleanup_result(), pgfdw_report_error(), PGRES_COMMAND_OK, PQclear(), PQconsumeInput(), PQresultStatus(), and WARNING.

Referenced by pgfdw_exec_cleanup_query(), and pgfdw_finish_abort_cleanup().

◆ pgfdw_exec_query()

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

Definition at line 853 of file connection.c.

 {
     /* First, process a pending asynchronous request, if any. */
     if (state && state->pendingAreq)
         process_pending_request(state->pendingAreq);
  
     if (!PQsendQuery(conn, query))
         return NULL;
     return pgfdw_get_result(conn);
 }

References conn, pgfdw_get_result(), PQsendQuery(), and process_pending_request().

Referenced by close_cursor(), deallocate_query(), fetch_more_data(), get_remote_estimate(), pgfdw_xact_callback(), postgresAcquireSampleRowsFunc(), postgresAnalyzeForeignTable(), postgresGetAnalyzeInfoForForeignTable(), postgresImportForeignSchema(), and postgresReScanForeignScan().

◆ pgfdw_finish_abort_cleanup()

static void pgfdw_finish_abort_cleanup	(	List *	pending_entries,
		List *	cancel_requested,
		bool	toplevel
	)

static

Definition at line 1841 of file connection.c.

 {
     List       *pending_deallocs = NIL;
     ListCell   *lc;
  
     /*
      * For each of the pending cancel requests (if any), get and discard the
      * result of the query, and submit an abort command to the remote server.
      */
     if (cancel_requested)
     {
         foreach(lc, cancel_requested)
         {
             ConnCacheEntry *entry = (ConnCacheEntry *) lfirst(lc);
             TimestampTz endtime;
             char        sql[100];
  
             Assert(entry->changing_xact_state);
  
             /*
              * Set end time.  You might think we should do this before issuing
              * cancel request like in normal mode, but that is problematic,
              * because if, for example, it took longer than 30 seconds to
              * process the first few entries in the cancel_requested list, it
              * would cause a timeout error when processing each of the
              * remaining entries in the list, leading to slamming that entry's
              * connection shut.
              */
             endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(),
                                                   CONNECTION_CLEANUP_TIMEOUT);
  
             if (!pgfdw_cancel_query_end(entry->conn, endtime, true))
             {
                 /* Unable to cancel running query */
                 pgfdw_reset_xact_state(entry, toplevel);
                 continue;
             }
  
             /* Send an abort command in parallel if needed */
             CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel);
             if (!pgfdw_exec_cleanup_query_begin(entry->conn, sql))
             {
                 /* Unable to abort remote (sub)transaction */
                 pgfdw_reset_xact_state(entry, toplevel);
             }
             else
                 pending_entries = lappend(pending_entries, entry);
         }
     }
  
     /* No further work if no pending entries */
     if (!pending_entries)
         return;
  
     /*
      * Get the result of the abort command for each of the pending entries
      */
     foreach(lc, pending_entries)
     {
         ConnCacheEntry *entry = (ConnCacheEntry *) lfirst(lc);
         TimestampTz endtime;
         char        sql[100];
  
         Assert(entry->changing_xact_state);
  
         /*
          * Set end time.  We do this now, not before issuing the command like
          * in normal mode, for the same reason as for the cancel_requested
          * entries.
          */
         endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(),
                                               CONNECTION_CLEANUP_TIMEOUT);
  
         CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel);
         if (!pgfdw_exec_cleanup_query_end(entry->conn, sql, endtime,
                                           true, false))
         {
             /* Unable to abort remote (sub)transaction */
             pgfdw_reset_xact_state(entry, toplevel);
             continue;
         }
  
         if (toplevel)
         {
             /* Do a DEALLOCATE ALL in parallel if needed */
             if (entry->have_prep_stmt && entry->have_error)
             {
                 if (!pgfdw_exec_cleanup_query_begin(entry->conn,
                                                     "DEALLOCATE ALL"))
                 {
                     /* Trouble clearing prepared statements */
                     pgfdw_reset_xact_state(entry, toplevel);
                 }
                 else
                     pending_deallocs = lappend(pending_deallocs, entry);
                 continue;
             }
             entry->have_prep_stmt = false;
             entry->have_error = false;
         }
  
         /* Reset the per-connection state if needed */
         if (entry->state.pendingAreq)
             memset(&entry->state, 0, sizeof(entry->state));
  
         /* We're done with this entry; unset the changing_xact_state flag */
         entry->changing_xact_state = false;
         pgfdw_reset_xact_state(entry, toplevel);
     }
  
     /* No further work if no pending entries */
     if (!pending_deallocs)
         return;
     Assert(toplevel);
  
     /*
      * Get the result of the DEALLOCATE command for each of the pending
      * entries
      */
     foreach(lc, pending_deallocs)
     {
         ConnCacheEntry *entry = (ConnCacheEntry *) lfirst(lc);
         TimestampTz endtime;
  
         Assert(entry->changing_xact_state);
         Assert(entry->have_prep_stmt);
         Assert(entry->have_error);
  
         /*
          * Set end time.  We do this now, not before issuing the command like
          * in normal mode, for the same reason as for the cancel_requested
          * entries.
          */
         endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(),
                                               CONNECTION_CLEANUP_TIMEOUT);
  
         if (!pgfdw_exec_cleanup_query_end(entry->conn, "DEALLOCATE ALL",
                                           endtime, true, true))
         {
             /* Trouble clearing prepared statements */
             pgfdw_reset_xact_state(entry, toplevel);
             continue;
         }
         entry->have_prep_stmt = false;
         entry->have_error = false;
  
         /* Reset the per-connection state if needed */
         if (entry->state.pendingAreq)
             memset(&entry->state, 0, sizeof(entry->state));
  
         /* We're done with this entry; unset the changing_xact_state flag */
         entry->changing_xact_state = false;
         pgfdw_reset_xact_state(entry, toplevel);
     }
 }

References Assert, ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, CONNECTION_CLEANUP_TIMEOUT, CONSTRUCT_ABORT_COMMAND, GetCurrentTimestamp(), ConnCacheEntry::have_error, ConnCacheEntry::have_prep_stmt, lappend(), lfirst, NIL, PgFdwConnState::pendingAreq, pgfdw_cancel_query_end(), pgfdw_exec_cleanup_query_begin(), pgfdw_exec_cleanup_query_end(), pgfdw_reset_xact_state(), ConnCacheEntry::state, and TimestampTzPlusMilliseconds.

Referenced by pgfdw_subxact_callback(), and pgfdw_xact_callback().

◆ pgfdw_finish_pre_commit_cleanup()

static void pgfdw_finish_pre_commit_cleanup ( List * pending_entries )

static

Definition at line 1733 of file connection.c.

 {
     ConnCacheEntry *entry;
     List       *pending_deallocs = NIL;
     ListCell   *lc;
  
     Assert(pending_entries);
  
     /*
      * Get the result of the COMMIT command for each of the pending entries
      */
     foreach(lc, pending_entries)
     {
         entry = (ConnCacheEntry *) lfirst(lc);
  
         Assert(entry->changing_xact_state);
  
         /*
          * We might already have received the result on the socket, so pass
          * consume_input=true to try to consume it first
          */
         do_sql_command_end(entry->conn, "COMMIT TRANSACTION", true);
         entry->changing_xact_state = false;
  
         /* Do a DEALLOCATE ALL in parallel if needed */
         if (entry->have_prep_stmt && entry->have_error)
         {
             /* Ignore errors (see notes in pgfdw_xact_callback) */
             if (PQsendQuery(entry->conn, "DEALLOCATE ALL"))
             {
                 pending_deallocs = lappend(pending_deallocs, entry);
                 continue;
             }
         }
         entry->have_prep_stmt = false;
         entry->have_error = false;
  
         pgfdw_reset_xact_state(entry, true);
     }
  
     /* No further work if no pending entries */
     if (!pending_deallocs)
         return;
  
     /*
      * Get the result of the DEALLOCATE command for each of the pending
      * entries
      */
     foreach(lc, pending_deallocs)
     {
         PGresult   *res;
  
         entry = (ConnCacheEntry *) lfirst(lc);
  
         /* Ignore errors (see notes in pgfdw_xact_callback) */
         while ((res = PQgetResult(entry->conn)) != NULL)
         {
             PQclear(res);
             /* Stop if the connection is lost (else we'll loop infinitely) */
             if (PQstatus(entry->conn) == CONNECTION_BAD)
                 break;
         }
         entry->have_prep_stmt = false;
         entry->have_error = false;
  
         pgfdw_reset_xact_state(entry, true);
     }
 }

References Assert, ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, CONNECTION_BAD, do_sql_command_end(), ConnCacheEntry::have_error, ConnCacheEntry::have_prep_stmt, lappend(), lfirst, NIL, pgfdw_reset_xact_state(), PQclear(), PQgetResult(), PQsendQuery(), PQstatus(), and res.

Referenced by pgfdw_xact_callback().

◆ pgfdw_finish_pre_subcommit_cleanup()

static void pgfdw_finish_pre_subcommit_cleanup	(	List *	pending_entries,
		int	curlevel
	)

static

Definition at line 1807 of file connection.c.

 {
     ConnCacheEntry *entry;
     char        sql[100];
     ListCell   *lc;
  
     Assert(pending_entries);
  
     /*
      * Get the result of the RELEASE command for each of the pending entries
      */
     snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel);
     foreach(lc, pending_entries)
     {
         entry = (ConnCacheEntry *) lfirst(lc);
  
         Assert(entry->changing_xact_state);
  
         /*
          * We might already have received the result on the socket, so pass
          * consume_input=true to try to consume it first
          */
         do_sql_command_end(entry->conn, sql, true);
         entry->changing_xact_state = false;
  
         pgfdw_reset_xact_state(entry, false);
     }
 }

References Assert, ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, do_sql_command_end(), lfirst, pgfdw_reset_xact_state(), and snprintf.

Referenced by pgfdw_subxact_callback().

◆ pgfdw_get_cleanup_result()

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

static

Definition at line 1513 of file connection.c.

 {
     volatile bool failed = false;
     PGresult   *volatile last_res = NULL;
  
     *timed_out = false;
  
     /* In what follows, do not leak any PGresults on an error. */
     PG_TRY();
     {
         for (;;)
         {
             PGresult   *res;
  
             while (PQisBusy(conn))
             {
                 int         wc;
                 TimestampTz now = GetCurrentTimestamp();
                 long        cur_timeout;
  
                 /* If timeout has expired, give up, else get sleep time. */
                 cur_timeout = TimestampDifferenceMilliseconds(now, endtime);
                 if (cur_timeout <= 0)
                 {
                     *timed_out = true;
                     failed = true;
                     goto exit;
                 }
  
                 /* first time, allocate or get the custom wait event */
                 if (pgfdw_we_cleanup_result == 0)
                     pgfdw_we_cleanup_result = WaitEventExtensionNew("PostgresFdwCleanupResult");
  
                 /* Sleep until there's something to do */
                 wc = WaitLatchOrSocket(MyLatch,
                                        WL_LATCH_SET | WL_SOCKET_READABLE |
                                        WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                                        PQsocket(conn),
                                        cur_timeout, pgfdw_we_cleanup_result);
                 ResetLatch(MyLatch);
  
                 CHECK_FOR_INTERRUPTS();
  
                 /* Data available in socket? */
                 if (wc & WL_SOCKET_READABLE)
                 {
                     if (!PQconsumeInput(conn))
                     {
                         /* connection trouble */
                         failed = true;
                         goto exit;
                     }
                 }
             }
  
             res = PQgetResult(conn);
             if (res == NULL)
                 break;          /* query is complete */
  
             PQclear(last_res);
             last_res = res;
         }
 exit:   ;
     }
     PG_CATCH();
     {
         PQclear(last_res);
         PG_RE_THROW();
     }
     PG_END_TRY();
  
     if (failed)
         PQclear(last_res);
     else
         *result = last_res;
     return failed;
 }

References CHECK_FOR_INTERRUPTS, conn, exit(), GetCurrentTimestamp(), MyLatch, now(), PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, pgfdw_we_cleanup_result, PQclear(), PQconsumeInput(), PQgetResult(), PQisBusy(), PQsocket(), res, ResetLatch(), TimestampDifferenceMilliseconds(), WaitEventExtensionNew(), WaitLatchOrSocket(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, WL_SOCKET_READABLE, and WL_TIMEOUT.

Referenced by pgfdw_cancel_query_end(), and pgfdw_exec_cleanup_query_end().

◆ pgfdw_get_result()

PGresult* pgfdw_get_result ( PGconn * conn )

Definition at line 870 of file connection.c.

 {
     return libpqsrv_get_result_last(conn, pgfdw_we_get_result);
 }

References conn, libpqsrv_get_result_last(), and pgfdw_we_get_result.

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

◆ pgfdw_inval_callback()

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

static

Definition at line 1208 of file connection.c.

 {
     HASH_SEQ_STATUS scan;
     ConnCacheEntry *entry;
  
     Assert(cacheid == FOREIGNSERVEROID || cacheid == USERMAPPINGOID);
  
     /* ConnectionHash must exist already, if we're registered */
     hash_seq_init(&scan, ConnectionHash);
     while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
     {
         /* Ignore invalid entries */
         if (entry->conn == NULL)
             continue;
  
         /* hashvalue == 0 means a cache reset, must clear all state */
         if (hashvalue == 0 ||
             (cacheid == FOREIGNSERVEROID &&
              entry->server_hashvalue == hashvalue) ||
             (cacheid == USERMAPPINGOID &&
              entry->mapping_hashvalue == hashvalue))
         {
             /*
              * Close the connection immediately if it's not used yet in this
              * transaction. Otherwise mark it as invalid so that
              * pgfdw_xact_callback() can close it at the end of this
              * transaction.
              */
             if (entry->xact_depth == 0)
             {
                 elog(DEBUG3, "discarding connection %p", entry->conn);
                 disconnect_pg_server(entry);
             }
             else
                 entry->invalidated = true;
         }
     }
 }

References Assert, ConnCacheEntry::conn, ConnectionHash, DEBUG3, disconnect_pg_server(), elog, hash_seq_init(), hash_seq_search(), ConnCacheEntry::invalidated, ConnCacheEntry::mapping_hashvalue, ConnCacheEntry::server_hashvalue, and ConnCacheEntry::xact_depth.

Referenced by GetConnection().

◆ pgfdw_reject_incomplete_xact_state_change()

static void pgfdw_reject_incomplete_xact_state_change ( ConnCacheEntry * entry )

static

Definition at line 1258 of file connection.c.

 {
     ForeignServer *server;
  
     /* nothing to do for inactive entries and entries of sane state */
     if (entry->conn == NULL || !entry->changing_xact_state)
         return;
  
     /* make sure this entry is inactive */
     disconnect_pg_server(entry);
  
     /* find server name to be shown in the message below */
     server = GetForeignServer(entry->serverid);
  
     ereport(ERROR,
             (errcode(ERRCODE_CONNECTION_EXCEPTION),
              errmsg("connection to server \"%s\" was lost",
                     server->servername)));
 }

References ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, disconnect_pg_server(), ereport, errcode(), errmsg(), ERROR, GetForeignServer(), ConnCacheEntry::serverid, and ForeignServer::servername.

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

◆ pgfdw_report_error()

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

Definition at line 889 of file connection.c.

 {
     /* If requested, PGresult must be released before leaving this function. */
     PG_TRY();
     {
         char       *diag_sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
         char       *message_primary = PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY);
         char       *message_detail = PQresultErrorField(res, PG_DIAG_MESSAGE_DETAIL);
         char       *message_hint = PQresultErrorField(res, PG_DIAG_MESSAGE_HINT);
         char       *message_context = PQresultErrorField(res, PG_DIAG_CONTEXT);
         int         sqlstate;
  
         if (diag_sqlstate)
             sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
                                      diag_sqlstate[1],
                                      diag_sqlstate[2],
                                      diag_sqlstate[3],
                                      diag_sqlstate[4]);
         else
             sqlstate = ERRCODE_CONNECTION_FAILURE;
  
         /*
          * If we don't get a message from the PGresult, try the PGconn.  This
          * is needed because for connection-level failures, PQgetResult may
          * just return NULL, not a PGresult at all.
          */
         if (message_primary == NULL)
             message_primary = pchomp(PQerrorMessage(conn));
  
         ereport(elevel,
                 (errcode(sqlstate),
                  (message_primary != NULL && message_primary[0] != '\0') ?
                  errmsg_internal("%s", message_primary) :
                  errmsg("could not obtain message string for remote error"),
                  message_detail ? errdetail_internal("%s", message_detail) : 0,
                  message_hint ? errhint("%s", message_hint) : 0,
                  message_context ? errcontext("%s", message_context) : 0,
                  sql ? errcontext("remote SQL command: %s", sql) : 0));
     }
     PG_FINALLY();
     {
         if (clear)
             PQclear(res);
     }
     PG_END_TRY();
 }

References conn, 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(), PQresultErrorField(), and res.

Referenced by close_cursor(), create_cursor(), deallocate_query(), do_sql_command_begin(), do_sql_command_end(), execute_dml_stmt(), execute_foreign_modify(), fetch_more_data(), fetch_more_data_begin(), get_remote_estimate(), pgfdw_exec_cleanup_query_begin(), pgfdw_exec_cleanup_query_end(), postgresAcquireSampleRowsFunc(), postgresAnalyzeForeignTable(), postgresForeignAsyncNotify(), postgresGetAnalyzeInfoForForeignTable(), postgresImportForeignSchema(), postgresReScanForeignScan(), and prepare_foreign_modify().

◆ pgfdw_reset_xact_state()

static void pgfdw_reset_xact_state	(	ConnCacheEntry *	entry,
		bool	toplevel
	)

static

Definition at line 1282 of file connection.c.

 {
     if (toplevel)
     {
         /* Reset state to show we're out of a transaction */
         entry->xact_depth = 0;
  
         /*
          * If the connection isn't in a good idle state, it is marked as
          * invalid or keep_connections option of its server is disabled, then
          * discard it to recover. Next GetConnection will open a new
          * connection.
          */
         if (PQstatus(entry->conn) != CONNECTION_OK ||
             PQtransactionStatus(entry->conn) != PQTRANS_IDLE ||
             entry->changing_xact_state ||
             entry->invalidated ||
             !entry->keep_connections)
         {
             elog(DEBUG3, "discarding connection %p", entry->conn);
             disconnect_pg_server(entry);
         }
     }
     else
     {
         /* Reset state to show we're out of a subtransaction */
         entry->xact_depth--;
     }
 }

References ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, CONNECTION_OK, DEBUG3, disconnect_pg_server(), elog, ConnCacheEntry::invalidated, ConnCacheEntry::keep_connections, PQstatus(), PQTRANS_IDLE, PQtransactionStatus(), and ConnCacheEntry::xact_depth.

Referenced by pgfdw_finish_abort_cleanup(), pgfdw_finish_pre_commit_cleanup(), pgfdw_finish_pre_subcommit_cleanup(), pgfdw_subxact_callback(), and pgfdw_xact_callback().

◆ pgfdw_security_check()

static void pgfdw_security_check	(	const char **	keywords,
		const char **	values,
		UserMapping *	user,
		PGconn *	conn
	)

static

Definition at line 424 of file connection.c.

 {
     /* Superusers bypass the check */
     if (superuser_arg(user->userid))
         return;
  
 #ifdef ENABLE_GSS
     /* Connected via GSSAPI with delegated credentials- all good. */
     if (PQconnectionUsedGSSAPI(conn) && be_gssapi_get_delegation(MyProcPort))
         return;
 #endif
  
     /* Ok if superuser set PW required false. */
     if (!UserMappingPasswordRequired(user))
         return;
  
     /* Connected via PW, with PW required true, and provided non-empty PW. */
     if (PQconnectionUsedPassword(conn))
     {
         /* ok if params contain a non-empty password */
         for (int i = 0; keywords[i] != NULL; i++)
         {
             if (strcmp(keywords[i], "password") == 0 && values[i][0] != '\0')
                 return;
         }
     }
  
     ereport(ERROR,
             (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
              errmsg("password or GSSAPI delegated credentials required"),
              errdetail("Non-superuser cannot connect if the server does not request a password or use GSSAPI with delegated credentials."),
              errhint("Target server's authentication method must be changed or password_required=false set in the user mapping attributes.")));
 }

References be_gssapi_get_delegation(), conn, ereport, errcode(), errdetail(), errhint(), errmsg(), ERROR, i, MyProcPort, PQconnectionUsedGSSAPI(), PQconnectionUsedPassword(), superuser_arg(), user, UserMappingPasswordRequired(), and values.

Referenced by connect_pg_server().

◆ pgfdw_subxact_callback()

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

static

Definition at line 1095 of file connection.c.

 {
     HASH_SEQ_STATUS scan;
     ConnCacheEntry *entry;
     int         curlevel;
     List       *pending_entries = NIL;
     List       *cancel_requested = NIL;
  
     /* Nothing to do at subxact start, nor after commit. */
     if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB ||
           event == SUBXACT_EVENT_ABORT_SUB))
         return;
  
     /* Quick exit if no connections were touched in this transaction. */
     if (!xact_got_connection)
         return;
  
     /*
      * Scan all connection cache entries to find open remote subtransactions
      * of the current level, and close them.
      */
     curlevel = GetCurrentTransactionNestLevel();
     hash_seq_init(&scan, ConnectionHash);
     while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
     {
         char        sql[100];
  
         /*
          * We only care about connections with open remote subtransactions of
          * the current level.
          */
         if (entry->conn == NULL || entry->xact_depth < curlevel)
             continue;
  
         if (entry->xact_depth > curlevel)
             elog(ERROR, "missed cleaning up remote subtransaction at level %d",
                  entry->xact_depth);
  
         if (event == SUBXACT_EVENT_PRE_COMMIT_SUB)
         {
             /*
              * If abort cleanup previously failed for this connection, we
              * can't issue any more commands against it.
              */
             pgfdw_reject_incomplete_xact_state_change(entry);
  
             /* Commit all remote subtransactions during pre-commit */
             snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel);
             entry->changing_xact_state = true;
             if (entry->parallel_commit)
             {
                 do_sql_command_begin(entry->conn, sql);
                 pending_entries = lappend(pending_entries, entry);
                 continue;
             }
             do_sql_command(entry->conn, sql);
             entry->changing_xact_state = false;
         }
         else
         {
             /* Rollback all remote subtransactions during abort */
             if (entry->parallel_abort)
             {
                 if (pgfdw_abort_cleanup_begin(entry, false,
                                               &pending_entries,
                                               &cancel_requested))
                     continue;
             }
             else
                 pgfdw_abort_cleanup(entry, false);
         }
  
         /* OK, we're outta that level of subtransaction */
         pgfdw_reset_xact_state(entry, false);
     }
  
     /* If there are any pending connections, finish cleaning them up */
     if (pending_entries || cancel_requested)
     {
         if (event == SUBXACT_EVENT_PRE_COMMIT_SUB)
         {
             Assert(cancel_requested == NIL);
             pgfdw_finish_pre_subcommit_cleanup(pending_entries, curlevel);
         }
         else
         {
             Assert(event == SUBXACT_EVENT_ABORT_SUB);
             pgfdw_finish_abort_cleanup(pending_entries, cancel_requested,
                                        false);
         }
     }
 }

References Assert, ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, ConnectionHash, do_sql_command(), do_sql_command_begin(), elog, ERROR, GetCurrentTransactionNestLevel(), hash_seq_init(), hash_seq_search(), lappend(), NIL, ConnCacheEntry::parallel_abort, ConnCacheEntry::parallel_commit, pgfdw_abort_cleanup(), pgfdw_abort_cleanup_begin(), pgfdw_finish_abort_cleanup(), pgfdw_finish_pre_subcommit_cleanup(), pgfdw_reject_incomplete_xact_state_change(), pgfdw_reset_xact_state(), snprintf, SUBXACT_EVENT_ABORT_SUB, SUBXACT_EVENT_PRE_COMMIT_SUB, ConnCacheEntry::xact_depth, and xact_got_connection.

Referenced by GetConnection().

◆ pgfdw_xact_callback()

static void pgfdw_xact_callback	(	XactEvent	event,
		void *	arg
	)

static

Definition at line 945 of file connection.c.

 {
     HASH_SEQ_STATUS scan;
     ConnCacheEntry *entry;
     List       *pending_entries = NIL;
     List       *cancel_requested = NIL;
  
     /* Quick exit if no connections were touched in this transaction. */
     if (!xact_got_connection)
         return;
  
     /*
      * Scan all connection cache entries to find open remote transactions, and
      * close them.
      */
     hash_seq_init(&scan, ConnectionHash);
     while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
     {
         PGresult   *res;
  
         /* Ignore cache entry if no open connection right now */
         if (entry->conn == NULL)
             continue;
  
         /* If it has an open remote transaction, try to close it */
         if (entry->xact_depth > 0)
         {
             elog(DEBUG3, "closing remote transaction on connection %p",
                  entry->conn);
  
             switch (event)
             {
                 case XACT_EVENT_PARALLEL_PRE_COMMIT:
                 case XACT_EVENT_PRE_COMMIT:
  
                     /*
                      * If abort cleanup previously failed for this connection,
                      * we can't issue any more commands against it.
                      */
                     pgfdw_reject_incomplete_xact_state_change(entry);
  
                     /* Commit all remote transactions during pre-commit */
                     entry->changing_xact_state = true;
                     if (entry->parallel_commit)
                     {
                         do_sql_command_begin(entry->conn, "COMMIT TRANSACTION");
                         pending_entries = lappend(pending_entries, entry);
                         continue;
                     }
                     do_sql_command(entry->conn, "COMMIT TRANSACTION");
                     entry->changing_xact_state = false;
  
                     /*
                      * If there were any errors in subtransactions, and we
                      * made prepared statements, do a DEALLOCATE ALL to make
                      * sure we get rid of all prepared statements. This is
                      * annoying and not terribly bulletproof, but it's
                      * probably not worth trying harder.
                      *
                      * DEALLOCATE ALL only exists in 8.3 and later, so this
                      * constrains how old a server postgres_fdw can
                      * communicate with.  We intentionally ignore errors in
                      * the DEALLOCATE, so that we can hobble along to some
                      * extent with older servers (leaking prepared statements
                      * as we go; but we don't really support update operations
                      * pre-8.3 anyway).
                      */
                     if (entry->have_prep_stmt && entry->have_error)
                     {
                         res = pgfdw_exec_query(entry->conn, "DEALLOCATE ALL",
                                                NULL);
                         PQclear(res);
                     }
                     entry->have_prep_stmt = false;
                     entry->have_error = false;
                     break;
                 case XACT_EVENT_PRE_PREPARE:
  
                     /*
                      * We disallow any remote transactions, since it's not
                      * very reasonable to hold them open until the prepared
                      * transaction is committed.  For the moment, throw error
                      * unconditionally; later we might allow read-only cases.
                      * Note that the error will cause us to come right back
                      * here with event == XACT_EVENT_ABORT, so we'll clean up
                      * the connection state at that point.
                      */
                     ereport(ERROR,
                             (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                              errmsg("cannot PREPARE a transaction that has operated on postgres_fdw foreign tables")));
                     break;
                 case XACT_EVENT_PARALLEL_COMMIT:
                 case XACT_EVENT_COMMIT:
                 case XACT_EVENT_PREPARE:
                     /* Pre-commit should have closed the open transaction */
                     elog(ERROR, "missed cleaning up connection during pre-commit");
                     break;
                 case XACT_EVENT_PARALLEL_ABORT:
                 case XACT_EVENT_ABORT:
                     /* Rollback all remote transactions during abort */
                     if (entry->parallel_abort)
                     {
                         if (pgfdw_abort_cleanup_begin(entry, true,
                                                       &pending_entries,
                                                       &cancel_requested))
                             continue;
                     }
                     else
                         pgfdw_abort_cleanup(entry, true);
                     break;
             }
         }
  
         /* Reset state to show we're out of a transaction */
         pgfdw_reset_xact_state(entry, true);
     }
  
     /* If there are any pending connections, finish cleaning them up */
     if (pending_entries || cancel_requested)
     {
         if (event == XACT_EVENT_PARALLEL_PRE_COMMIT ||
             event == XACT_EVENT_PRE_COMMIT)
         {
             Assert(cancel_requested == NIL);
             pgfdw_finish_pre_commit_cleanup(pending_entries);
         }
         else
         {
             Assert(event == XACT_EVENT_PARALLEL_ABORT ||
                    event == XACT_EVENT_ABORT);
             pgfdw_finish_abort_cleanup(pending_entries, cancel_requested,
                                        true);
         }
     }
  
     /*
      * Regardless of the event type, we can now mark ourselves as out of the
      * transaction.  (Note: if we are here during PRE_COMMIT or PRE_PREPARE,
      * this saves a useless scan of the hashtable during COMMIT or PREPARE.)
      */
     xact_got_connection = false;
  
     /* Also reset cursor numbering for next transaction */
     cursor_number = 0;
 }

References Assert, ConnCacheEntry::changing_xact_state, ConnCacheEntry::conn, ConnectionHash, cursor_number, DEBUG3, do_sql_command(), do_sql_command_begin(), elog, ereport, errcode(), errmsg(), ERROR, hash_seq_init(), hash_seq_search(), ConnCacheEntry::have_error, ConnCacheEntry::have_prep_stmt, lappend(), NIL, ConnCacheEntry::parallel_abort, ConnCacheEntry::parallel_commit, pgfdw_abort_cleanup(), pgfdw_abort_cleanup_begin(), pgfdw_exec_query(), pgfdw_finish_abort_cleanup(), pgfdw_finish_pre_commit_cleanup(), pgfdw_reject_incomplete_xact_state_change(), pgfdw_reset_xact_state(), PQclear(), res, 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().

◆ postgres_fdw_disconnect()

Datum postgres_fdw_disconnect ( PG_FUNCTION_ARGS )

Definition at line 2205 of file connection.c.

 {
     ForeignServer *server;
     char       *servername;
  
     servername = text_to_cstring(PG_GETARG_TEXT_PP(0));
     server = GetForeignServerByName(servername, false);
  
     PG_RETURN_BOOL(disconnect_cached_connections(server->serverid));
 }

References disconnect_cached_connections(), GetForeignServerByName(), PG_GETARG_TEXT_PP, PG_RETURN_BOOL, ForeignServer::serverid, and text_to_cstring().

◆ postgres_fdw_disconnect_all()

Datum postgres_fdw_disconnect_all ( PG_FUNCTION_ARGS )

Definition at line 2226 of file connection.c.

 {
     PG_RETURN_BOOL(disconnect_cached_connections(InvalidOid));
 }

References disconnect_cached_connections(), InvalidOid, and PG_RETURN_BOOL.

◆ postgres_fdw_get_connections()

Datum postgres_fdw_get_connections ( PG_FUNCTION_ARGS )

Definition at line 2185 of file connection.c.

 {
     postgres_fdw_get_connections_internal(fcinfo, PGFDW_V1_1);
  
     PG_RETURN_VOID();
 }

References PG_RETURN_VOID, PGFDW_V1_1, and postgres_fdw_get_connections_internal().

◆ postgres_fdw_get_connections_1_2()

Datum postgres_fdw_get_connections_1_2 ( PG_FUNCTION_ARGS )

Definition at line 2177 of file connection.c.

 {
     postgres_fdw_get_connections_internal(fcinfo, PGFDW_V1_2);
  
     PG_RETURN_VOID();
 }

References PG_RETURN_VOID, PGFDW_V1_2, and postgres_fdw_get_connections_internal().

◆ postgres_fdw_get_connections_internal()

static void postgres_fdw_get_connections_internal	(	FunctionCallInfo	fcinfo,
		enum pgfdwVersion	api_version
	)

static

Definition at line 2028 of file connection.c.

 {
     ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
     HASH_SEQ_STATUS scan;
     ConnCacheEntry *entry;
  
     InitMaterializedSRF(fcinfo, 0);
  
     /* If cache doesn't exist, we return no records */
     if (!ConnectionHash)
         return;
  
     /* Check we have the expected number of output arguments */
     switch (rsinfo->setDesc->natts)
     {
         case POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_1:
             if (api_version != PGFDW_V1_1)
                 elog(ERROR, "incorrect number of output arguments");
             break;
         case POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_2:
             if (api_version != PGFDW_V1_2)
                 elog(ERROR, "incorrect number of output arguments");
             break;
         default:
             elog(ERROR, "incorrect number of output arguments");
     }
  
     hash_seq_init(&scan, ConnectionHash);
     while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
     {
         ForeignServer *server;
         Datum       values[POSTGRES_FDW_GET_CONNECTIONS_COLS] = {0};
         bool        nulls[POSTGRES_FDW_GET_CONNECTIONS_COLS] = {0};
         int         i = 0;
  
         /* We only look for open remote connections */
         if (!entry->conn)
             continue;
  
         server = GetForeignServerExtended(entry->serverid, FSV_MISSING_OK);
  
         /*
          * The foreign server may have been dropped in current explicit
          * transaction. It is not possible to drop the server from another
          * session when the connection associated with it is in use in the
          * current transaction, if tried so, the drop query in another session
          * blocks until the current transaction finishes.
          *
          * Even though the server is dropped in the current transaction, the
          * cache can still have associated active connection entry, say we
          * call such connections dangling. Since we can not fetch the server
          * name from system catalogs for dangling connections, instead we show
          * NULL value for server name in output.
          *
          * We could have done better by storing the server name in the cache
          * entry instead of server oid so that it could be used in the output.
          * But the server name in each cache entry requires 64 bytes of
          * memory, which is huge, when there are many cached connections and
          * the use case i.e. dropping the foreign server within the explicit
          * current transaction seems rare. So, we chose to show NULL value for
          * server name in output.
          *
          * Such dangling connections get closed either in next use or at the
          * end of current explicit transaction in pgfdw_xact_callback.
          */
         if (!server)
         {
             /*
              * If the server has been dropped in the current explicit
              * transaction, then this entry would have been invalidated in
              * pgfdw_inval_callback at the end of drop server command. Note
              * that this connection would not have been closed in
              * pgfdw_inval_callback because it is still being used in the
              * current explicit transaction. So, assert that here.
              */
             Assert(entry->conn && entry->xact_depth > 0 && entry->invalidated);
  
             /* Show null, if no server name was found */
             nulls[i++] = true;
         }
         else
             values[i++] = CStringGetTextDatum(server->servername);
  
         if (api_version >= PGFDW_V1_2)
         {
             HeapTuple   tp;
  
             /* Use the system cache to obtain the user mapping */
             tp = SearchSysCache1(USERMAPPINGOID, ObjectIdGetDatum(entry->key));
  
             /*
              * Just like in the foreign server case, user mappings can also be
              * dropped in the current explicit transaction. Therefore, the
              * similar check as in the server case is required.
              */
             if (!HeapTupleIsValid(tp))
             {
                 /*
                  * If we reach here, this entry must have been invalidated in
                  * pgfdw_inval_callback, same as in the server case.
                  */
                 Assert(entry->conn && entry->xact_depth > 0 &&
                        entry->invalidated);
  
                 nulls[i++] = true;
             }
             else
             {
                 Oid         userid;
  
                 userid = ((Form_pg_user_mapping) GETSTRUCT(tp))->umuser;
                 values[i++] = CStringGetTextDatum(MappingUserName(userid));
                 ReleaseSysCache(tp);
             }
         }
  
         values[i++] = BoolGetDatum(!entry->invalidated);
  
         if (api_version >= PGFDW_V1_2)
         {
             bool        check_conn = PG_GETARG_BOOL(0);
  
             /* Is this connection used in the current transaction? */
             values[i++] = BoolGetDatum(entry->xact_depth > 0);
  
             /*
              * If a connection status check is requested and supported, return
              * whether the connection is closed. Otherwise, return NULL.
              */
             if (check_conn && pgfdw_conn_checkable())
                 values[i++] = BoolGetDatum(pgfdw_conn_check(entry->conn) != 0);
             else
                 nulls[i++] = true;
         }
  
         tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls);
     }
 }

References Assert, BoolGetDatum(), ConnectionHash, CStringGetTextDatum, elog, ERROR, FSV_MISSING_OK, GetForeignServerExtended(), GETSTRUCT, hash_seq_init(), hash_seq_search(), HeapTupleIsValid, i, InitMaterializedSRF(), MappingUserName, TupleDescData::natts, ObjectIdGetDatum(), PG_GETARG_BOOL, pgfdw_conn_check(), pgfdw_conn_checkable(), PGFDW_V1_1, PGFDW_V1_2, POSTGRES_FDW_GET_CONNECTIONS_COLS, POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_1, POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_2, ReleaseSysCache(), FunctionCallInfoBaseData::resultinfo, SearchSysCache1(), ForeignServer::servername, ReturnSetInfo::setDesc, ReturnSetInfo::setResult, tuplestore_putvalues(), and values.

Referenced by postgres_fdw_get_connections(), and postgres_fdw_get_connections_1_2().

◆ ReleaseConnection()

void ReleaseConnection ( PGconn * conn )

Definition at line 803 of file connection.c.

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

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

◆ UserMappingPasswordRequired()

static bool UserMappingPasswordRequired ( UserMapping * user )

static

Definition at line 612 of file connection.c.

 {
     ListCell   *cell;
  
     foreach(cell, user->options)
     {
         DefElem    *def = (DefElem *) lfirst(cell);
  
         if (strcmp(def->defname, "password_required") == 0)
             return defGetBoolean(def);
     }
  
     return true;
 }

References defGetBoolean(), DefElem::defname, lfirst, and user.

Referenced by check_conn_params(), and pgfdw_security_check().

Variable Documentation

◆ ConnectionHash

HTAB* ConnectionHash = NULL

static

Definition at line 81 of file connection.c.

Referenced by disconnect_cached_connections(), GetConnection(), pgfdw_inval_callback(), pgfdw_subxact_callback(), pgfdw_xact_callback(), and postgres_fdw_get_connections_internal().

◆ cursor_number

unsigned int cursor_number = 0

static

Definition at line 84 of file connection.c.

Referenced by close_cursor(), GetCursorNumber(), pgfdw_xact_callback(), and postgresAcquireSampleRowsFunc().

◆ pgfdw_we_cleanup_result

uint32 pgfdw_we_cleanup_result = 0

static

Definition at line 91 of file connection.c.

Referenced by pgfdw_get_cleanup_result().

◆ pgfdw_we_connect

uint32 pgfdw_we_connect = 0

static

Definition at line 92 of file connection.c.

Referenced by connect_pg_server().

◆ pgfdw_we_get_result

uint32 pgfdw_we_get_result = 0

static

Definition at line 93 of file connection.c.

Referenced by GetConnection(), and pgfdw_get_result().

◆ prep_stmt_number

unsigned int prep_stmt_number = 0

static

Definition at line 85 of file connection.c.

Referenced by GetPrepStmtNumber().

◆ xact_got_connection

bool xact_got_connection = false

static

Definition at line 88 of file connection.c.

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

Data Structures

Macros

Typedefs

Enumerations

Functions

Variables

Macro Definition Documentation

◆ CONNECTION_CLEANUP_TIMEOUT

◆ CONSTRUCT_ABORT_COMMAND

◆ POSTGRES_FDW_GET_CONNECTIONS_COLS

◆ POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_1

◆ POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_2

Typedef Documentation

◆ ConnCacheEntry

◆ ConnCacheKey

Enumeration Type Documentation

◆ pgfdwVersion

Function Documentation

◆ begin_remote_xact()

◆ check_conn_params()

◆ configure_remote_session()

◆ connect_pg_server()

◆ disconnect_cached_connections()

◆ disconnect_pg_server()

◆ do_sql_command()

◆ do_sql_command_begin()

◆ do_sql_command_end()

◆ GetConnection()

◆ GetCursorNumber()

◆ GetPrepStmtNumber()

◆ make_new_connection()

◆ PG_FUNCTION_INFO_V1() [1/4]

◆ PG_FUNCTION_INFO_V1() [2/4]

◆ PG_FUNCTION_INFO_V1() [3/4]

◆ PG_FUNCTION_INFO_V1() [4/4]

◆ pgfdw_abort_cleanup()

◆ pgfdw_abort_cleanup_begin()

◆ pgfdw_cancel_query()

◆ pgfdw_cancel_query_begin()

◆ pgfdw_cancel_query_end()

◆ pgfdw_conn_check()

◆ pgfdw_conn_checkable()

◆ pgfdw_exec_cleanup_query()

◆ pgfdw_exec_cleanup_query_begin()

◆ pgfdw_exec_cleanup_query_end()

◆ pgfdw_exec_query()

◆ pgfdw_finish_abort_cleanup()

◆ pgfdw_finish_pre_commit_cleanup()

◆ pgfdw_finish_pre_subcommit_cleanup()

◆ pgfdw_get_cleanup_result()

◆ pgfdw_get_result()

◆ pgfdw_inval_callback()

◆ pgfdw_reject_incomplete_xact_state_change()

◆ pgfdw_report_error()

◆ pgfdw_reset_xact_state()

◆ pgfdw_security_check()

◆ pgfdw_subxact_callback()

◆ pgfdw_xact_callback()

◆ postgres_fdw_disconnect()

◆ postgres_fdw_disconnect_all()

◆ postgres_fdw_get_connections()

◆ postgres_fdw_get_connections_1_2()

◆ postgres_fdw_get_connections_internal()

◆ ReleaseConnection()

◆ UserMappingPasswordRequired()

Variable Documentation

◆ ConnectionHash

◆ cursor_number

◆ pgfdw_we_cleanup_result

◆ pgfdw_we_connect

◆ pgfdw_we_get_result

◆ prep_stmt_number

◆ xact_got_connection