schemachanger: Enable ADD FOREIGN KEY by default and add tests

pkg/bench/rttanalysis/testdata/benchmark_expectations

@@ -9,8 +9,8 @@ exp,benchmark
 12,AlterTableAddColumn/alter_table_add_2_columns
 12,AlterTableAddColumn/alter_table_add_3_columns
 13,AlterTableAddForeignKey/alter_table_add_1_foreign_key
-17,AlterTableAddForeignKey/alter_table_add_2_foreign_keys
-21,AlterTableAddForeignKey/alter_table_add_3_foreign_keys
+13,AlterTableAddForeignKey/alter_table_add_2_foreign_keys
+13,AlterTableAddForeignKey/alter_table_add_3_foreign_keys
 13,AlterTableAddForeignKey/alter_table_add_foreign_key_with_3_columns
 8,AlterTableConfigureZone/alter_table_configure_zone_5_replicas
 8,AlterTableConfigureZone/alter_table_configure_zone_7_replicas_

pkg/sql/logictest/testdata/logic_test/fk

@@ -2715,7 +2715,7 @@ statement ok
 CREATE TABLE child_duplicate_cols (id INT, parent_id INT, PRIMARY KEY (id))
 
 # The fk constraint is invalid because it has duplicate columns, so automatically adding the index fails
-statement error foreign key contains duplicate column \"parent_id\"
+statement error pgcode 42830 foreign key contains duplicate column ".*parent_id"
 ALTER TABLE child_duplicate_cols ADD CONSTRAINT fk FOREIGN KEY (parent_id, parent_id) references parent
 
 statement ok

pkg/sql/schema_changer_test.go

