BUG #14738: ALTER SERVER for foregin servers not working

*** a/contrib/postgres_fdw/connection.c
--- b/contrib/postgres_fdw/connection.c
***************
*** 53,58 **** typedef struct ConnCacheEntry
--- 53,61 ----
  	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;
  
  /*
***************
*** 150,160 **** GetConnection(UserMapping *user, bool will_prep_stmt)
--- 153,180 ----
  		entry->have_prep_stmt = false;
  		entry->have_error = false;
  		entry->changing_xact_state = false;
+ 		entry->invalidated = false;
+ 		entry->server_hashvalue = 0;
+ 		entry->mapping_hashvalue = 0;
  	}
  
  	/* 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. We could avoid suporious disconnection by
+ 	 * examining individual option values but it would be too-much for the
+ 	 * gain.
+ 	 */
+ 	if (entry->conn != NULL && entry->invalidated && entry->xact_depth == 0)
+ 	{
+ 		PQfinish(entry->conn);
+ 		entry->conn = NULL;
+ 		entry->invalidated = false;
+ 	}
+ 
  	/*
  	 * 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)
--- 193,202 ----
  		entry->xact_depth = 0;	/* just to be sure */
  		entry->have_prep_stmt = false;
  		entry->have_error = 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)",
***************
*** 429,434 **** ReleaseConnection(PGconn *conn)
--- 453,514 ----
  }
  
  /*
+  * Connection invalidation functions.
+  *
+  * Changes of some options of foreign server or user mapping that a connection
+  * depends on requires the connection to be disconnected at an oppotunity. The
+  * parameter is the hash value of target syscache entry given through syscache
+  * invalidation.
+  */
+ 
+ /* Connection invalidation by modifying foreign server options. */
+ void
+ InvalidateConnectionForServer(uint32 server_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 &&
+ 			entry->server_hashvalue == server_hashvalue)
+ 		{
+ 			entry->invalidated = true;
+ 			hash_seq_term(&scan);
+ 			break;
+ 		}
+ 	}
+ }
+ 
+ /* Connection invalidation by modifying user mapping options. */
+ void
+ InvalidateConnectionForMapping(uint32 mapping_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 &&
+ 			entry->mapping_hashvalue == mapping_hashvalue)
+ 		{
+ 			entry->invalidated = true;
+ 			hash_seq_term(&scan);
+ 			break;
+ 		}
+ 	}
+ }
+ 
+ 
+ /*
   * Assign a "unique" number for a cursor.
   *
   * These really only need to be unique per connection within a transaction.
*** a/contrib/postgres_fdw/postgres_fdw.c
--- b/contrib/postgres_fdw/postgres_fdw.c
***************
*** 36,41 ****
--- 36,43 ----
  #include "parser/parsetree.h"
  #include "utils/builtins.h"
  #include "utils/guc.h"
+ #include "utils/inval.h"
+ #include "utils/syscache.h"
  #include "utils/lsyscache.h"
  #include "utils/memutils.h"
  #include "utils/rel.h"
***************
*** 420,425 **** static void merge_fdw_options(PgFdwRelationInfo *fpinfo,
--- 422,429 ----
  				  const PgFdwRelationInfo *fpinfo_o,
  				  const PgFdwRelationInfo *fpinfo_i);
  
+ void		_PG_init(void);
+ void		_PG_fini(void);
  
  /*
   * Foreign-data wrapper handler function: return a struct with pointers
***************
*** 476,481 **** postgres_fdw_handler(PG_FUNCTION_ARGS)
--- 480,514 ----
  }
  
  /*
+  *  Callback functions for connection invalidation by syscache invalidation
+  */
+ static void
+ postgresServerSysCallback(Datum arg, int cacheid, uint32 hashvalue)
+ {
+ 	InvalidateConnectionForServer(hashvalue);
+ }
+ 
+ static void
+ postgresMappingSysCallback(Datum arg, int cacheid, uint32 hashvalue)
+ {
+ 	InvalidateConnectionForMapping(hashvalue);
+ }
+ 
+ void
+ _PG_init(void)
+ {
+ 	CacheRegisterSyscacheCallback(FOREIGNSERVEROID,
+ 								  postgresServerSysCallback, (Datum)0);
+ 	CacheRegisterSyscacheCallback(USERMAPPINGOID,
+ 								  postgresMappingSysCallback, (Datum)0);
+ }
+ 
+ void
+ _PG_fini(void)
+ {
+ }
+ 
+ /*
   * postgresGetForeignRelSize
   *		Estimate # of rows and width of the result of the scan
   *
*** a/contrib/postgres_fdw/postgres_fdw.h
--- b/contrib/postgres_fdw/postgres_fdw.h
***************
*** 117,122 **** extern void reset_transmission_modes(int nestlevel);
--- 117,124 ----
  /* in connection.c */
  extern PGconn *GetConnection(UserMapping *user, bool will_prep_stmt);
  extern void ReleaseConnection(PGconn *conn);
