From da557aa6b26c40e50ca517cd598c68074f9db22b Mon Sep 17 00:00:00 2001 From: Amul Sul Date: Thu, 2 Jan 2025 17:15:30 +0530 Subject: [PATCH v2 1/2] Refactor: Split ATExecValidateConstraint() Splitting ATExecValidateConstraint() into two separate functions to handle FOREIGN KEY and CHECK constraints respectively. The next patch requires the FOREIGN KEY validation code in a separate function so it can be called directly. Additionally, the function needs to be recursive to handle NOT VALID constraints on declarative partitions. Although moving the CHECK constraint-related code is not strictly necessary for this task, maintaining separate code for FOREIGN KEY constraints makes it logical to do the same for CHECK constraints. This separation will also simplify adding support for other constraints (e.g., NOT NULL) in ATExecValidateConstraint() in the future. --- src/backend/commands/tablecmds.c | 282 +++++++++++++++++++------------ 1 file changed, 171 insertions(+), 111 deletions(-) diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index 4fc54bd6eba..c0881832eb1 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -394,6 +394,11 @@ static ObjectAddress ATExecAlterConstraint(Relation rel, AlterTableCmd *cmd, static bool ATExecAlterConstrRecurse(Constraint *cmdcon, Relation conrel, Relation tgrel, Relation rel, HeapTuple contuple, List **otherrelids, LOCKMODE lockmode); +static void QueueFKConstraintValidation(List **wqueue, Relation conrel, Relation rel, + HeapTuple contuple, LOCKMODE lockmode); +static void QueueCheckConstraintValidation(List **wqueue, Relation conrel, Relation rel, + char *constrName, HeapTuple contuple, + bool recurse, bool recursing, LOCKMODE lockmode); static ObjectAddress ATExecValidateConstraint(List **wqueue, Relation rel, char *constrName, bool recurse, bool recursing, LOCKMODE lockmode); @@ -11973,6 +11978,169 @@ ATExecAlterConstrRecurse(Constraint *cmdcon, Relation conrel, Relation tgrel, return changed; } +/* + * QueueFKConstraintValidation + * + * Add an entry to the wqueue to validate the given foreign key constraint in + * Phase 3 and update the convalidated field in the pg_constraint catalog + * for the specified relation. + */ +static void +QueueFKConstraintValidation(List **wqueue, Relation conrel, Relation rel, + HeapTuple contuple, LOCKMODE lockmode) +{ + Form_pg_constraint con; + AlteredTableInfo *tab; + HeapTuple copyTuple; + Form_pg_constraint copy_con; + + con = (Form_pg_constraint) GETSTRUCT(contuple); + Assert(con->contype == CONSTRAINT_FOREIGN); + + if (rel->rd_rel->relkind == RELKIND_RELATION) + { + NewConstraint *newcon; + Constraint *fkconstraint; + + /* Queue validation for phase 3 */ + fkconstraint = makeNode(Constraint); + /* for now this is all we need */ + fkconstraint->conname = pstrdup(NameStr(con->conname)); + + newcon = (NewConstraint *) palloc0(sizeof(NewConstraint)); + newcon->name = fkconstraint->conname; + newcon->contype = CONSTR_FOREIGN; + newcon->refrelid = con->confrelid; + newcon->refindid = con->conindid; + newcon->conid = con->oid; + newcon->qual = (Node *) fkconstraint; + + /* Find or create work queue entry for this table */ + tab = ATGetQueueEntry(wqueue, rel); + tab->constraints = lappend(tab->constraints, newcon); + } + + /* + * We disallow creating invalid foreign keys to or from partitioned + * tables, so ignoring the recursion bit is okay. + */ + + /* + * Now update the catalog, while we have the door open. + */ + copyTuple = heap_copytuple(contuple); + copy_con = (Form_pg_constraint) GETSTRUCT(copyTuple); + copy_con->convalidated = true; + CatalogTupleUpdate(conrel, ©Tuple->t_self, copyTuple); + + InvokeObjectPostAlterHook(ConstraintRelationId, con->oid, 0); + + heap_freetuple(copyTuple); +} + +/* + * QueueCheckConstraintValidation + * + * Add an entry to the wqueue to validate the given check constraint in Phase 3 + * and update the convalidated field in the pg_constraint catalog for the + * specified relation and all its inheriting children. + */ +static void +QueueCheckConstraintValidation(List **wqueue, Relation conrel, Relation rel, + char *constrName, HeapTuple contuple, + bool recurse, bool recursing, LOCKMODE lockmode) +{ + Form_pg_constraint con; + AlteredTableInfo *tab; + HeapTuple copyTuple; + Form_pg_constraint copy_con; + + List *children = NIL; + ListCell *child; + NewConstraint *newcon; + Datum val; + char *conbin; + + con = (Form_pg_constraint) GETSTRUCT(contuple); + Assert(con->contype == CONSTRAINT_CHECK); + + /* + * If we're recursing, the parent has already done this, so skip it. + * Also, if the constraint is a NO INHERIT constraint, we shouldn't try to + * look for it in the children. + */ + if (!recursing && !con->connoinherit) + children = find_all_inheritors(RelationGetRelid(rel), + lockmode, NULL); + + /* + * For CHECK constraints, we must ensure that we only mark the constraint + * as validated on the parent if it's already validated on the children. + * + * We recurse before validating on the parent, to reduce risk of + * deadlocks. + */ + foreach(child, children) + { + Oid childoid = lfirst_oid(child); + Relation childrel; + + if (childoid == RelationGetRelid(rel)) + continue; + + /* + * If we are told not to recurse, there had better not be any child + * tables, because we can't mark the constraint on the parent valid + * unless it is valid for all child tables. + */ + if (!recurse) + ereport(ERROR, + (errcode(ERRCODE_INVALID_TABLE_DEFINITION), + errmsg("constraint must be validated on child tables too"))); + + /* find_all_inheritors already got lock */ + childrel = table_open(childoid, NoLock); + + ATExecValidateConstraint(wqueue, childrel, constrName, false, + true, lockmode); + table_close(childrel, NoLock); + } + + /* Queue validation for phase 3 */ + newcon = (NewConstraint *) palloc0(sizeof(NewConstraint)); + newcon->name = constrName; + newcon->contype = CONSTR_CHECK; + newcon->refrelid = InvalidOid; + newcon->refindid = InvalidOid; + newcon->conid = con->oid; + + val = SysCacheGetAttrNotNull(CONSTROID, contuple, + Anum_pg_constraint_conbin); + conbin = TextDatumGetCString(val); + newcon->qual = (Node *) stringToNode(conbin); + + /* Find or create work queue entry for this table */ + tab = ATGetQueueEntry(wqueue, rel); + tab->constraints = lappend(tab->constraints, newcon); + + /* + * Invalidate relcache so that others see the new validated constraint. + */ + CacheInvalidateRelcache(rel); + + /* + * Now update the catalog, while we have the door open. + */ + copyTuple = heap_copytuple(contuple); + copy_con = (Form_pg_constraint) GETSTRUCT(copyTuple); + copy_con->convalidated = true; + CatalogTupleUpdate(conrel, ©Tuple->t_self, copyTuple); + + InvokeObjectPostAlterHook(ConstraintRelationId, con->oid, 0); + + heap_freetuple(copyTuple); +} + /* * ALTER TABLE VALIDATE CONSTRAINT * @@ -12037,124 +12205,16 @@ ATExecValidateConstraint(List **wqueue, Relation rel, char *constrName, if (!con->convalidated) { - AlteredTableInfo *tab; - HeapTuple copyTuple; - Form_pg_constraint copy_con; - if (con->contype == CONSTRAINT_FOREIGN) { - NewConstraint *newcon; - Constraint *fkconstraint; - - /* Queue validation for phase 3 */ - fkconstraint = makeNode(Constraint); - /* for now this is all we need */ - fkconstraint->conname = constrName; - - newcon = (NewConstraint *) palloc0(sizeof(NewConstraint)); - newcon->name = constrName; - newcon->contype = CONSTR_FOREIGN; - newcon->refrelid = con->confrelid; - newcon->refindid = con->conindid; - newcon->conid = con->oid; - newcon->qual = (Node *) fkconstraint; - - /* Find or create work queue entry for this table */ - tab = ATGetQueueEntry(wqueue, rel); - tab->constraints = lappend(tab->constraints, newcon); - - /* - * We disallow creating invalid foreign keys to or from - * partitioned tables, so ignoring the recursion bit is okay. - */ + QueueFKConstraintValidation(wqueue, conrel, rel, tuple, lockmode); } else if (con->contype == CONSTRAINT_CHECK) { - List *children = NIL; - ListCell *child; - NewConstraint *newcon; - Datum val; - char *conbin; - - /* - * If we're recursing, the parent has already done this, so skip - * it. Also, if the constraint is a NO INHERIT constraint, we - * shouldn't try to look for it in the children. - */ - if (!recursing && !con->connoinherit) - children = find_all_inheritors(RelationGetRelid(rel), - lockmode, NULL); - - /* - * For CHECK constraints, we must ensure that we only mark the - * constraint as validated on the parent if it's already validated - * on the children. - * - * We recurse before validating on the parent, to reduce risk of - * deadlocks. - */ - foreach(child, children) - { - Oid childoid = lfirst_oid(child); - Relation childrel; - - if (childoid == RelationGetRelid(rel)) - continue; - - /* - * If we are told not to recurse, there had better not be any - * child tables, because we can't mark the constraint on the - * parent valid unless it is valid for all child tables. - */ - if (!recurse) - ereport(ERROR, - (errcode(ERRCODE_INVALID_TABLE_DEFINITION), - errmsg("constraint must be validated on child tables too"))); - - /* find_all_inheritors already got lock */ - childrel = table_open(childoid, NoLock); - - ATExecValidateConstraint(wqueue, childrel, constrName, false, - true, lockmode); - table_close(childrel, NoLock); - } - - /* Queue validation for phase 3 */ - newcon = (NewConstraint *) palloc0(sizeof(NewConstraint)); - newcon->name = constrName; - newcon->contype = CONSTR_CHECK; - newcon->refrelid = InvalidOid; - newcon->refindid = InvalidOid; - newcon->conid = con->oid; - - val = SysCacheGetAttrNotNull(CONSTROID, tuple, - Anum_pg_constraint_conbin); - conbin = TextDatumGetCString(val); - newcon->qual = (Node *) stringToNode(conbin); - - /* Find or create work queue entry for this table */ - tab = ATGetQueueEntry(wqueue, rel); - tab->constraints = lappend(tab->constraints, newcon); - - /* - * Invalidate relcache so that others see the new validated - * constraint. - */ - CacheInvalidateRelcache(rel); + QueueCheckConstraintValidation(wqueue, conrel, rel, constrName, + tuple, recurse, recursing, lockmode); } - /* - * Now update the catalog, while we have the door open. - */ - copyTuple = heap_copytuple(tuple); - copy_con = (Form_pg_constraint) GETSTRUCT(copyTuple); - copy_con->convalidated = true; - CatalogTupleUpdate(conrel, ©Tuple->t_self, copyTuple); - - InvokeObjectPostAlterHook(ConstraintRelationId, con->oid, 0); - - heap_freetuple(copyTuple); - ObjectAddressSet(address, ConstraintRelationId, con->oid); } else -- 2.43.5