sql/schemachanger: add additional DML injection for testing

Previously, we lacked DML injection testing with
UPDATES/DELETES and non-empty tables. We also lacked
DML injection for new functionality in this release.
To address this, this patch will start populating tables
in certain tasks and add new DML injection.

Additionally, the backup/restore tests will fully validate
job state on completion as a sanity check. Job failures are allowed
if backfilling schema changes with DML (during the setup).

Fixes: cockroachdb#98635
Release note: None

pkg/ccl/schemachangerccl/testdata/end_to_end/create_index

@@ -4,6 +4,39 @@ CREATE TABLE defaultdb.t1 (id INT PRIMARY KEY, name VARCHAR(256), money INT)
 ...
 +object {100 101 t1} -> 104
 
+stage-exec phase=PostCommitPhase stage=:
+INSERT INTO t VALUES($stageKey, 'a');
+INSERT INTO t VALUES($stageKey + 1, 'b');
+INSERT INTO t VALUES($stageKey + 2, 'c');
+DELETE FROM t WHERE v = 'a' and k=$stageKey;
+INSERT INTO t VALUES($stageKey, 'a');
+UPDATE t SET v='a' WHERE k > 0;
+----
+
+# Each insert will be injected thrice per stage, so we should always,
+# see a count of 2.
+stage-query phase=PostCommitPhase stage=:
+SELECT count(*)=$successfulStageCount*3 FROM t where v='a';
+----
+true
+
+
+stage-exec phase=PostCommitNonRevertiblePhase stage=:
+INSERT INTO t VALUES($stageKey, 'a');
+INSERT INTO t VALUES($stageKey + 1, 'b');
+INSERT INTO t VALUES($stageKey + 2, 'c');
+DELETE FROM t WHERE v = 'a' and k=$stageKey;
+INSERT INTO t VALUES($stageKey, 'a');
+UPDATE t SET v='a' WHERE k > 0;
+----
+
+# Each insert will be injected twice per stage, so we should always,
+# see a count of 2.
+stage-query phase=PostCommitNonRevertiblePhase stage=:
+SELECT count(*)=$successfulStageCount*3 FROM t where v='a';
+----
+true
+
 test
 CREATE INDEX id1
   ON defaultdb.t1 (id, name)

pkg/sql/schemachanger/sctest/cumulative.go

