schemachanger: Implement ADD CONSTRAINT NOT VALID

pkg/sql/logictest/testdata/logic_test/drop_table

@@ -114,3 +114,27 @@ COMMIT;
 
 statement error pgcode 42P01 relation "to_drop" does not exist
 DROP TABLE to_drop;
+
+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/builder_test.go

@@ -93,6 +93,7 @@ func TestBuildDataDriven(t *testing.T) {
 										// changer will allow non-fully implemented operations.
 										sd.NewSchemaChangerMode = sessiondatapb.UseNewSchemaChangerUnsafe
 										sd.ApplicationName = ""
+										sd.EnableUniqueWithoutIndexConstraints = true
 									},
 								),
 							),

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

@@ -52,20 +52,21 @@ var supportedAlterTableStatements = map[reflect.Type]supportedAlterTableCommand{
 	reflect.TypeOf((*tree.AlterTableAddConstraint)(nil)): {fn: alterTableAddConstraint, on: true, extraChecks: func(
 		t *tree.AlterTableAddConstraint,
 	) bool {
-		// Support ALTER TABLE ... ADD PRIMARY KEY
 		if d, ok := t.ConstraintDef.(*tree.UniqueConstraintTableDef); ok && d.PrimaryKey && t.ValidationBehavior == tree.ValidationDefault {
+			// Support ALTER TABLE ... ADD PRIMARY KEY
 			return true
-		} else if ok && d.WithoutIndex && t.ValidationBehavior == tree.ValidationDefault {
+		} else if ok && d.WithoutIndex {
+			// Support ALTER TABLE ... ADD UNIQUE WITHOUT INDEX [NOT VALID]
 			return true
 		}
 
-		// Support ALTER TABLE ... ADD CONSTRAINT CHECK
-		if _, ok := t.ConstraintDef.(*tree.CheckConstraintTableDef); ok && t.ValidationBehavior == tree.ValidationDefault {
+		// Support ALTER TABLE ... ADD CONSTRAINT CHECK [NOT VALID]
+		if _, ok := t.ConstraintDef.(*tree.CheckConstraintTableDef); ok {
 			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
 		}
 
@@ -83,6 +84,9 @@ var alterTableAddConstraintMinSupportedClusterVersion = map[string]clusterversio
 	"ADD_CHECK_DEFAULT":                clusterversion.V23_1Start,
 	"ADD_FOREIGN_KEY_DEFAULT":          clusterversion.V23_1Start,
 	"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() {
@@ -186,7 +190,10 @@ func alterTableAddConstraintSupportedInCurrentClusterVersion(
 		cmdKey += "_SKIP"
 	}
 
-	minSupportedClusterVersion := alterTableAddConstraintMinSupportedClusterVersion[cmdKey]
+	minSupportedClusterVersion, ok := alterTableAddConstraintMinSupportedClusterVersion[cmdKey]
+	if !ok {
+		return false
+	}
 	return b.EvalCtx().Settings.Version.IsActive(b, minSupportedClusterVersion)
 }
 

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

@@ -41,17 +41,13 @@ func alterTableAddConstraint(
 	case *tree.UniqueConstraintTableDef:
 		if d.PrimaryKey && t.ValidationBehavior == tree.ValidationDefault {
 			alterTableAddPrimaryKey(b, tn, tbl, t)
-		} else if d.WithoutIndex && t.ValidationBehavior == tree.ValidationDefault {
+		} else if d.WithoutIndex {
 			alterTableAddUniqueWithoutIndex(b, tn, tbl, t)
 		}
 	case *tree.CheckConstraintTableDef:
-		if t.ValidationBehavior == tree.ValidationDefault {
-			alterTableAddCheck(b, tn, tbl, t)
-		}
+		alterTableAddCheck(b, tn, tbl, t)
 	case *tree.ForeignKeyConstraintTableDef:
-		if t.ValidationBehavior == tree.ValidationDefault {
-			alterTableAddForeignKey(b, tn, tbl, t)
-		}
+		alterTableAddForeignKey(b, tn, tbl, t)
 	}
 }
 
@@ -80,7 +76,7 @@ func alterTableAddPrimaryKey(
 }
 
 // alterTableAddCheck contains logic for building
-// `ALTER TABLE ... ADD CONSTRAINT ... CHECK`.
+// `ALTER TABLE ... ADD CHECK ... [NOT VALID]`.
 // It assumes `t` is such a command.
 func alterTableAddCheck(
 	b BuildCtx, tn *tree.TableName, tbl *scpb.Table, t *tree.AlterTableAddConstraint,
@@ -113,18 +109,31 @@ func alterTableAddCheck(
 		panic(err)
 	}
 
-	// 3. Add relevant check constraint element: CheckConstraint and ConstraintName.
+	// 3. Add relevant check constraint element:
+	// - CheckConstraint or CheckConstraintUnvalidated
+	// - ConstraintName
 	constraintID := b.NextTableConstraintID(tbl.TableID)
-	ck := &scpb.CheckConstraint{
-		TableID:               tbl.TableID,
-		ConstraintID:          constraintID,
-		ColumnIDs:             colIDs.Ordered(),
-		Expression:            *b.WrapExpression(tbl.TableID, typedCkExpr),
-		FromHashShardedColumn: ckDef.FromHashShardedColumn,
-		IndexIDForValidation:  getIndexIDForValidationForConstraint(b, tbl.TableID),
+	if t.ValidationBehavior == tree.ValidationDefault {
+		ck := &scpb.CheckConstraint{
+			TableID:               tbl.TableID,
+			ConstraintID:          constraintID,
+			ColumnIDs:             colIDs.Ordered(),
+			Expression:            *b.WrapExpression(tbl.TableID, typedCkExpr),
+			FromHashShardedColumn: ckDef.FromHashShardedColumn,
+			IndexIDForValidation:  getIndexIDForValidationForConstraint(b, tbl.TableID),
+		}
+		b.Add(ck)
+		b.LogEventForExistingTarget(ck)
+	} else {
+		ck := &scpb.CheckConstraintUnvalidated{
+			TableID:      tbl.TableID,
+			ConstraintID: constraintID,
+			ColumnIDs:    colIDs.Ordered(),
+			Expression:   *b.WrapExpression(tbl.TableID, typedCkExpr),
+		}
+		b.Add(ck)
+		b.LogEventForExistingTarget(ck)
 	}
-	b.Add(ck)
-	b.LogEventForExistingTarget(ck)
 
 	constraintName := string(ckDef.Name)
 	if constraintName == "" {
@@ -162,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,
@@ -358,27 +367,43 @@ 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,
 		Name:         string(fkDef.Name),
 	})
 }
 
-// alterTableAddUniqueWithoutIndex contains logic for building ALTER TABLE ... ADD CONSTRAINT ... UNIQUE WITHOUT INDEX.
+// alterTableAddUniqueWithoutIndex contains logic for building
+// `ALTER TABLE ... ADD UNIQUE WITHOUT INDEX ... [NOT VALID]`.
 // It assumes `t` is such a command.
 func alterTableAddUniqueWithoutIndex(
 	b BuildCtx, tn *tree.TableName, tbl *scpb.Table, t *tree.AlterTableAddConstraint,
@@ -462,17 +487,30 @@ func alterTableAddUniqueWithoutIndex(
 
 	// 5. (Finally!) Add a UniqueWithoutIndex, ConstraintName element to builder state.
 	constraintID := b.NextTableConstraintID(tbl.TableID)
-	uwi := &scpb.UniqueWithoutIndexConstraint{
-		TableID:              tbl.TableID,
-		ConstraintID:         constraintID,
-		ColumnIDs:            colIDs,
-		IndexIDForValidation: getIndexIDForValidationForConstraint(b, tbl.TableID),
-	}
-	if d.Predicate != nil {
-		uwi.Predicate = b.WrapExpression(tbl.TableID, d.Predicate)
+	if t.ValidationBehavior == tree.ValidationDefault {
+		uwi := &scpb.UniqueWithoutIndexConstraint{
+			TableID:              tbl.TableID,
+			ConstraintID:         constraintID,
+			ColumnIDs:            colIDs,
+			IndexIDForValidation: getIndexIDForValidationForConstraint(b, tbl.TableID),
+		}
+		if d.Predicate != nil {
+			uwi.Predicate = b.WrapExpression(tbl.TableID, d.Predicate)
+		}
+		b.Add(uwi)
+		b.LogEventForExistingTarget(uwi)
+	} else {
+		uwi := &scpb.UniqueWithoutIndexConstraintUnvalidated{
+			TableID:      tbl.TableID,
+			ConstraintID: constraintID,
+			ColumnIDs:    colIDs,
+		}
+		if d.Predicate != nil {
+			uwi.Predicate = b.WrapExpression(tbl.TableID, d.Predicate)
+		}
+		b.Add(uwi)
+		b.LogEventForExistingTarget(uwi)
 	}
-	b.Add(uwi)
-	b.LogEventForExistingTarget(uwi)
 	b.Add(&scpb.ConstraintWithoutIndexName{
 		TableID:      tbl.TableID,
 		ConstraintID: constraintID,

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

@@ -197,5 +197,4 @@ func stmtSupportedInCurrentClusterVersion(b BuildCtx, n tree.Statement) bool {
 	}
 	minSupportedClusterVersion := supportedStatements[reflect.TypeOf(n)].minSupportedClusterVersion
 	return b.EvalCtx().Settings.Version.IsActive(b, minSupportedClusterVersion)
-
 }

pkg/sql/schemachanger/scbuild/testdata/alter_table_add_check

@@ -4,7 +4,6 @@ CREATE TABLE t (i INT PRIMARY KEY)
 
 build
 ALTER TABLE t ADD CHECK (i > 0)
----
 ----
 - [[CheckConstraint:{DescID: 104, IndexID: 0, ConstraintID: 2}, PUBLIC], ABSENT]
   {columnIds: [1], constraintId: 2, expr: 'i > 0:::INT8', referencedColumnIds: [1], tableId: 104}

pkg/sql/schemachanger/scbuild/testdata/alter_table_add_check_unvalidated

@@ -0,0 +1,11 @@
+setup
+CREATE TABLE t (i INT PRIMARY KEY)
+----
+
+build
+ALTER TABLE t ADD CHECK (i > 0) NOT VALID
+----
+- [[CheckConstraintUnvalidated:{DescID: 104, ConstraintID: 2}, PUBLIC], ABSENT]
+  {columnIds: [1], constraintId: 2, expr: 'i > 0:::INT8', referencedColumnIds: [1], tableId: 104}
+- [[ConstraintWithoutIndexName:{DescID: 104, Name: check_i, ConstraintID: 2}, PUBLIC], ABSENT]
+  {constraintId: 2, name: check_i, tableId: 104}

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/scbuild/testdata/alter_table_add_unique_without_index

@@ -0,0 +1,11 @@
+setup
+CREATE TABLE t (i INT PRIMARY KEY, j INT);
+----
+
+build
+ALTER TABLE t ADD UNIQUE WITHOUT INDEX (j);
+----
+- [[UniqueWithoutIndexConstraint:{DescID: 104, ConstraintID: 2}, PUBLIC], ABSENT]
+  {columnIds: [2], constraintId: 2, tableId: 104}
+- [[ConstraintWithoutIndexName:{DescID: 104, Name: unique_j, ConstraintID: 2}, PUBLIC], ABSENT]
+  {constraintId: 2, name: unique_j, tableId: 104}

pkg/sql/schemachanger/scbuild/testdata/alter_table_add_unique_without_index_unvalidated

@@ -0,0 +1,11 @@
+setup
+CREATE TABLE t (i INT PRIMARY KEY, j INT);
+----
+
+build
+ALTER TABLE t ADD UNIQUE WITHOUT INDEX (j) NOT VALID;
+----
+- [[UniqueWithoutIndexConstraintUnvalidated:{DescID: 104, ConstraintID: 2}, PUBLIC], ABSENT]
+  {columnIds: [2], constraintId: 2, tableId: 104}
+- [[ConstraintWithoutIndexName:{DescID: 104, Name: unique_j, ConstraintID: 2}, PUBLIC], ABSENT]
+  {constraintId: 2, name: unique_j, tableId: 104}