database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-06-16 06:01:02 +03:00
Code Activity

Extend ALTER TABLE to allow Foreign Keys to be added without initial validation.

Browse Source 
FK constraints that are marked NOT VALID may later be VALIDATED, which uses an
ShareUpdateExclusiveLock on constraint table and RowShareLock on referenced
table. Significantly reduces lock strength and duration when adding FKs.
New state visible from psql.

Simon Riggs, with reviews from Marko Tiikkaja and Robert Haas
This commit is contained in:
Simon Riggs
2011-02-08 12:23:20 +00:00
parent 7202ad7b8d
commit 722bf7017b
15 changed files with 215 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
src/backend/commands/tablecmds.c
View File

@ -254,6 +254,7 @@ static void AlterIndexNamespaces(Relation classRel, Relation rel,
static void AlterSeqNamespaces(Relation classRel, Relation rel,
Oid oldNspOid, Oid newNspOid,
const char *newNspName, LOCKMODE lockmode);
static void ATExecValidateConstraint(Relation rel, const char *constrName);
static int transformColumnNameList(Oid relId, List *colList,
int16 *attnums, Oid *atttypids);
static int transformFkeyGetPrimaryKey(Relation pkrel, Oid *indexOid,
@ -264,7 +265,7 @@ static Oid transformFkeyCheckAttrs(Relation pkrel,
int numattrs, int16 *attnums,
Oid *opclasses);
static void checkFkeyPermissions(Relation rel, int16 *attnums, int natts);
static void validateForeignKeyConstraint(Constraint *fkconstraint,
static void validateForeignKeyConstraint(char *conname,
Relation rel, Relation pkrel,
Oid pkindOid, Oid constraintOid);
static void createForeignKeyTriggers(Relation rel, Constraint *fkconstraint,
@ -2649,7 +2650,7 @@ AlterTableGetLockLevel(List *cmds)
* though don't change the semantic results from normal data reads and writes.
* Delaying an ALTER TABLE behind currently active writes only delays the point
* where the new strategy begins to take effect, so there is no benefit in waiting.
* In thise case the minimum restriction applies: we don't currently allow
* In this case the minimum restriction applies: we don't currently allow
* concurrent catalog updates.
*/
case AT_SetStatistics:
@ -2660,6 +2661,7 @@ AlterTableGetLockLevel(List *cmds)
case AT_SetOptions:
case AT_ResetOptions:
case AT_SetStorage:
case AT_ValidateConstraint:
cmd_lockmode = ShareUpdateExclusiveLock;
break;
@ -2887,6 +2889,7 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
ATPrepAddInherit(rel);
pass = AT_PASS_MISC;
break;
case AT_ValidateConstraint:
case AT_EnableTrig: /* ENABLE TRIGGER variants */
case AT_EnableAlwaysTrig:
case AT_EnableReplicaTrig:
@ -3054,6 +3057,9 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
case AT_AddIndexConstraint: /* ADD CONSTRAINT USING INDEX */
ATExecAddIndexConstraint(tab, rel, (IndexStmt *) cmd->def, lockmode);
break;
case AT_ValidateConstraint:
ATExecValidateConstraint(rel, cmd->name);
break;
case AT_DropConstraint: /* DROP CONSTRAINT */
ATExecDropConstraint(rel, cmd->name, cmd->behavior,
false, false,
@ -3307,10 +3313,15 @@ ATRewriteTables(List **wqueue, LOCKMODE lockmode)
*/
refrel = heap_open(con->refrelid, ShareRowExclusiveLock);
validateForeignKeyConstraint(fkconstraint, rel, refrel,
validateForeignKeyConstraint(fkconstraint->conname, rel, refrel,
con->refindid,
con->conid);
/*
* No need to mark the constraint row as validated,
* we did that when we inserted the row earlier.
*/
heap_close(refrel, NoLock);
}
}
@ -5509,6 +5520,7 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation rel,
CONSTRAINT_FOREIGN,
fkconstraint->deferrable,
fkconstraint->initdeferred,
!fkconstraint->skip_validation,
RelationGetRelid(rel),
fkattnum,
numfks,
@ -5538,7 +5550,8 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation rel,
/*
* Tell Phase 3 to check that the constraint is satisfied by existing rows
* (we can skip this during table creation).
* We can skip this during table creation or if requested explicitly
* by specifying NOT VALID on an alter table statement.
*/
if (!fkconstraint->skip_validation)
{
@ -5561,6 +5574,86 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation rel,
heap_close(pkrel, NoLock);
}
/*
* ALTER TABLE VALIDATE CONSTRAINT
*/
static void
ATExecValidateConstraint(Relation rel, const char *constrName)
{
Relation conrel;
Form_pg_constraint con;
SysScanDesc scan;
ScanKeyData key;
HeapTuple tuple;
bool found = false;
Oid conid;
conrel = heap_open(ConstraintRelationId, RowExclusiveLock);
/*
* Find and the target constraint
*/
ScanKeyInit(&key,
Anum_pg_constraint_conrelid,
BTEqualStrategyNumber, F_OIDEQ,
ObjectIdGetDatum(RelationGetRelid(rel)));
scan = systable_beginscan(conrel, ConstraintRelidIndexId,
true, SnapshotNow, 1, &key);
while (HeapTupleIsValid(tuple = systable_getnext(scan)))
{
con = (Form_pg_constraint) GETSTRUCT(tuple);
if (strcmp(NameStr(con->conname), constrName) != 0)
continue;
conid = HeapTupleGetOid(tuple);
found = true;
break;
}
if (found && con->contype == CONSTRAINT_FOREIGN && !con->convalidated)
{
HeapTuple copyTuple = heap_copytuple(tuple);
Form_pg_constraint copy_con = (Form_pg_constraint) GETSTRUCT(copyTuple);
Relation refrel;
/*
* Triggers are already in place on both tables, so a
* concurrent write that alters the result here is not
* possible. Normally we can run a query here to do the
* validation, which would only require AccessShareLock.
* In some cases, it is possible that we might need to
* fire triggers to perform the check, so we take a lock
* at RowShareLock level just in case.
*/
refrel = heap_open(con->confrelid, RowShareLock);
validateForeignKeyConstraint((char *)constrName, rel, refrel,
con->conindid,
conid);
/*
* Now update the catalog, while we have the door open.
*/
copy_con->convalidated = true;
simple_heap_update(conrel, &copyTuple->t_self, copyTuple);
CatalogUpdateIndexes(conrel, copyTuple);
heap_freetuple(copyTuple);
heap_close(refrel, NoLock);
}
systable_endscan(scan);
heap_close(conrel, RowExclusiveLock);
if (!found)
{
ereport(ERROR,
(errcode(ERRCODE_UNDEFINED_OBJECT),
errmsg("foreign key constraint \"%s\" of relation \"%s\" does not exist",
constrName, RelationGetRelationName(rel))));
}
}
/*
* transformColumnNameList - transform list of column names
@ -5866,7 +5959,7 @@ checkFkeyPermissions(Relation rel, int16 *attnums, int natts)
* Caller must have opened and locked both relations appropriately.
*/
static void
validateForeignKeyConstraint(Constraint *fkconstraint,
validateForeignKeyConstraint(char *conname,
Relation rel,
Relation pkrel,
Oid pkindOid,
@ -5881,7 +5974,7 @@ validateForeignKeyConstraint(Constraint *fkconstraint,
*/
MemSet(&trig, 0, sizeof(trig));
trig.tgoid = InvalidOid;
trig.tgname = fkconstraint->conname;
trig.tgname = conname;
trig.tgenabled = TRIGGER_FIRES_ON_ORIGIN;
trig.tgisinternal = TRUE;
trig.tgconstrrelid = RelationGetRelid(pkrel);