100 #define CONNECTION_CLEANUP_TIMEOUT 30000
103 #define CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel) \
106 snprintf((sql), sizeof(sql), \
107 "ABORT TRANSACTION"); \
109 snprintf((sql), sizeof(sql), \
110 "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", \
111 (entry)->xact_depth, (entry)->xact_depth); \
161 PGresult **result,
bool *timed_out);
164 List **pending_entries,
165 List **cancel_requested);
170 List *cancel_requested,
258 elog(
DEBUG3,
"closing connection %p for option changes to take effect",
268 if (entry->
conn == NULL)
303 if (errdata->
sqlerrcode != ERRCODE_CONNECTION_FAILURE ||
330 (
errmsg_internal(
"could not start remote transaction on connection %p",
334 elog(
DEBUG3,
"closing connection %p to reestablish a new one",
378 memset(&entry->
state, 0,
sizeof(entry->
state));
402 if (strcmp(def->
defname,
"keep_connections") == 0)
404 else if (strcmp(def->
defname,
"parallel_commit") == 0)
406 else if (strcmp(def->
defname,
"parallel_abort") == 0)
413 elog(
DEBUG3,
"new postgres_fdw connection %p for server \"%s\" (user mapping oid %u, userid %u)",
444 for (
int i = 0; keywords[
i] != NULL;
i++)
446 if (strcmp(keywords[
i],
"password") == 0 &&
values[
i][0] !=
'\0')
452 (
errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
453 errmsg(
"password or GSSAPI delegated credentials required"),
454 errdetail(
"Non-superuser cannot connect if the server does not request a password or use GSSAPI with delegated credentials."),
455 errhint(
"Target server's authentication method must be changed or password_required=false set in the user mapping attributes.")));
471 const char **keywords;
473 char *appname = NULL;
484 keywords = (
const char **)
palloc(n *
sizeof(
char *));
489 keywords + n,
values + n);
491 keywords + n,
values + n);
504 keywords[n] =
"application_name";
515 for (
int i = n - 1;
i >= 0;
i--)
517 if (strcmp(keywords[
i],
"application_name") == 0 &&
525 if (appname[0] !=
'\0')
543 keywords[n] =
"fallback_application_name";
544 values[n] =
"postgres_fdw";
548 keywords[n] =
"client_encoding";
552 keywords[n] =
values[n] = NULL;
568 (
errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
569 errmsg(
"could not connect to server \"%s\"",
600 if (entry->
conn != NULL)
616 foreach(cell,
user->options)
620 if (strcmp(def->
defname,
"password_required") == 0)
651 for (
i = 0; keywords[
i] != NULL;
i++)
653 if (strcmp(keywords[
i],
"password") == 0 &&
values[
i][0] !=
'\0')
662 (
errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
663 errmsg(
"password or GSSAPI delegated credentials required"),
664 errdetail(
"Non-superusers must delegate GSSAPI credentials or provide a password in the user mapping.")));
705 if (remoteversion >= 80400)
707 if (remoteversion >= 90000)
769 elog(
DEBUG3,
"starting remote transaction on connection %p",
773 sql =
"START TRANSACTION ISOLATION LEVEL SERIALIZABLE";
775 sql =
"START TRANSACTION ISOLATION LEVEL REPEATABLE READ";
890 bool clear,
const char *sql)
909 sqlstate = ERRCODE_CONNECTION_FAILURE;
916 if (message_primary == NULL)
921 (message_primary != NULL && message_primary[0] !=
'\0') ?
923 errmsg(
"could not obtain message string for remote error"),
925 message_hint ?
errhint(
"%s", message_hint) : 0,
926 message_context ?
errcontext(
"%s", message_context) : 0,
927 sql ?
errcontext(
"remote SQL command: %s", sql) : 0));
966 if (entry->
conn == NULL)
972 elog(
DEBUG3,
"closing remote transaction on connection %p",
991 pending_entries =
lappend(pending_entries, entry);
1033 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1034 errmsg(
"cannot PREPARE a transaction that has operated on postgres_fdw foreign tables")));
1040 elog(
ERROR,
"missed cleaning up connection during pre-commit");
1063 if (pending_entries || cancel_requested)
1131 elog(
ERROR,
"missed cleaning up remote subtransaction at level %d",
1143 snprintf(sql,
sizeof(sql),
"RELEASE SAVEPOINT s%d", curlevel);
1148 pending_entries =
lappend(pending_entries, entry);
1173 if (pending_entries || cancel_requested)
1213 Assert(cacheid == FOREIGNSERVEROID || cacheid == USERMAPPINGOID);
1220 if (entry->
conn == NULL)
1224 if (hashvalue == 0 ||
1225 (cacheid == FOREIGNSERVEROID &&
1227 (cacheid == USERMAPPINGOID &&
1273 (
errcode(ERRCODE_CONNECTION_EXCEPTION),
1274 errmsg(
"connection to server \"%s\" was lost",
1357 if (errormsg != NULL)
1359 errcode(ERRCODE_CONNECTION_FAILURE),
1360 errmsg(
"could not send cancel request: %s", errormsg));
1362 return errormsg == NULL;
1380 (
errcode(ERRCODE_CONNECTION_FAILURE),
1381 errmsg(
"could not get result of cancel request: %s",
1391 (
errmsg(
"could not get result of cancel request due to timeout")));
1394 (
errcode(ERRCODE_CONNECTION_FAILURE),
1395 errmsg(
"could not get result of cancel request: %s",
1434 false, ignore_errors);
1482 (
errmsg(
"could not get query result due to timeout"),
1483 errcontext(
"remote SQL command: %s", query)));
1494 return ignore_errors;
1516 volatile bool failed =
false;
1517 PGresult *
volatile last_res = NULL;
1536 if (cur_timeout <= 0)
1659 memset(&entry->
state, 0,
sizeof(entry->
state));
1676 List **pending_entries,
List **cancel_requested)
1713 *cancel_requested =
lappend(*cancel_requested, entry);
1722 *pending_entries =
lappend(*pending_entries, entry);
1744 foreach(lc, pending_entries)
1763 pending_deallocs =
lappend(pending_deallocs, entry);
1774 if (!pending_deallocs)
1781 foreach(lc, pending_deallocs)
1818 snprintf(sql,
sizeof(sql),
"RELEASE SAVEPOINT s%d", curlevel);
1819 foreach(lc, pending_entries)
1851 if (cancel_requested)
1853 foreach(lc, cancel_requested)
1888 pending_entries =
lappend(pending_entries, entry);
1893 if (!pending_entries)
1899 foreach(lc, pending_entries)
1936 pending_deallocs =
lappend(pending_deallocs, entry);
1945 memset(&entry->
state, 0,
sizeof(entry->
state));
1953 if (!pending_deallocs)
1961 foreach(lc, pending_deallocs)
1979 endtime,
true,
true))
1990 memset(&entry->
state, 0,
sizeof(entry->
state));
1999 #define POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_1 2
2000 #define POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_2 5
2001 #define POSTGRES_FDW_GET_CONNECTIONS_COLS 5
2046 elog(
ERROR,
"incorrect number of output arguments");
2050 elog(
ERROR,
"incorrect number of output arguments");
2053 elog(
ERROR,
"incorrect number of output arguments");
2104 Assert(entry->conn && entry->xact_depth > 0 && entry->invalidated);
2130 Assert(entry->conn && entry->xact_depth > 0 &&
2131 entry->invalidated);
2259 bool result =
false;
2275 if (all || entry->
serverid == serverid)
2299 (
errmsg(
"cannot close dropped server connection because it is still in use")));
2303 (
errmsg(
"cannot close connection for server \"%s\" because it is still in use",
2333 #if (defined(HAVE_POLL) && defined(POLLRDHUP))
2335 struct pollfd input_fd;
2339 input_fd.events = POLLRDHUP;
2340 input_fd.revents = 0;
2343 result = poll(&input_fd, 1, 0);
2344 while (result < 0 && errno ==
EINTR);
2349 return (input_fd.revents &
2350 (POLLRDHUP | POLLHUP | POLLERR | POLLNVAL)) ? 1 : 0;
2365 #if (defined(HAVE_POLL) && defined(POLLRDHUP))
long TimestampDifferenceMilliseconds(TimestampTz start_time, TimestampTz stop_time)
TimestampTz GetCurrentTimestamp(void)
Datum now(PG_FUNCTION_ARGS)
bool be_gssapi_get_delegation(Port *port)
static Datum values[MAXATTR]
#define CStringGetTextDatum(s)
#define Assert(condition)
#define OidIsValid(objectId)
static unsigned int prep_stmt_number
unsigned int GetCursorNumber(PGconn *conn)
static bool UserMappingPasswordRequired(UserMapping *user)
Datum postgres_fdw_get_connections(PG_FUNCTION_ARGS)
void do_sql_command(PGconn *conn, const char *sql)
#define POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_2
static void pgfdw_finish_pre_subcommit_cleanup(List *pending_entries, int curlevel)
static void disconnect_pg_server(ConnCacheEntry *entry)
void ReleaseConnection(PGconn *conn)
static uint32 pgfdw_we_get_result
PG_FUNCTION_INFO_V1(postgres_fdw_get_connections)
static bool pgfdw_cancel_query_begin(PGconn *conn, TimestampTz endtime)
static void pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, bool toplevel)
static void pgfdw_reset_xact_state(ConnCacheEntry *entry, bool toplevel)
static int pgfdw_conn_check(PGconn *conn)
#define POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_1
static void configure_remote_session(PGconn *conn)
void pgfdw_report_error(int elevel, PGresult *res, PGconn *conn, bool clear, const char *sql)
PGresult * pgfdw_exec_query(PGconn *conn, const char *query, PgFdwConnState *state)
static bool xact_got_connection
#define POSTGRES_FDW_GET_CONNECTIONS_COLS
struct ConnCacheEntry ConnCacheEntry
Datum postgres_fdw_disconnect_all(PG_FUNCTION_ARGS)
static bool pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime, bool consume_input)
static void do_sql_command_end(PGconn *conn, const char *sql, bool consume_input)
static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result, bool *timed_out)
#define CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel)
static bool pgfdw_conn_checkable(void)
static uint32 pgfdw_we_cleanup_result
static bool pgfdw_abort_cleanup_begin(ConnCacheEntry *entry, bool toplevel, List **pending_entries, List **cancel_requested)
static HTAB * ConnectionHash
static bool pgfdw_exec_cleanup_query_end(PGconn *conn, const char *query, TimestampTz endtime, bool consume_input, bool ignore_errors)
static unsigned int cursor_number
static void make_new_connection(ConnCacheEntry *entry, UserMapping *user)
static void pgfdw_security_check(const char **keywords, const char **values, UserMapping *user, PGconn *conn)
Datum postgres_fdw_disconnect(PG_FUNCTION_ARGS)
static void pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, SubTransactionId parentSubid, void *arg)
static PGconn * connect_pg_server(ForeignServer *server, UserMapping *user)
static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors)
PGconn * GetConnection(UserMapping *user, bool will_prep_stmt, PgFdwConnState **state)
unsigned int GetPrepStmtNumber(PGconn *conn)
Datum postgres_fdw_get_connections_1_2(PG_FUNCTION_ARGS)
static bool pgfdw_exec_cleanup_query_begin(PGconn *conn, const char *query)
static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry)
static void check_conn_params(const char **keywords, const char **values, UserMapping *user)
static uint32 pgfdw_we_connect
static void pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue)
static void pgfdw_xact_callback(XactEvent event, void *arg)
static void postgres_fdw_get_connections_internal(FunctionCallInfo fcinfo, enum pgfdwVersion api_version)
#define CONNECTION_CLEANUP_TIMEOUT
static void do_sql_command_begin(PGconn *conn, const char *sql)
static void pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel)
PGresult * pgfdw_get_result(PGconn *conn)
static void begin_remote_xact(ConnCacheEntry *entry)
static bool pgfdw_cancel_query(PGconn *conn)
static void pgfdw_finish_pre_commit_cleanup(List *pending_entries)
static bool disconnect_cached_connections(Oid serverid)
char * process_pgfdw_appname(const char *appname)
int ExtractConnectionOptions(List *defelems, const char **keywords, const char **values)
char * pgfdw_application_name
static void PGresult * res
bool defGetBoolean(DefElem *def)
void * hash_search(HTAB *hashp, const void *keyPtr, HASHACTION action, bool *foundPtr)
HTAB * hash_create(const char *tabname, long nelem, const HASHCTL *info, int flags)
void * hash_seq_search(HASH_SEQ_STATUS *status)
void hash_seq_init(HASH_SEQ_STATUS *status, HTAB *hashp)
int errmsg_internal(const char *fmt,...)
void FreeErrorData(ErrorData *edata)
int errdetail_internal(const char *fmt,...)
int errdetail(const char *fmt,...)
void FlushErrorState(void)
int errhint(const char *fmt,...)
int errcode(int sqlerrcode)
int errmsg(const char *fmt,...)
bool in_error_recursion_trouble(void)
ErrorData * CopyErrorData(void)
#define MAKE_SQLSTATE(ch1, ch2, ch3, ch4, ch5)
#define ereport(elevel,...)
int PQserverVersion(const PGconn *conn)
PGTransactionStatusType PQtransactionStatus(const PGconn *conn)
int PQconnectionUsedPassword(const PGconn *conn)
int PQconnectionUsedGSSAPI(const PGconn *conn)
char * PQerrorMessage(const PGconn *conn)
ConnStatusType PQstatus(const PGconn *conn)
int PQsocket(const PGconn *conn)
ExecStatusType PQresultStatus(const PGresult *res)
int PQconsumeInput(PGconn *conn)
int PQsendQuery(PGconn *conn, const char *query)
int PQisBusy(PGconn *conn)
char * PQresultErrorField(const PGresult *res, int fieldcode)
PGresult * PQgetResult(PGconn *conn)
#define PG_GETARG_TEXT_PP(n)
#define PG_GETARG_BOOL(n)
#define PG_RETURN_BOOL(x)
ForeignServer * GetForeignServerExtended(Oid serverid, bits16 flags)
ForeignServer * GetForeignServer(Oid serverid)
ForeignServer * GetForeignServerByName(const char *srvname, bool missing_ok)
#define MappingUserName(userid)
void InitMaterializedSRF(FunctionCallInfo fcinfo, bits32 flags)
#define HeapTupleIsValid(tuple)
void CacheRegisterSyscacheCallback(int cacheid, SyscacheCallbackFunction func, Datum arg)
int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
void ResetLatch(Latch *latch)
#define WL_SOCKET_READABLE
#define WL_EXIT_ON_PM_DEATH
static const char * libpqsrv_cancel(PGconn *conn, TimestampTz endtime)
static void libpqsrv_disconnect(PGconn *conn)
static PGconn * libpqsrv_connect_params(const char *const *keywords, const char *const *values, int expand_dbname, uint32 wait_event_info)
static PGresult * libpqsrv_get_result_last(PGconn *conn, uint32 wait_event_info)
List * lappend(List *list, void *datum)
const char * GetDatabaseEncodingName(void)
char * pchomp(const char *in)
void pfree(void *pointer)
MemoryContext CurrentMemoryContext
#define CHECK_FOR_INTERRUPTS()
static int list_length(const List *l)
FormData_pg_user_mapping * Form_pg_user_mapping
static Datum BoolGetDatum(bool X)
static Datum ObjectIdGetDatum(Oid X)
#define PG_DIAG_MESSAGE_HINT
#define PG_DIAG_MESSAGE_PRIMARY
#define PG_DIAG_MESSAGE_DETAIL
void process_pending_request(AsyncRequest *areq)
MemoryContextSwitchTo(old_ctx)
AsyncRequest * pendingAreq
Tuplestorestate * setResult
bool superuser_arg(Oid roleid)
void ReleaseSysCache(HeapTuple tuple)
HeapTuple SearchSysCache1(int cacheId, Datum key1)
#define GetSysCacheHashValue1(cacheId, key1)
void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, const Datum *values, const bool *isnull)
#define TimestampTzPlusMilliseconds(tz, ms)
char * text_to_cstring(const text *t)
uint32 WaitEventExtensionNew(const char *wait_event_name)
int GetCurrentTransactionNestLevel(void)
void RegisterXactCallback(XactCallback callback, void *arg)
void RegisterSubXactCallback(SubXactCallback callback, void *arg)
@ SUBXACT_EVENT_PRE_COMMIT_SUB
@ SUBXACT_EVENT_ABORT_SUB
@ XACT_EVENT_PARALLEL_PRE_COMMIT
@ XACT_EVENT_PARALLEL_COMMIT
@ XACT_EVENT_PARALLEL_ABORT
#define IsolationIsSerializable()