diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index 630c3ab..2d86266 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -979,7 +979,7 @@ relation_openrv(const RangeVar *relation, LOCKMODE lockmode) Oid relOid; /* Look up and lock the appropriate relation using namespace search */ - relOid = RangeVarGetRelid(relation, lockmode, false, false); + relOid = RangeVarGetRelid(relation, lockmode, false); /* Let relation_open do the rest */ return relation_open(relOid, NoLock); @@ -1001,7 +1001,7 @@ relation_openrv_extended(const RangeVar *relation, LOCKMODE lockmode, Oid relOid; /* Look up and lock the appropriate relation using namespace search */ - relOid = RangeVarGetRelid(relation, lockmode, missing_ok, false); + relOid = RangeVarGetRelid(relation, lockmode, missing_ok); /* Return NULL on not-found */ if (!OidIsValid(relOid)) diff --git a/src/backend/catalog/aclchk.c b/src/backend/catalog/aclchk.c index 9a125bd..60c0b8a 100644 --- a/src/backend/catalog/aclchk.c +++ b/src/backend/catalog/aclchk.c @@ -606,7 +606,7 @@ objectNamesToOids(GrantObjectType objtype, List *objnames) RangeVar *relvar = (RangeVar *) lfirst(cell); Oid relOid; - relOid = RangeVarGetRelid(relvar, NoLock, false, false); + relOid = RangeVarGetRelid(relvar, NoLock, false); objects = lappend_oid(objects, relOid); } break; diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c index 99e130c..758872f 100644 --- a/src/backend/catalog/index.c +++ b/src/backend/catalog/index.c @@ -111,7 +111,6 @@ static void validate_index_heapscan(Relation heapRelation, IndexInfo *indexInfo, Snapshot snapshot, v_i_state *state); -static Oid IndexGetRelation(Oid indexId); static bool ReindexIsCurrentlyProcessingIndex(Oid indexOid); static void SetReindexProcessing(Oid heapOid, Oid indexOid); static void ResetReindexProcessing(void); @@ -1301,7 +1300,7 @@ index_drop(Oid indexId) * proceeding until we commit and send out a shared-cache-inval notice * that will make them update their index lists. */ - heapId = IndexGetRelation(indexId); + heapId = IndexGetRelation(indexId, false); userHeapRelation = heap_open(heapId, AccessExclusiveLock); userIndexRelation = index_open(indexId, AccessExclusiveLock); @@ -2763,8 +2762,8 @@ validate_index_heapscan(Relation heapRelation, * IndexGetRelation: given an index's relation OID, get the OID of the * relation it is an index on. Uses the system cache. */ -static Oid -IndexGetRelation(Oid indexId) +Oid +IndexGetRelation(Oid indexId, bool missing_ok) { HeapTuple tuple; Form_pg_index index; @@ -2772,7 +2771,11 @@ IndexGetRelation(Oid indexId) tuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(indexId)); if (!HeapTupleIsValid(tuple)) + { + if (missing_ok) + return InvalidOid; elog(ERROR, "cache lookup failed for index %u", indexId); + } index = (Form_pg_index) GETSTRUCT(tuple); Assert(index->indexrelid == indexId); @@ -2800,7 +2803,7 @@ reindex_index(Oid indexId, bool skip_constraint_checks) * Open and lock the parent heap relation. ShareLock is sufficient since * we only need to be sure no schema or data changes are going on. */ - heapId = IndexGetRelation(indexId); + heapId = IndexGetRelation(indexId, false); heapRelation = heap_open(heapId, ShareLock); /* diff --git a/src/backend/catalog/namespace.c b/src/backend/catalog/namespace.c index 6d4d4b1..25cad3d 100644 --- a/src/backend/catalog/namespace.c +++ b/src/backend/catalog/namespace.c @@ -219,10 +219,14 @@ Datum pg_is_other_temp_schema(PG_FUNCTION_ARGS); * otherwise raise an error. * * If nowait = true, throw an error if we'd have to wait for a lock. + * + * Callback allows caller to check permissions or acquire additional locks + * prior to grabbing the relation lock. */ Oid -RangeVarGetRelid(const RangeVar *relation, LOCKMODE lockmode, bool missing_ok, - bool nowait) +RangeVarGetRelidExtended(const RangeVar *relation, LOCKMODE lockmode, + bool missing_ok, bool nowait, + RangeVarGetRelidCallback callback, void *callback_arg) { uint64 inval_count; Oid relId; @@ -309,6 +313,19 @@ RangeVarGetRelid(const RangeVar *relation, LOCKMODE lockmode, bool missing_ok, } /* + * Invoke caller-supplied callback, if any. + * + * This callback is a good place to check permissions: we haven't taken + * the table lock yet (and it's really best to check permissions before + * locking anything!), but we've gotten far enough to know what OID we + * think we should lock. Of course, concurrent DDL might things while + * we're waiting for the lock, but in that case the callback will be + * invoked again for the new OID. + */ + if (callback) + callback(relation, relId, oldRelId, callback_arg); + + /* * If no lock requested, we assume the caller knows what they're * doing. They should have already acquired a heavyweight lock on * this relation earlier in the processing of this same statement, diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.c index c321224..eb0584e 100644 --- a/src/backend/commands/alter.c +++ b/src/backend/commands/alter.c @@ -111,7 +111,7 @@ ExecRenameStmt(RenameStmt *stmt) * in RenameRelation, renameatt, or renametrig. */ relid = RangeVarGetRelid(stmt->relation, AccessExclusiveLock, - false, false); + false); switch (stmt->renameType) { diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c index 69aa5bf..386b95b 100644 --- a/src/backend/commands/indexcmds.c +++ b/src/backend/commands/indexcmds.c @@ -63,7 +63,10 @@ static void ComputeIndexAttrs(IndexInfo *indexInfo, static Oid GetIndexOpClass(List *opclass, Oid attrType, char *accessMethodName, Oid accessMethodId); static char *ChooseIndexNameAddition(List *colnames); - +static void RangeVarCallbackForReindexTable(const RangeVar *relation, + Oid relId, Oid oldRelId, void *arg); +static void RangeVarCallbackForReindexIndex(const RangeVar *relation, + Oid relId, Oid oldRelId, void *arg); /* * CheckIndexCompatible @@ -129,7 +132,7 @@ CheckIndexCompatible(Oid oldId, Datum d; /* Caller should already have the relation locked in some way. */ - relationId = RangeVarGetRelid(heapRelation, NoLock, false, false); + relationId = RangeVarGetRelid(heapRelation, NoLock, false); /* * We can pretend isconstraint = false unconditionally. It only serves to * decide the text of an error message that should never happen for us. @@ -1725,33 +1728,74 @@ void ReindexIndex(RangeVar *indexRelation) { Oid indOid; - HeapTuple tuple; + Oid heapOid = InvalidOid; + + /* lock level used here should match index lock reindex_index() */ + indOid = RangeVarGetRelidExtended(indexRelation, AccessExclusiveLock, + false, false, + RangeVarCallbackForReindexIndex, + (void *) &heapOid); + + reindex_index(indOid, false); +} + +/* + * Check permissions on table before acquiring relation lock; also lock + * the heap before the RangeVarGetRelidExtended takes the index lock, to avoid + * deadlocks. + */ +static void +RangeVarCallbackForReindexIndex(const RangeVar *relation, + Oid relId, Oid oldRelId, void *arg) +{ + char relkind; + Oid *heapOid = (Oid *) arg; /* - * XXX: This is not safe in the presence of concurrent DDL. We should - * take AccessExclusiveLock here, but that would violate the rule that - * indexes should only be locked after their parent tables. For now, - * we live with it. + * If we previously locked some other index's heap, and the name we're + * looking up no longer refers to that relation, release the now-useless + * lock. */ - indOid = RangeVarGetRelid(indexRelation, NoLock, false, false); - tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(indOid)); - if (!HeapTupleIsValid(tuple)) - elog(ERROR, "cache lookup failed for relation %u", indOid); + if (relId != oldRelId && OidIsValid(oldRelId)) + { + /* lock level here should match reindex_index() heap lock */ + UnlockRelationOid(*heapOid, ShareLock); + *heapOid = InvalidOid; + } + + /* If the relation does not exist, there's nothing more to do. */ + if (!OidIsValid(relId)) + return; - if (((Form_pg_class) GETSTRUCT(tuple))->relkind != RELKIND_INDEX) + /* + * If the relation does exist, check whether it's an index. But note + * that the relation might have been dropped between the time we did the + * name lookup and now. In that case, there's nothing to do. + */ + relkind = get_rel_relkind(relId); + if (!relkind) + return; + if (relkind != RELKIND_INDEX) ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), - errmsg("\"%s\" is not an index", - indexRelation->relname))); + errmsg("\"%s\" is not an index", relation->relname))); /* Check permissions */ - if (!pg_class_ownercheck(indOid, GetUserId())) - aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS, - indexRelation->relname); - - ReleaseSysCache(tuple); + if (!pg_class_ownercheck(relId, GetUserId())) + aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS, relation->relname); - reindex_index(indOid, false); + /* Lock heap before index to avoid deadlock. */ + if (relId != oldRelId) + { + /* + * Lock level here should match reindex_index() heap lock. + * If the OID isn't valid, it means the index as concurrently dropped, + * which is not a problem for us; just return normally. + */ + *heapOid = IndexGetRelation(relId, true); + if (OidIsValid(*heapOid)) + LockRelationOid(*heapOid, ShareLock); + } } /* @@ -1762,27 +1806,10 @@ void ReindexTable(RangeVar *relation) { Oid heapOid; - HeapTuple tuple; /* The lock level used here should match reindex_relation(). */ - heapOid = RangeVarGetRelid(relation, ShareLock, false, false); - tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(heapOid)); - if (!HeapTupleIsValid(tuple)) /* shouldn't happen */ - elog(ERROR, "cache lookup failed for relation %u", heapOid); - - if (((Form_pg_class) GETSTRUCT(tuple))->relkind != RELKIND_RELATION && - ((Form_pg_class) GETSTRUCT(tuple))->relkind != RELKIND_TOASTVALUE) - ereport(ERROR, - (errcode(ERRCODE_WRONG_OBJECT_TYPE), - errmsg("\"%s\" is not a table", - relation->relname))); - - /* Check permissions */ - if (!pg_class_ownercheck(heapOid, GetUserId())) - aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS, - relation->relname); - - ReleaseSysCache(tuple); + heapOid = RangeVarGetRelidExtended(relation, ShareLock, false, false, + RangeVarCallbackForReindexTable, NULL); if (!reindex_relation(heapOid, REINDEX_REL_PROCESS_TOAST)) ereport(NOTICE, @@ -1791,6 +1818,37 @@ ReindexTable(RangeVar *relation) } /* + * Check permissions on table before acquiring relation lock. + */ +static void +RangeVarCallbackForReindexTable(const RangeVar *relation, + Oid relId, Oid oldRelId, void *arg) +{ + char relkind; + + /* Nothing to do if the relation was not found. */ + if (!OidIsValid(relId)) + return; + + /* + * If the relation does exist, check whether it's an index. But note + * that the relation might have been dropped between the time we did the + * name lookup and now. In that case, there's nothing to do. + */ + relkind = get_rel_relkind(relId); + if (!relkind) + return; + if (relkind != RELKIND_RELATION && relkind != RELKIND_TOASTVALUE) + ereport(ERROR, + (errcode(ERRCODE_WRONG_OBJECT_TYPE), + errmsg("\"%s\" is not a table", relation->relname))); + + /* Check permissions */ + if (!pg_class_ownercheck(relId, GetUserId())) + aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS, relation->relname); +} + +/* * ReindexDatabase * Recreate indexes of a database. * diff --git a/src/backend/commands/lockcmds.c b/src/backend/commands/lockcmds.c index 875f868..7898b3c 100644 --- a/src/backend/commands/lockcmds.c +++ b/src/backend/commands/lockcmds.c @@ -55,7 +55,8 @@ LockTableCommand(LockStmt *lockstmt) bool recurse = interpretInhOption(rv->inhOpt); Oid reloid; - reloid = RangeVarGetRelid(rv, lockstmt->mode, false, lockstmt->nowait); + reloid = RangeVarGetRelidExtended(rv, lockstmt->mode, false, + lockstmt->nowait, NULL, NULL); rel = relation_open(reloid, NoLock); LockTableRecurse(rel, lockstmt->mode, lockstmt->nowait, recurse); diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c index 54660f4..41e5301 100644 --- a/src/backend/commands/sequence.c +++ b/src/backend/commands/sequence.c @@ -425,7 +425,7 @@ AlterSequence(AlterSeqStmt *stmt) List *owned_by; /* Open and lock sequence. */ - relid = RangeVarGetRelid(stmt->sequence, AccessShareLock, false, false); + relid = RangeVarGetRelid(stmt->sequence, AccessShareLock, false); init_sequence(relid, &elm, &seqrel); /* allow ALTER to sequence owner only */ @@ -513,7 +513,7 @@ nextval(PG_FUNCTION_ARGS) * whether the performance penalty is material in practice, but for now, * we do it this way. */ - relid = RangeVarGetRelid(sequence, NoLock, false, false); + relid = RangeVarGetRelid(sequence, NoLock, false); PG_RETURN_INT64(nextval_internal(relid)); } diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index c4622c0..b63dc0e 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -775,7 +775,7 @@ RemoveRelations(DropStmt *drop) * that a table must be locked before its corresponding index. * So, for now, we ignore the hazard. */ - relOid = RangeVarGetRelid(rel, NoLock, true, false); + relOid = RangeVarGetRelid(rel, NoLock, true); /* Not there? */ if (!OidIsValid(relOid)) diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c index 5589528..b205dec 100644 --- a/src/backend/commands/trigger.c +++ b/src/backend/commands/trigger.c @@ -201,8 +201,7 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString, * we might end up creating a pg_constraint entry referencing a * nonexistent table. */ - constrrelid = RangeVarGetRelid(stmt->constrrel, AccessShareLock, false, - false); + constrrelid = RangeVarGetRelid(stmt->constrrel, AccessShareLock, false); } /* permission checks */ diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c index f42504c..9d0a331 100644 --- a/src/backend/commands/vacuum.c +++ b/src/backend/commands/vacuum.c @@ -327,7 +327,7 @@ get_rel_oids(Oid relid, const RangeVar *vacrel) * alternative, since we're going to commit this transaction and * begin a new one between now and then. */ - relid = RangeVarGetRelid(vacrel, NoLock, false, false); + relid = RangeVarGetRelid(vacrel, NoLock, false); /* Make a relation list entry for this guy */ oldcontext = MemoryContextSwitchTo(vac_context); diff --git a/src/backend/parser/parse_relation.c b/src/backend/parser/parse_relation.c index 9ac28c9..35e4baa 100644 --- a/src/backend/parser/parse_relation.c +++ b/src/backend/parser/parse_relation.c @@ -282,7 +282,7 @@ searchRangeTable(ParseState *pstate, RangeVar *relation) if (!relation->schemaname) cte = scanNameSpaceForCTE(pstate, refname, &ctelevelsup); if (!cte) - relId = RangeVarGetRelid(relation, NoLock, true, false); + relId = RangeVarGetRelid(relation, NoLock, true); /* Now look for RTEs matching either the relation/CTE or the alias */ for (levelsup = 0; diff --git a/src/backend/parser/parse_type.c b/src/backend/parser/parse_type.c index 2122032..91028c4 100644 --- a/src/backend/parser/parse_type.c +++ b/src/backend/parser/parse_type.c @@ -115,7 +115,7 @@ LookupTypeName(ParseState *pstate, const TypeName *typeName, * of concurrent DDL. But taking a lock would carry a performance * penalty and would also require a permissions check. */ - relid = RangeVarGetRelid(rel, NoLock, false, false); + relid = RangeVarGetRelid(rel, NoLock, false); attnum = get_attnum(relid, field); if (attnum == InvalidAttrNumber) ereport(ERROR, diff --git a/src/backend/rewrite/rewriteDefine.c b/src/backend/rewrite/rewriteDefine.c index 17db70e..a9bb8d5 100644 --- a/src/backend/rewrite/rewriteDefine.c +++ b/src/backend/rewrite/rewriteDefine.c @@ -203,8 +203,7 @@ DefineRule(RuleStmt *stmt, const char *queryString) * Find and lock the relation. Lock level should match * DefineQueryRewrite. */ - relId = RangeVarGetRelid(stmt->relation, AccessExclusiveLock, false, - false); + relId = RangeVarGetRelid(stmt->relation, AccessExclusiveLock, false); /* ... and execute */ DefineQueryRewrite(stmt->rulename, diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c index 6f88c47..36d9edc 100644 --- a/src/backend/tcop/utility.c +++ b/src/backend/tcop/utility.c @@ -84,7 +84,7 @@ CheckRelationOwnership(RangeVar *rel, bool noCatalogs) * AccessExclusiveLock) before verifying that the user has permissions * is not appealing either. */ - relOid = RangeVarGetRelid(rel, NoLock, false, false); + relOid = RangeVarGetRelid(rel, NoLock, false); tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relOid)); if (!HeapTupleIsValid(tuple)) /* should not happen */ diff --git a/src/backend/utils/adt/acl.c b/src/backend/utils/adt/acl.c index e79ba50..48fb673 100644 --- a/src/backend/utils/adt/acl.c +++ b/src/backend/utils/adt/acl.c @@ -1940,7 +1940,7 @@ convert_table_name(text *tablename) relrv = makeRangeVarFromNameList(textToQualifiedNameList(tablename)); /* We might not even have permissions on this relation; don't lock it. */ - return RangeVarGetRelid(relrv, NoLock, false, false); + return RangeVarGetRelid(relrv, NoLock, false); } /* diff --git a/src/backend/utils/adt/regproc.c b/src/backend/utils/adt/regproc.c index 0d42a39..c050f5e 100644 --- a/src/backend/utils/adt/regproc.c +++ b/src/backend/utils/adt/regproc.c @@ -824,8 +824,7 @@ regclassin(PG_FUNCTION_ARGS) names = stringToQualifiedNameList(class_name_or_oid); /* We might not even have permissions on this relation; don't lock it. */ - result = RangeVarGetRelid(makeRangeVarFromNameList(names), NoLock, false, - false); + result = RangeVarGetRelid(makeRangeVarFromNameList(names), NoLock, false); PG_RETURN_OID(result); } @@ -1298,7 +1297,7 @@ text_regclass(PG_FUNCTION_ARGS) rv = makeRangeVarFromNameList(textToQualifiedNameList(relname)); /* We might not even have permissions on this relation; don't lock it. */ - result = RangeVarGetRelid(rv, NoLock, false, false); + result = RangeVarGetRelid(rv, NoLock, false); PG_RETURN_OID(result); } diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c index 75923a6..29df748 100644 --- a/src/backend/utils/adt/ruleutils.c +++ b/src/backend/utils/adt/ruleutils.c @@ -390,7 +390,7 @@ pg_get_viewdef_name(PG_FUNCTION_ARGS) /* Look up view name. Can't lock it - we might not have privileges. */ viewrel = makeRangeVarFromNameList(textToQualifiedNameList(viewname)); - viewoid = RangeVarGetRelid(viewrel, NoLock, false, false); + viewoid = RangeVarGetRelid(viewrel, NoLock, false); PG_RETURN_TEXT_P(string_to_text(pg_get_viewdef_worker(viewoid, 0))); } @@ -410,7 +410,7 @@ pg_get_viewdef_name_ext(PG_FUNCTION_ARGS) /* Look up view name. Can't lock it - we might not have privileges. */ viewrel = makeRangeVarFromNameList(textToQualifiedNameList(viewname)); - viewoid = RangeVarGetRelid(viewrel, NoLock, false, false); + viewoid = RangeVarGetRelid(viewrel, NoLock, false); PG_RETURN_TEXT_P(string_to_text(pg_get_viewdef_worker(viewoid, prettyFlags))); } @@ -1576,7 +1576,7 @@ pg_get_serial_sequence(PG_FUNCTION_ARGS) /* Look up table name. Can't lock it - we might not have privileges. */ tablerv = makeRangeVarFromNameList(textToQualifiedNameList(tablename)); - tableOid = RangeVarGetRelid(tablerv, NoLock, false, false); + tableOid = RangeVarGetRelid(tablerv, NoLock, false); /* Get the number of the column */ column = text_to_cstring(columnname); diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h index 8b78b05..9efd9af 100644 --- a/src/include/catalog/index.h +++ b/src/include/catalog/index.h @@ -99,5 +99,6 @@ extern bool reindex_relation(Oid relid, int flags); extern bool ReindexIsProcessingHeap(Oid heapOid); extern bool ReindexIsProcessingIndex(Oid indexOid); +extern Oid IndexGetRelation(Oid indexId, bool missing_ok); #endif /* INDEX_H */ diff --git a/src/include/catalog/namespace.h b/src/include/catalog/namespace.h index 904c6fd..d537a48 100644 --- a/src/include/catalog/namespace.h +++ b/src/include/catalog/namespace.h @@ -47,9 +47,16 @@ typedef struct OverrideSearchPath bool addTemp; /* implicitly prepend temp schema? */ } OverrideSearchPath; +typedef void (*RangeVarGetRelidCallback)(const RangeVar *relation, Oid relId, + Oid oldRelId, void *callback_arg); -extern Oid RangeVarGetRelid(const RangeVar *relation, LOCKMODE lockmode, - bool missing_ok, bool nowait); +#define RangeVarGetRelid(relation, lockmode, missing_ok) \ + RangeVarGetRelidExtended(relation, lockmode, missing_ok, false, NULL, NULL) + +extern Oid RangeVarGetRelidExtended(const RangeVar *relation, + LOCKMODE lockmode, bool missing_ok, bool nowait, + RangeVarGetRelidCallback callback, + void *callback_arg); extern Oid RangeVarGetCreationNamespace(const RangeVar *newRelation); extern Oid RangeVarGetAndCheckCreationNamespace(const RangeVar *newRelation); extern void RangeVarAdjustRelationPersistence(RangeVar *newRelation, Oid nspid); diff --git a/src/pl/plpgsql/src/pl_comp.c b/src/pl/plpgsql/src/pl_comp.c index 275cad7..d1be6c9 100644 --- a/src/pl/plpgsql/src/pl_comp.c +++ b/src/pl/plpgsql/src/pl_comp.c @@ -1707,7 +1707,7 @@ plpgsql_parse_cwordtype(List *idents) strVal(lsecond(idents)), -1); /* Can't lock relation - we might not have privileges. */ - classOid = RangeVarGetRelid(relvar, NoLock, true, false); + classOid = RangeVarGetRelid(relvar, NoLock, true); if (!OidIsValid(classOid)) goto done; fldname = strVal(lthird(idents)); @@ -1808,7 +1808,7 @@ plpgsql_parse_cwordrowtype(List *idents) relvar = makeRangeVar(strVal(linitial(idents)), strVal(lsecond(idents)), -1); - classOid = RangeVarGetRelid(relvar, NoLock, false, false); + classOid = RangeVarGetRelid(relvar, NoLock, false); MemoryContextSwitchTo(oldCxt);