@@ -988,18 +988,29 @@ func Backup(t *testing.T, path string, newCluster NewClusterFunc) {
 	testBackupRestoreCase := func(
 		t *testing.T, setup, stmts []parser.Statement, ord int,
 	) {
+		// If tables are empty then, backfills should never lead to rollbacks
+		// in restores.
+		hasDMLInSetup := false
+		for _, setupStmt := range setup {
+			hasDMLInSetup = hasDMLInSetup || setupStmt.IsANSIDML()
+		}
 		type stage struct {
 			p        scplan.Plan
 			stageIdx int
 			resume   chan error
 		}
-
+		successExpected := atomic.Bool{}
 		stageChan := make(chan stage)
 		var closedStageChan bool // mark when stageChan is closed in the callback
 		ctx, cancel := context.WithCancel(context.Background())
 		_, db, cleanup := newCluster(t, &scexec.TestingKnobs{
 			BeforeStage: func(p scplan.Plan, stageIdx int) error {
-
+				// If this plan contains any backfills, we will not
+				// back up or restore those indexes, so failure can occur
+				if p.Stages[stageIdx].Type() == scop.BackfillType &&
+					hasDMLInSetup {
+					successExpected.Store(false)
+				}
 				// If the plan has no post-commit stages, we'll close the
 				// stageChan eagerly.
 				if p.Stages[len(p.Stages)-1].Phase < scop.PostCommitPhase {
@@ -1058,10 +1069,11 @@ func Backup(t *testing.T, path string, newCluster NewClusterFunc) {
 			)
 		})
 		type backup struct {
-			name       string
-			isRollback bool
-			url        string
-			s          stage
+			name            string
+			isRollback      bool
+			successExpected bool
+			url             string
+			s               stage
 		}
 		var backups []backup
 		var done bool
@@ -1131,10 +1143,11 @@ func Backup(t *testing.T, path string, newCluster NewClusterFunc) {
 			tdb.Exec(t, fmt.Sprintf(
 				"BACKUP DATABASE %s INTO '%s'", dbName, backupURL))
 			backups = append(backups, backup{
-				name:       dbName,
-				isRollback: rollbackStage > 0,
-				url:        backupURL,
-				s:          s,
+				name:            dbName,
+				isRollback:      rollbackStage > 0,
+				successExpected: successExpected.Load(),
+				url:             backupURL,
+				s:               s,
 			})
 
 			if s.p.InRollback {
@@ -1279,10 +1292,16 @@ func Backup(t *testing.T, path string, newCluster NewClusterFunc) {
 					tdb.Exec(t, flavor.restoreQuery)
 					tdb.Exec(t, fmt.Sprintf("USE %q", dbName))
 					waitForSchemaChangesToFinish(t, tdb)
+					wasSchemaChangeSuccessful := schemaChangeQueryLatestStatus(t, tdb) == "succeeded"
+					// Validate if the schema change was successful after the restore,
+					// only if no backfill's are required.
+					if !b.isRollback && !wasSchemaChangeSuccessful && b.successExpected {
+						t.Fatalf("schema change failed when successful completion was expected")
+					}
 					afterRestore := tdb.QueryStr(t, fetchDescriptorStateQuery)
 
 					if flavor.name != "restore all tables in database" {
-						if b.isRollback {
+						if b.isRollback || !wasSchemaChangeSuccessful {
 							require.Equal(t, before, afterRestore)
 						} else {
 							require.Equal(t, after, afterRestore)
@@ -1293,7 +1312,7 @@ func Backup(t *testing.T, path string, newCluster NewClusterFunc) {
 						// dropped through RESTORE due to missing UDFs. We need to remove
 						// those kind of table elements from original AST.
 						var expected [][]string
-						if b.isRollback {
+						if b.isRollback || !wasSchemaChangeSuccessful {
 							expected = before
 						} else {
 							expected = after
@@ -1689,6 +1708,13 @@ func BackupMixedVersionElements(t *testing.T, path string, newCluster NewMixedCl
 	testBackupRestoreCase := func(
 		t *testing.T, setup, stmts []parser.Statement, ord int,
 	) {
+		// If tables are empty then, backfills should never lead to rollbacks
+		// in restores.
+		hasDMLInSetup := false
+		for _, setupStmt := range setup {
+			hasDMLInSetup = hasDMLInSetup || setupStmt.IsANSIDML()
+		}
+		successExpected := atomic.Bool{}
 		type stage struct {
 			p        scplan.Plan
 			stageIdx int
@@ -1699,6 +1725,11 @@ func BackupMixedVersionElements(t *testing.T, path string, newCluster NewMixedCl
 		ctx, cancel := context.WithCancel(context.Background())
 		_, db, cleanup := newCluster(t, &scexec.TestingKnobs{
 			BeforeStage: func(p scplan.Plan, stageIdx int) error {
+				if p.Stages[stageIdx].Type() == scop.BackfillType && hasDMLInSetup {
+					// Restores for anything with a backfill can potentially,
+					// fail.
+					successExpected.Store(false)
+				}
 				if p.Stages[len(p.Stages)-1].Phase < scop.PostCommitPhase {
 					if stageChan != nil {
 						close(stageChan)
@@ -1753,10 +1784,11 @@ func BackupMixedVersionElements(t *testing.T, path string, newCluster NewMixedCl
 			)
 		})
 		type backup struct {
-			name       string
-			isRollback bool
-			url        string
-			s          stage
+			name            string
+			isRollback      bool
+			successExpected bool
+			url             string
+			s               stage
 		}
 		var backups []backup
 		var done bool
@@ -1830,10 +1862,11 @@ func BackupMixedVersionElements(t *testing.T, path string, newCluster NewMixedCl
 			tdb.Exec(t, fmt.Sprintf(
 				"BACKUP DATABASE %s INTO '%s'", dbName, backupURL))
 			backups = append(backups, backup{
-				name:       dbName,
-				isRollback: rollbackStage > 0,
-				url:        backupURL,
-				s:          s,
+				name:            dbName,
+				isRollback:      rollbackStage > 0,
+				url:             backupURL,
+				successExpected: successExpected.Load(),
+				s:               s,
 			})
 
 			if s.p.InRollback {
@@ -1932,8 +1965,14 @@ func BackupMixedVersionElements(t *testing.T, path string, newCluster NewMixedCl
 					tdb.Exec(t, flavor.restoreQuery)
 					tdb.Exec(t, fmt.Sprintf("USE %q", dbName))
 					waitForSchemaChangesToFinish(t, tdb)
+					wasSchemaChangeSuccessful := schemaChangeQueryLatestStatus(t, tdb) == "succeeded"
+					// Validate if the schema change was successful after the restore,
+					// only if no backfill's are required.
+					if !b.isRollback && !wasSchemaChangeSuccessful && b.successExpected {
+						t.Fatalf("schema change failed when successful completion was expected")
+					}
 					afterRestore := tdb.QueryStr(t, fetchDescriptorStateQuery)
-					if b.isRollback {
+					if b.isRollback || !wasSchemaChangeSuccessful {
 						require.Equal(t, before, afterRestore)
 					} else {
 						require.Equal(t, after, afterRestore)

pkg/sql/schemachanger/sctest/end_to_end.go

@@ -444,6 +444,18 @@ func waitForSchemaChangesToFinish(t *testing.T, tdb *sqlutils.SQLRunner) {
 	)
 }
 
+func schemaChangeQueryLatestStatus(t *testing.T, tdb *sqlutils.SQLRunner) string {
+	q := fmt.Sprintf(
+		`SELECT status FROM [SHOW JOBS] WHERE job_type IN ('%s', '%s', '%s') ORDER BY finished DESC LIMIT 1`,
+		jobspb.TypeSchemaChange,
+		jobspb.TypeTypeSchemaChange,
+		jobspb.TypeNewSchemaChange,
+	)
+	result := tdb.QueryStr(t, q)
+	require.Len(t, result, 1)
+	require.Len(t, result[0], 1)
+	return result[0][0]
+}
 func schemaChangeWaitQuery(statusInString string) string {
 	q := fmt.Sprintf(
 		`SELECT status, job_type, description FROM [SHOW JOBS] WHERE job_type IN ('%s', '%s', '%s') AND status NOT IN %s`,

pkg/sql/schemachanger/testdata/end_to_end/add_column

@@ -2,6 +2,9 @@ setup
 CREATE DATABASE db;
 CREATE TABLE db.public.tbl (i INT PRIMARY KEY, k INT);
 CREATE SEQUENCE db.public.sq1;
+INSERT INTO db.public.tbl VALUES(-1);
+INSERT INTO db.public.tbl VALUES(-2);
+INSERT INTO db.public.tbl VALUES(-3);
 ----
 ...
 +database {0 0 db} -> 104
@@ -22,7 +25,7 @@ INSERT INTO db.public.tbl VALUES(-1);
 
 # Each insert will be injected twice per stage, plus 1 extra.
 stage-query phase=PostCommitPhase stage=:
-SELECT count(*)=($successfulStageCount*2)+1 FROM db.public.tbl;
+SELECT count(*)=($successfulStageCount*2)+3 FROM db.public.tbl;
 ----
 true
 
@@ -40,7 +43,7 @@ INSERT INTO db.public.tbl VALUES(-1);
 
 # Each insert will be injected twice per stage, , plus 1 extra.
 stage-query phase=PostCommitNonRevertiblePhase stage=:
-SELECT count(*)=($successfulStageCount*2)+1 FROM db.public.tbl;
+SELECT count(*)=($successfulStageCount*2)+3 FROM db.public.tbl;
 ----
 true
 

pkg/sql/schemachanger/testdata/end_to_end/drop_column_basic

@@ -2,6 +2,9 @@ setup
 CREATE TABLE t (i INT PRIMARY KEY, j INT, k int);
 COMMENT ON TABLE t IS 't has a comment';
 COMMENT ON COLUMN t.j IS 'j has a comment';
+INSERT INTO t VALUES(-1);
+INSERT INTO t VALUES(-2);
+INSERT INTO t VALUES(-3);
 ----
 ...
 +object {100 101 t} -> 104
@@ -20,9 +23,9 @@ INSERT INTO t VALUES(-1);
 # Each insert will be injected twice per stage, plus 3 injected
 # at the start.
 stage-query phase=PostCommitPhase stage=:
-SELECT count(*)=($successfulStageCount*2) FROM t;
+SELECT count(*)=($successfulStageCount*2)+3 FROM t;
 ----
-false
+true
 
 stage-exec phase=PostCommitNonRevertiblePhase stage=:
 INSERT INTO t VALUES($stageKey);
@@ -38,9 +41,9 @@ INSERT INTO t VALUES(-1);
 # Each insert will be injected twice per stage, plus 3 injected
 # at the start.
 stage-query phase=PostCommitNonRevertiblePhase stage=:
-SELECT count(*)=($successfulStageCount*2) FROM t;
+SELECT count(*)=($successfulStageCount*2)+3 FROM t;
 ----
-false
+true
 
 test
 ALTER TABLE t DROP COLUMN j

pkg/sql/schemachanger/testdata/end_to_end/drop_index_hash_sharded_index

@@ -1,6 +1,9 @@
 setup
 CREATE TABLE t (i INT PRIMARY KEY, j INT);
-CREATE INDEX idx ON t(j) USING HASH
+CREATE INDEX idx ON t(j) USING HASH;
+INSERT INTO t VALUES(-1);
+INSERT INTO t VALUES(-2);
+INSERT INTO t VALUES(-3);
 ----
 ...
 +object {100 101 t} -> 104
@@ -13,7 +16,7 @@ INSERT INTO t (i, j) VALUES($stageKey + 1, $stageKey +1);
 # Each insert will be injected twice per stage, so we should always,
 # see a count of 2.
 stage-query phase=PostCommitNonRevertiblePhase stage=:
-SELECT count(*)=$successfulStageCount*2 FROM t;
+SELECT count(*)=($successfulStageCount*2)+3 FROM t;
 ----
 true
 

pkg/sql/schemachanger/testdata/end_to_end/drop_multiple_columns_separate_statements

@@ -1,3 +1,6 @@
+# This test also helps provide implicit coverage for dropping
+# normal and virtual computed columns, which can be interesting
+# from an IUD perpective when mutations are active.
 setup
 CREATE TABLE t (i INT PRIMARY KEY, j INT, k INT DEFAULT 32 ON UPDATE 42, INDEX((j+1), k));
 INSERT INTO t VALUES(-1);

pkg/sql/schemachanger/testdata/explain/add_column

@@ -2,6 +2,9 @@
 CREATE DATABASE db;
 CREATE TABLE db.public.tbl (i INT PRIMARY KEY, k INT);
 CREATE SEQUENCE db.public.sq1;
+INSERT INTO db.public.tbl VALUES(-1);
+INSERT INTO db.public.tbl VALUES(-2);
+INSERT INTO db.public.tbl VALUES(-3);
 
 /* test */
 EXPLAIN (ddl) ALTER TABLE db.public.tbl ADD COLUMN j INT NOT NULL DEFAULT 42;

pkg/sql/schemachanger/testdata/explain/add_column.rollback_1_of_7

@@ -2,6 +2,9 @@
 CREATE DATABASE db;
 CREATE TABLE db.public.tbl (i INT PRIMARY KEY, k INT);
 CREATE SEQUENCE db.public.sq1;
+INSERT INTO db.public.tbl VALUES(-1);
+INSERT INTO db.public.tbl VALUES(-2);
+INSERT INTO db.public.tbl VALUES(-3);
 
 /* test */
 ALTER TABLE db.public.tbl ADD COLUMN j INT NOT NULL DEFAULT 42;

pkg/sql/schemachanger/testdata/explain/add_column.rollback_2_of_7

@@ -2,6 +2,9 @@
 CREATE DATABASE db;
 CREATE TABLE db.public.tbl (i INT PRIMARY KEY, k INT);
 CREATE SEQUENCE db.public.sq1;
+INSERT INTO db.public.tbl VALUES(-1);
+INSERT INTO db.public.tbl VALUES(-2);
+INSERT INTO db.public.tbl VALUES(-3);
 
 /* test */
 ALTER TABLE db.public.tbl ADD COLUMN j INT NOT NULL DEFAULT 42;

pkg/sql/schemachanger/testdata/explain/add_column.rollback_3_of_7

@@ -2,6 +2,9 @@
 CREATE DATABASE db;
 CREATE TABLE db.public.tbl (i INT PRIMARY KEY, k INT);
 CREATE SEQUENCE db.public.sq1;
+INSERT INTO db.public.tbl VALUES(-1);
+INSERT INTO db.public.tbl VALUES(-2);
+INSERT INTO db.public.tbl VALUES(-3);
 
 /* test */
 ALTER TABLE db.public.tbl ADD COLUMN j INT NOT NULL DEFAULT 42;

pkg/sql/schemachanger/testdata/explain/add_column.rollback_4_of_7

@@ -2,6 +2,9 @@
 CREATE DATABASE db;
 CREATE TABLE db.public.tbl (i INT PRIMARY KEY, k INT);
 CREATE SEQUENCE db.public.sq1;
+INSERT INTO db.public.tbl VALUES(-1);
+INSERT INTO db.public.tbl VALUES(-2);
+INSERT INTO db.public.tbl VALUES(-3);
 
 /* test */
 ALTER TABLE db.public.tbl ADD COLUMN j INT NOT NULL DEFAULT 42;

pkg/sql/schemachanger/testdata/explain/add_column.rollback_5_of_7

@@ -2,6 +2,9 @@
 CREATE DATABASE db;
 CREATE TABLE db.public.tbl (i INT PRIMARY KEY, k INT);
 CREATE SEQUENCE db.public.sq1;
+INSERT INTO db.public.tbl VALUES(-1);
+INSERT INTO db.public.tbl VALUES(-2);
+INSERT INTO db.public.tbl VALUES(-3);
 
 /* test */
 ALTER TABLE db.public.tbl ADD COLUMN j INT NOT NULL DEFAULT 42;

pkg/sql/schemachanger/testdata/explain/add_column.rollback_6_of_7

@@ -2,6 +2,9 @@
 CREATE DATABASE db;
 CREATE TABLE db.public.tbl (i INT PRIMARY KEY, k INT);
 CREATE SEQUENCE db.public.sq1;
+INSERT INTO db.public.tbl VALUES(-1);
+INSERT INTO db.public.tbl VALUES(-2);
+INSERT INTO db.public.tbl VALUES(-3);
 
 /* test */
 ALTER TABLE db.public.tbl ADD COLUMN j INT NOT NULL DEFAULT 42;

pkg/sql/schemachanger/testdata/explain/add_column.rollback_7_of_7

@@ -2,6 +2,9 @@
 CREATE DATABASE db;
 CREATE TABLE db.public.tbl (i INT PRIMARY KEY, k INT);
 CREATE SEQUENCE db.public.sq1;
+INSERT INTO db.public.tbl VALUES(-1);
+INSERT INTO db.public.tbl VALUES(-2);
+INSERT INTO db.public.tbl VALUES(-3);
 
 /* test */
 ALTER TABLE db.public.tbl ADD COLUMN j INT NOT NULL DEFAULT 42;