96 #define CONNECTION_CLEANUP_TIMEOUT 30000
99 #define CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel) \
102 snprintf((sql), sizeof(sql), \
103 "ABORT TRANSACTION"); \
105 snprintf((sql), sizeof(sql), \
106 "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", \
107 (entry)->xact_depth, (entry)->xact_depth); \
147 PGresult **result,
bool *timed_out);
150 List **pending_entries,
151 List **cancel_requested);
156 List *cancel_requested,
240 elog(
DEBUG3,
"closing connection %p for option changes to take effect",
250 if (entry->
conn == NULL)
285 if (errdata->
sqlerrcode != ERRCODE_CONNECTION_FAILURE ||
312 (
errmsg_internal(
"could not start remote transaction on connection %p",
316 elog(
DEBUG3,
"closing connection %p to reestablish a new one",
360 memset(&entry->
state, 0,
sizeof(entry->
state));
384 if (strcmp(def->
defname,
"keep_connections") == 0)
386 else if (strcmp(def->
defname,
"parallel_commit") == 0)
388 else if (strcmp(def->
defname,
"parallel_abort") == 0)
395 elog(
DEBUG3,
"new postgres_fdw connection %p for server \"%s\" (user mapping oid %u, userid %u)",
426 for (
int i = 0; keywords[
i] != NULL;
i++)
428 if (strcmp(keywords[
i],
"password") == 0 &&
values[
i][0] !=
'\0')
434 (
errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
435 errmsg(
"password or GSSAPI delegated credentials required"),
436 errdetail(
"Non-superuser cannot connect if the server does not request a password or use GSSAPI with delegated credentials."),
437 errhint(
"Target server's authentication method must be changed or password_required=false set in the user mapping attributes.")));
453 const char **keywords;
455 char *appname = NULL;
466 keywords = (
const char **)
palloc(n *
sizeof(
char *));
471 keywords + n,
values + n);
473 keywords + n,
values + n);
486 keywords[n] =
"application_name";
497 for (
int i = n - 1;
i >= 0;
i--)
499 if (strcmp(keywords[
i],
"application_name") == 0 &&
507 if (appname[0] !=
'\0')
525 keywords[n] =
"fallback_application_name";
526 values[n] =
"postgres_fdw";
530 keywords[n] =
"client_encoding";
534 keywords[n] =
values[n] = NULL;
550 (
errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
551 errmsg(
"could not connect to server \"%s\"",
582 if (entry->
conn != NULL)
598 foreach(cell,
user->options)
602 if (strcmp(def->
defname,
"password_required") == 0)
633 for (
i = 0; keywords[
i] != NULL;
i++)
635 if (strcmp(keywords[
i],
"password") == 0 &&
values[
i][0] !=
'\0')
644 (
errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
645 errmsg(
"password or GSSAPI delegated credentials required"),
646 errdetail(
"Non-superusers must delegate GSSAPI credentials or provide a password in the user mapping.")));
685 if (remoteversion >= 80400)
687 if (remoteversion >= 90000)
749 elog(
DEBUG3,
"starting remote transaction on connection %p",
753 sql =
"START TRANSACTION ISOLATION LEVEL SERIALIZABLE";
755 sql =
"START TRANSACTION ISOLATION LEVEL REPEATABLE READ";
870 bool clear,
const char *sql)
889 sqlstate = ERRCODE_CONNECTION_FAILURE;
896 if (message_primary == NULL)
901 (message_primary != NULL && message_primary[0] !=
'\0') ?
903 errmsg(
"could not obtain message string for remote error"),
905 message_hint ?
errhint(
"%s", message_hint) : 0,
906 message_context ?
errcontext(
"%s", message_context) : 0,
907 sql ?
errcontext(
"remote SQL command: %s", sql) : 0));
946 if (entry->
conn == NULL)
952 elog(
DEBUG3,
"closing remote transaction on connection %p",
971 pending_entries =
lappend(pending_entries, entry);
1013 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1014 errmsg(
"cannot PREPARE a transaction that has operated on postgres_fdw foreign tables")));
1020 elog(
ERROR,
"missed cleaning up connection during pre-commit");
1043 if (pending_entries || cancel_requested)
1111 elog(
ERROR,
"missed cleaning up remote subtransaction at level %d",
1123 snprintf(sql,
sizeof(sql),
"RELEASE SAVEPOINT s%d", curlevel);
1128 pending_entries =
lappend(pending_entries, entry);
1153 if (pending_entries || cancel_requested)
1193 Assert(cacheid == FOREIGNSERVEROID || cacheid == USERMAPPINGOID);
1200 if (entry->
conn == NULL)
1204 if (hashvalue == 0 ||
1205 (cacheid == FOREIGNSERVEROID &&
1207 (cacheid == USERMAPPINGOID &&
1253 (
errcode(ERRCODE_CONNECTION_EXCEPTION),
1254 errmsg(
"connection to server \"%s\" was lost",
1337 if (errormsg != NULL)
1339 errcode(ERRCODE_CONNECTION_FAILURE),
1340 errmsg(
"could not send cancel request: %s", errormsg));
1342 return errormsg == NULL;
1360 (
errcode(ERRCODE_CONNECTION_FAILURE),
1361 errmsg(
"could not get result of cancel request: %s",
1371 (
errmsg(
"could not get result of cancel request due to timeout")));
1374 (
errcode(ERRCODE_CONNECTION_FAILURE),
1375 errmsg(
"could not get result of cancel request: %s",
1414 false, ignore_errors);
1458 (
errmsg(
"could not get query result due to timeout"),
1459 query ?
errcontext(
"remote SQL command: %s", query) : 0));
1470 return ignore_errors;
1492 volatile bool failed =
false;
1493 PGresult *
volatile last_res = NULL;
1512 if (cur_timeout <= 0)
1635 memset(&entry->
state, 0,
sizeof(entry->
state));
1652 List **pending_entries,
List **cancel_requested)
1689 *cancel_requested =
lappend(*cancel_requested, entry);
1698 *pending_entries =
lappend(*pending_entries, entry);
1720 foreach(lc, pending_entries)
1739 pending_deallocs =
lappend(pending_deallocs, entry);
1750 if (!pending_deallocs)
1757 foreach(lc, pending_deallocs)
1794 snprintf(sql,
sizeof(sql),
"RELEASE SAVEPOINT s%d", curlevel);
1795 foreach(lc, pending_entries)
1827 if (cancel_requested)
1829 foreach(lc, cancel_requested)
1864 pending_entries =
lappend(pending_entries, entry);
1869 if (!pending_entries)
1875 foreach(lc, pending_entries)
1912 pending_deallocs =
lappend(pending_deallocs, entry);
1921 memset(&entry->
state, 0,
sizeof(entry->
state));
1929 if (!pending_deallocs)
1937 foreach(lc, pending_deallocs)
1955 endtime,
true,
true))
1966 memset(&entry->
state, 0,
sizeof(entry->
state));
1990 #define POSTGRES_FDW_GET_CONNECTIONS_COLS 2
2048 Assert(entry->conn && entry->xact_depth > 0 && entry->invalidated);
2131 bool result =
false;
2147 if (all || entry->
serverid == serverid)
2171 (
errmsg(
"cannot close dropped server connection because it is still in use")));
2175 (
errmsg(
"cannot close connection for server \"%s\" because it is still in use",
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 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)
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 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 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)
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)
#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_RETURN_BOOL(x)
ForeignServer * GetForeignServerExtended(Oid serverid, bits16 flags)
ForeignServer * GetForeignServer(Oid serverid)
ForeignServer * GetForeignServerByName(const char *srvname, bool missing_ok)
void InitMaterializedSRF(FunctionCallInfo fcinfo, bits32 flags)
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 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)
static char * libpqsrv_cancel(PGconn *conn, TimestampTz endtime)
Assert(fmt[strlen(fmt) - 1] !='\n')
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)
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)
#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()