Patch: plan invalidation vs stored procedures - Mailing list pgsql-hackers

This is a followup for thread "plan invalidation vs stored procedures".

The background is that it is impossible to change function return type without
dropping and recreating. Unfortunately dropping a function ruins all of the
prepared statements that reference that function (including other functions).
To make matters worse the ruined plans are never invalidated and keep returning
"cache lookup failed" error until replanned (requires admin intervention). Also
the DBA that dropped the function probably has no clue that something is wrong
- not before looking at the server logs at least. This is NOT good, especially
if the database is supporting paid services.

I have prepared proof of concept patch to support plan invalidation on function
DROP (will add ALTER, REPLACE, etc. later). Currently the invalidation is
handled by just dropping all the plans when invalidation message is received.
The desired behaviour would be of course to drop only the affected plans. This
needs function oid list to be present in PlannedStmt -- will look into this later.
Probably a job for the planner.

Changing statement result type is also currently prohibited in
StorePreparedStatement. There maybe good reasons for this, but for the
invalidation to be really useful, it should be enabled.  Right now the attempt
to change type renders the plan unusable -- "ERROR:  cached plan must not
change result type". Besides that, the patch could already be useful in some
environments - if you are willing to trade the errors for some extra planning
CPU.

There are probably a lot of details that I have overlooked. I'd be really
thankful for some constructive comments and criticism. Especially, what needs
to be done to have this in the core.  Feedback appreciated.

regards,
Martin
*** ./src/backend/commands/functioncmds.c.orig    2008-08-06 17:01:28.000000000 +0300
--- ./src/backend/commands/functioncmds.c    2008-08-06 19:10:51.000000000 +0300
***************
*** 59,64 ****
--- 59,65 ----
  #include "utils/rel.h"
  #include "utils/syscache.h"
  #include "utils/tqual.h"
+ #include "utils/inval.h"


  static void AlterFunctionOwner_internal(Relation rel, HeapTuple tup,
***************
*** 906,911 ****
--- 907,915 ----
      object.objectSubId = 0;

      performDeletion(&object, stmt->behavior);
+
+     /* Notify that cached plans should be replanned */
+     CacheInvalidateProcedure(funcOid);
  }

  /*
*** ./src/backend/utils/cache/inval.c.orig    2008-08-06 16:12:17.000000000 +0300
--- ./src/backend/utils/cache/inval.c    2008-08-06 18:01:24.000000000 +0300
***************
*** 115,121 ****
  typedef struct InvalidationListHeader
  {
      InvalidationChunk *cclist;    /* list of chunks holding catcache msgs */
!     InvalidationChunk *rclist;    /* list of chunks holding relcache/smgr msgs */
  } InvalidationListHeader;

  /*----------------
--- 115,121 ----
  typedef struct InvalidationListHeader
  {
      InvalidationChunk *cclist;    /* list of chunks holding catcache msgs */
!     InvalidationChunk *rclist;    /* list of chunks holding relcache/smgr/proc msgs */
  } InvalidationListHeader;

  /*----------------
***************
*** 177,182 ****
--- 177,183 ----
  #define TWOPHASE_INFO_FILE_AFTER    2    /* relcache file inval */

  static void PersistInvalidationMessage(SharedInvalidationMessage *msg);
+ static void CacheRegisterCallback(int cacheid, CacheCallbackFunction func, Datum arg);


  /* ----------------------------------------------------------------
***************
*** 363,368 ****
--- 364,393 ----
  }

  /*
+  * Add a proc inval entry
+  */
+ static void
+ AddProcInvalidationMessage(InvalidationListHeader *hdr,
+                            Oid dbId, Oid procId)
+ {
+     SharedInvalidationMessage msg;
+
+     /* Don't add a duplicate item */
+     /* We assume dbId need not be checked because it will never change */
+
+     /* XXX: for now, only keep one proc invalidation message per database */
+     ProcessMessageList(hdr->rclist,
+                        if (msg->pm.id == SHAREDINVALPROC_ID)
+                        return);
+
+     /* OK, add the item */
+     msg.pm.id = SHAREDINVALPROC_ID;
+     msg.pm.dbId = dbId;
+     msg.pm.procId = procId;
+     AddInvalidationMessage(&hdr->rclist, &msg);
+ }
+
+ /*
   * Append one list of invalidation messages to another, resetting
   * the source list to empty.
   */
