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.ForeignKeyConstraintNotValid{
+			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_not_valid

@@ -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;
+----
+- [[ForeignKeyConstraintNotValid:{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

@@ -667,17 +667,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.ForeignKeyConstraintNotValid{
+			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
@@ -378,6 +379,18 @@ ElementState:
     predicate: null
     tableId: 105
   Status: PUBLIC
+- ForeignKeyConstraintNotValid:
+    columnIds:
+    - 1
+    compositeKeyMatchMethod: SIMPLE
+    constraintId: 7
+    onDeleteAction: NO_ACTION
+    onUpdateAction: NO_ACTION
+    referencedColumnIds:
+    - 1
+    referencedTableId: 104
+    tableId: 105
+  Status: PUBLIC
 - TableZoneConfig:
     tableId: 105
   Status: PUBLIC
@@ -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

@@ -428,6 +428,40 @@ func (i *immediateVisitor) MakeAbsentForeignKeyConstraintWriteOnly(
 	return nil
 }
 
+func (i *immediateVisitor) MakeAbsentForeignKeyConstraintNotValidPublic(
+	ctx context.Context, op scop.MakeAbsentForeignKeyConstraintNotValidPublic,
+) error {
+	out, err := i.checkOutTable(ctx, op.TableID)
+	if err != nil || out.Dropped() {
+		return err
+	}
+	if op.ConstraintID >= out.NextConstraintID {
+		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_Unvalidated,
+		OnDelete:            op.OnDeleteAction,
+		OnUpdate:            op.OnUpdateAction,
+		Match:               op.CompositeKeyMatchMethod,
+		ConstraintID:        op.ConstraintID,
+	}
+	out.OutboundFKs = append(out.OutboundFKs, fk)
+	// Add an entry in "InboundFKs" in the referenced table as company.
+	in, err := i.checkOutTable(ctx, op.ReferencedTableID)
+	if err != nil {
+		return err
+	}
+	in.InboundFKs = append(in.InboundFKs, fk)
+	return nil
+}
+
 func (i *immediateVisitor) MakeValidatedForeignKeyConstraintPublic(
 	ctx context.Context, op scop.MakeValidatedForeignKeyConstraintPublic,
 ) error {

pkg/sql/schemachanger/scop/immediate_mutation.go

@@ -341,6 +341,20 @@ type MakeAbsentForeignKeyConstraintWriteOnly struct {
 	CompositeKeyMatchMethod semenumpb.Match
 }
 
+// MakeAbsentForeignKeyConstraintNotValidPublic adds a non-existent, not-valid
+// foreign key constraint to the table.
+type MakeAbsentForeignKeyConstraintNotValidPublic struct {
+	immediateMutationOp
+	TableID                 descpb.ID
+	ConstraintID            descpb.ConstraintID
+	ColumnIDs               []descpb.ColumnID
+	ReferencedTableID       descpb.ID
+	ReferencedColumnIDs     []descpb.ColumnID
+	OnUpdateAction          semenumpb.ForeignKeyAction
+	OnDeleteAction          semenumpb.ForeignKeyAction
+	CompositeKeyMatchMethod semenumpb.Match
+}
+
 // MakeValidatedForeignKeyConstraintPublic moves a new, validated foreign key
 // constraint from mutation to public.
 type MakeValidatedForeignKeyConstraintPublic struct {

pkg/sql/schemachanger/scpb/elements.proto

@@ -77,6 +77,7 @@ message ElementProto {
   RowLevelTTL row_level_ttl = 29 [(gogoproto.customname) = "RowLevelTTL", (gogoproto.moretags) = "parent:\"Table\""];
   CheckConstraintNotValid check_constraint_not_valid = 170 [(gogoproto.moretags) = "parent:\"Table\""];
   UniqueWithoutIndexConstraintNotValid unique_without_index_constraint_not_valid = 171 [(gogoproto.moretags) = "parent:\"Table\""];
+  ForeignKeyConstraintNotValid foreign_key_constraint_not_valid = 172 [(gogoproto.moretags) = "parent:\"Table\""];
   TableZoneConfig table_zone_config = 121 [(gogoproto.moretags) = "parent:\"Table, View\""];
   TableData table_data = 131 [(gogoproto.customname) = "TableData", (gogoproto.moretags) = "parent:\"Table, View, Sequence\""];
   TablePartitioning table_partitioning = 132 [(gogoproto.customname) = "TablePartitioning", (gogoproto.moretags) = "parent:\"Table\""];
@@ -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 ForeignKeyConstraintNotValid {
+  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"];

pkg/sql/schemachanger/scpb/uml/table.puml

@@ -135,6 +135,17 @@ UniqueWithoutIndexConstraintNotValid :  ConstraintID
 UniqueWithoutIndexConstraintNotValid : []ColumnIDs
 UniqueWithoutIndexConstraintNotValid :  Predicate
 
+object ForeignKeyConstraintNotValid
+
+ForeignKeyConstraintNotValid :  TableID
+ForeignKeyConstraintNotValid :  ConstraintID
+ForeignKeyConstraintNotValid : []ColumnIDs
+ForeignKeyConstraintNotValid :  ReferencedTableID
+ForeignKeyConstraintNotValid : []ReferencedColumnIDs
+ForeignKeyConstraintNotValid :  OnUpdateAction
+ForeignKeyConstraintNotValid :  OnDeleteAction
+ForeignKeyConstraintNotValid :  CompositeKeyMatchMethod
+
 object TableZoneConfig
 
 TableZoneConfig :  TableID
@@ -390,6 +401,7 @@ Sequence <|-- TableComment
 Table <|-- RowLevelTTL
 Table <|-- CheckConstraintNotValid
 Table <|-- UniqueWithoutIndexConstraintNotValid
+Table <|-- ForeignKeyConstraintNotValid
 Table <|-- TableZoneConfig
 View <|-- TableZoneConfig
 Table <|-- TableData