[HACKERS] PgFDW connection invalidation by ALTER SERVER/ALTER USER MAPPING

Hello, moved to pgsql-hackers.

This is the revased and revised version of the previous patch.

At Thu, 13 Jul 2017 13:42:49 +0900 (Tokyo Standard Time), Kyotaro HORIGUCHI <horiguchi.kyotaro@lab.ntt.co.jp> wrote in
<20170713.134249.97825982.horiguchi.kyotaro@lab.ntt.co.jp>
> At Tue, 11 Jul 2017 15:39:14 -0400, Tom Lane <tgl@sss.pgh.pa.us> wrote in <6234.1499801954@sss.pgh.pa.us>
> > Amit Langote <Langote_Amit_f8@lab.ntt.co.jp> writes:
> > > Horiguchi-san,
> > > On 2017/07/11 10:28, Kyotaro HORIGUCHI wrote:
> > >> I faintly recall such discussion was held aroud that time and
> > >> maybe we concluded that we don't do that but I haven't find such
> > >> a thread in pgsql-hackers..
> > 
> > > I mentioned it in my reply.  Here again:
> > > https://www.postgresql.org/message-id/20160405.184408.166437663.horiguchi.kyotaro%40lab.ntt.co.jp
> > 
> > The followup discussion noted that that approach was no good because it
> > would only close connections in the same session that had done the ALTER
> > SERVER.  I think the basic idea of marking postgres_fdw connections as
> > needing to be remade when next possible is OK, but we have to drive it
> > off catcache invalidation events, the same as we did in c52d37c8b.  An
> > advantage of that way is we don't need any new hooks in the core code.
> > 
> > Kyotaro-san, are you planning to update your old patch?
> 
> I'm pleased to do that. I will reconsider the way shown in a mail
> in the thread soon.

done.

(As a recap) Changing of some options such as host of a foreign
server or password of a user mapping make the existing
connections stale, but postgres_fdw continue using them.

The attached patch does the following things.

- postgres_fdw registers two invalidation callbacks on loading.

- On any change on a foreign server or a user mapping, the callbacks mark the affected connection as 'invalid'

- The invalidated connections will be once disconnected before the next execution if no transaction exists.

As the consequence, changes of options properly affects
subsequent queries in the next transaction on any session. It
occurs after whatever option has been modifed.

======
create server sv1 foreign data wrapper postgres_fdw options (host '/tmp', port '5432', dbname 'postgres');
create user mapping for public server sv1;
create table t (a int);
create foreign table ft1 (a int) server sv1 options (table_name 't1');
begin;
select * from ft1; -- no error
alter server sv1 options (set host '/tmpe');
select * from ft1; -- no error because transaction is living.
commit;
select * from ft1;
ERROR:  could not connect to server "sv1"  - NEW
======

This patch is postgres_fdw-private but it's annoying that hash
value of syscache is handled in connection.c. If we allow to add
something to the core for this feature, I could add a new member
in FdwRoutine to notify invalidation of mapping and server by
oid. (Of course it is not back-patcheable, though)

Does anyone have opinitons or suggestions?

regards,

-- 
Kyotaro Horiguchi
NTT Open Source Software Center
*** 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
changein 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.
Brokenconnection 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
voidReleaseConnection(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);