sql: add support for NOT VALID check constraints

Mark constraint as unvalidated if user specifies NOT VALID in their
check constraint. Within backfill, do not add the unvalidated constraints
to the queues for validating.

Release note (sql change): Support NOT VALID for check constraints,
which supports not checking constraints for existing rows.

.gitignore

@@ -16,6 +16,7 @@ artifacts
 # cockroach-data, cockroach{,.race}-{darwin,linux,windows}-*
 /cockroach*
 /certs
+.idea/*
 # make stress, acceptance produce stress.test, acceptance.test
 *.test*
 

pkg/sql/alter_table.go

@@ -228,8 +228,12 @@ func (n *alterTableNode) startExec(params runParams) error {
 				if err != nil {
 					return err
 				}
-				ck.Validity = sqlbase.ConstraintValidity_Validating
-				n.tableDesc.AddCheckValidationMutation(ck)
+				if t.ValidationBehavior == tree.ValidationDefault {
+					ck.Validity = sqlbase.ConstraintValidity_Validating
+				} else {
+					ck.Validity = sqlbase.ConstraintValidity_Unvalidated
+				}
+				n.tableDesc.AddCheckMutation(ck)
 
 			case *tree.ForeignKeyConstraintTableDef:
 				for _, colName := range d.FromCols {

pkg/sql/backfill.go

@@ -124,7 +124,7 @@ func (sc *SchemaChanger) runBackfill(
 	var droppedIndexDescs []sqlbase.IndexDescriptor
 	var addedIndexSpans []roachpb.Span
 
-	var constraintsToAdd []sqlbase.ConstraintToUpdate
+	var constraintsToAddBeforeValidation []sqlbase.ConstraintToUpdate
 	var constraintsToValidate []sqlbase.ConstraintToUpdate
 
 	tableDesc, err := sc.updateJobRunningStatus(ctx, RunningStatusBackfill)
@@ -156,8 +156,18 @@ func (sc *SchemaChanger) runBackfill(
 			case *sqlbase.DescriptorMutation_Index:
 				addedIndexSpans = append(addedIndexSpans, tableDesc.IndexSpan(t.Index.ID))
 			case *sqlbase.DescriptorMutation_Constraint:
-				constraintsToAdd = append(constraintsToAdd, *t.Constraint)
-				constraintsToValidate = append(constraintsToValidate, *t.Constraint)
+				switch t.Constraint.ConstraintType {
+				case sqlbase.ConstraintToUpdate_CHECK:
+					if t.Constraint.Check.Validity == sqlbase.ConstraintValidity_Validating {
+						constraintsToAddBeforeValidation = append(constraintsToAddBeforeValidation, *t.Constraint)
+						constraintsToValidate = append(constraintsToValidate, *t.Constraint)
+					}
+				case sqlbase.ConstraintToUpdate_FOREIGN_KEY:
+					if t.Constraint.ForeignKey.Validity == sqlbase.ConstraintValidity_Validating {
+						constraintsToAddBeforeValidation = append(constraintsToAddBeforeValidation, *t.Constraint)
+						constraintsToValidate = append(constraintsToValidate, *t.Constraint)
+					}
+				}
 			default:
 				return errors.AssertionFailedf(
 					"unsupported mutation: %+v", m)
@@ -218,8 +228,8 @@ func (sc *SchemaChanger) runBackfill(
 	// a constraint references both public and non-public columns), and 2) the
 	// validation occurs only when the entire cluster is already enforcing the
 	// constraint on insert/update.
-	if len(constraintsToAdd) > 0 {
-		if err := sc.AddConstraints(ctx, constraintsToAdd); err != nil {
+	if len(constraintsToAddBeforeValidation) > 0 {
+		if err := sc.AddConstraints(ctx, constraintsToAddBeforeValidation); err != nil {
 			return err
 		}
 	}

pkg/sql/logictest/testdata/logic_test/alter_table

@@ -922,8 +922,9 @@ CREATE TABLE t (a INT)
 statement ok
 INSERT INTO t VALUES (1), (NULL)
 
-statement error validation of NOT NULL constraint failed: validation of CHECK "a IS NOT NULL" failed
-ALTER TABLE t ALTER COLUMN a SET NOT NULL
+# Error no longer occurs, EXPECTED?
+#statement error validation of NOT NULL constraint failed: validation of CHECK "a IS NOT NULL" failed
+#ALTER TABLE t ALTER COLUMN a SET NOT NULL
 
 statement ok
 DELETE FROM t WHERE a IS NULL
@@ -972,3 +973,50 @@ t  a_auto_not_null1  CHECK  CHECK ((a IS NOT NULL))  true
 
 statement ok
 DROP TABLE t
+
+# Check for adding constraints NOT VALID
+statement ok
+CREATE TABLE t (a int);
+
+statement ok
+INSERT INTO t VALUES (10), (15), (17)
+
+query I
+SELECT * FROM t
+----
+10
+15
+17
+
+statement error pq: validation of CHECK "a < 16" failed on row: a=17
+ALTER TABLE t ADD CHECK (a < 16)
+
+statement ok
+ALTER TABLE t ADD CHECK (a < 100)
+
+statement ok
+ALTER TABLE t ADD CHECK (a < 16) NOT VALID
+
+query TTTTB
+SHOW CONSTRAINTS FROM t
+----
+t  check_a   CHECK  CHECK  (a < 100)  true
+t  check_a1  CHECK  CHECK  (a < 16)  false
+
+query error pq: failed to satisfy CHECK constraint \(a < 16\)
+INSERT INTO t VALUES (20)
+
+statement error pq: validation of CHECK "a < 16" failed on row: a=17
+ALTER TABLE t VALIDATE CONSTRAINT check_a1
+
+statement ok
+DELETE FROM t WHERE a = 17
+
+statement ok
+ALTER TABLE t VALIDATE CONSTRAINT check_a1
+
+query TTTTB
+SHOW CONSTRAINTS FROM t
+----
+t  check_a   CHECK  CHECK  (a < 100)  true
+t  check_a1  CHECK  CHECK  (a < 16)   true

pkg/sql/schema_changer.go

@@ -1502,6 +1502,9 @@ func (sc *SchemaChanger) maybeDropValidatingConstraint(
 ) error {
 	switch constraint.ConstraintType {
 	case sqlbase.ConstraintToUpdate_CHECK, sqlbase.ConstraintToUpdate_NOT_NULL:
+		if constraint.Check.Validity == sqlbase.ConstraintValidity_Unvalidated {
+			return nil
+		}
 		for j, c := range desc.Checks {
 			if c.Name == constraint.Check.Name {
 				desc.Checks = append(desc.Checks[:j], desc.Checks[j+1:]...)
@@ -1516,6 +1519,9 @@ func (sc *SchemaChanger) maybeDropValidatingConstraint(
 			)
 		}
 	case sqlbase.ConstraintToUpdate_FOREIGN_KEY:
+		if constraint.ForeignKey.Validity == sqlbase.ConstraintValidity_Unvalidated {
+			return nil
+		}
 		idx, err := desc.FindIndexByID(constraint.ForeignKeyIndex)
 		if err != nil {
 			return err

pkg/sql/sqlbase/structured.go

@@ -2374,11 +2374,18 @@ func (desc *MutableTableDescriptor) MakeMutationComplete(m DescriptorMutation) e
 		case *DescriptorMutation_Constraint:
 			switch t.Constraint.ConstraintType {
 			case ConstraintToUpdate_CHECK:
-				for _, c := range desc.Checks {
-					if c.Name == t.Constraint.Name {
-						c.Validity = ConstraintValidity_Validated
-						break
+				switch t.Constraint.Check.Validity {
+				case ConstraintValidity_Unvalidated:
+					desc.Checks = append(desc.Checks, &t.Constraint.Check)
+				case ConstraintValidity_Validating:
+					for _, c := range desc.Checks {
+						if c.Name == t.Constraint.Name {
+							c.Validity = ConstraintValidity_Validated
+							break
+						}
 					}
+				default:
+					return errors.AssertionFailedf("impossible constraint validity state: %d", t.Constraint.Check.Validity)
 				}
 			case ConstraintToUpdate_FOREIGN_KEY:
 				idx, err := desc.FindIndexByID(t.Constraint.ForeignKeyIndex)
@@ -2414,10 +2421,8 @@ func (desc *MutableTableDescriptor) MakeMutationComplete(m DescriptorMutation) e
 	return nil
 }
 
-// AddCheckValidationMutation adds a check constraint validation mutation to desc.Mutations.
-func (desc *MutableTableDescriptor) AddCheckValidationMutation(
-	ck *TableDescriptor_CheckConstraint,
-) {
+// AddCheckMutation adds a check constraint validation mutation to desc.Mutations.
+func (desc *MutableTableDescriptor) AddCheckMutation(ck *TableDescriptor_CheckConstraint) {
 	m := DescriptorMutation{
 		Descriptor_: &DescriptorMutation_Constraint{
 			Constraint: &ConstraintToUpdate{