Merge #37433

37433: sql: async validation of foreign keys in ADD CONSTRAINT r=lucy-zhang a=lucy-zhang

Currently, validating an unvalidated foreign key constraint requires running
`VALIDATE CONSTRAINT`, which executes a potentially long-running query in the
user transaction. In this PR, `ADD CONSTRAINT` for foreign keys is now
processed by the schema changer, with the validation for existing rows running
in a separate transaction from the original user transaction.

The foreign key mutation (represented by the new `FOREIGN_KEY` enum value in
`ConstraintToUpdate`) is processed in the same way as check constraint
mutations: the foreign key is in the mutations list while other columns and
indexes with the same mutation ID are backfilled, then added to the appropriate
index in the `Validating` state before being validated, and is finalized when
the validation query for existing rows completes successfully. If validation
fails, the transaction is rolled back, with the foreign key (and backreference)
removed from the table descriptor(s) as part of the rollback.

Adding foreign keys to columns or indexes being added is still not supported
and is left for later work. Also unsupported is adding a foreign key constraint
in the same transaction as `CREATE TABLE` that is either validated or that
rolls back the entire transaction on failure. In this PR, the constraint is
just left unvalidated; This needs a follow-up PR to queue a separate mutation
for validating the constraint after it's been added.

Release note (sql change): Foreign keys that are added to an existing table
using `ALTER TABLE` will now be validated for existing rows, with improved
performance compared to running `ADD CONSTRAINT` followed by `VALIDATE
CONSTRAINT` previously.

Co-authored-by: Lucy Zhang <[email protected]>

pkg/sql/alter_table.go

@@ -498,56 +498,55 @@ func (n *alterTableNode) startExec(params runParams) error {
 			switch constraint.Kind {
 			case sqlbase.ConstraintTypeCheck:
 				found := false
-				var idx int
-				for idx = range n.tableDesc.Checks {
-					if n.tableDesc.Checks[idx].Name == name {
+				var ck *sqlbase.TableDescriptor_CheckConstraint
+				for _, c := range n.tableDesc.Checks {
+					// If the constraint is still being validated, don't allow VALIDATE CONSTRAINT to run
+					if c.Name == name && c.Validity != sqlbase.ConstraintValidity_Validating {
 						found = true
+						ck = c
 						break
 					}
 				}
 				if !found {
 					return pgerror.Newf(pgerror.CodeObjectNotInPrerequisiteStateError,
 						"constraint %q in the middle of being added, try again later", t.Constraint)
 				}
-
-				ck := n.tableDesc.Checks[idx]
-				if err := validateCheckExpr(
-					params.ctx, ck.Expr, n.tableDesc.TableDesc(), params.EvalContext().InternalExecutor, params.EvalContext().Txn,
+				if err := validateCheckInTxn(
+					params.ctx, params.p.LeaseMgr(), params.EvalContext(), n.tableDesc, params.EvalContext().Txn, name,
 				); err != nil {
 					return err
 				}
-				n.tableDesc.Checks[idx].Validity = sqlbase.ConstraintValidity_Validated
-				descriptorChanged = true
+				ck.Validity = sqlbase.ConstraintValidity_Validated
 
 			case sqlbase.ConstraintTypeFK:
 				found := false
-				var id sqlbase.IndexID
+				var fkIdx *sqlbase.IndexDescriptor
 				for _, idx := range n.tableDesc.AllNonDropIndexes() {
-					if idx.ForeignKey.IsSet() && idx.ForeignKey.Name == name {
+					fk := &idx.ForeignKey
+					// If the constraint is still being validated, don't allow VALIDATE CONSTRAINT to run
+					if fk.IsSet() && fk.Name == name && fk.Validity != sqlbase.ConstraintValidity_Validating {
 						found = true
-						id = idx.ID
+						fkIdx = idx
 						break
 					}
 				}
 				if !found {
-					return pgerror.AssertionFailedf(
-						"constraint returned by GetConstraintInfo not found")
-				}
-				idx, err := n.tableDesc.FindIndexByID(id)
-				if err != nil {
-					return pgerror.NewAssertionErrorWithWrappedErrf(err, "")
+					return pgerror.Newf(pgerror.CodeObjectNotInPrerequisiteStateError,
+						"constraint %q in the middle of being added, try again later", t.Constraint)
 				}
-				if err := params.p.validateForeignKey(params.ctx, n.tableDesc.TableDesc(), idx); err != nil {
+				if err := validateFkInTxn(
+					params.ctx, params.p.LeaseMgr(), params.EvalContext(), n.tableDesc, params.EvalContext().Txn, name,
+				); err != nil {
 					return err
 				}
-				idx.ForeignKey.Validity = sqlbase.ConstraintValidity_Validated
-				descriptorChanged = true
+				fkIdx.ForeignKey.Validity = sqlbase.ConstraintValidity_Validated
 
 			default:
 				return pgerror.Newf(pgerror.CodeWrongObjectTypeError,
 					"constraint %q of relation %q is not a foreign key or check constraint",
 					tree.ErrString(&t.Constraint), tree.ErrString(n.n.Table))
 			}
+			descriptorChanged = true
 
 		case tree.ColumnMutationCmd:
 			// Column mutations