Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commit1575fbc

Browse files
committed
Prevent adding relations to a concurrently dropped schema.
In the previous coding, it was possible for a relation to be createdvia CREATE TABLE, CREATE VIEW, CREATE SEQUENCE, CREATE FOREIGN TABLE,etc. in a schema while that schema was meanwhile being concurrentlydropped. This led to a pg_class entry with an invalid relnamespacevalue. The same problem could occur if a relation was moved usingALTER .. SET SCHEMA while the target schema was being concurrentlydropped. This patch prevents both of those scenarios by locking theschema to which the relation is being added using AccessShareLock,which conflicts with the AccessExclusiveLock taken by DROP.As a desirable side effect, this also prevents the use of CREATE ORREPLACE VIEW to queue for an AccessExclusiveLock on a relation on whichyou have no rights: that will now fail immediately with a permissionserror, before trying to obtain a lock.We need similar protection for all other object types, but as everythingother than relations uses a slightly different set of code paths, I'mleaving that for a separate commit.Original complaint (as far as I could find) about CREATE by NikhilSontakke; risk for ALTER .. SET SCHEMA pointed out by Tom Lane;further details by Dan Farina; patch by me; review by Hitoshi Harada.
1 parent01d83ff commit1575fbc

File tree

7 files changed

+158
-59
lines changed

7 files changed

+158
-59
lines changed

‎src/backend/catalog/namespace.c

Lines changed: 111 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -480,31 +480,131 @@ RangeVarGetCreationNamespace(const RangeVar *newRelation)
480480

481481
/*
482482
* RangeVarGetAndCheckCreationNamespace
483-
*As RangeVarGetCreationNamespace, but with a permissions check.
483+
*
484+
* This function returns the OID of the namespace in which a new relation
485+
* with a given name should be created. If the user does not have CREATE
486+
* permission on the target namespace, this function will instead signal
487+
* an ERROR.
488+
*
489+
* If non-NULL, *existing_oid is set to the OID of any existing relation with
490+
* the same name which already exists in that namespace, or to InvalidOid if
491+
* no such relation exists.
492+
*
493+
* If lockmode != NoLock, the specified lock mode is acquire on the existing
494+
* relation, if any, provided that the current user owns the target relation.
495+
* However, if lockmode != NoLock and the user does not own the target
496+
* relation, we throw an ERROR, as we must not try to lock relations the
497+
* user does not have permissions on.
498+
*
499+
* As a side effect, this function acquires AccessShareLock on the target
500+
* namespace. Without this, the namespace could be dropped before our
501+
* transaction commits, leaving behind relations with relnamespace pointing
502+
* to a no-longer-exstant namespace.
503+
*
504+
* As a further side-effect, if the select namespace is a temporary namespace,
505+
* we mark the RangeVar as RELPERSISTENCE_TEMP.
484506
*/
485507
Oid
486-
RangeVarGetAndCheckCreationNamespace(constRangeVar*newRelation)
508+
RangeVarGetAndCheckCreationNamespace(RangeVar*relation,
509+
LOCKMODElockmode,
510+
Oid*existing_relation_id)
487511
{
488-
OidnamespaceId;
512+
uint64inval_count;
513+
Oidrelid;
514+
Oidoldrelid=InvalidOid;
515+
Oidnspid;
516+
Oidoldnspid=InvalidOid;
517+
boolretry= false;
489518

490-
namespaceId=RangeVarGetCreationNamespace(newRelation);
519+
/*
520+
* We check the catalog name and then ignore it.
521+
*/
522+
if (relation->catalogname)
523+
{
524+
if (strcmp(relation->catalogname,get_database_name(MyDatabaseId))!=0)
525+
ereport(ERROR,
526+
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
527+
errmsg("cross-database references are not implemented: \"%s.%s.%s\"",
528+
relation->catalogname,relation->schemaname,
529+
relation->relname)));
530+
}
491531

