schemacmds.c

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * 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);
}