@@ -5550,6 +5550,7 @@ func TestTableValidityWhileAddingFK(t *testing.T) {
 CREATE DATABASE t;
 CREATE TABLE t.child (a INT PRIMARY KEY, b INT, INDEX (b));
 CREATE TABLE t.parent (a INT PRIMARY KEY);
+SET use_declarative_schema_changer = off;
 `); err != nil {
 		t.Fatal(err)
 	}
@@ -6769,6 +6770,7 @@ func TestRollbackForeignKeyAddition(t *testing.T) {
 	tdb.Exec(t, `CREATE DATABASE db`)
 	tdb.Exec(t, `CREATE TABLE db.t (a INT PRIMARY KEY)`)
 	tdb.Exec(t, `CREATE TABLE db.t2 (a INT)`)
+	tdb.Exec(t, `SET use_declarative_schema_changer = off`)
 
 	g := ctxgroup.WithContext(ctx)
 	g.GoCtx(func(ctx context.Context) error {

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

@@ -61,6 +61,11 @@ 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 {
+			return true
+		}
+
 		return false
 	}},
 }
@@ -72,6 +77,7 @@ var supportedAlterTableStatements = map[reflect.Type]supportedAlterTableCommand{
 var alterTableAddConstraintMinSupportedClusterVersion = map[string]clusterversion.Key{
 	"ADD_PRIMARY_KEY_DEFAULT": clusterversion.V22_2Start,
 	"ADD_CHECK_DEFAULT":       clusterversion.V23_1Start,
+	"ADD_FOREIGN_KEY_DEFAULT": clusterversion.V23_1Start,
 }
 
 func init() {
@@ -163,6 +169,8 @@ func alterTableAddConstraintSupportedInCurrentClusterVersion(
 		}
 	case *tree.CheckConstraintTableDef:
 		cmdKey = "ADD_CHECK"
+	case *tree.ForeignKeyConstraintTableDef:
+		cmdKey = "ADD_FOREIGN_KEY"
 	}
 	// Figure out command validation behavior: DEFAULT or SKIP
 	if constraint.ValidationBehavior == tree.ValidationDefault {

pkg/sql/schemachanger/scbuild/testdata/drop_table

@@ -121,13 +121,13 @@ DROP TABLE defaultdb.shipments CASCADE;
   {comment: pkey is good, indexId: 1, tableId: 109}
 - [[IndexData:{DescID: 109, IndexID: 1}, ABSENT], PUBLIC]
   {indexId: 1, tableId: 109}
-- [[ForeignKeyConstraint:{DescID: 109, ConstraintID: 2, ReferencedDescID: 104}, ABSENT], PUBLIC]
+- [[ForeignKeyConstraint:{DescID: 109, IndexID: 0, ConstraintID: 2, ReferencedDescID: 104}, ABSENT], PUBLIC]
   {columnIds: [4], constraintId: 2, referencedColumnIds: [1], referencedTableId: 104, tableId: 109}
 - [[ConstraintWithoutIndexName:{DescID: 109, Name: fk_customers, ConstraintID: 2}, ABSENT], PUBLIC]
   {constraintId: 2, name: fk_customers, tableId: 109}
 - [[ConstraintComment:{DescID: 109, ConstraintID: 2, Comment: customer is important}, ABSENT], PUBLIC]
   {comment: customer is important, constraintId: 2, tableId: 109}
-- [[ForeignKeyConstraint:{DescID: 109, ConstraintID: 3, ReferencedDescID: 105}, ABSENT], PUBLIC]
+- [[ForeignKeyConstraint:{DescID: 109, IndexID: 0, ConstraintID: 3, ReferencedDescID: 105}, ABSENT], PUBLIC]
   {columnIds: [4], constraintId: 3, referencedColumnIds: [2], referencedTableId: 105, tableId: 109}
 - [[ConstraintWithoutIndexName:{DescID: 109, Name: fk_orders, ConstraintID: 3}, ABSENT], PUBLIC]
   {constraintId: 3, name: fk_orders, tableId: 109}

pkg/sql/schemachanger/scbuild/testdata/unimplemented_alter_table

@@ -12,10 +12,6 @@ CREATE TABLE defaultdb.foo (
 );
 ----
 
-unimplemented
-ALTER TABLE defaultdb.foo ADD COLUMN j INT REFERENCES defaultdb.foo(i)
-----
-
 unimplemented
 ALTER TABLE defaultdb.foo ADD COLUMN j SERIAL
 ----
@@ -92,10 +88,6 @@ unimplemented
 ALTER TABLE defaultdb.foo ADD UNIQUE(i);
 ----
 
-unimplemented
-ALTER TABLE defaultdb.foo ADD FOREIGN KEY (i) REFERENCES defaultdb.foo(i);
-----
-
 unimplemented
 ALTER TABLE defaultdb.foo ADD PRIMARY KEY (l);
 ----

pkg/sql/schemachanger/scplan/internal/rules/assertions_test.go

@@ -58,10 +58,15 @@ func nonNilElement(element scpb.Element) scpb.Element {
 
 // Assert that only simple dependents (non-descriptor, non-index, non-column)
 // and data elements have screl.ReferencedDescID attributes.
+// One exception is foreign key constraint, which is not simple dependent nor data
+// element but it has a screl.ReferencedDescID attribute.
 func checkSimpleDependentsReferenceDescID(e scpb.Element) error {
 	if isSimpleDependent(e) || isData(e) {
 		return nil
 	}
+	if _, ok := e.(*scpb.ForeignKeyConstraint); ok {
+		return nil
+	}
 	if _, err := screl.Schema.GetAttribute(screl.ReferencedDescID, e); err == nil {
 		return errors.New("unexpected screl.ReferencedDescID attr")
 	}

pkg/sql/schemachanger/scplan/internal/rules/helpers.go

@@ -374,8 +374,11 @@ func isIndexDependent(e scpb.Element) bool {
 // TODO (xiang): Expand this predicate to include other non-index-backed constraints
 // when we properly support adding/dropping them in the new schema changer.
 func isSupportedNonIndexBackedConstraint(e scpb.Element) bool {
-	_, ok := e.(*scpb.CheckConstraint)
-	return ok
+	switch e.(type) {
+	case *scpb.CheckConstraint, *scpb.ForeignKeyConstraint:
+		return true
+	}
+	return false
 }
 
 func isConstraint(e scpb.Element) bool {

pkg/sql/schemachanger/scplan/internal/rules/op_drop.go

@@ -169,6 +169,7 @@ func init() {
 			),
 			constraint.Type(
 				(*scpb.CheckConstraint)(nil),
+				(*scpb.ForeignKeyConstraint)(nil),
 				(*scpb.UniqueWithoutIndexConstraint)(nil),
 			),