492532
/*
493-
* Check we have permission to create there. Skip check if bootstrapping,
494-
* since permissions machinery may not be working yet.
533+
* As in RangeVarGetRelidExtended(), we guard against concurrent DDL
534+
* operations by tracking whether any invalidation messages are processed
535+
* while we're doing the name lookups and acquiring locks. See comments
536+
* in that function for a more detailed explanation of this logic.
495537
*/
496-
if (!IsBootstrapProcessingMode())
538+
for (;;)
497539
{
498540
AclResultaclresult;
499541

500-
aclresult=pg_namespace_aclcheck(namespaceId,GetUserId(),
501-
ACL_CREATE);
542+
inval_count=SharedInvalidMessageCounter;
543+
544+
/* Look up creation namespace and check for existing relation. */
545+
nspid=RangeVarGetCreationNamespace(relation);
546+
Assert(OidIsValid(nspid));
547+
if (existing_relation_id!=NULL)
548+
relid=get_relname_relid(relation->relname,nspid);
549+
else
550+
relid=InvalidOid;
551+
552+
/*
553+
* In bootstrap processing mode, we don't bother with permissions
554+
* or locking. Permissions might not be working yet, and locking is
555+
* unnecessary.
556+
*/
557+
if (IsBootstrapProcessingMode())
558+
break;
559+
560+
/* Check namespace permissions. */
561+
aclresult=pg_namespace_aclcheck(nspid,GetUserId(),ACL_CREATE);
502562
if (aclresult!=ACLCHECK_OK)
503563
aclcheck_error(aclresult,ACL_KIND_NAMESPACE,
504-
get_namespace_name(namespaceId));
564+
get_namespace_name(nspid));
565+
566+
if (retry)
567+
{
568+
/* If nothing changed, we're done. */
569+
if (relid==oldrelid&&nspid==oldnspid)
570+
break;
571+
/* If creation namespace has changed, give up old lock. */
572+
if (nspid!=oldnspid)
573+
UnlockDatabaseObject(NamespaceRelationId,oldnspid,0,
574+
AccessShareLock);
575+
/* If name points to something different, give up old lock. */
576+
if (relid!=oldrelid&&OidIsValid(oldrelid)&&lockmode!=NoLock)
577+
UnlockRelationOid(oldrelid,lockmode);
578+
}
579+
580+
/* Lock namespace. */
581+
if (nspid!=oldnspid)
582+
LockDatabaseObject(NamespaceRelationId,nspid,0,AccessShareLock);
583+
584+
/* Lock relation, if required if and we have permission. */
585+
if (lockmode!=NoLock&&OidIsValid(relid))
586+
{
587+
if (!pg_class_ownercheck(relid,GetUserId()))
588+
aclcheck_error(ACLCHECK_NOT_OWNER,ACL_KIND_CLASS,
589+
relation->relname);
590+
if (relid!=oldrelid)
591+
LockRelationOid(relid,lockmode);
592+
}
593+
594+
/* If no invalidation message were processed, we're done! */
595+
if (inval_count==SharedInvalidMessageCounter)
596+
break;
597+
598+
/* Something may have changed, so recheck our work. */
599+
retry= true;
600+
oldrelid=relid;
601+
oldnspid=nspid;
505602
}
506603

507-
returnnamespaceId;
604+
RangeVarAdjustRelationPersistence(relation,nspid);
605+
if (existing_relation_id!=NULL)
606+
*existing_relation_id=relid;
607+
returnnspid;
508608
}
509609

510610
/*

‎src/backend/commands/tablecmds.c

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -451,10 +451,12 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId)
451451

452452
/*
453453
* Look up the namespace in which we are supposed to create the relation,
454-
* and check we have permission to create there.
454+
* check we have permission to create there, lock it against concurrent
455+
* drop, and mark stmt->relation as RELPERSISTENCE_TEMP if a temporary
456+
* namespace is selected.
455457
*/
456-
namespaceId=RangeVarGetAndCheckCreationNamespace(stmt->relation);
457-
RangeVarAdjustRelationPersistence(stmt->relation,namespaceId);
458+
namespaceId=
459+
RangeVarGetAndCheckCreationNamespace(stmt->relation,NoLock,NULL);
458460

459461
/*
460462
* Security check: disallow creating temp tables from security-restricted
@@ -9417,6 +9419,7 @@ AlterTableNamespace(AlterObjectSchemaStmt *stmt)
94179419
OidoldNspOid;
94189420
OidnspOid;
94199421
RelationclassRel;
9422+
RangeVar*newrv;
94209423

94219424
relid=RangeVarGetRelidExtended(stmt->relation,AccessExclusiveLock,
94229425
false, false,
@@ -9441,8 +9444,9 @@ AlterTableNamespace(AlterObjectSchemaStmt *stmt)
94419444
get_rel_name(tableId))));
94429445
}
94439446

9444-
/* get schema OID and check its permissions */
9445-
nspOid=LookupCreationNamespace(stmt->newschema);
9447+
/* Get and lock schema OID and check its permissions. */
9448+
newrv=makeRangeVar(stmt->newschema,RelationGetRelationName(rel),-1);
9449+
nspOid=RangeVarGetAndCheckCreationNamespace(newrv,NoLock,NULL);
94469450

