importccl: Set FK and CHECK constraints to unvalidated after IMPORT INTO

pkg/ccl/importccl/import_stmt.go

@@ -970,6 +970,38 @@ func (r *importResumer) OnSuccess(ctx context.Context, txn *client.Txn) error {
 		tableDesc := *tbl.Desc
 		tableDesc.Version++
 		tableDesc.State = sqlbase.TableDescriptor_PUBLIC
+
+		if !tbl.IsNew {
+			// NB: This is not using AllNonDropIndexes or directly mutating the
+			// constraints returned by the other usual helpers because we need to
+			// replace the `Indexes` and `Checks` slices of tableDesc with copies that
+			// we can mutate. We need to do that because tableDesc is a shallow copy
+			// of tbl.Desc that we'll be asserting is the current version when we
+			// CPut below.
+			//
+			// Set FK constraints to unvalidated before publishing the table imported
+			// into.
+			if tableDesc.PrimaryIndex.ForeignKey.IsSet() {
+				tableDesc.PrimaryIndex.ForeignKey.Validity = sqlbase.ConstraintValidity_Unvalidated
+			}
+
+			tableDesc.Indexes = make([]sqlbase.IndexDescriptor, len(tbl.Desc.Indexes))
+			copy(tableDesc.Indexes, tbl.Desc.Indexes)
+			for i := range tableDesc.Indexes {
+				if tableDesc.Indexes[i].ForeignKey.IsSet() {
+					tableDesc.Indexes[i].ForeignKey.Validity = sqlbase.ConstraintValidity_Unvalidated
+				}
+			}
+
+			// Set CHECK constraints to unvalidated before publishing the table imported into.
+			tableDesc.Checks = make([]*sqlbase.TableDescriptor_CheckConstraint, len(tbl.Desc.Checks))
+			for i, c := range tbl.Desc.AllActiveAndInactiveChecks() {
+				ck := *c
+				ck.Validity = sqlbase.ConstraintValidity_Unvalidated
+				tableDesc.Checks[i] = &ck
+			}
+		}
+
 		// TODO(dt): re-validate any FKs?
 		// Note that this CPut is safe with respect to mixed-version descriptor
 		// upgrade and downgrade, because IMPORT does not operate in mixed-version

pkg/ccl/importccl/import_stmt_test.go

@@ -1828,6 +1828,33 @@ func TestImportIntoCSV(t *testing.T) {
 			t.Fatalf("expected %d rows, got %d", expect, result)
 		}
 	})
+
+	// Tests that IMPORT INTO invalidates FK and CHECK constraints.
+	t.Run("import-into-invalidate-constraints", func(t *testing.T) {
+		sqlDB.Exec(t, "CREATE DATABASE invalidateconstraints; USE invalidateconstraints")
+
+		sqlDB.Exec(t, `CREATE TABLE ref (b STRING PRIMARY KEY)`)
+		sqlDB.Exec(t, `CREATE TABLE t (a INT CHECK (a >= 0), b STRING, CONSTRAINT fk_ref FOREIGN KEY (b) REFERENCES ref)`)
+
+		var checkValidated, fkValidated bool
+		sqlDB.QueryRow(t, fmt.Sprintf(`SELECT validated from [SHOW CONSTRAINT FROM t] WHERE constraint_name = 'check_a'`)).Scan(&checkValidated)
+		sqlDB.QueryRow(t, fmt.Sprintf(`SELECT validated from [SHOW CONSTRAINT FROM t] WHERE constraint_name = 'fk_ref'`)).Scan(&fkValidated)
+
+		// Prior to import all constraints should be validated.
+		if !checkValidated || !fkValidated {
+			t.Fatal("Constraints not validated on creation.\n")
+		}
+
+		sqlDB.Exec(t, fmt.Sprintf(`IMPORT INTO t (a, b) CSV DATA (%s)`, testFiles.files[0]))
+
+		sqlDB.QueryRow(t, fmt.Sprintf(`SELECT validated from [SHOW CONSTRAINT FROM t] WHERE constraint_name = 'check_a'`)).Scan(&checkValidated)
+		sqlDB.QueryRow(t, fmt.Sprintf(`SELECT validated from [SHOW CONSTRAINT FROM t] WHERE constraint_name = 'fk_ref'`)).Scan(&fkValidated)
+
+		// Following an import the constraints should be unvalidated.
+		if checkValidated || fkValidated {
+			t.Fatal("FK and CHECK constraints not unvalidated after IMPORT INTO\n")
+		}
+	})
 }
 
 func BenchmarkImport(b *testing.B) {