Re: PGXS: REGRESS_OPTS=--load-language=plpgsql - Mailing list pgsql-hackers

Robert Haas <robertmhaas@gmail.com> writes:
> On Feb 20, 2010, at 10:56 PM, Tom Lane <tgl@sss.pgh.pa.us> wrote:
>> There is a very clear set of behaviors that CORL ought to have given
>> the precedents of our other COR commands.  If we don't make it do
>> things that way then we are going to surprise users, and we are also
>> going to paint ourselves into a corner because we won't be able to
>> fix it later without creating compatibility gotchas.

> Exactly.  I agree completely.

Attached is a draft patch (no doc changes) that implements CREATE OR
REPLACE LANGUAGE following the semantics used in CREATE OR REPLACE
FUNCTION, namely that in addition to whatever privileges you need to
do the CREATE, you need to be owner of the existing entry if any;
and the recorded ownership and permissions don't change.  It's not bad
at all --- net addition of 40 lines.  So if we want to go at it this
way, it's certainly feasible.

I've got mixed feelings about the ownership check.  If you get past
the normal CREATE LANGUAGE permission checks, then either you are
superuser, or you are database owner and you are trying to recreate
a language from a pg_pltemplate entry with tmpldbacreate true.
So it would fail only for a database owner who's trying to do
C.O.R.L. on a superuser-installed language.  Which arguably is a case
we ought to allow.  On the other hand, the case where not throwing an
error would really matter is in trying to do pg_restore --single, and
in that case even if we allowed the C.O.R.L. it would still spit up on
the ALTER LANGUAGE OWNER that pg_dump is presumably going to emit right
afterwards (except if using --no-owner, I guess).  So I'm not sure
we'd really be gaining much by omitting the ownership check, and it
would certainly be less consistent with other C.O.R. commands if we
don't apply such a check.

Comments?

            regards, tom lane

Index: src/backend/commands/proclang.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/commands/proclang.c,v
retrieving revision 1.89
diff -c -r1.89 proclang.c
*** src/backend/commands/proclang.c    14 Feb 2010 18:42:14 -0000    1.89
--- src/backend/commands/proclang.c    21 Feb 2010 17:08:15 -0000
***************
*** 17,23 ****
  #include "access/heapam.h"
  #include "catalog/dependency.h"
  #include "catalog/indexing.h"
- #include "catalog/pg_authid.h"
  #include "catalog/pg_language.h"
  #include "catalog/pg_namespace.h"
  #include "catalog/pg_pltemplate.h"
--- 17,22 ----
***************
*** 49,55 ****
      char       *tmpllibrary;    /* path of shared library */
  } PLTemplate;

! static void create_proc_lang(const char *languageName,
                   Oid languageOwner, Oid handlerOid, Oid inlineOid,
                   Oid valOid, bool trusted);
  static PLTemplate *find_language_template(const char *languageName);
--- 48,54 ----
      char       *tmpllibrary;    /* path of shared library */
  } PLTemplate;

! static void create_proc_lang(const char *languageName, bool replace,
                   Oid languageOwner, Oid handlerOid, Oid inlineOid,
                   Oid valOid, bool trusted);
  static PLTemplate *find_language_template(const char *languageName);
***************
*** 73,88 ****
      Oid            funcargtypes[1];

      /*
!      * Translate the language name and check that this language doesn't
!      * already exist
       */
      languageName = case_translate_language_name(stmt->plname);

-     if (SearchSysCacheExists1(LANGNAME, PointerGetDatum(languageName)))
-         ereport(ERROR,
-                 (errcode(ERRCODE_DUPLICATE_OBJECT),
-                  errmsg("language \"%s\" already exists", languageName)));
-
      /*
       * If we have template information for the language, ignore the supplied
       * parameters (if any) and use the template information.
--- 72,81 ----
      Oid            funcargtypes[1];

      /*
!      * Translate the language name to lower case
       */
      languageName = case_translate_language_name(stmt->plname);

      /*
       * If we have template information for the language, ignore the supplied
       * parameters (if any) and use the template information.
***************
*** 232,238 ****
              valOid = InvalidOid;

          /* ok, create it */
!         create_proc_lang(languageName, GetUserId(), handlerOid, inlineOid,
                           valOid, pltemplate->tmpltrusted);
      }
      else
--- 225,232 ----
              valOid = InvalidOid;

          /* ok, create it */
!         create_proc_lang(languageName, stmt->replace, GetUserId(),
!                          handlerOid, inlineOid,
                           valOid, pltemplate->tmpltrusted);
      }
      else
***************
*** 306,312 ****
              valOid = InvalidOid;

          /* ok, create it */
!         create_proc_lang(languageName, GetUserId(), handlerOid, inlineOid,
                           valOid, stmt->pltrusted);
      }
  }
--- 300,307 ----
              valOid = InvalidOid;

          /* ok, create it */
!         create_proc_lang(languageName, stmt->replace, GetUserId(),
!                          handlerOid, inlineOid,
                           valOid, stmt->pltrusted);
      }
  }
***************
*** 315,321 ****
   * Guts of language creation.
   */
  static void