94479451
/* common checks on switching namespaces */
94489452
CheckSetNamespace(oldNspOid,nspOid,RelationRelationId,relid);

‎src/backend/commands/typecmds.c

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2005,7 +2005,8 @@ DefineCompositeType(const RangeVar *typevar, List *coldeflist)
20052005
* check is here mainly to get a better error message about a "type"
20062006
* instead of below about a "relation".
20072007
*/
2008-
typeNamespace=RangeVarGetCreationNamespace(createStmt->relation);
2008+
typeNamespace=RangeVarGetAndCheckCreationNamespace(createStmt->relation,
2009+
NoLock,NULL);
20092010
RangeVarAdjustRelationPersistence(createStmt->relation,typeNamespace);
20102011
old_type_oid=
20112012
GetSysCacheOid2(TYPENAMENSP,

‎src/backend/commands/view.c

Lines changed: 15 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -98,10 +98,12 @@ isViewOnTempTable_walker(Node *node, void *context)
9898
*---------------------------------------------------------------------
9999
*/
100100
staticOid
101-
DefineVirtualRelation(constRangeVar*relation,List*tlist,boolreplace,
102-
OidnamespaceId,List*options)
101+
DefineVirtualRelation(RangeVar*relation,List*tlist,boolreplace,
102+
List*options)
103103
{
104104
OidviewOid;
105+
OidnamespaceId;
106+
LOCKMODElockmode;
105107
CreateStmt*createStmt=makeNode(CreateStmt);
106108
List*attrList;
107109
ListCell*t;
@@ -159,9 +161,14 @@ DefineVirtualRelation(const RangeVar *relation, List *tlist, bool replace,
159161
errmsg("view must have at least one column")));
160162

161163
/*
162-
* Check to see if we want to replace an existing view.
164+
* Look up, check permissions on, and lock the creation namespace; also
165+
* check for a preexisting view with the same name. This will also set
166+
* relation->relpersistence to RELPERSISTENCE_TEMP if the selected
167+
* namespace is temporary.
163168
*/
164-
viewOid=get_relname_relid(relation->relname,namespaceId);
169+
lockmode=replace ?AccessExclusiveLock :NoLock;
170+
namespaceId=
171+
RangeVarGetAndCheckCreationNamespace(relation,lockmode,&viewOid);
165172

166173
if (OidIsValid(viewOid)&&replace)
167174
{
@@ -170,24 +177,16 @@ DefineVirtualRelation(const RangeVar *relation, List *tlist, bool replace,
170177
List*atcmds=NIL;
171178
AlterTableCmd*atcmd;
172179

173-
/*
174-
* Yes. Get exclusive lock on the existing view ...
175-
*/
176-
rel=relation_open(viewOid,AccessExclusiveLock);
180+
/* Relation is already locked, but we must build a relcache entry. */
181+
rel=relation_open(viewOid,NoLock);
177182

178-
/*
179-
* Make sure it *is* a view, and do permissions checks.
180-
*/
183+
/* Make sure it *is* a view. */
181184
if (rel->rd_rel->relkind!=RELKIND_VIEW)
182185
ereport(ERROR,
183186
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
184187
errmsg("\"%s\" is not a view",
185188
RelationGetRelationName(rel))));
186189

187-
if (!pg_class_ownercheck(viewOid,GetUserId()))
188-
aclcheck_error(ACLCHECK_NOT_OWNER,ACL_KIND_CLASS,
189-
RelationGetRelationName(rel));
190-
191190
/* Also check it's not in use already */
192191
CheckTableNotInUse(rel,"CREATE OR REPLACE VIEW");
193192

@@ -428,7 +427,6 @@ DefineView(ViewStmt *stmt, const char *queryString)
428427
{
429428
Query*viewParse;
430429
OidviewOid;
431-
OidnamespaceId;
432430
RangeVar*view;
433431

434432
/*
@@ -514,18 +512,14 @@ DefineView(ViewStmt *stmt, const char *queryString)
514512
view->relname)));
515513
}
516514

517-
/* Might also need to make it temporary if placed in temp schema. */
518-
namespaceId=RangeVarGetCreationNamespace(view);
519-
RangeVarAdjustRelationPersistence(view,namespaceId);
520-
521515
/*
522516
* Create the view relation
523517
*
524518
* NOTE: if it already exists and replace is false, the xact will be
525519
* aborted.
526520
*/
527521
viewOid=DefineVirtualRelation(view,viewParse->targetList,
528-
stmt->replace,namespaceId,stmt->options);
522+
stmt->replace,stmt->options);
529523

530524
/*
531525
* The relation we have just created is not visible to any other commands

‎src/backend/executor/execMain.c

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2532,11 +2532,13 @@ OpenIntoRel(QueryDesc *queryDesc)
25322532
}
25332533

25342534
/*
2535-
* Find namespace to create in, check its permissions
2535+
* Find namespace to create in, check its permissions, lock it against
2536+
* concurrent drop, and mark into->rel as RELPERSISTENCE_TEMP if the
2537+
* selected namespace is temporary.
25362538
*/
25372539
intoName=into->rel->relname;
2538-
namespaceId=RangeVarGetAndCheckCreationNamespace(into->rel);
2539-
RangeVarAdjustRelationPersistence(into->rel,namespaceId);
2540+
namespaceId=RangeVarGetAndCheckCreationNamespace(into->rel,NoLock,
2541+
NULL);
25402542

25412543
/*
25422544
* Security check: disallow creating temp tables from security-restricted

‎src/backend/parser/parse_utilcmd.c

Lines changed: 13 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -146,6 +146,7 @@ transformCreateStmt(CreateStmt *stmt, const char *queryString)
146146
List*save_alist;
147147
ListCell*elements;
148148
Oidnamespaceid;
149+
Oidexisting_relid;
149150

150151
/*
151152
* We must not scribble on the passed-in CreateStmt, so copy it. (This is
@@ -155,30 +156,25 @@ transformCreateStmt(CreateStmt *stmt, const char *queryString)
155156

156157
/*
157158
* Look up the creation namespace.This also checks permissions on the
158-
* target namespace, so that we throw any permissions error as early as
159-
* possible.
159+
* target namespace, locks it against concurrent drops, checks for a
160+
* preexisting relation in that namespace with the same name, and updates
161+
* stmt->relation->relpersistence if the select namespace is temporary.
160162
*/
161-
namespaceid=RangeVarGetAndCheckCreationNamespace(stmt->relation);
162-
RangeVarAdjustRelationPersistence(stmt->relation,namespaceid);
163+
namespaceid=
164+
RangeVarGetAndCheckCreationNamespace(stmt->relation,NoLock,
165+
&existing_relid);
163166

164167
/*
165168
* If the relation already exists and the user specified "IF NOT EXISTS",
166169
* bail out with a NOTICE.
167170
*/
168-
if (stmt->if_not_exists)
171+
if (stmt->if_not_exists&&OidIsValid(existing_relid))
169172
{
170-
Oidexisting_relid;
171-
172-
existing_relid=get_relname_relid(stmt->relation->relname,
173-
namespaceid);
174-
if (existing_relid!=InvalidOid)
175-
{
176-
ereport(NOTICE,
177-
(errcode(ERRCODE_DUPLICATE_TABLE),
178-
errmsg("relation \"%s\" already exists, skipping",
179-
stmt->relation->relname)));
180-
returnNIL;
181-
}
173+
ereport(NOTICE,
174+
(errcode(ERRCODE_DUPLICATE_TABLE),
175+
errmsg("relation \"%s\" already exists, skipping",
176+
stmt->relation->relname)));
177+
returnNIL;
182178
}
183179

184180
/*

‎src/include/catalog/namespace.h

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,9 @@ extern OidRangeVarGetRelidExtended(const RangeVar *relation,
5858
RangeVarGetRelidCallbackcallback,
5959
void*callback_arg);
6060
externOidRangeVarGetCreationNamespace(constRangeVar*newRelation);
61-
externOidRangeVarGetAndCheckCreationNamespace(constRangeVar*newRelation);
61+
externOidRangeVarGetAndCheckCreationNamespace(RangeVar*newRelation,
62+
LOCKMODElockmode,
63+
Oid*existing_relation_id);
6264
externvoidRangeVarAdjustRelationPersistence(RangeVar*newRelation,Oidnspid);
6365
externOidRelnameGetRelid(constchar*relname);
6466
externboolRelationIsVisible(Oidrelid);

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp