schemachanger: Implement ADD FOREIGN KEY NOT VALID and add tests

pkg/sql/logictest/testdata/logic_test/drop_table

@@ -120,12 +120,21 @@ subtest drop_table_with_not_valid_constraints
 statement ok
 CREATE TABLE t_with_not_valid_constraints_1 (i INT PRIMARY KEY, j INT);
 
+statement ok
+CREATE TABLE t_with_not_valid_constraints_2 (i INT PRIMARY KEY, j INT);
+
 statement ok
 ALTER TABLE t_with_not_valid_constraints_1 ADD CHECK (i > 0) NOT VALID;
 
 statement ok
 SET experimental_enable_unique_without_index_constraints = true;
 ALTER TABLE t_with_not_valid_constraints_1 ADD UNIQUE WITHOUT INDEX (j) NOT VALID;
 
+statement ok
+ALTER TABLE t_with_not_valid_constraints_1 ADD FOREIGN KEY (i) REFERENCES t_with_not_valid_constraints_2 (i) NOT VALID;
+
 statement ok
 DROP TABLE t_with_not_valid_constraints_1;
+
+statement ok
+DROP TABLE t_with_not_valid_constraints_2;

pkg/sql/schemachanger/scbuild/internal/scbuildstmt/alter_table.go

@@ -65,8 +65,8 @@ var supportedAlterTableStatements = map[reflect.Type]supportedAlterTableCommand{
 			return true
 		}
 
-		// Support ALTER TABLE ... ADD CONSTRAINT FOREIGN KEY
-		if _, ok := t.ConstraintDef.(*tree.ForeignKeyConstraintTableDef); ok && t.ValidationBehavior == tree.ValidationDefault {
+		// Support ALTER TABLE ... ADD CONSTRAINT FOREIGN KEY [NOT VALID]
+		if _, ok := t.ConstraintDef.(*tree.ForeignKeyConstraintTableDef); ok {
 			return true
 		}
 
@@ -86,6 +86,7 @@ var alterTableAddConstraintMinSupportedClusterVersion = map[string]clusterversio
 	"ADD_UNIQUE_WITHOUT_INDEX_DEFAULT": clusterversion.V23_1Start,
 	"ADD_CHECK_SKIP":                   clusterversion.V23_1,
 	"ADD_UNIQUE_WITHOUT_INDEX_SKIP":    clusterversion.V23_1,
+	"ADD_FOREIGN_KEY_SKIP":             clusterversion.V23_1,
 }
 
 func init() {

pkg/sql/schemachanger/scbuild/internal/scbuildstmt/alter_table_add_constraint.go

@@ -47,9 +47,7 @@ func alterTableAddConstraint(
 	case *tree.CheckConstraintTableDef:
 		alterTableAddCheck(b, tn, tbl, t)
 	case *tree.ForeignKeyConstraintTableDef:
-		if t.ValidationBehavior == tree.ValidationDefault {
-			alterTableAddForeignKey(b, tn, tbl, t)
-		}
+		alterTableAddForeignKey(b, tn, tbl, t)
 	}
 }
 
@@ -173,7 +171,7 @@ func getIndexIDForValidationForConstraint(b BuildCtx, tableID catid.DescID) (ret
 }
 
 // alterTableAddForeignKey contains logic for building
-// `ALTER TABLE ... ADD CONSTRAINT ... FOREIGN KEY`.
+// `ALTER TABLE ... ADD FOREIGN KEY ... [NOT VALID]`.
 // It assumes `t` is such a command.
 func alterTableAddForeignKey(
 	b BuildCtx, tn *tree.TableName, tbl *scpb.Table, t *tree.AlterTableAddConstraint,
@@ -369,19 +367,34 @@ func alterTableAddForeignKey(
 	// 12. (Finally!) Add a ForeignKey_Constraint, ConstraintName element to
 	// builder state.
 	constraintID := b.NextTableConstraintID(tbl.TableID)
-	fk := &scpb.ForeignKeyConstraint{
-		TableID:                 tbl.TableID,
-		ConstraintID:            constraintID,
-		ColumnIDs:               originColIDs,
-		ReferencedTableID:       referencedTableID,
-		ReferencedColumnIDs:     referencedColIDs,
-		OnUpdateAction:          tree.ForeignKeyReferenceActionValue[fkDef.Actions.Update],
-		OnDeleteAction:          tree.ForeignKeyReferenceActionValue[fkDef.Actions.Delete],
-		CompositeKeyMatchMethod: tree.CompositeKeyMatchMethodValue[fkDef.Match],
-		IndexIDForValidation:    getIndexIDForValidationForConstraint(b, tbl.TableID),
-	}
-	b.Add(fk)
-	b.LogEventForExistingTarget(fk)
+	if t.ValidationBehavior == tree.ValidationDefault {
+		fk := &scpb.ForeignKeyConstraint{
+			TableID:                 tbl.TableID,
+			ConstraintID:            constraintID,
+			ColumnIDs:               originColIDs,
+			ReferencedTableID:       referencedTableID,
+			ReferencedColumnIDs:     referencedColIDs,
+			OnUpdateAction:          tree.ForeignKeyReferenceActionValue[fkDef.Actions.Update],
+			OnDeleteAction:          tree.ForeignKeyReferenceActionValue[fkDef.Actions.Delete],
+			CompositeKeyMatchMethod: tree.CompositeKeyMatchMethodValue[fkDef.Match],
+			IndexIDForValidation:    getIndexIDForValidationForConstraint(b, tbl.TableID),
+		}
+		b.Add(fk)
+		b.LogEventForExistingTarget(fk)
+	} else {
+		fk := &scpb.ForeignKeyConstraintUnvalidated{
+			TableID:                 tbl.TableID,
+			ConstraintID:            constraintID,
+			ColumnIDs:               originColIDs,
+			ReferencedTableID:       referencedTableID,
+			ReferencedColumnIDs:     referencedColIDs,
+			OnUpdateAction:          tree.ForeignKeyReferenceActionValue[fkDef.Actions.Update],
+			OnDeleteAction:          tree.ForeignKeyReferenceActionValue[fkDef.Actions.Delete],
+			CompositeKeyMatchMethod: tree.CompositeKeyMatchMethodValue[fkDef.Match],
+		}
+		b.Add(fk)
+		b.LogEventForExistingTarget(fk)
+	}
 	b.Add(&scpb.ConstraintWithoutIndexName{
 		TableID:      tbl.TableID,
 		ConstraintID: constraintID,

pkg/sql/schemachanger/scbuild/testdata/alter_table_add_foreign_key

@@ -0,0 +1,12 @@
+setup
+CREATE TABLE t2 (i INT PRIMARY KEY);
+CREATE TABLE t1 (i INT PRIMARY KEY);
+----
+
+build
+ALTER TABLE t1 ADD FOREIGN KEY (i) REFERENCES t2(i);
+----
+- [[ForeignKeyConstraint:{DescID: 105, IndexID: 0, ConstraintID: 2, ReferencedDescID: 104}, PUBLIC], ABSENT]
+  {columnIds: [1], constraintId: 2, referencedColumnIds: [1], referencedTableId: 104, tableId: 105}
+- [[ConstraintWithoutIndexName:{DescID: 105, Name: t1_i_fkey, ConstraintID: 2}, PUBLIC], ABSENT]
+  {constraintId: 2, name: t1_i_fkey, tableId: 105}

pkg/sql/schemachanger/scbuild/testdata/alter_table_add_foreign_key_unvalidated

@@ -0,0 +1,12 @@
+setup
+CREATE TABLE t2 (i INT PRIMARY KEY);
+CREATE TABLE t1 (i INT PRIMARY KEY);
+----
+
+build
+ALTER TABLE t1 ADD FOREIGN KEY (i) REFERENCES t2(i) NOT VALID;
+----
+- [[ForeignKeyConstraintUnvalidated:{DescID: 105, ConstraintID: 2, ReferencedDescID: 104}, PUBLIC], ABSENT]
+  {columnIds: [1], constraintId: 2, referencedColumnIds: [1], referencedTableId: 104, tableId: 105}
+- [[ConstraintWithoutIndexName:{DescID: 105, Name: t1_i_fkey, ConstraintID: 2}, PUBLIC], ABSENT]
+  {constraintId: 2, name: t1_i_fkey, tableId: 105}

pkg/sql/schemachanger/scdecomp/decomp.go

@@ -680,17 +680,29 @@ func (w *walkCtx) walkCheckConstraint(tbl catalog.TableDescriptor, c catalog.Che
 func (w *walkCtx) walkForeignKeyConstraint(
 	tbl catalog.TableDescriptor, c catalog.ForeignKeyConstraint,
 ) {
-	// TODO(postamar): proper handling of constraint status
-	w.ev(scpb.Status_PUBLIC, &scpb.ForeignKeyConstraint{
-		TableID:                 tbl.GetID(),
-		ConstraintID:            c.GetConstraintID(),
-		ColumnIDs:               c.ForeignKeyDesc().OriginColumnIDs,
-		ReferencedTableID:       c.GetReferencedTableID(),
-		ReferencedColumnIDs:     c.ForeignKeyDesc().ReferencedColumnIDs,
-		OnUpdateAction:          c.OnUpdate(),
-		OnDeleteAction:          c.OnDelete(),
-		CompositeKeyMatchMethod: c.Match(),
-	})
+	if c.IsConstraintUnvalidated() && w.clusterVersion.IsActive(clusterversion.V23_1) {
+		w.ev(scpb.Status_PUBLIC, &scpb.ForeignKeyConstraintUnvalidated{
+			TableID:                 tbl.GetID(),
+			ConstraintID:            c.GetConstraintID(),
+			ColumnIDs:               c.ForeignKeyDesc().OriginColumnIDs,
+			ReferencedTableID:       c.GetReferencedTableID(),
+			ReferencedColumnIDs:     c.ForeignKeyDesc().ReferencedColumnIDs,
+			OnUpdateAction:          c.OnUpdate(),
+			OnDeleteAction:          c.OnDelete(),
+			CompositeKeyMatchMethod: c.Match(),
+		})
+	} else {
+		w.ev(scpb.Status_PUBLIC, &scpb.ForeignKeyConstraint{
+			TableID:                 tbl.GetID(),
+			ConstraintID:            c.GetConstraintID(),
+			ColumnIDs:               c.ForeignKeyDesc().OriginColumnIDs,
+			ReferencedTableID:       c.GetReferencedTableID(),
+			ReferencedColumnIDs:     c.ForeignKeyDesc().ReferencedColumnIDs,
+			OnUpdateAction:          c.OnUpdate(),
+			OnDeleteAction:          c.OnDelete(),
+			CompositeKeyMatchMethod: c.Match(),
+		})
+	}
 	w.ev(scpb.Status_PUBLIC, &scpb.ConstraintWithoutIndexName{
 		TableID:      tbl.GetID(),
 		ConstraintID: c.GetConstraintID(),

pkg/sql/schemachanger/scdecomp/testdata/table

@@ -18,6 +18,7 @@ COMMENT ON CONSTRAINT tbl_pkey ON tbl IS 'primary key';
 ALTER TABLE tbl ADD CONSTRAINT mycheck2 CHECK (id < 10) NOT VALID;
 ALTER TABLE tbl ADD CONSTRAINT myuwi1 UNIQUE WITHOUT INDEX (price);
 ALTER TABLE tbl ADD CONSTRAINT myuwi2 UNIQUE WITHOUT INDEX (price) NOT VALID;
+ALTER TABLE tbl ADD CONSTRAINT myfk2 FOREIGN KEY (id) REFERENCES parent (id) NOT VALID;
 ----
 
 decompose
@@ -374,6 +375,18 @@ ElementState:
     referencedTableId: 104
     tableId: 105
   Status: PUBLIC
+- ForeignKeyConstraintUnvalidated:
+    columnIds:
+    - 1
+    compositeKeyMatchMethod: SIMPLE
+    constraintId: 7
+    onDeleteAction: NO_ACTION
+    onUpdateAction: NO_ACTION
+    referencedColumnIds:
+    - 1
+    referencedTableId: 104
+    tableId: 105
+  Status: PUBLIC
 - TableComment:
     comment: tbl is good table
     tableId: 105
@@ -678,6 +691,11 @@ ElementState:
     name: myuwi2
     tableId: 105
   Status: PUBLIC
+- ConstraintWithoutIndexName:
+    constraintId: 7
+    name: myfk2
+    tableId: 105
+  Status: PUBLIC
 - ConstraintComment:
     comment: must have a parent
     constraintId: 2

pkg/sql/schemachanger/scexec/scmutationexec/constraint.go

@@ -367,8 +367,8 @@ func (i *immediateVisitor) RemoveUniqueWithoutIndexConstraint(
 	return nil
 }
 
-func (i *immediateVisitor) MakeAbsentForeignKeyConstraintWriteOnly(
-	ctx context.Context, op scop.MakeAbsentForeignKeyConstraintWriteOnly,
+func (i *immediateVisitor) AddForeignKeyConstraint(
+	ctx context.Context, op scop.AddForeignKeyConstraint,
 ) error {
 	out, err := i.checkOutTable(ctx, op.TableID)
 	if err != nil || out.Dropped() {
@@ -378,21 +378,27 @@ func (i *immediateVisitor) MakeAbsentForeignKeyConstraintWriteOnly(
 		out.NextConstraintID = op.ConstraintID + 1
 	}
 
-	// Enqueue a mutation in `out` to signal this mutation is now enforced.
 	fk := &descpb.ForeignKeyConstraint{
 		OriginTableID:       op.TableID,
 		OriginColumnIDs:     op.ColumnIDs,
 		ReferencedColumnIDs: op.ReferencedColumnIDs,
 		ReferencedTableID:   op.ReferencedTableID,
 		Name:                tabledesc.ConstraintNamePlaceholder(op.ConstraintID),
-		Validity:            descpb.ConstraintValidity_Validating,
+		Validity:            op.Validity,
 		OnDelete:            op.OnDeleteAction,
 		OnUpdate:            op.OnUpdateAction,
 		Match:               op.CompositeKeyMatchMethod,
 		ConstraintID:        op.ConstraintID,
 	}
-	enqueueNonIndexMutation(out, out.AddForeignKeyMutation, fk, descpb.DescriptorMutation_ADD)
-	out.Mutations[len(out.Mutations)-1].State = descpb.DescriptorMutation_WRITE_ONLY
+	if op.Validity == descpb.ConstraintValidity_Unvalidated {
+		// Unvalidated constraint doesn't need to transition through an intermediate
+		// state, so we don't enqueue a mutation for it.
+		out.OutboundFKs = append(out.OutboundFKs, *fk)
+	} else {
+		enqueueNonIndexMutation(out, out.AddForeignKeyMutation, fk, descpb.DescriptorMutation_ADD)
+		out.Mutations[len(out.Mutations)-1].State = descpb.DescriptorMutation_WRITE_ONLY
+	}
+
 	// Add an entry in "InboundFKs" in the referenced table as company.
 	in, err := i.checkOutTable(ctx, op.ReferencedTableID)
 	if err != nil {
@@ -454,14 +460,16 @@ func (i *immediateVisitor) MakePublicForeignKeyConstraintValidated(
 	ctx context.Context, op scop.MakePublicForeignKeyConstraintValidated,
 ) error {
 	// A helper function to update the inbound FK in referenced table to DROPPING.
-	updateInboundFKAsDropping := func(referencedTableID descpb.ID) error {
+	// Note that the constraintID in the referencedTable might not match that in
+	// in the referencing table, so we pass in the FK name as identifier.
+	updateInboundFKAsDropping := func(referencedTableID descpb.ID, fkName string) error {
 		foundInReferencedTable := false
 		in, err := i.checkOutTable(ctx, referencedTableID)
 		if err != nil {
 			return err
 		}
 		for idx, inboundFk := range in.InboundFKs {
-			if inboundFk.OriginTableID == op.TableID && inboundFk.ConstraintID == op.ConstraintID {
+			if inboundFk.OriginTableID == op.TableID && inboundFk.Name == fkName {
 				in.InboundFKs[idx].Validity = descpb.ConstraintValidity_Dropping
 				foundInReferencedTable = true
 				break
@@ -486,7 +494,7 @@ func (i *immediateVisitor) MakePublicForeignKeyConstraintValidated(
 			}
 			fk.Validity = descpb.ConstraintValidity_Dropping
 			enqueueNonIndexMutation(out, out.AddForeignKeyMutation, &fk, descpb.DescriptorMutation_DROP)
-			return updateInboundFKAsDropping(fk.ReferencedTableID)
+			return updateInboundFKAsDropping(fk.ReferencedTableID, fk.Name)
 		}
 	}
 

pkg/sql/schemachanger/scexec/scmutationexec/references.go

@@ -20,6 +20,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/catid"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/iterutil"
+	"github.com/cockroachdb/errors"
 )
 
 func (i *immediateVisitor) RemoveSchemaParent(
@@ -73,6 +74,29 @@ func (i *immediateVisitor) RemoveForeignKeyBackReference(
 		// Exit early if the foreign key back-reference holder is getting dropped.
 		return err
 	}
+	// Retrieve foreign key name in origin table to identify it in the referenced
+	// table.
+	var name string
+	{
+		out, err := i.getDescriptor(ctx, op.OriginTableID)
+		if err != nil {
+			return err
+		}
+		tbl, err := catalog.AsTableDescriptor(out)
+		if err != nil {
+			return err
+		}
+		for _, fk := range tbl.OutboundForeignKeys() {
+			if fk.GetConstraintID() == op.OriginConstraintID {
+				name = fk.GetName()
+				break
+			}
+		}
+		if name == "" {
+			return errors.AssertionFailedf("foreign key with ID %d not found in origin table %q (%d)",
+				op.OriginConstraintID, out.GetName(), out.GetID())
+		}
+	}
 	// Attempt to remove back reference.
 	// Note how we
 	//  1. only check to remove from `in.InboundFKs` but not from `in.Mutations`:
@@ -83,7 +107,7 @@ func (i *immediateVisitor) RemoveForeignKeyBackReference(
 	//  this is because if we roll back before the adding FK is published in `out`,
 	//  such a back-reference won't exist in `in` yet.
 	for idx, fk := range in.InboundFKs {
-		if fk.OriginTableID == op.OriginTableID && fk.ConstraintID == op.OriginConstraintID {
+		if fk.OriginTableID == op.OriginTableID && fk.Name == name {
 			in.InboundFKs = append(in.InboundFKs[:idx], in.InboundFKs[idx+1:]...)
 			if len(in.InboundFKs) == 0 {
 				in.InboundFKs = nil

pkg/sql/schemachanger/scop/immediate_mutation.go

@@ -318,9 +318,9 @@ type MakeValidatedColumnNotNullPublic struct {
 	ColumnID descpb.ColumnID
 }
 
-// MakeAbsentForeignKeyConstraintWriteOnly adds a non-existent foreign key
-// constraint to the table in the WRITE_ONLY state.
-type MakeAbsentForeignKeyConstraintWriteOnly struct {
+// AddForeignKeyConstraint adds a non-existent foreign key
+// constraint to the table.
+type AddForeignKeyConstraint struct {
 	immediateMutationOp
 	TableID                 descpb.ID
 	ConstraintID            descpb.ConstraintID
@@ -330,6 +330,7 @@ type MakeAbsentForeignKeyConstraintWriteOnly struct {
 	OnUpdateAction          semenumpb.ForeignKeyAction
 	OnDeleteAction          semenumpb.ForeignKeyAction
 	CompositeKeyMatchMethod semenumpb.Match
+	Validity                descpb.ConstraintValidity
 }
 
 // MakeValidatedForeignKeyConstraintPublic moves a new, validated foreign key

pkg/sql/schemachanger/scpb/elements.proto

@@ -75,6 +75,7 @@ message ElementProto {
   CheckConstraint check_constraint = 26 [(gogoproto.moretags) = "parent:\"Table\""];
   CheckConstraintUnvalidated check_constraint_unvalidated = 170 [(gogoproto.moretags) = "parent:\"Table\""];
   ForeignKeyConstraint foreign_key_constraint = 27 [(gogoproto.moretags) = "parent:\"Table\""];
+  ForeignKeyConstraintUnvalidated foreign_key_constraint_not_valid = 172 [(gogoproto.moretags) = "parent:\"Table\""];
   TableComment table_comment = 28 [(gogoproto.moretags) = "parent:\"Table, View, Sequence\""];
   RowLevelTTL row_level_ttl = 29 [(gogoproto.customname) = "RowLevelTTL", (gogoproto.moretags) = "parent:\"Table\""];
   TableZoneConfig table_zone_config = 121 [(gogoproto.moretags) = "parent:\"Table, View\""];
@@ -398,6 +399,17 @@ message ForeignKeyConstraint {
   uint32 index_id_for_validation = 9 [(gogoproto.customname) = "IndexIDForValidation", (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/sql/sem/catid.IndexID"];
 }
 
+message ForeignKeyConstraintUnvalidated {
+  uint32 table_id = 1 [(gogoproto.customname) = "TableID", (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/sql/sem/catid.DescID"];
+  uint32 constraint_id = 2 [(gogoproto.customname) = "ConstraintID", (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/sql/sem/catid.ConstraintID"];
+  repeated uint32 column_ids = 3 [(gogoproto.customname) = "ColumnIDs", (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/sql/sem/catid.ColumnID"];
+  uint32 referenced_table_id = 4 [(gogoproto.customname) = "ReferencedTableID", (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/sql/sem/catid.DescID"];
+  repeated uint32 referenced_column_ids = 5 [(gogoproto.customname) = "ReferencedColumnIDs", (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/sql/sem/catid.ColumnID"];
+  cockroach.sql.sem.semenumpb.ForeignKeyAction on_update_action = 6 [(gogoproto.customname) = "OnUpdateAction"];
+  cockroach.sql.sem.semenumpb.ForeignKeyAction on_delete_action = 7 [(gogoproto.customname) = "OnDeleteAction"];
+  cockroach.sql.sem.semenumpb.Match composite_key_match_method = 8 [(gogoproto.customname) = "CompositeKeyMatchMethod"];
+}
+
 message EnumType {
   uint32 type_id = 1 [(gogoproto.customname) = "TypeID", (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/sql/sem/catid.DescID"];
   uint32 array_type_id = 2 [(gogoproto.customname) = "ArrayTypeID", (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/sql/sem/catid.DescID"];