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

Commit4240e42

Browse files
committed
Try to acquire relation locks in RangeVarGetRelid.
In the previous coding, we would look up a relation in RangeVarGetRelid,lock the resulting OID, and then AcceptInvalidationMessages(). Whilethis was sufficient to ensure that we noticed any changes to therelation definition before building the relcache entry, it didn'thandle the possibility that the name we looked up no longer referencedthe same OID. This was particularly problematic in the case where atable had been dropped and recreated: we'd latch on to the entry forthe old relation and fail later on. Now, we acquire the relation lockinside RangeVarGetRelid, and retry the name lookup if we notice thatinvalidation messages have been processed meanwhile. Many operationsthat would previously have failed with an error in the presence ofconcurrent DDL will now succeed.There is a good deal of work remaining to be done here: many callersof RangeVarGetRelid still pass NoLock for one reason or another. Inaddition, nothing in this patch guards against the possibility thatthe meaning of an unqualified name might change due to the creationof a relation in a schema earlier in the user's search path than theone where it was previously found. Furthermore, there's nothing atall here to guard against similar race conditions for non-relations.For all that, it's a start.Noah Misch and Robert Haas
1 parent9d522cb commit4240e42

File tree

26 files changed

+337
-195
lines changed

26 files changed

+337
-195
lines changed

‎src/backend/access/heap/heapam.c

Lines changed: 6 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -975,26 +975,11 @@ relation_openrv(const RangeVar *relation, LOCKMODE lockmode)
975975
{
976976
OidrelOid;
977977

978-
/*
979-
* Check for shared-cache-inval messages before trying to open the
980-
* relation. This is needed to cover the case where the name identifies a
981-
* rel that has been dropped and recreated since the start of our
982-
* transaction: if we don't flush the old syscache entry then we'll latch
983-
* onto that entry and suffer an error when we do RelationIdGetRelation.
984-
* Note that relation_open does not need to do this, since a relation's
985-
* OID never changes.
986-
*
987-
* We skip this if asked for NoLock, on the assumption that the caller has
988-
* already ensured some appropriate lock is held.
989-
*/
990-
if (lockmode!=NoLock)
991-
AcceptInvalidationMessages();
992-
993-
/* Look up the appropriate relation using namespace search */
994-
relOid=RangeVarGetRelid(relation, false);
978+
/* Look up and lock the appropriate relation using namespace search */
979+
relOid=RangeVarGetRelid(relation,lockmode, false, false);
995980

996981
/* Let relation_open do the rest */
997-
returnrelation_open(relOid,lockmode);
982+
returnrelation_open(relOid,NoLock);
998983
}
999984

1000985
/* ----------------
@@ -1012,30 +997,15 @@ relation_openrv_extended(const RangeVar *relation, LOCKMODE lockmode,
1012997
{
1013998
OidrelOid;
1014999

1015-
/*
1016-
* Check for shared-cache-inval messages before trying to open the
1017-
* relation. This is needed to cover the case where the name identifies a
1018-
* rel that has been dropped and recreated since the start of our
1019-
* transaction: if we don't flush the old syscache entry then we'll latch
1020-
* onto that entry and suffer an error when we do RelationIdGetRelation.
1021-
* Note that relation_open does not need to do this, since a relation's
1022-
* OID never changes.
1023-
*
1024-
* We skip this if asked for NoLock, on the assumption that the caller has
1025-
* already ensured some appropriate lock is held.
1026-
*/
1027-
if (lockmode!=NoLock)
1028-
AcceptInvalidationMessages();
1029-
1030-
/* Look up the appropriate relation using namespace search */
1031-
relOid=RangeVarGetRelid(relation,missing_ok);
1000+
/* Look up and lock the appropriate relation using namespace search */
1001+
relOid=RangeVarGetRelid(relation,lockmode,missing_ok, false);
10321002

10331003
/* Return NULL on not-found */
10341004
if (!OidIsValid(relOid))
10351005
returnNULL;
10361006

10371007
/* Let relation_open do the rest */
1038-
returnrelation_open(relOid,lockmode);
1008+
returnrelation_open(relOid,NoLock);
10391009
}
10401010

10411011
/* ----------------

‎src/backend/catalog/aclchk.c

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -583,6 +583,11 @@ ExecGrantStmt_oids(InternalGrant *istmt)
583583
* objectNamesToOids
584584
*
585585
* Turn a list of object names of a given type into an Oid list.
586+
*
587+
* XXX: This function doesn't take any sort of locks on the objects whose
588+
* names it looks up. In the face of concurrent DDL, we might easily latch
589+
* onto an old version of an object, causing the GRANT or REVOKE statement
590+
* to fail.
586591
*/
587592
staticList*
588593
objectNamesToOids(GrantObjectTypeobjtype,List*objnames)
@@ -601,7 +606,7 @@ objectNamesToOids(GrantObjectType objtype, List *objnames)
601606
RangeVar*relvar= (RangeVar*)lfirst(cell);
602607
OidrelOid;
603608

604-
relOid=RangeVarGetRelid(relvar, false);
609+
relOid=RangeVarGetRelid(relvar,NoLock, false,false);
605610
objects=lappend_oid(objects,relOid);
606611
}
607612
break;

