Quelle  schemacmds.c  Sprache: unbekannt

/*-------------------------------------------------------------------------
 *
 * schemacmds.c
 *   schema creation/manipulation commands
 *
 * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *   src/backend/commands/schemacmds.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/htup_details.h"
#include "access/table.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/objectaccess.h"
#include "catalog/pg_authid.h"
#include "catalog/pg_database.h"
#include "catalog/pg_namespace.h"
#include "commands/dbcommands.h"
#include "commands/event_trigger.h"
#include "commands/schemacmds.h"
#include "miscadmin.h"
#include "parser/parse_utilcmd.h"
#include "parser/scansup.h"
#include "tcop/utility.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/rel.h"
#include "utils/syscache.h"

static void AlterSchemaOwner_internal(HeapTuple tup, Relation rel, Oid newOwnerId);

/*
 * CREATE SCHEMA
 *
 * Note: caller should pass in location information for the whole
 * CREATE SCHEMA statement, which in turn we pass down as the location
 * of the component commands.  This comports with our general plan of
 * reporting location/len for the whole command even when executing
 * a subquery.
 */
Oid
CreateSchemaCommand(CreateSchemaStmt *stmt, const char *queryString,
     int stmt_location, int stmt_len)
{
 const char *schemaName = stmt->schemaname;
 Oid   namespaceId;
 List    *parsetree_list;
 ListCell   *parsetree_item;
 Oid   owner_uid;
 Oid   saved_uid;
 int   save_sec_context;
 int   save_nestlevel;
 char    *nsp = namespace_search_path;
 AclResult aclresult;
 ObjectAddress address;
 StringInfoData pathbuf;

 GetUserIdAndSecContext(&saved_uid, &save_sec_context);

 /*
  * Who is supposed to own the new schema?
 */
 if (stmt->authrole)
  owner_uid = get_rolespec_oid(stmt->authrole, false);
 else
  owner_uid = saved_uid;

 /* fill schema name with the user name if not specified */
 if (!schemaName)
 {
  HeapTuple tuple;

  tuple = SearchSysCache1(AUTHOID, ObjectIdGetDatum(owner_uid));
  if (!HeapTupleIsValid(tuple))
   elog(ERROR, "cache lookup failed for role %u", owner_uid);
  schemaName =
   pstrdup(NameStr(((Form_pg_authid) GETSTRUCT(tuple))->rolname));
  ReleaseSysCache(tuple);
 }

 /*
  * To create a schema, must have schema-create privilege on the current
  * database and must be able to become the target role (this does not
  * imply that the target role itself must have create-schema privilege).
  * The latter provision guards against "giveaway" attacks.  Note that a
  * superuser will always have both of these privileges a fortiori.
 */
 aclresult = object_aclcheck(DatabaseRelationId, MyDatabaseId, saved_uid, ACL_CREATE);
 if (aclresult != ACLCHECK_OK)
  aclcheck_error(aclresult, OBJECT_DATABASE,
        get_database_name(MyDatabaseId));

 check_can_set_role(saved_uid, owner_uid);

 /* Additional check to protect reserved schema names */
 if (!allowSystemTableMods && IsReservedName(schemaName))
  ereport(ERROR,
    (errcode(ERRCODE_RESERVED_NAME),
     errmsg("unacceptable schema name \"%s\"", schemaName),
     errdetail("The prefix \"pg_\" is reserved for system schemas.")));

 /*
  * If if_not_exists was given and the schema already exists, bail out.
  * (Note: we needn't check this when not if_not_exists, because
  * NamespaceCreate will complain anyway.)  We could do this before making
  * the permissions checks, but since CREATE TABLE IF NOT EXISTS makes its
  * creation-permission check first, we do likewise.
 */
 if (stmt->if_not_exists)
 {
  namespaceId = get_namespace_oid(schemaName, true);
  if (OidIsValid(namespaceId))
  {
   /*
    * If we are in an extension script, insist that the pre-existing
    * object be a member of the extension, to avoid security risks.
 */
   ObjectAddressSet(address, NamespaceRelationId, namespaceId);
   checkMembershipInCurrentExtension(&address);

   /* OK to skip */
   ereport(NOTICE,
     (errcode(ERRCODE_DUPLICATE_SCHEMA),
      errmsg("schema \"%s\" already exists, skipping",
       schemaName)));
   return InvalidOid;
  }
 }

 /*
  * If the requested authorization is different from the current user,
  * temporarily set the current user so that the object(s) will be created
  * with the correct ownership.
  *
  * (The setting will be restored at the end of this routine, or in case of
  * error, transaction abort will clean things up.)
 */
 if (saved_uid != owner_uid)
  SetUserIdAndSecContext(owner_uid,
          save_sec_context | SECURITY_LOCAL_USERID_CHANGE);

 /* Create the schema's namespace */
 namespaceId = NamespaceCreate(schemaName, owner_uid, false);

 /* Advance cmd counter to make the namespace visible */
 CommandCounterIncrement();

 /*
  * Prepend the new schema to the current search path.
  *
  * We use the equivalent of a function SET option to allow the setting to
  * persist for exactly the duration of the schema creation.  guc.c also
  * takes care of undoing the setting on error.
 */
 save_nestlevel = NewGUCNestLevel();

 initStringInfo(&pathbuf);
 appendStringInfoString(&pathbuf, quote_identifier(schemaName));

 while (scanner_isspace(*nsp))
  nsp++;

 if (*nsp != '\0')
  appendStringInfo(&pathbuf, ", %s", nsp);

 (void) set_config_option("search_path", pathbuf.data,
        PGC_USERSET, PGC_S_SESSION,
        GUC_ACTION_SAVE, true, 0, false);

 /*
  * Report the new schema to possibly interested event triggers.  Note we
  * must do this here and not in ProcessUtilitySlow because otherwise the
  * objects created below are reported before the schema, which would be
  * wrong.
 */
 ObjectAddressSet(address, NamespaceRelationId, namespaceId);
 EventTriggerCollectSimpleCommand(address, InvalidObjectAddress,
          (Node *) stmt);

 /*
  * Examine the list of commands embedded in the CREATE SCHEMA command, and
  * reorganize them into a sequentially executable order with no forward
  * references.  Note that the result is still a list of raw parsetrees ---
  * we cannot, in general, run parse analysis on one statement until we
  * have actually executed the prior ones.
 */
 parsetree_list = transformCreateSchemaStmtElements(stmt->schemaElts,
                schemaName);

 /*
  * Execute each command contained in the CREATE SCHEMA.  Since the grammar
  * allows only utility commands in CREATE SCHEMA, there is no need to pass
  * them through parse_analyze_*() or the rewriter; we can just hand them
  * straight to ProcessUtility.
 */
 foreach(parsetree_item, parsetree_list)
 {
  Node    *stmt = (Node *) lfirst(parsetree_item);
  PlannedStmt *wrapper;

  /* need to make a wrapper PlannedStmt */
  wrapper = makeNode(PlannedStmt);
  wrapper->commandType = CMD_UTILITY;
  wrapper->canSetTag = false;
  wrapper->utilityStmt = stmt;
  wrapper->stmt_location = stmt_location;
  wrapper->stmt_len = stmt_len;

  /* do this step */
  ProcessUtility(wrapper,
        queryString,
        false,
        PROCESS_UTILITY_SUBCOMMAND,
        NULL,
        NULL,
        None_Receiver,
        NULL);

  /* make sure later steps can see the object created here */
  CommandCounterIncrement();
 }

 /*
  * Restore the GUC variable search_path we set above.
 */
 AtEOXact_GUC(true, save_nestlevel);

 /* Reset current user and security context */
 SetUserIdAndSecContext(saved_uid, save_sec_context);

 return namespaceId;
}