***************
*** 465,470 ****
--- 490,512 ----
  }

  /*
+  * RegisterProcInvalidation
+  *
+  * As above, but register a procedure invalidation event.
+  */
+ static void
+ RegisterProcInvalidation(Oid dbId, Oid procId)
+ {
+     AddProcInvalidationMessage(&transInvalInfo->CurrentCmdInvalidMsgs,
+                                    dbId, procId);
+
+     /*
+      * As above, just in case there is not an associated catalog change.
+      */
+     (void) GetCurrentCommandId(true);
+ }
+
+ /*
   * LocalExecuteInvalidationMessage
   *
   * Process a single invalidation message (which could be of any type).
***************
*** 516,521 ****
--- 558,577 ----
           */
          smgrclosenode(msg->sm.rnode);
      }
+     else if (msg->id == SHAREDINVALPROC_ID)
+     {
+         if (msg->rc.dbId == MyDatabaseId)
+         {
+             /* for now we just need to handle callback functions */
+             for (i = 0; i < cache_callback_count; i++)
+             {
+                 struct CACHECALLBACK *ccitem = cache_callback_list + i;
+
+                 if (ccitem->id == SHAREDINVALPROC_ID)
+                     (*ccitem->function) (ccitem->arg, msg->rc.relId);
+             }
+         }
+     }
      else
          elog(FATAL, "unrecognized SI message id: %d", msg->id);
  }
***************
*** 1175,1191 ****
  }

  /*
!  * CacheRegisterSyscacheCallback
   *        Register the specified function to be called for all future
   *        invalidation events in the specified cache.
   *
-  * NOTE: currently, the OID argument to the callback routine is not
-  * provided for syscache callbacks; the routine doesn't really get any
-  * useful info as to exactly what changed.    It should treat every call
-  * as a "cache flush" request.
   */
! void
! CacheRegisterSyscacheCallback(int cacheid,
                                CacheCallbackFunction func,
                                Datum arg)
  {
--- 1231,1254 ----
  }

  /*
!  * CacheInvalidateProcedure
!  *        Register invalidation of the specified stored procedure.
!  *
!  */
! void
! CacheInvalidateProcedure(Oid procId)
! {
!     RegisterProcInvalidation(MyDatabaseId, procId);
! }
!
! /*
!  * CacheRegisterCallback
   *        Register the specified function to be called for all future
   *        invalidation events in the specified cache.
   *
   */
! static void
! CacheRegisterCallback(int cacheid,
                                CacheCallbackFunction func,
                                Datum arg)
  {
***************
*** 1199,1204 ****
--- 1262,1286 ----
      ++cache_callback_count;
  }

+
+ /*
+  * CacheRegisterSyscacheCallback
+  *        Register the specified function to be called for all future
+  *        invalidation events in the specified cache.
+  *
+  * NOTE: currently, the OID argument to the callback routine is not
+  * provided for syscache callbacks; the routine doesn't really get any
+  * useful info as to exactly what changed.    It should treat every call
+  * as a "cache flush" request.
+  */
+ void
+ CacheRegisterSyscacheCallback(int cacheid,
+                               CacheCallbackFunction func,
+                               Datum arg)
+ {
+     CacheRegisterCallback(cacheid, func, arg);
+ }
+
  /*
   * CacheRegisterRelcacheCallback
   *        Register the specified function to be called for all future
***************
*** 1212,1223 ****
  CacheRegisterRelcacheCallback(CacheCallbackFunction func,
                                Datum arg)
  {
!     if (cache_callback_count >= MAX_CACHE_CALLBACKS)
!         elog(FATAL, "out of cache_callback_list slots");
!
!     cache_callback_list[cache_callback_count].id = SHAREDINVALRELCACHE_ID;
!     cache_callback_list[cache_callback_count].function = func;
!     cache_callback_list[cache_callback_count].arg = arg;

!     ++cache_callback_count;
  }
--- 1294,1311 ----
  CacheRegisterRelcacheCallback(CacheCallbackFunction func,
                                Datum arg)
  {
!     CacheRegisterCallback(SHAREDINVALRELCACHE_ID, func, arg);
! }

! /*
!  * CacheRegisterProcCallback
!  *        Register the specified function to be called for all future
!  *        proccache invalidation events.  The OID of the procedure being
!  *        invalidated will be passed to the function.
!  */
! void
! CacheRegisterProcCallback(CacheCallbackFunction func, Datum arg)
! {
!     CacheRegisterCallback(SHAREDINVALPROC_ID, func, arg);
  }
