*** a/contrib/postgres_fdw/connection.c --- b/contrib/postgres_fdw/connection.c *************** *** 22,27 **** --- 22,28 ---- #include "pgstat.h" #include "storage/latch.h" #include "utils/hsearch.h" + #include "utils/inval.h" #include "utils/memutils.h" #include "utils/syscache.h" *************** *** 53,58 **** typedef struct ConnCacheEntry --- 54,62 ---- bool have_prep_stmt; /* have we prepared any stmts in this xact? */ bool have_error; /* have any subxacts aborted in this xact? */ bool changing_xact_state; /* xact state change in process */ + bool invalidated; /* true if reconnect is requried */ + uint32 server_hashvalue; /* hash value of foreign server oid */ + uint32 mapping_hashvalue; /* hash value of user mapping oid */ } ConnCacheEntry; /* *************** *** 69,74 **** static bool xact_got_connection = false; --- 73,79 ---- /* prototypes of private functions */ 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); static void configure_remote_session(PGconn *conn); static void do_sql_command(PGconn *conn, const char *sql); *************** *** 78,83 **** static void pgfdw_subxact_callback(SubXactEvent event, --- 83,89 ---- SubTransactionId mySubid, SubTransactionId parentSubid, void *arg); + static void pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue); static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry); static bool pgfdw_cancel_query(PGconn *conn); static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, *************** *** 130,135 **** GetConnection(UserMapping *user, bool will_prep_stmt) --- 136,145 ---- */ 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 */ *************** *** 144,160 **** GetConnection(UserMapping *user, bool will_prep_stmt) entry = hash_search(ConnectionHash, &key, HASH_ENTER, &found); if (!found) { ! /* initialize new hashtable entry (key is already filled in) */ entry->conn = NULL; - entry->xact_depth = 0; - entry->have_prep_stmt = false; - entry->have_error = false; - entry->changing_xact_state = false; } /* Reject further use of connections which failed abort cleanup. */ pgfdw_reject_incomplete_xact_state_change(entry); /* * We don't check the health of cached connection here, because it would * require some overhead. Broken connection will be detected when the --- 154,182 ---- entry = hash_search(ConnectionHash, &key, HASH_ENTER, &found); if (!found) { ! /* ! * key is already filled in, flags well be initialized at the time of ! * making a new connection, so just clear conn here. ! */ entry->conn = NULL; } /* Reject further use of connections which failed abort cleanup. */ pgfdw_reject_incomplete_xact_state_change(entry); + + /* + * This connection is no longer valid. Disconnect such connections if no + * transaction is running. + */ + if (entry->conn != NULL && entry->invalidated && entry->xact_depth == 0) + { + /* reconneced immediately, so the messages is "reconnecting" */ + elog(DEBUG3, "closing connection %p for option changes to take effect", + entry->conn); + disconnect_pg_server(entry); + } + /* * We don't check the health of cached connection here, because it would * require some overhead. Broken connection will be detected when the *************** *** 173,178 **** GetConnection(UserMapping *user, bool will_prep_stmt) --- 195,206 ---- entry->xact_depth = 0; /* just to be sure */ entry->have_prep_stmt = false; entry->have_error = false; + entry->invalidated = false; + entry->changing_xact_state = false; + entry->server_hashvalue = + GetSysCacheHashValue1(FOREIGNSERVEROID, server->serverid); + entry->mapping_hashvalue = + GetSysCacheHashValue1(USERMAPPINGOID, user->umid); entry->conn = connect_pg_server(server, user); elog(DEBUG3, "new postgres_fdw connection %p for server \"%s\" (user mapping oid %u, userid %u)", *************** *** 276,281 **** connect_pg_server(ForeignServer *server, UserMapping *user) --- 304,321 ---- return conn; } + /* disconnect the connection for a connection cache entry */ + static void + disconnect_pg_server(ConnCacheEntry *entry) + { + if (entry->conn != NULL) + { + PQfinish(entry->conn); + entry->conn = NULL; + } + } + + /* * For non-superusers, insist that the connstr specify a password. This * prevents a password from being picked up from .pgpass, a service file, *************** *** 429,434 **** ReleaseConnection(PGconn *conn) --- 469,519 ---- } /* + * Connection invalidation callback function + * + * Mark the connections to be disconnected if it depends on a foreign server + * or user mapping any options on which have been modified. + * + * Although most invalidate callbacks blow away all the related stuff + * regardless of the given hashvalue, if we blow away all existing connection + * from this server, it can cause a massive number of simultaneous + * reconnection to all the remotes. We restrict the connection to break to + * avoid such a mess. + * + * NB: We could avoid unnecessary disconnection more strictly by examining + * individual option values but it would be too-much for the gain. + */ + static void + pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue) + { + HASH_SEQ_STATUS scan; + ConnCacheEntry *entry; + + if (!ConnectionHash) + return; + + hash_seq_init(&scan, ConnectionHash); + while ((entry = (ConnCacheEntry *) hash_seq_search(&scan))) + { + if (entry->conn == NULL) + continue; + + if (hashvalue == 0) + entry->invalidated = true; + else + { + Assert(cacheid == FOREIGNSERVEROID || cacheid == USERMAPPINGOID); + + if ((cacheid == FOREIGNSERVEROID && + entry->server_hashvalue == hashvalue) || + (cacheid == USERMAPPINGOID && + entry->mapping_hashvalue == hashvalue)) + entry->invalidated = true; + } + } + } + + /* * Assign a "unique" number for a cursor. * * These really only need to be unique per connection within a transaction. *************** *** 777,785 **** pgfdw_xact_callback(XactEvent event, void *arg) entry->changing_xact_state) { elog(DEBUG3, "discarding connection %p", entry->conn); ! PQfinish(entry->conn); ! entry->conn = NULL; ! entry->changing_xact_state = false; } } --- 862,868 ---- entry->changing_xact_state) { elog(DEBUG3, "discarding connection %p", entry->conn); ! disconnect_pg_server(entry); } } *************** *** 913,921 **** pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry) Form_pg_user_mapping umform; ForeignServer *server; ! if (!entry->changing_xact_state) return; tup = SearchSysCache1(USERMAPPINGOID, ObjectIdGetDatum(entry->key)); if (!HeapTupleIsValid(tup)) --- 996,1009 ---- Form_pg_user_mapping umform; 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 */ tup = SearchSysCache1(USERMAPPINGOID, ObjectIdGetDatum(entry->key)); if (!HeapTupleIsValid(tup))