/*
 * Rename schema
 */
ObjectAddress
RenameSchema(const char *oldname, const char *newname)
{
 Oid   nspOid;
 HeapTuple tup;
 Relation rel;
 AclResult aclresult;
 ObjectAddress address;
 Form_pg_namespace nspform;

 rel = table_open(NamespaceRelationId, RowExclusiveLock);

 tup = SearchSysCacheCopy1(NAMESPACENAME, CStringGetDatum(oldname));
 if (!HeapTupleIsValid(tup))
  ereport(ERROR,
    (errcode(ERRCODE_UNDEFINED_SCHEMA),
     errmsg("schema \"%s\" does not exist", oldname)));

 nspform = (Form_pg_namespace) GETSTRUCT(tup);
 nspOid = nspform->oid;

 /* make sure the new name doesn't exist */
 if (OidIsValid(get_namespace_oid(newname, true)))
  ereport(ERROR,
    (errcode(ERRCODE_DUPLICATE_SCHEMA),
     errmsg("schema \"%s\" already exists", newname)));

 /* must be owner */
 if (!object_ownercheck(NamespaceRelationId, nspOid, GetUserId()))
  aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SCHEMA,
        oldname);

 /* must have CREATE privilege on database */
 aclresult = object_aclcheck(DatabaseRelationId, MyDatabaseId, GetUserId(), ACL_CREATE);
 if (aclresult != ACLCHECK_OK)
  aclcheck_error(aclresult, OBJECT_DATABASE,
        get_database_name(MyDatabaseId));

 if (!allowSystemTableMods && IsReservedName(newname))
  ereport(ERROR,
    (errcode(ERRCODE_RESERVED_NAME),
     errmsg("unacceptable schema name \"%s\"", newname),
     errdetail("The prefix \"pg_\" is reserved for system schemas.")));

 /* rename */
 namestrcpy(&nspform->nspname, newname);
 CatalogTupleUpdate(rel, &tup->t_self, tup);

 InvokeObjectPostAlterHook(NamespaceRelationId, nspOid, 0);

 ObjectAddressSet(address, NamespaceRelationId, nspOid);

 table_close(rel, NoLock);
 heap_freetuple(tup);

 return address;
}