+
*** ./src/backend/utils/cache/plancache.c.orig    2008-08-06 16:43:11.000000000 +0300
--- ./src/backend/utils/cache/plancache.c    2008-08-06 18:34:30.000000000 +0300
***************
*** 82,87 ****
--- 82,88 ----
  static bool rowmark_member(List *rowMarks, int rt_index);
  static bool plan_list_is_transient(List *stmt_list);
  static void PlanCacheCallback(Datum arg, Oid relid);
+ static void PlanCacheProcCallback(Datum arg, Oid procId);
  static void InvalRelid(Oid relid, LOCKMODE lockmode,
             InvalRelidContext *context);

***************
*** 95,100 ****
--- 96,102 ----
  InitPlanCache(void)
  {
      CacheRegisterRelcacheCallback(PlanCacheCallback, (Datum) 0);
+     CacheRegisterProcCallback(PlanCacheProcCallback, (Datum) 0);
  }

  /*
***************
*** 925,930 ****
--- 927,951 ----
  }

  /*
+  * PlanCacheProcCallback
+  *        Procedure inval callback function
+  *
+  * XXX: For now we just throw away everything.
+  */
+ static void
+ PlanCacheProcCallback(Datum arg, Oid procId)
+ {
+     ListCell   *lc;
+
+     foreach(lc, cached_plans_list)
+     {
+         CachedPlanSource *ps = (CachedPlanSource *)lfirst(lc);
+         if (ps && ps->plan)
+             ps->plan->dead = true;
+     }
+ }
+
+ /*
   * ResetPlanCache: drop all cached plans.
   */
  void
*** ./src/include/storage/sinval.h.orig    2008-08-06 16:19:41.000000000 +0300
--- ./src/include/storage/sinval.h    2008-08-06 16:32:50.000000000 +0300
***************
*** 74,85 ****
--- 74,95 ----
      RelFileNode rnode;            /* physical file ID */
  } SharedInvalSmgrMsg;

+ #define SHAREDINVALPROC_ID        (-3)
+
+ typedef struct
+ {
+     int16        id;                /* type field --- must be first */
+     Oid            dbId;            /* database ID */
+     Oid            procId;            /* procedure ID */
+ } SharedInvalProcMsg;
+
  typedef union
  {
      int16        id;                /* type field --- must be first */
      SharedInvalCatcacheMsg cc;
      SharedInvalRelcacheMsg rc;
      SharedInvalSmgrMsg sm;
+     SharedInvalProcMsg pm;
  } SharedInvalidationMessage;


*** ./src/include/utils/inval.h.orig    2008-08-06 16:27:58.000000000 +0300
--- ./src/include/utils/inval.h    2008-08-06 16:55:40.000000000 +0300
***************
*** 49,54 ****
--- 49,56 ----

  extern void CacheInvalidateRelcacheByRelid(Oid relid);

+ extern void CacheInvalidateProcedure(Oid procId);
+
  extern void CacheRegisterSyscacheCallback(int cacheid,
                                CacheCallbackFunction func,
                                Datum arg);
***************
*** 56,61 ****
--- 58,66 ----
  extern void CacheRegisterRelcacheCallback(CacheCallbackFunction func,
                                Datum arg);

+ extern void CacheRegisterProcCallback(CacheCallbackFunction func,
+                               Datum arg);
+
  extern void inval_twophase_postcommit(TransactionId xid, uint16 info,
                            void *recdata, uint32 len);