+ extern void InvalidateConnectionForServer(uint32 server_hashvalue);
+ extern void InvalidateConnectionForMapping(uint32 mapping_hashvalue);
  extern unsigned int GetCursorNumber(PGconn *conn);
  extern unsigned int GetPrepStmtNumber(PGconn *conn);
  extern PGresult *pgfdw_get_result(PGconn *conn, const char *query);

*** a/contrib/postgres_fdw/connection.c
--- b/contrib/postgres_fdw/connection.c
***************
*** 53,58 **** typedef struct ConnCacheEntry
--- 53,61 ----
  	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;
--- 72,78 ----
  
  /* 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);
***************
*** 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
--- 148,176 ----
  	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)
--- 189,200 ----
  		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)
--- 298,315 ----
  	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)
--- 463,527 ----
  }
  
  /*
+  * Connection invalidation functions.
+  *
+  * Changes of some options of foreign server or user mapping that a connection
+  * depends on requires the connection to be disconnected at an oppotunity. The
+  * parameter is the hash value of target syscache entry given through syscache
+  * invalidation.
+  *
+  * NB: We could avoid unnecessary disconnection by examining individual option
+  * values but it would be too-much for the gain.
+  */
+ 
+ /* Connection invalidation by modifying foreign server options. */
+ void
+ InvalidateConnectionForServer(uint32 server_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 &&
+ 			entry->server_hashvalue == server_hashvalue)
+ 		{
+ 			entry->invalidated = true;
+ 			hash_seq_term(&scan);
+ 			break;
+ 		}
+ 	}
+ }
+ 
+ /* Connection invalidation by modifying user mapping options. */
+ void
+ InvalidateConnectionForMapping(uint32 mapping_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 &&
+ 			entry->mapping_hashvalue == mapping_hashvalue)
+ 		{
+ 			entry->invalidated = true;
+ 			hash_seq_term(&scan);
+ 			break;
+ 		}
+ 	}
+ }
+ 
+ 
+ /*
   * 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;
  		}
  	}
  
--- 870,876 ----
  			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))
--- 1004,1017 ----
  	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))
*** a/contrib/postgres_fdw/postgres_fdw.c
--- b/contrib/postgres_fdw/postgres_fdw.c
***************
*** 36,41 ****
--- 36,43 ----
  #include "parser/parsetree.h"
  #include "utils/builtins.h"
  #include "utils/guc.h"
+ #include "utils/inval.h"
+ #include "utils/syscache.h"
  #include "utils/lsyscache.h"
  #include "utils/memutils.h"
  #include "utils/rel.h"
***************
*** 420,425 **** static void merge_fdw_options(PgFdwRelationInfo *fpinfo,
--- 422,429 ----
  				  const PgFdwRelationInfo *fpinfo_o,
  				  const PgFdwRelationInfo *fpinfo_i);
  
+ void		_PG_init(void);
+ void		_PG_fini(void);
  
  /*
   * Foreign-data wrapper handler function: return a struct with pointers
***************
*** 476,481 **** postgres_fdw_handler(PG_FUNCTION_ARGS)
--- 480,514 ----
  }
  
  /*
+  *  Callback functions for connection invalidation by syscache invalidation
+  */
+ static void
+ postgresServerSysCallback(Datum arg, int cacheid, uint32 hashvalue)
+ {
+ 	InvalidateConnectionForServer(hashvalue);
+ }
+ 
+ static void
+ postgresMappingSysCallback(Datum arg, int cacheid, uint32 hashvalue)
+ {
+ 	InvalidateConnectionForMapping(hashvalue);
+ }
+ 
+ void
+ _PG_init(void)
+ {
+ 	CacheRegisterSyscacheCallback(FOREIGNSERVEROID,
+ 								  postgresServerSysCallback, (Datum)0);
+ 	CacheRegisterSyscacheCallback(USERMAPPINGOID,
+ 								  postgresMappingSysCallback, (Datum)0);
+ }
+ 
+ void
+ _PG_fini(void)
+ {
+ }
+ 
+ /*
   * postgresGetForeignRelSize
   *		Estimate # of rows and width of the result of the scan
   *
*** a/contrib/postgres_fdw/postgres_fdw.h
--- b/contrib/postgres_fdw/postgres_fdw.h
***************
*** 117,122 **** extern void reset_transmission_modes(int nestlevel);
--- 117,124 ----
  /* in connection.c */
  extern PGconn *GetConnection(UserMapping *user, bool will_prep_stmt);
  extern void ReleaseConnection(PGconn *conn);
