97#define CONNECTION_CLEANUP_TIMEOUT 30000
104#define RETRY_CANCEL_TIMEOUT 1000
107#define CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel) \
110 snprintf((sql), sizeof(sql), \
111 "ABORT TRANSACTION"); \
113 snprintf((sql), sizeof(sql), \
114 "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", \
115 (entry)->xact_depth, (entry)->xact_depth); \
167 PGresult **result,
bool *timed_out);
170 List **pending_entries,
171 List **cancel_requested);
176 List *cancel_requested,
265 elog(
DEBUG3,
"closing connection %p for option changes to take effect",
275 if (entry->
conn == NULL)
310 if (errdata->
sqlerrcode != ERRCODE_CONNECTION_FAILURE ||
337 (
errmsg_internal(
"could not start remote transaction on connection %p",
341 elog(
DEBUG3,
"closing connection %p to reestablish a new one",
385 memset(&entry->
state, 0,
sizeof(entry->
state));
409 if (strcmp(def->
defname,
"keep_connections") == 0)
411 else if (strcmp(def->
defname,
"parallel_commit") == 0)
413 else if (strcmp(def->
defname,
"parallel_abort") == 0)
420 elog(
DEBUG3,
"new postgres_fdw connection %p for server \"%s\" (user mapping oid %u, userid %u)",
459 (
errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
460 errmsg(
"password or GSSAPI delegated credentials required"),
461 errdetail(
"Non-superuser cannot connect if the server does not request a password or use GSSAPI with delegated credentials."),
462 errhint(
"Target server's authentication method must be changed or password_required=false set in the user mapping attributes.")));
480 char *appname = NULL;
522 for (
int i = n - 1;
i >= 0;
i--)
524 if (strcmp(
keywords[
i],
"application_name") == 0 &&
532 if (appname[0] !=
'\0')
550 keywords[n] =
"fallback_application_name";
551 values[n] =
"postgres_fdw";
572 elog(
ERROR,
"could not encode SCRAM client key");
583 elog(
ERROR,
"could not encode SCRAM server key");
607 (
errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
608 errmsg(
"could not connect to server \"%s\"",
643 if (entry->
conn != NULL)
659 foreach(cell,
user->options)
663 if (strcmp(def->
defname,
"password_required") == 0)
679 if (strcmp(def->
defname,
"use_scram_passthrough") == 0)
683 foreach(cell,
user->options)
687 if (strcmp(def->
defname,
"use_scram_passthrough") == 0)
729 (
errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
730 errmsg(
"password or GSSAPI delegated credentials required"),
731 errdetail(
"Non-superusers must delegate GSSAPI credentials, provide a password, or enable SCRAM pass-through in user mapping.")));
772 if (remoteversion >= 80400)
774 if (remoteversion >= 90000)
836 elog(
DEBUG3,
"starting remote transaction on connection %p",
840 sql =
"START TRANSACTION ISOLATION LEVEL SERIALIZABLE";
842 sql =
"START TRANSACTION ISOLATION LEVEL REPEATABLE READ";
957 bool clear,
const char *sql)
976 sqlstate = ERRCODE_CONNECTION_FAILURE;
983 if (message_primary == NULL)
988 (message_primary != NULL && message_primary[0] !=
'\0') ?
990 errmsg(
"could not obtain message string for remote error"),
992 message_hint ?
errhint(
"%s", message_hint) : 0,
993 message_context ?
errcontext(
"%s", message_context) : 0,
994 sql ?
errcontext(
"remote SQL command: %s", sql) : 0));
1033 if (entry->
conn == NULL)
1039 elog(
DEBUG3,
"closing remote transaction on connection %p",
1058 pending_entries =
lappend(pending_entries, entry);
1100 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1101 errmsg(
"cannot PREPARE a transaction that has operated on postgres_fdw foreign tables")));
1107 elog(
ERROR,
"missed cleaning up connection during pre-commit");
1130 if (pending_entries || cancel_requested)
1198 elog(
ERROR,
"missed cleaning up remote subtransaction at level %d",
1210 snprintf(sql,
sizeof(sql),
"RELEASE SAVEPOINT s%d", curlevel);
1215 pending_entries =
lappend(pending_entries, entry);
1240 if (pending_entries || cancel_requested)
1280 Assert(cacheid == FOREIGNSERVEROID || cacheid == USERMAPPINGOID);
1287 if (entry->
conn == NULL)
1291 if (hashvalue == 0 ||
1292 (cacheid == FOREIGNSERVEROID &&
1294 (cacheid == USERMAPPINGOID &&
1340 (
errcode(ERRCODE_CONNECTION_EXCEPTION),
1341 errmsg(
"connection to server \"%s\" was lost",
1431 if (errormsg != NULL)
1433 errcode(ERRCODE_CONNECTION_FAILURE),
1434 errmsg(
"could not send cancel request: %s", errormsg));
1436 return errormsg == NULL;
1455 (
errcode(ERRCODE_CONNECTION_FAILURE),
1456 errmsg(
"could not get result of cancel request: %s",
1463 &result, &timed_out))
1467 (
errmsg(
"could not get result of cancel request due to timeout")));
1470 (
errcode(ERRCODE_CONNECTION_FAILURE),
1471 errmsg(
"could not get result of cancel request: %s",
1510 false, ignore_errors);
1558 (
errmsg(
"could not get query result due to timeout"),
1559 errcontext(
"remote SQL command: %s", query)));
1570 return ignore_errors;
1597 volatile bool failed =
false;
1598 PGresult *
volatile last_res = NULL;
1627 if (
now >= retrycanceltime)
1638 canceldelta += canceldelta;
1645 if (cur_timeout <= 0)
1768 memset(&entry->
state, 0,
sizeof(entry->
state));
1785 List **pending_entries,
List **cancel_requested)
1822 *cancel_requested =
lappend(*cancel_requested, entry);
1831 *pending_entries =
lappend(*pending_entries, entry);
1853 foreach(lc, pending_entries)
1872 pending_deallocs =
lappend(pending_deallocs, entry);
1883 if (!pending_deallocs)
1890 foreach(lc, pending_deallocs)
1927 snprintf(sql,
sizeof(sql),
"RELEASE SAVEPOINT s%d", curlevel);
1928 foreach(lc, pending_entries)
1960 if (cancel_requested)
1962 foreach(lc, cancel_requested)
1987 retrycanceltime,
true))
2002 pending_entries =
lappend(pending_entries, entry);
2007 if (!pending_entries)
2013 foreach(lc, pending_entries)
2050 pending_deallocs =
lappend(pending_deallocs, entry);
2059 memset(&entry->
state, 0,
sizeof(entry->
state));
2067 if (!pending_deallocs)
2075 foreach(lc, pending_deallocs)
2093 endtime,
true,
true))
2104 memset(&entry->
state, 0,
sizeof(entry->
state));
2113#define POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_1 2
2114#define POSTGRES_FDW_GET_CONNECTIONS_COLS_V1_2 5
2115#define POSTGRES_FDW_GET_CONNECTIONS_COLS 5
2160 elog(
ERROR,
"incorrect number of output arguments");
2164 elog(
ERROR,
"incorrect number of output arguments");
2167 elog(
ERROR,
"incorrect number of output arguments");
2218 Assert(entry->conn && entry->xact_depth > 0 && entry->invalidated);
2244 Assert(entry->conn && entry->xact_depth > 0 &&
2245 entry->invalidated);
2373 bool result =
false;
2389 if (all || entry->
serverid == serverid)
2413 (
errmsg(
"cannot close dropped server connection because it is still in use")));
2417 (
errmsg(
"cannot close connection for server \"%s\" because it is still in use",
2447#if (defined(HAVE_POLL) && defined(POLLRDHUP))
2449 struct pollfd input_fd;
2453 input_fd.events = POLLRDHUP;
2454 input_fd.revents = 0;
2457 result = poll(&input_fd, 1, 0);
2458 while (result < 0 && errno ==
EINTR);
2463 return (input_fd.revents &
2464 (POLLRDHUP | POLLHUP | POLLERR | POLLNVAL)) ? 1 : 0;
2479#if (defined(HAVE_POLL) && defined(POLLRDHUP))
long TimestampDifferenceMilliseconds(TimestampTz start_time, TimestampTz stop_time)
TimestampTz GetCurrentTimestamp(void)
Datum now(PG_FUNCTION_ARGS)
int pg_b64_enc_len(int srclen)
int pg_b64_encode(const char *src, int len, char *dst, int dstlen)
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
PGresult * pgfdw_exec_query(PGconn *conn, const char *query, PgFdwConnState *state)
static void pgfdw_finish_pre_subcommit_cleanup(List *pending_entries, int curlevel)
static void disconnect_pg_server(ConnCacheEntry *entry)
void ReleaseConnection(PGconn *conn)
PGconn * GetConnection(UserMapping *user, bool will_prep_stmt, PgFdwConnState **state)
static uint32 pgfdw_we_get_result
PG_FUNCTION_INFO_V1(postgres_fdw_get_connections)
static bool UseScramPassthrough(ForeignServer *server, UserMapping *user)
#define RETRY_CANCEL_TIMEOUT
PGresult * pgfdw_get_result(PGconn *conn)
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)
static bool pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime, TimestampTz retrycanceltime, bool consume_input)
void pgfdw_report_error(int elevel, PGresult *res, PGconn *conn, bool clear, const char *sql)
static bool xact_got_connection
#define POSTGRES_FDW_GET_CONNECTIONS_COLS
struct ConnCacheEntry ConnCacheEntry
Datum postgres_fdw_disconnect_all(PG_FUNCTION_ARGS)
static void do_sql_command_end(PGconn *conn, const char *sql, bool consume_input)
#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)
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)
static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, TimestampTz retrycanceltime, PGresult **result, bool *timed_out)
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)
int ExtractConnectionOptions(List *defelems, const char **keywords, const char **values)
char * pgfdw_application_name
char * process_pgfdw_appname(const char *appname)
static void PGresult * res
bool defGetBoolean(DefElem *def)
void * hash_search(HTAB *hashp, const void *keyPtr, HASHACTION action, bool *foundPtr)
void * hash_seq_search(HASH_SEQ_STATUS *status)
HTAB * hash_create(const char *tabname, long nelem, const HASHCTL *info, int flags)
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,...)
ErrorData * CopyErrorData(void)
void FlushErrorState(void)
int errhint(const char *fmt,...)
int errcode(int sqlerrcode)
int errmsg(const char *fmt,...)
bool in_error_recursion_trouble(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)
ConnStatusType PQstatus(const PGconn *conn)
char * PQerrorMessage(const PGconn *conn)
int PQsocket(const PGconn *conn)
PGresult * PQgetResult(PGconn *conn)
ExecStatusType PQresultStatus(const PGresult *res)
int PQconsumeInput(PGconn *conn)
char * PQresultErrorField(const PGresult *res, int fieldcode)
int PQsendQuery(PGconn *conn, const char *query)
int PQisBusy(PGconn *conn)
#define PG_GETARG_TEXT_PP(n)
#define PG_GETARG_BOOL(n)
#define PG_RETURN_BOOL(x)
ForeignServer * GetForeignServerByName(const char *srvname, bool missing_ok)
ForeignServer * GetForeignServer(Oid serverid)
ForeignServer * GetForeignServerExtended(Oid serverid, bits16 flags)
#define MappingUserName(userid)
void InitMaterializedSRF(FunctionCallInfo fcinfo, bits32 flags)
#define HeapTupleIsValid(tuple)
static void * GETSTRUCT(const HeapTupleData *tuple)
void CacheRegisterSyscacheCallback(int cacheid, SyscacheCallbackFunction func, Datum arg)
static const JsonPathKeyword keywords[]
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 PGconn * libpqsrv_connect_params(const char *const *keywords, const char *const *values, int expand_dbname, uint32 wait_event_info)
static void libpqsrv_disconnect(PGconn *conn)
static PGresult * libpqsrv_get_result_last(PGconn *conn, uint32 wait_event_info)
List * lappend(List *list, void *datum)
const char * GetDatabaseEncodingName(void)
void pfree(void *pointer)
void * palloc0(Size size)
char * pchomp(const char *in)
MemoryContext CurrentMemoryContext
#define CHECK_FOR_INTERRUPTS()
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
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)
AsyncRequest * pendingAreq
uint8 scram_ServerKey[SCRAM_MAX_KEY_LEN]
uint8 scram_ClientKey[SCRAM_MAX_KEY_LEN]
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()