‎src/backend/catalog/index.c

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2942,7 +2942,8 @@ reindex_relation(Oid relid, int flags)
29422942

29432943
/*
29442944
* Open and lock the relation.ShareLock is sufficient since we only need
2945-
* to prevent schema and data changes in it.
2945+
* to prevent schema and data changes in it. The lock level used here
2946+
* should match ReindexTable().
29462947
*/
29472948
rel=heap_open(relid,ShareLock);
29482949

‎src/backend/catalog/namespace.c

Lines changed: 120 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,8 @@
4444
#include"parser/parse_func.h"
4545
#include"storage/backendid.h"
4646
#include"storage/ipc.h"
47+
#include"storage/lmgr.h"
48+
#include"storage/sinval.h"
4749
#include"utils/acl.h"
4850
#include"utils/builtins.h"
4951
#include"utils/guc.h"
@@ -215,14 +217,20 @@ Datumpg_is_other_temp_schema(PG_FUNCTION_ARGS);
215217
*Given a RangeVar describing an existing relation,
216218
*select the proper namespace and look up the relation OID.
217219
*
218-
* If the relation is not found, return InvalidOid iffailOK = true,
220+
* If the relation is not found, return InvalidOid ifmissing_ok = true,
219221
* otherwise raise an error.
222+
*
223+
* If nowait = true, throw an error if we'd have to wait for a lock.
220224
*/
221225
Oid
222-
RangeVarGetRelid(constRangeVar*relation,boolfailOK)
226+
RangeVarGetRelid(constRangeVar*relation,LOCKMODElockmode,boolmissing_ok,
227+
boolnowait)
223228
{
229+
uint64inval_count;
224230
OidnamespaceId;
225231
OidrelId;
232+
OidoldRelId=InvalidOid;
233+
boolretry= false;
226234

227235
/*
228236
* We check the catalog name and then ignore it.
@@ -238,37 +246,120 @@ RangeVarGetRelid(const RangeVar *relation, bool failOK)
238246
}
239247

240248
/*
241-
* Some non-default relpersistence value may have been specified. The
242-
* parser never generates such a RangeVar in simple DML, but it can happen
243-
* in contexts such as "CREATE TEMP TABLE foo (f1 int PRIMARY KEY)". Such
244-
* a command will generate an added CREATE INDEX operation, which must be
245-
* careful to find the temp table, even when pg_temp is not first in the
246-
* search path.
249+
* DDL operations can change the results of a name lookup. Since all
250+
* such operations will generate invalidation messages, we keep track
251+
* of whether any such messages show up while we're performing the
252+
* operation, and retry until either (1) no more invalidation messages
253+
* show up or (2) the answer doesn't change.
254+
*
255+
* But if lockmode = NoLock, then we assume that either the caller is OK
256+
* with the answer changing under them, or that they already hold some
257+
* appropriate lock, and therefore return the first answer we get without
258+
* checking for invalidation messages. Also, if the requested lock is
259+
* already held, no LockRelationOid will not AcceptInvalidationMessages,
260+
* so we may fail to notice a change. We could protect against that case
261+
* by calling AcceptInvalidationMessages() before beginning this loop,
262+
* but that would add a significant amount overhead, so for now we don't.
247263
*/
248-
if (relation->relpersistence==RELPERSISTENCE_TEMP)
264+
for (;;)
249265
{
250-
if (relation->schemaname)
251-
ereport(ERROR,
252-
(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
253-
errmsg("temporary tables cannot specify a schema name")));
254-
if (OidIsValid(myTempNamespace))
255-
relId=get_relname_relid(relation->relname,myTempNamespace);
256-
else/* this probably can't happen? */
257-
relId=InvalidOid;
258-
}
259-
elseif (relation->schemaname)
260-
{
261-
/* use exact schema given */
262-
namespaceId=LookupExplicitNamespace(relation->schemaname);
263-
relId=get_relname_relid(relation->relname,namespaceId);
264-
}
265-
else
266-
{
267-
/* search the namespace path */
268-
relId=RelnameGetRelid(relation->relname);
266+
/*
267+
* Remember this value, so that, after looking up the relation name
268+
* and locking its OID, we can check whether any invalidation messages
269+
* have been processed that might require a do-over.
270+
*/
271+
inval_count=SharedInvalidMessageCounter;
272+
273+
/*
274+
* Some non-default relpersistence value may have been specified. The
275+
* parser never generates such a RangeVar in simple DML, but it can
276+
* happen in contexts such as "CREATE TEMP TABLE foo (f1 int PRIMARY
277+
* KEY)". Such a command will generate an added CREATE INDEX
278+
* operation, which must be careful to find the temp table, even when
279+
* pg_temp is not first in the search path.
280+
*/
281+
if (relation->relpersistence==RELPERSISTENCE_TEMP)
282+
{
283+
if (relation->schemaname)
284+
ereport(ERROR,
285+
(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
286+
errmsg("temporary tables cannot specify a schema name")));
287+
if (OidIsValid(myTempNamespace))
288+
relId=get_relname_relid(relation->relname,myTempNamespace);
289+
else/* this probably can't happen? */
290+
relId=InvalidOid;
291+
}
292+
elseif (relation->schemaname)
293+
{
294+
/* use exact schema given */
295+
namespaceId=LookupExplicitNamespace(relation->schemaname);
296+
relId=get_relname_relid(relation->relname,namespaceId);
297+
}
298+
else
299+
{
300+
/* search the namespace path */
301+
relId=RelnameGetRelid(relation->relname);
302+
}
303+
304+
/*
305+
* If no lock requested, we assume the caller knows what they're
306+
* doing. They should have already acquired a heavyweight lock on
307+
* this relation earlier in the processing of this same statement,
308+
* so it wouldn't be appropriate to AcceptInvalidationMessages()
309+
* here, as that might pull the rug out from under them.
310+
*/
311+
if (lockmode==NoLock)
312+
break;
313+
314+
/*
315+
* If, upon retry, we get back the same OID we did last time, then
316+
* the invalidation messages we processed did not change the final
317+
* answer. So we're done.
318+
*/
319+
if (retry&&relId==oldRelId)
320+
break;
321+
322+
/*
323+
* Lock relation. This will also accept any pending invalidation
324+
* messages. If we got back InvalidOid, indicating not found, then
325+
* there's nothing to lock, but we accept invalidation messages
326+
* anyway, to flush any negative catcache entries that may be
327+
* lingering.
328+
*/
329+
if (!OidIsValid(relId))
330+
AcceptInvalidationMessages();
331+
elseif (!nowait)
332+
LockRelationOid(relId,lockmode);
333+
elseif (!ConditionalLockRelationOid(relId,lockmode))
334+
{
335+
if (relation->schemaname)
336+
ereport(ERROR,
337+
(errcode(ERRCODE_LOCK_NOT_AVAILABLE),
338+
errmsg("could not obtain lock on relation \"%s.%s\"",
339+
relation->schemaname,relation->relname)));
340+
else
341+
ereport(ERROR,
342+
(errcode(ERRCODE_LOCK_NOT_AVAILABLE),
343+
errmsg("could not obtain lock on relation \"%s\"",
344+
relation->relname)));
345+
}
346+
347+
/*
348+
* If no invalidation message were processed, we're done!
349+
*/
350+
if (inval_count==SharedInvalidMessageCounter)
351+
break;
352+
353+
/*
354+
* Something may have changed. Let's repeat the name lookup, to
355+
* make sure this name still references the same relation it did
356+
* previously.
357+
*/
358+
retry= true;
359+
oldRelId=relId;
269360
}
270361

271-
if (!OidIsValid(relId)&& !failOK)
362+
if (!OidIsValid(relId)&& !missing_ok)
272363
{
273364
if (relation->schemaname)
274365
ereport(ERROR,

‎src/backend/commands/alter.c

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -108,7 +108,12 @@ ExecRenameStmt(RenameStmt *stmt)
108108

109109
CheckRelationOwnership(stmt->relation, true);
110110

111-
relid=RangeVarGetRelid(stmt->relation, false);
111+
/*
112+
* Lock level used here should match what will be taken later,
113+
* in RenameRelation, renameatt, or renametrig.
114+
*/
115+
relid=RangeVarGetRelid(stmt->relation,AccessExclusiveLock,
116+
false, false);
112117

113118
switch (stmt->renameType)
114119
{

‎src/backend/commands/indexcmds.c

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1531,9 +1531,15 @@ ReindexIndex(RangeVar *indexRelation)
15311531
OidindOid;
15321532
HeapTupletuple;
15331533

1534-
indOid=RangeVarGetRelid(indexRelation, false);
1534+
/*
1535+
* XXX: This is not safe in the presence of concurrent DDL. We should
1536+
* take AccessExclusiveLock here, but that would violate the rule that
1537+
* indexes should only be locked after their parent tables. For now,
1538+
* we live with it.
1539+
*/
1540+
indOid=RangeVarGetRelid(indexRelation,NoLock, false, false);
15351541
tuple=SearchSysCache1(RELOID,ObjectIdGetDatum(indOid));
1536-
if (!HeapTupleIsValid(tuple))/* shouldn't happen */
1542+
if (!HeapTupleIsValid(tuple))
15371543
elog(ERROR,"cache lookup failed for relation %u",indOid);
15381544

15391545
if (((Form_pg_class)GETSTRUCT(tuple))->relkind!=RELKIND_INDEX)
@@ -1562,7 +1568,8 @@ ReindexTable(RangeVar *relation)
15621568
OidheapOid;
15631569
HeapTupletuple;
15641570

1565-
heapOid=RangeVarGetRelid(relation, false);
1571+
/* The lock level used here should match reindex_relation(). */
1572+
heapOid=RangeVarGetRelid(relation,ShareLock, false, false);
15661573
tuple=SearchSysCache1(RELOID,ObjectIdGetDatum(heapOid));
15671574
if (!HeapTupleIsValid(tuple))/* shouldn't happen */
15681575
elog(ERROR,"cache lookup failed for relation %u",heapOid);

0 commit comments

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