+ extern void InvalidateConnectionForServer(uint32 server_hashvalue);
+ extern void InvalidateConnectionForMapping(uint32 mapping_hashvalue);
  extern unsigned int GetCursorNumber(PGconn *conn);
  extern unsigned int GetPrepStmtNumber(PGconn *conn);
  extern PGresult *pgfdw_get_result(PGconn *conn, const char *query);

*** 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))

From fdb5cbab3375d9d2e4da078cf6ee7eaf7de5c8fd Mon Sep 17 00:00:00 2001
From: Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
Date: Thu, 20 Jul 2017 14:56:51 +0900
Subject: [PATCH 1/2] Add new TAP test library PsqlSession.pm

PostgreNode::psql makes temporary session to run commands so it is not
usable when more interactive operation on a continued session. This
library offers continuous sessions feature that can execute multiple
sql commands separately.
---
 src/test/perl/PsqlSession.pm | 341 +++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 341 insertions(+)
 create mode 100644 src/test/perl/PsqlSession.pm

diff --git a/src/test/perl/PsqlSession.pm b/src/test/perl/PsqlSession.pm
new file mode 100644
index 0000000..d69fd14
--- /dev/null
+++ b/src/test/perl/PsqlSession.pm
@@ -0,0 +1,341 @@
+
+=pod
+
+=head1 NAME
+
+PsqlSession - class representing PostgreSQL psql instance
+
+=head1 SYNOPSIS
+
+  use PsqlSession;
+
+  my $session = get_new_session('session1', $server);
+
+  to connect to a PostgreNode as $server, or
+
+  my $session = get_new_session('session1', 'localhost', '5432', 'postgres');
+
+  to specify each options explicitly.
+
+  # Connect to the server
+  $session->open();
+
+  # Execute an SQL query
+  $ret = $session->execsql('SELECT now();');
+
+  Returns a pair of output of stdout, and stderr in array context.
+
+  ($out, $err) = $session->execsql('SELECT now();');
+
+  $session->execsql_multi('SELECT 1;', 'SELECT now();');
+
+  is just a shortcut of writing many execsqls.
+
+  # End the session
+  $session->close();
+
+=head1 DESCRIPTION
+
+PsqlSession contains a set of routines able to work on a psql session,
+allowing to connect, send a command and receive the result and close.
+
+The IPC::Run module is required.
+
+=cut
+
+package PsqlSession;
+
+use strict;
+use warnings;
+
+use Exporter 'import';
+use Test::More;
+use TestLib ();
+use Scalar::Util qw(blessed);
+
+our @EXPORT = qw(
+  get_new_session
+);
+
+
+=pod
+
+=head1 METHODS
+
+=over
+
+=item PsqlSession::new($class, $name, $pghost, $pgport, $dbname)
+
+Create a new PsqlSession instance. Does not connect.
+
+You should generally prefer to use get_new_session() instead since it
+takes care of finding host name, port number or database name.
+
+=cut
+
+sub new
+{
+	my ($class, $name, $pghost, $pgport, $dbname) = @_;
+
+	my $self = {
+		_name => $name,
+		_host => $pghost,
+		_port => $pgport,
+		_dbname => $dbname };
+
+	bless $self, $class;
+
+#	$self->dump_info;
+
+	return $self;
+}
+
+=pod
+
+=item $session->name()
+
+The name assigned to the session at creation time.
+
+=cut
+
+sub name
+{
+	return $_[0]->{_name};
+}
+
+=pod
+
+=item $session->host()
+
+Return the host (like PGHOST) for this instance. May be a UNIX socket path.
+
+=cut
+
+sub host
+{
+	return $_[0]->{_host};
+}
+
+=pod
+
+=item $session->port()
+
+Get the port number connects to. This won't necessarily be a TCP port
+open on the local host since we prefer to use unix sockets if
+possible.
+
+=cut
+
+sub port
+{
+	return $_[0]->{_port};
+}
+
+=pod
+
+=item $session->dbname()
+
+Get the database name this session connects to.
+
+=cut
+
+sub dbname
+{
+	return $_[0]->{_dbname};
+}
+
+=pod
+
+=item $session->errstate()
+
+Get the error state of this session. 0 means no error and 1 means
+error. This value is reset at the starting of every execution of an
+SQL query.
+
+=cut
+
+sub errstate
+{
+	return $_[0]->{_errstate};
+}
+
+=pod
+
+=item $session->open()
+
+Open this session.
+
+=cut
+
+sub open
+{
+	my ($self) = @_;
+
+	# Create anonymous scalar references to be passed to IPC::Run::start
+	my $in = do {\my $anon};
+	my $out = do {\my $anon};
+	my $err = do {\my $anon};
+
+	# create and store the harness
+	$self->{_run} = IPC::Run::start
+		["psql", $self->dbname, "-p" ,$self->port, "-h", $self->host, "-n"],
+		$in, $out, $err, IPC::Run::timeout(10);
+
+	$self->{_in} = $in;
+	$self->{_out} = $out;
+	$self->{_err} = $err;
+	$self->{_errstate} = 0;
+}
+
+=pod
+
+=item $session->clearerr()
+
+Clear error status.
+
+=cut
+
+sub clearerr
+{
+	$_[0]->errstate = 0;
+}
+
+=pod
+
+=item $session->close()
+
+Close this session. This session can be relaunched by open().
+
+=cut
+
+sub close
+{
+	my ($self) = @_;
+
+	if (defined $self->{_run})
+	{
+		IPC::Run::finish($self->{_run});
+		delete $self->{_run};
+	}
+}
+
+=pod
+
+=item $session->execsql($sqlstr)
+
+Execute sql commands in $sqlstr then returns the
+result. Multistatement is allowed.
+
+In scalar context, this function returns a string that contains the
+result of the query. In array context, returns an array consists of
+the output of stdout and stderr.
+
+For implment reasons, this function adds a sentinel query that gives a
+end marker for reading the result at the end of the query. The stderr
+may contain the log for the sentinel query on certain setting, such as
+log_min_duration_statement = 0.
+
+If the stderr contained the string 'ERROR:', this session is marked as
+error. This state can be checked by $session->errstate();
+
+=cut
+
+sub execsql
+{
+	my ($self, $cmd) = @_;
+	my $run = $self->{_run};
+
+	die if (!defined $run);
+	${$self->{_out}} = '';
+	${$self->{_err}} = '';
+
+	# add a query to receive an end marker
+	$cmd =~ s/[;\n]+$//;
+	${$self->{_in}} .= $cmd.";select '###END###' as end;\n";
+
+	# wait for the end marker or an error
+	$self->{_run}->pump until
+		(${$self->{_out}} =~ /###END###\n\(1 row\)\n/ ||
+		 ${$self->{_err}} =~ /ERROR:/);
+
+	# set error state
+	$self->{_errstate} = 1 if (${$self->{_err}} =~ /ERROR:/);
+
+	# remove useless result
+	my $out = ${$self->{_out}};
+	$out =~ s/ *end *\n-+\n *###END### *\n\(1 row\)\n\n//g;
+
+	return wantarray ? ($out, ${$self->{_err}}) : $out;
+}
+
+=pod
+
+=item $session->execsql_multi("query", "query",...)
+
+Run multiple queries. This is just a convenient function using instead
+of a series of execsql(). Returns concatenated output of each
+query. Returns the output of stdout in scalar context and a pair of
+stdout and stderr in array context.
+
+If any query fails, this function returns immediately.
+
+=cut
+
+sub execsql_multi
+{
+	my ($self) = shift(@_);
+	my ($retout, $reterr, $out, $err);
+
+	foreach my $cmd (@_)
+    {
+		($out, $err) = $self->execsql($cmd);
+		$retout .= $out;
+		$reterr .= $err;
+		last if ($self->errstate);
+	}
+
+	return wantarray ? ($retout, $reterr) : $retout;
+}
+
+=pod
+
+=item get_new_sesion($name, $host, $port, $dbname) or
+     get_new_sesion($name, $PostgreNode)
+
+Build a new session object to connect to the specified
+server. Standalone function that's automatically imported.
+
+In the first form, returns a session according to the parameters. In
+the second form, returns a session connects to the node.
+
+=cut
+
+
+sub get_new_session
+{
+	my ($name, $host_or_ref, $pgport, $dbname) = @_;
+	my $pghost = $host_or_ref;
+
+	if (blessed($host_or_ref) && $host_or_ref->isa('PostgresNode'))
+	{
+		$pghost = $host_or_ref->host;
+		$pgport = $host_or_ref->port;
+	}
+
+	# Fill in default values
+	$pghost = 'localhost' if (! defined $pghost);
+	$pgport = 5432 if (! defined $pgport);
+	$dbname = 'postgres' if (! defined $dbname);
+
+	my $session = new PsqlSession($name, $pghost, $pgport, $dbname);
+
+	return $session;
+}
+
+=pod
+
+=back
+
+=cut
+
+1;
-- 
2.9.2