void
AlterSchemaOwner_oid(Oid schemaoid, Oid newOwnerId)
{
 HeapTuple tup;
 Relation rel;

 rel = table_open(NamespaceRelationId, RowExclusiveLock);

 tup = SearchSysCache1(NAMESPACEOID, ObjectIdGetDatum(schemaoid));
 if (!HeapTupleIsValid(tup))
  elog(ERROR, "cache lookup failed for schema %u", schemaoid);

 AlterSchemaOwner_internal(tup, rel, newOwnerId);

 ReleaseSysCache(tup);

 table_close(rel, RowExclusiveLock);
}


/*
 * Change schema owner
 */
ObjectAddress
AlterSchemaOwner(const char *name, Oid newOwnerId)
{
 Oid   nspOid;
 HeapTuple tup;
 Relation rel;
 ObjectAddress address;
 Form_pg_namespace nspform;

 rel = table_open(NamespaceRelationId, RowExclusiveLock);

 tup = SearchSysCache1(NAMESPACENAME, CStringGetDatum(name));
 if (!HeapTupleIsValid(tup))
  ereport(ERROR,
    (errcode(ERRCODE_UNDEFINED_SCHEMA),
     errmsg("schema \"%s\" does not exist", name)));

 nspform = (Form_pg_namespace) GETSTRUCT(tup);
 nspOid = nspform->oid;

 AlterSchemaOwner_internal(tup, rel, newOwnerId);

 ObjectAddressSet(address, NamespaceRelationId, nspOid);

 ReleaseSysCache(tup);

 table_close(rel, RowExclusiveLock);

 return address;
}

static void
AlterSchemaOwner_internal(HeapTuple tup, Relation rel, Oid newOwnerId)
{
 Form_pg_namespace nspForm;

 Assert(tup->t_tableOid == NamespaceRelationId);
 Assert(RelationGetRelid(rel) == NamespaceRelationId);

 nspForm = (Form_pg_namespace) GETSTRUCT(tup);

 /*
  * If the new owner is the same as the existing owner, consider the
  * command to have succeeded.  This is for dump restoration purposes.
 */
 if (nspForm->nspowner != newOwnerId)
 {
  Datum  repl_val[Natts_pg_namespace];
  bool  repl_null[Natts_pg_namespace];
  bool  repl_repl[Natts_pg_namespace];
  Acl     *newAcl;
  Datum  aclDatum;
  bool  isNull;
  HeapTuple newtuple;
  AclResult aclresult;

  /* Otherwise, must be owner of the existing object */
  if (!object_ownercheck(NamespaceRelationId, nspForm->oid, GetUserId()))
   aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SCHEMA,
         NameStr(nspForm->nspname));

  /* Must be able to become new owner */
  check_can_set_role(GetUserId(), newOwnerId);

  /*
   * must have create-schema rights
   *
   * NOTE: This is different from other alter-owner checks in that the
   * current user is checked for create privileges instead of the
   * destination owner.  This is consistent with the CREATE case for
   * schemas.  Because superusers will always have this right, we need
   * no special case for them.
 */
  aclresult = object_aclcheck(DatabaseRelationId, MyDatabaseId, GetUserId(),
         ACL_CREATE);
  if (aclresult != ACLCHECK_OK)
   aclcheck_error(aclresult, OBJECT_DATABASE,
         get_database_name(MyDatabaseId));

  memset(repl_null, false, sizeof(repl_null));
  memset(repl_repl, false, sizeof(repl_repl));

  repl_repl[Anum_pg_namespace_nspowner - 1] = true;
  repl_val[Anum_pg_namespace_nspowner - 1] = ObjectIdGetDatum(newOwnerId);

  /*
   * Determine the modified ACL for the new owner.  This is only
   * necessary when the ACL is non-null.
 */
  aclDatum = SysCacheGetAttr(NAMESPACENAME, tup,
           Anum_pg_namespace_nspacl,
           &isNull);
  if (!isNull)
  {
   newAcl = aclnewowner(DatumGetAclP(aclDatum),
         nspForm->nspowner, newOwnerId);
   repl_repl[Anum_pg_namespace_nspacl - 1] = true;
   repl_val[Anum_pg_namespace_nspacl - 1] = PointerGetDatum(newAcl);
  }

  newtuple = heap_modify_tuple(tup, RelationGetDescr(rel), repl_val, repl_null, repl_repl);

  CatalogTupleUpdate(rel, &newtuple->t_self, newtuple);

  heap_freetuple(newtuple);

  /* Update owner dependency reference */
  changeDependencyOnOwner(NamespaceRelationId, nspForm->oid,
        newOwnerId);
 }

 InvokeObjectPostAlterHook(NamespaceRelationId,
         nspForm->oid, 0);
}