! create_proc_lang(const char *languageName,
                   Oid languageOwner, Oid handlerOid, Oid inlineOid,
                   Oid valOid, bool trusted)
  {
--- 310,316 ----
   * Guts of language creation.
   */
  static void
! create_proc_lang(const char *languageName, bool replace,
                   Oid languageOwner, Oid handlerOid, Oid inlineOid,
                   Oid valOid, bool trusted)
  {
***************
*** 323,341 ****
      TupleDesc    tupDesc;
      Datum        values[Natts_pg_language];
      bool        nulls[Natts_pg_language];
      NameData    langname;
      HeapTuple    tup;
      ObjectAddress myself,
                  referenced;

-     /*
-      * Insert the new language into pg_language
-      */
      rel = heap_open(LanguageRelationId, RowExclusiveLock);
!     tupDesc = rel->rd_att;

      memset(values, 0, sizeof(values));
      memset(nulls, false, sizeof(nulls));

      namestrcpy(&langname, languageName);
      values[Anum_pg_language_lanname - 1] = NameGetDatum(&langname);
--- 318,338 ----
      TupleDesc    tupDesc;
      Datum        values[Natts_pg_language];
      bool        nulls[Natts_pg_language];
+     bool        replaces[Natts_pg_language];
      NameData    langname;
+     HeapTuple    oldtup;
      HeapTuple    tup;
+     bool        is_update;
      ObjectAddress myself,
                  referenced;

      rel = heap_open(LanguageRelationId, RowExclusiveLock);
!     tupDesc = RelationGetDescr(rel);

+     /* Prepare data to be inserted */
      memset(values, 0, sizeof(values));
      memset(nulls, false, sizeof(nulls));
+     memset(replaces, true, sizeof(replaces));

      namestrcpy(&langname, languageName);
      values[Anum_pg_language_lanname - 1] = NameGetDatum(&langname);
***************
*** 347,370 ****
      values[Anum_pg_language_lanvalidator - 1] = ObjectIdGetDatum(valOid);
      nulls[Anum_pg_language_lanacl - 1] = true;

!     tup = heap_form_tuple(tupDesc, values, nulls);

!     simple_heap_insert(rel, tup);

      CatalogUpdateIndexes(rel, tup);

      /*
!      * Create dependencies for language
       */
      myself.classId = LanguageRelationId;
      myself.objectId = HeapTupleGetOid(tup);
      myself.objectSubId = 0;

      /* dependency on owner of language */
!     referenced.classId = AuthIdRelationId;
!     referenced.objectId = languageOwner;
!     referenced.objectSubId = 0;
!     recordSharedDependencyOn(&myself, &referenced, SHARED_DEPENDENCY_OWNER);

      /* dependency on the PL handler function */
      referenced.classId = ProcedureRelationId;
--- 344,405 ----
      values[Anum_pg_language_lanvalidator - 1] = ObjectIdGetDatum(valOid);
      nulls[Anum_pg_language_lanacl - 1] = true;

!     /* Check for pre-existing definition */
!     oldtup = SearchSysCache1(LANGNAME, PointerGetDatum(languageName));

!     if (HeapTupleIsValid(oldtup))
!     {
!         /* There is one; okay to replace it? */
!         if (!replace)
!             ereport(ERROR,
!                     (errcode(ERRCODE_DUPLICATE_OBJECT),
!                      errmsg("language \"%s\" already exists", languageName)));
!         if (!pg_language_ownercheck(HeapTupleGetOid(oldtup), languageOwner))
!             aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_LANGUAGE,
!                            languageName);

+         /*
+          * Do not change existing ownership or permissions.  Note
+          * dependency-update code below has to agree with this decision.
+          */
+         replaces[Anum_pg_language_lanowner - 1] = false;
+         replaces[Anum_pg_language_lanacl - 1] = false;
+
+         /* Okay, do it... */
+         tup = heap_modify_tuple(oldtup, tupDesc, values, nulls, replaces);
+         simple_heap_update(rel, &tup->t_self, tup);
+
+         ReleaseSysCache(oldtup);
+         is_update = true;
+     }
+     else
+     {
+         /* Creating a new language */
+         tup = heap_form_tuple(tupDesc, values, nulls);
+         simple_heap_insert(rel, tup);
+         is_update = false;
+     }
+
+     /* Need to update indexes for either the insert or update case */
      CatalogUpdateIndexes(rel, tup);

      /*
!      * Create dependencies for the new language.  If we are updating an
!      * existing language, first delete any existing pg_depend entries.
!      * (However, since we are not changing ownership or permissions, the
!      * shared dependencies do *not* need to change, and we leave them alone.)
       */
      myself.classId = LanguageRelationId;
      myself.objectId = HeapTupleGetOid(tup);
      myself.objectSubId = 0;

+     if (is_update)
+         deleteDependencyRecordsFor(myself.classId, myself.objectId);
+
      /* dependency on owner of language */
!     if (!is_update)
!         recordDependencyOnOwner(myself.classId, myself.objectId,
!                                 languageOwner);

      /* dependency on the PL handler function */
      referenced.classId = ProcedureRelationId;
Index: src/backend/nodes/copyfuncs.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/nodes/copyfuncs.c,v
retrieving revision 1.462
diff -c -r1.462 copyfuncs.c
*** src/backend/nodes/copyfuncs.c    16 Feb 2010 22:34:43 -0000    1.462
--- src/backend/nodes/copyfuncs.c    21 Feb 2010 17:08:15 -0000
***************
*** 3237,3242 ****
--- 3237,3243 ----
  {
      CreatePLangStmt *newnode = makeNode(CreatePLangStmt);

+     COPY_SCALAR_FIELD(replace);
      COPY_STRING_FIELD(plname);
      COPY_NODE_FIELD(plhandler);
      COPY_NODE_FIELD(plinline);
Index: src/backend/nodes/equalfuncs.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/nodes/equalfuncs.c,v
retrieving revision 1.383
diff -c -r1.383 equalfuncs.c
*** src/backend/nodes/equalfuncs.c    16 Feb 2010 22:34:43 -0000    1.383
--- src/backend/nodes/equalfuncs.c    21 Feb 2010 17:08:15 -0000
***************
*** 1710,1715 ****
--- 1710,1716 ----
  static bool
  _equalCreatePLangStmt(CreatePLangStmt *a, CreatePLangStmt *b)
  {
+     COMPARE_SCALAR_FIELD(replace);
      COMPARE_STRING_FIELD(plname);
      COMPARE_NODE_FIELD(plhandler);
      COMPARE_NODE_FIELD(plinline);
Index: src/backend/parser/gram.y
===================================================================
RCS file: /cvsroot/pgsql/src/backend/parser/gram.y,v
retrieving revision 2.710
diff -c -r2.710 gram.y
*** src/backend/parser/gram.y    17 Feb 2010 04:19:39 -0000    2.710
--- src/backend/parser/gram.y    21 Feb 2010 17:08:16 -0000
***************
*** 2882,2897 ****
  /*****************************************************************************
   *
   *        QUERIES :
!  *                CREATE PROCEDURAL LANGUAGE ...
   *                DROP PROCEDURAL LANGUAGE ...
   *
   *****************************************************************************/

  CreatePLangStmt:
!             CREATE opt_trusted opt_procedural LANGUAGE ColId_or_Sconst
              {
                  CreatePLangStmt *n = makeNode(CreatePLangStmt);
!                 n->plname = $5;
                  /* parameters are all to be supplied by system */
                  n->plhandler = NIL;
                  n->plinline = NIL;
--- 2882,2898 ----
  /*****************************************************************************
   *
   *        QUERIES :
!  *                CREATE [OR REPLACE] PROCEDURAL LANGUAGE ...
   *                DROP PROCEDURAL LANGUAGE ...
   *
   *****************************************************************************/

  CreatePLangStmt:
!             CREATE opt_or_replace opt_trusted opt_procedural LANGUAGE ColId_or_Sconst
              {
                  CreatePLangStmt *n = makeNode(CreatePLangStmt);
!                 n->replace = $2;
!                 n->plname = $6;
                  /* parameters are all to be supplied by system */
                  n->plhandler = NIL;
                  n->plinline = NIL;
***************
*** 2899,2913 ****
                  n->pltrusted = false;
                  $$ = (Node *)n;
              }
!             | CREATE opt_trusted opt_procedural LANGUAGE ColId_or_Sconst
                HANDLER handler_name opt_inline_handler opt_validator
              {
                  CreatePLangStmt *n = makeNode(CreatePLangStmt);
!                 n->plname = $5;
!                 n->plhandler = $7;
!                 n->plinline = $8;
!                 n->plvalidator = $9;
!                 n->pltrusted = $2;
                  $$ = (Node *)n;
              }
          ;
--- 2900,2915 ----
                  n->pltrusted = false;
                  $$ = (Node *)n;
              }
!             | CREATE opt_or_replace opt_trusted opt_procedural LANGUAGE ColId_or_Sconst
                HANDLER handler_name opt_inline_handler opt_validator
              {
                  CreatePLangStmt *n = makeNode(CreatePLangStmt);
!                 n->replace = $2;
!                 n->plname = $6;
!                 n->plhandler = $8;
!                 n->plinline = $9;
!                 n->plvalidator = $10;
!                 n->pltrusted = $3;
                  $$ = (Node *)n;
              }
          ;
Index: src/include/nodes/parsenodes.h
===================================================================
RCS file: /cvsroot/pgsql/src/include/nodes/parsenodes.h,v
retrieving revision 1.430
diff -c -r1.430 parsenodes.h
*** src/include/nodes/parsenodes.h    16 Feb 2010 22:34:57 -0000    1.430
--- src/include/nodes/parsenodes.h    21 Feb 2010 17:08:16 -0000
***************
*** 1623,1628 ****
--- 1623,1629 ----
  typedef struct CreatePLangStmt
  {
      NodeTag        type;
+     bool        replace;        /* T => replace if already exists */
      char       *plname;            /* PL name */
      List       *plhandler;        /* PL call handler function (qual. name) */
      List       *plinline;        /* optional inline function (qual. name) */