From c47a3b348c5818e341668b074582eef6b23376ef Mon Sep 17 00:00:00 2001
From: Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
Date: Thu, 20 Jul 2017 15:02:12 +0900
Subject: [PATCH 2/2] Add test for reconnection feature

This patch adds the regression test for the session reconnection
feature of postgres_fdw. make prove-check runs the test.
---
 contrib/postgres_fdw/Makefile              |  3 ++
 contrib/postgres_fdw/t/001_reconnection.pl | 65 ++++++++++++++++++++++++++++++
 2 files changed, 68 insertions(+)
 create mode 100644 contrib/postgres_fdw/t/001_reconnection.pl

diff --git a/contrib/postgres_fdw/Makefile b/contrib/postgres_fdw/Makefile
index 3543312..1be5c9f 100644
--- a/contrib/postgres_fdw/Makefile
+++ b/contrib/postgres_fdw/Makefile
@@ -23,3 +23,6 @@ top_builddir = ../..
 include $(top_builddir)/src/Makefile.global
 include $(top_srcdir)/contrib/contrib-global.mk
 endif
+
+prove-check:
+	$(prove_check)
diff --git a/contrib/postgres_fdw/t/001_reconnection.pl b/contrib/postgres_fdw/t/001_reconnection.pl
new file mode 100644
index 0000000..b93725d
--- /dev/null
+++ b/contrib/postgres_fdw/t/001_reconnection.pl
@@ -0,0 +1,65 @@
+# Minimal test testing reconnection
+use strict;
+use warnings;
+use PostgresNode;
+use PsqlSession;
+use TestLib;
+use Test::More tests => 6;
+
+# start a server
+my $server = get_new_node('server');
+$server->init();
+$server->start;
+my $session1 = get_new_session('session1', $server);
+my $session2 = get_new_session('session2', $server);
+$session1->open;
+$session2->open;
+
+my $ret = $session1->execsql("SELECT current_database()");
+ok($ret =~ /^.*\n-.*-\n *(\S+) *\n/,
+   'retrieving connection setting');
+my $dbname = $1;
+my $port = $server->port;
+
+$session1->execsql_multi(
+	("CREATE EXTENSION postgres_fdw;",
+	 "CREATE SERVER loopback FOREIGN DATA WRAPPER postgres_fdw OPTIONS (dbname \'$dbname\', port \'$port\');",
+	"CREATE USER MAPPING FOR CURRENT_USER SERVER loopback;",
+	"CREATE TABLE lt1 (c1 int);",
+	"INSERT INTO lt1 VALUES (1);",
+	"CREATE FOREIGN TABLE ft1 (c1 int) SERVER loopback OPTIONS (table_name 'lt1');",
+	 "SET client_min_messages to DEBUG3;"));
+ok($session1->errstate == 0, 'setting up');
+
+$session1->execsql("BEGIN;");
+
+my ($stout, $sterr) = $session1->execsql("SELECT c1 FROM ft1 LIMIT 1;");
+ok($sterr =~ /DEBUG: *new postgres_fdw connection 0x[[:xdigit:]]+/,
+   "creating new connection");
+
+$session2->execsql_multi(
+	"ALTER SERVER loopback OPTIONS (ADD host 'hoge')",
+	"ALTER SERVER loopback OPTIONS (DROP host)");
+
+($stout, $sterr) =  $session1->execsql("SELECT c1 FROM ft1 LIMIT 1;");
+ok($sterr !~ /DEBUG: *closing connection 0x[[:xdigit:]]+ for option changes to take effect/,
+   'no disconnection within a transaction');
+
+$session1->execsql("COMMIT;");
+
+$session2->execsql_multi(
+	"ALTER USER MAPPING ft1 OPTIONS (host 'hoge')",
+	"ALTER SERVER loopback OPTIONS (DROP host)");
+
+
+($stout, $sterr) = $session1->execsql("SELECT c1 FROM ft1 LIMIT 1;");
+ok($sterr =~ /DEBUG: *closing connection 0x[[:xdigit:]]+ for option changes to take effect\nDEBUG: *new postgres_fdw connection 0x[[:xdigit:]]+/,
+   'reconnection by option change outside a transactin');
+
+($stout, $sterr) =  $session1->execsql("SELECT c1 FROM ft1 LIMIT 1;");
+ok($sterr !~ /DEBUG: *closing connection 0x[[:xdigit:]]+ for option changes to take effect/,
+   'no disconnection without option change');
+
+$session1->close;
+$session2->close;
+$server->stop;
-- 
2.9.2

diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c
index 8c33dea..8eb477b 100644
*** 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"
  
*************** typedef struct ConnCacheEntry
*** 48,58 ****
--- 49,63 ----
  {
  	ConnCacheKey key;			/* hash key (must be first) */
  	PGconn	   *conn;			/* connection to foreign server, or NULL */
+ 	/* Remaining fields are invalid when conn is NULL: */
  	int			xact_depth;		/* 0 = no xact open, 1 = main xact open, 2 =
  								 * one level of subxact open, etc */
  	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 pending */
+ 	uint32		server_hashvalue;	/* hash value of foreign server OID */
+ 	uint32		mapping_hashvalue;	/* hash value of user mapping OID */
  } ConnCacheEntry;
  
  /*
*************** static bool xact_got_connection = false;
*** 69,74 ****
--- 74,80 ----
  
  /* 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);
*************** static void pgfdw_subxact_callback(SubXa
*** 78,83 ****
--- 84,90 ----
  					   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,
*************** GetConnection(UserMapping *user, bool wi
*** 130,135 ****
--- 137,146 ----
  		 */
  		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 */
*************** GetConnection(UserMapping *user, bool wi
*** 144,161 ****
  	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
  	 * connection is actually used.
--- 155,182 ----
  	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);
+ 	}
+ 
+ 	/*
  	 * We don't check the health of cached connection here, because it would
  	 * require some overhead.  Broken connection will be detected when the
  	 * connection is actually used.
*************** GetConnection(UserMapping *user, bool wi
*** 164,178 ****
  	/*
  	 * 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 be left in a valid empty state.)
  	 */
  	if (entry->conn == NULL)
  	{
  		ForeignServer *server = GetForeignServer(user->serverid);
  
! 		entry->xact_depth = 0;	/* just to be sure */
  		entry->have_prep_stmt = false;
  		entry->have_error = false;
  		entry->conn = connect_pg_server(server, user);
  
  		elog(DEBUG3, "new postgres_fdw connection %p for server \"%s\" (user mapping oid %u, userid %u)",
--- 185,208 ----
  	/*
  	 * 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)
  	{
  		ForeignServer *server = GetForeignServer(user->serverid);
  
! 		/* 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->server_hashvalue =
+ 			GetSysCacheHashValue1(FOREIGNSERVEROID, server->serverid);
+ 		entry->mapping_hashvalue =
+ 			GetSysCacheHashValue1(USERMAPPINGOID, user->umid);
+ 
+ 		/* 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)",
*************** connect_pg_server(ForeignServer *server,
*** 277,282 ****
--- 307,325 ----
  }
  
  /*
+  * Disconnect any open 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,
   * the environment, etc.  We don't want the postgres user's passwords
*************** pgfdw_xact_callback(XactEvent event, voi
*** 777,785 ****
  			entry->changing_xact_state)
  		{
  			elog(DEBUG3, "discarding connection %p", entry->conn);
! 			PQfinish(entry->conn);
! 			entry->conn = NULL;
! 			entry->changing_xact_state = false;
  		}
  	}
  
--- 820,826 ----
  			entry->changing_xact_state)
  		{
  			elog(DEBUG3, "discarding connection %p", entry->conn);
! 			disconnect_pg_server(entry);
  		}
  	}
  
*************** pgfdw_subxact_callback(SubXactEvent even
*** 897,902 ****
--- 938,985 ----
  }
  
  /*
+  * Connection invalidation callback function
+  *
+  * After a change to a pg_foreign_server or pg_user_mapping catalog entry,
+  * mark connections depending on that entry as needing to be remade.
+  * We can't immediately destroy them, since they might be in the midst of
+  * a transaction, but we'll remake them at the next opportunity.
+  *
+  * Although most cache invalidation callbacks blow away all the related stuff
+  * regardless of the given hashvalue, connections are expensive enough that
+  * it's worth trying to avoid that.
+  *
+  * NB: We could avoid unnecessary disconnection more strictly by examining
+  * individual option values, but it seems too much effort for the gain.
+  */
+ static void
+ pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue)
+ {
+ 	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)
+ 			entry->invalidated = true;
+ 		else if ((cacheid == FOREIGNSERVEROID &&
+ 				  entry->server_hashvalue == hashvalue) ||
+ 				 (cacheid == USERMAPPINGOID &&
+ 				  entry->mapping_hashvalue == hashvalue))
+ 			entry->invalidated = true;
+ 	}
+ }
+ 
+ /*
   * Raise an error if the given connection cache entry is marked as being
   * in the middle of an xact state change.  This should be called at which no
   * such change is expected to be in progress; if one is found to be in
*************** pgfdw_reject_incomplete_xact_state_chang
*** 913,921 ****
  	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))
diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index b112c19..ee75c4b 100644
*** a/contrib/postgres_fdw/expected/postgres_fdw.out
--- b/contrib/postgres_fdw/expected/postgres_fdw.out
*************** ALTER FOREIGN TABLE ft2 ALTER COLUMN c1 
*** 191,196 ****
--- 191,233 ----
   public | ft_pg_type | loopback  | (schema_name 'pg_catalog', table_name 'pg_type') | 
  (6 rows)
  
+ -- Test that alteration of server options causes reconnection
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should work
+   c3   |              c4              
+ -------+------------------------------
+  00001 | Fri Jan 02 00:00:00 1970 PST
+ (1 row)
+ 
+ ALTER SERVER loopback OPTIONS (SET dbname 'no such database');
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should fail
+ ERROR:  could not connect to server "loopback"
+ DETAIL:  FATAL:  database "no such database" does not exist
+ DO $d$
+     BEGIN
+         EXECUTE $$ALTER SERVER loopback
+             OPTIONS (SET dbname '$$||current_database()||$$')$$;
+     END;
+ $d$;
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should work again
+   c3   |              c4              
+ -------+------------------------------
+  00001 | Fri Jan 02 00:00:00 1970 PST
+ (1 row)
+ 
+ -- Test that alteration of user mapping options causes reconnection
+ ALTER USER MAPPING FOR CURRENT_USER SERVER loopback
+   OPTIONS (ADD user 'no such user');
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should fail
+ ERROR:  could not connect to server "loopback"
+ DETAIL:  FATAL:  role "no such user" does not exist
+ ALTER USER MAPPING FOR CURRENT_USER SERVER loopback
+   OPTIONS (DROP user);
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should work again
+   c3   |              c4              
+ -------+------------------------------
+  00001 | Fri Jan 02 00:00:00 1970 PST
+ (1 row)
+ 
  -- Now we should be able to run ANALYZE.
  -- To exercise multiple code paths, we use local stats on ft1
  -- and remote-estimate mode on ft2.
diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql
index 509bb54..2eca879 100644
*** a/contrib/postgres_fdw/sql/postgres_fdw.sql
--- b/contrib/postgres_fdw/sql/postgres_fdw.sql
*************** ALTER FOREIGN TABLE ft1 ALTER COLUMN c1 
*** 195,200 ****
--- 195,220 ----
  ALTER FOREIGN TABLE ft2 ALTER COLUMN c1 OPTIONS (column_name 'C 1');
  \det+
  
+ -- Test that alteration of server options causes reconnection
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should work
+ ALTER SERVER loopback OPTIONS (SET dbname 'no such database');
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should fail
+ DO $d$
+     BEGIN
+         EXECUTE $$ALTER SERVER loopback
+             OPTIONS (SET dbname '$$||current_database()||$$')$$;
+     END;
+ $d$;
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should work again
+ 
+ -- Test that alteration of user mapping options causes reconnection
+ ALTER USER MAPPING FOR CURRENT_USER SERVER loopback
+   OPTIONS (ADD user 'no such user');
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should fail
+ ALTER USER MAPPING FOR CURRENT_USER SERVER loopback
+   OPTIONS (DROP user);
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should work again
+ 
  -- Now we should be able to run ANALYZE.
  -- To exercise multiple code paths, we use local stats on ft1
  -- and remote-estimate mode on ft2.

*** 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"
  
***************
*** 48,58 **** typedef struct ConnCacheEntry
--- 49,63 ----
  {
  	ConnCacheKey key;			/* hash key (must be first) */
  	PGconn	   *conn;			/* connection to foreign server, or NULL */
+ 	/* Remaining fields are invalid when conn is NULL: */
  	int			xact_depth;		/* 0 = no xact open, 1 = main xact open, 2 =
  								 * one level of subxact open, etc */
  	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 pending */
+ 	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;
--- 74,80 ----
  
  /* 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,
--- 84,90 ----
  					   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)
--- 137,146 ----
  		 */
  		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,161 **** 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
  	 * connection is actually used.
--- 155,182 ----
  	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);
+ 	}
+ 
+ 	/*
  	 * We don't check the health of cached connection here, because it would
  	 * require some overhead.  Broken connection will be detected when the
  	 * connection is actually used.
***************
*** 164,178 **** GetConnection(UserMapping *user, bool will_prep_stmt)
  	/*
  	 * 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 be left in a valid empty state.)
  	 */
  	if (entry->conn == NULL)
  	{
  		ForeignServer *server = GetForeignServer(user->serverid);
  
! 		entry->xact_depth = 0;	/* just to be sure */
  		entry->have_prep_stmt = false;
  		entry->have_error = false;
  		entry->conn = connect_pg_server(server, user);
  
  		elog(DEBUG3, "new postgres_fdw connection %p for server \"%s\" (user mapping oid %u, userid %u)",
--- 185,208 ----
  	/*
  	 * 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)
  	{
  		ForeignServer *server = GetForeignServer(user->serverid);
  
! 		/* 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->server_hashvalue =
+ 			GetSysCacheHashValue1(FOREIGNSERVEROID, server->serverid);
+ 		entry->mapping_hashvalue =
+ 			GetSysCacheHashValue1(USERMAPPINGOID, user->umid);
+ 
+ 		/* 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)",
***************
*** 277,282 **** connect_pg_server(ForeignServer *server, UserMapping *user)
--- 307,325 ----
  }
  
  /*
+  * Disconnect any open 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,
   * the environment, etc.  We don't want the postgres user's passwords
***************
*** 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;
  		}
  	}
  
--- 820,826 ----
  			entry->changing_xact_state)
  		{
  			elog(DEBUG3, "discarding connection %p", entry->conn);
! 			disconnect_pg_server(entry);
  		}
  	}
  
***************
*** 897,902 **** pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
--- 938,984 ----
  }
  
  /*
+  * Connection invalidation callback function
+  *
+  * After a change to a pg_foreign_server or pg_user_mapping catalog entry,
+  * mark connections depending on that entry as needing to be remade.
+  * We can't immediately destroy them, since they might be in the midst of
+  * a transaction, but we'll remake them at the next opportunity.
+  *
+  * Although most cache invalidation callbacks blow away all the related stuff
+  * regardless of the given hashvalue, connections are expensive enough that
+  * it's worth trying to avoid that.
+  *
+  * NB: We could avoid unnecessary disconnection more strictly by examining
+  * individual option values, but it seems too much effort for the gain.
+  */
+ static void
+ pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue)
+ {
+ 	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)))
+ 			entry->invalidated = true;
+ 	}
+ }
+ 
+ /*
   * Raise an error if the given connection cache entry is marked as being
   * in the middle of an xact state change.  This should be called at which no
   * such change is expected to be in progress; if one is found to be in
***************
*** 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))
--- 995,1008 ----
  	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))
*** a/contrib/postgres_fdw/expected/postgres_fdw.out
--- b/contrib/postgres_fdw/expected/postgres_fdw.out
***************
*** 191,196 **** ALTER FOREIGN TABLE ft2 ALTER COLUMN c1 OPTIONS (column_name 'C 1');
--- 191,233 ----
   public | ft_pg_type | loopback  | (schema_name 'pg_catalog', table_name 'pg_type') | 
  (6 rows)
  
+ -- Test that alteration of server options causes reconnection
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should work
+   c3   |              c4              
+ -------+------------------------------
+  00001 | Fri Jan 02 00:00:00 1970 PST
+ (1 row)
+ 
+ ALTER SERVER loopback OPTIONS (SET dbname 'no such database');
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should fail
+ ERROR:  could not connect to server "loopback"
+ DETAIL:  FATAL:  database "no such database" does not exist
+ DO $d$
+     BEGIN
+         EXECUTE $$ALTER SERVER loopback
+             OPTIONS (SET dbname '$$||current_database()||$$')$$;
+     END;
+ $d$;
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should work again
+   c3   |              c4              
+ -------+------------------------------
+  00001 | Fri Jan 02 00:00:00 1970 PST
+ (1 row)
+ 
+ -- Test that alteration of user mapping options causes reconnection
+ ALTER USER MAPPING FOR CURRENT_USER SERVER loopback
+   OPTIONS (ADD user 'no such user');
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should fail
+ ERROR:  could not connect to server "loopback"
+ DETAIL:  FATAL:  role "no such user" does not exist
+ ALTER USER MAPPING FOR CURRENT_USER SERVER loopback
+   OPTIONS (DROP user);
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should work again
+   c3   |              c4              
+ -------+------------------------------
+  00001 | Fri Jan 02 00:00:00 1970 PST
+ (1 row)
+ 
  -- Now we should be able to run ANALYZE.
  -- To exercise multiple code paths, we use local stats on ft1
  -- and remote-estimate mode on ft2.
*** a/contrib/postgres_fdw/sql/postgres_fdw.sql
--- b/contrib/postgres_fdw/sql/postgres_fdw.sql
***************
*** 195,200 **** ALTER FOREIGN TABLE ft1 ALTER COLUMN c1 OPTIONS (column_name 'C 1');
--- 195,220 ----
  ALTER FOREIGN TABLE ft2 ALTER COLUMN c1 OPTIONS (column_name 'C 1');
  \det+
  
+ -- Test that alteration of server options causes reconnection
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should work
+ ALTER SERVER loopback OPTIONS (SET dbname 'no such database');
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should fail
+ DO $d$
+     BEGIN
+         EXECUTE $$ALTER SERVER loopback
+             OPTIONS (SET dbname '$$||current_database()||$$')$$;
+     END;
+ $d$;
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should work again
+ 
+ -- Test that alteration of user mapping options causes reconnection
+ ALTER USER MAPPING FOR CURRENT_USER SERVER loopback
+   OPTIONS (ADD user 'no such user');
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should fail
+ ALTER USER MAPPING FOR CURRENT_USER SERVER loopback
+   OPTIONS (DROP user);
+ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1;  -- should work again
+ 
  -- Now we should be able to run ANALYZE.
  -- To exercise multiple code paths, we use local stats on ft1
  -- and remote-estimate mode on ft2.