schemachanger: Add an intergration-style test for concurrent schema changer behavior

This commit adds an integration style test for concurrent schema
changer behaviors where we run multiple DDLs for an extended period of
time on a few descriptors and assert that they all eventually finish
and the descriptors end up in the expected state.

Release note: None

Author: Xiang-Gu

pkg/sql/schemachanger/BUILD.bazel

@@ -49,6 +49,7 @@ go_test(
         "//pkg/sql/catalog",
         "//pkg/sql/catalog/desctestutils",
         "//pkg/sql/execinfra",
+        "//pkg/sql/pgwire/pgcode",
         "//pkg/sql/rowenc",
         "//pkg/sql/schemachanger/scexec",
         "//pkg/sql/schemachanger/scop",
@@ -57,6 +58,7 @@ go_test(
         "//pkg/sql/sessiondatapb",
         "//pkg/testutils",
         "//pkg/testutils/serverutils",
+        "//pkg/testutils/skip",
         "//pkg/testutils/sqlutils",
         "//pkg/util/ctxgroup",
         "//pkg/util/leaktest",
@@ -65,6 +67,7 @@ go_test(
         "//pkg/util/randutil",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_errors//errorspb",
+        "@com_github_lib_pq//:pq",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
     ],

pkg/sql/schemachanger/schemachanger_test.go

@@ -14,11 +14,13 @@ import (
 	"context"
 	"encoding/hex"
 	"fmt"
+	"math/rand"
 	"regexp"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
@@ -30,20 +32,23 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/desctestutils"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
 	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scexec"
 	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scop"
 	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scplan"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondatapb"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/errors/errorspb"
+	"github.com/lib/pq"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -517,3 +522,269 @@ func requireTableKeyCount(
 	require.NoError(t, err)
 	require.Equal(t, keyCount, len(kvs))
 }
+
+// TestConcurrentSchemaChanges is an integration style tests where we issue many
+// schema changes concurrently (renames, add/drop columns, and create/drop
+// indexes) for a period of time and assert that they all finish eventually and
+// we end up with expected names, columns, and indexes.
+func TestConcurrentSchemaChanges(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	skip.UnderShort(t, "this test is long running (>3 mins).")
+	skip.UnderStress(t, "test is already integration style and long running")
+	skip.UnderStressRace(t, "test is already integration style and long running")
+	skip.UnderRace(t, "the test knowingly has data race and has logic to account for that")
+
+	const testDuration = 3 * time.Minute
+	const renameDBInterval = 5 * time.Second
+	const renameSCInterval = 4 * time.Second
+	const renameTblInterval = 3 * time.Second
+	const addColInterval = 1 * time.Second
+	const dropColInterval = 1 * time.Second
+	const createIdxInterval = 1 * time.Second
+	const dropIdxInterval = 1 * time.Second
+
+	ctx, cancel := context.WithCancel(context.Background())
+	var params base.TestServerArgs
+	params.Knobs = base.TestingKnobs{
+		// Decrease the adopt loop interval so that retries happen quickly.
+		JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(),
+	}
+	s, sqlDB, kvDB := serverutils.StartServer(t, params)
+	defer s.Stopper().Stop(ctx)
+
+	tdb := sqlutils.MakeSQLRunner(sqlDB)
+	dbName, scName, tblName := "testdb", "testsc", "t"
+	allColToIndexes := make(map[string]map[string]struct{}) // colName -> indexes that uses that column
+	allColToIndexes["col"] = map[string]struct{}{"t_pkey": {}}
+	allNonPublicIdxToKeyCols := make(map[string]map[string]struct{}) // indexName -> its key column(s)
+	tdb.Exec(t, fmt.Sprintf("CREATE DATABASE %v;", dbName))
+	tdb.Exec(t, fmt.Sprintf("CREATE SCHEMA %v.%v;", dbName, scName))
+	tdb.Exec(t, fmt.Sprintf("CREATE TABLE %v.%v.%v (col INT PRIMARY KEY);", dbName, scName, tblName))
+	tdb.Exec(t, fmt.Sprintf("INSERT INTO %v.%v.%v SELECT generate_series(1,100);", dbName, scName, tblName))
+
+	// repeatFnWithInterval repeats `fn` indefinitely every `interval` until
+	// `ctx` is cancelled.
+	workerErrChan := make(chan error)
+	var wg sync.WaitGroup
+	repeatWorkWithInterval := func(workerName string, workInterval time.Duration, work func() error) {
+		wg.Add(1)
+		defer wg.Done()
+		for {
+			jitteredInterval := workInterval * time.Duration(0.8+0.4*rand.Float32())
+			select {
+			case <-ctx.Done():
+				t.Logf("%v is signaled to finish work", workerName)
+				return
+			case <-time.After(jitteredInterval):
+				if err := work(); err != nil {
+					t.Logf("%v encounters error %v; signal to main routine and finish working", workerName, err.Error())
+					workerErrChan <- err
+					return
+				}
+			}
+		}
+	}
+
+	// validate performs a few quick validations after all schema changes are finished:
+	// 1. Database, schema, and table indeed end up with the tracked name.
+	// 2. Table indeed has the tracked columns.
+	// 3. Table indeed has the tracked indexes.
+	codec := s.ApplicationLayer().Codec()
+	validate := func() {
+		dbDesc := desctestutils.TestingGetDatabaseDescriptor(kvDB, codec, dbName)
+		desctestutils.TestingGetSchemaDescriptor(kvDB, codec, dbDesc.GetID(), scName)
+		tblDesc := desctestutils.TestingGetTableDescriptor(kvDB, codec, dbName, scName, tblName)
+		require.Equal(t, len(allColToIndexes), len(tblDesc.PublicColumns())) // allColToIndexes does not include `col`
+		for _, col := range tblDesc.PublicColumns() {
+			_, ok := allColToIndexes[col.GetName()]
+			require.True(t, ok, "column %v does not exist in allColToIndexes=%v", col.GetName(), allColToIndexes)
+		}
+		require.Equal(t, len(allNonPublicIdxToKeyCols), len(tblDesc.PublicNonPrimaryIndexes()))
+		for _, idx := range tblDesc.PublicNonPrimaryIndexes() {
+			_, ok := allNonPublicIdxToKeyCols[idx.GetName()]
+			require.True(t, ok, "index %v does not exist in allNonPublicIdxToKeyCols=%v", idx.GetName(), allNonPublicIdxToKeyCols)
+		}
+	}
+
+	// A goroutine that repeatedly renames database `testdb` randomly.
+	go repeatWorkWithInterval("rename-db-worker", renameDBInterval, func() error {
+		newDBName := fmt.Sprintf("testdb_%v", rand.Intn(1000))
+		if newDBName == dbName {
+			return nil
+		}
+		if _, err := sqlDB.Exec(fmt.Sprintf("ALTER DATABASE %v RENAME TO %v", dbName, newDBName)); err != nil {
+			return err
+		}
+		dbName = newDBName
+		t.Logf("RENAME DATABASE TO %v", newDBName)
+		return nil
+	})
+
+	// A goroutine that renames schema `testdb.testsc` randomly.
+	go repeatWorkWithInterval("rename-schema-worker", renameSCInterval, func() error {
+		newSCName := fmt.Sprintf("testsc_%v", rand.Intn(1000))
+		if scName == newSCName {
+			return nil
+		}
+		_, err := sqlDB.Exec(fmt.Sprintf("ALTER SCHEMA %v.%v RENAME TO %v", dbName, scName, newSCName))
+		if err == nil {
+			scName = newSCName
+			t.Logf("RENAME SCHEMA TO %v", newSCName)
+		} else if isPQErrWithCode(err, pgcode.UndefinedDatabase) {
+			err = nil // mute those errors as they're expected
+			t.Logf("Parent database is renamed; skipping this schema renaming.")
+		}
+		return err
+	})
+
+	// A goroutine that renames table `testdb.testsc.t` randomly.
+	go repeatWorkWithInterval("rename-tbl-worker", renameTblInterval, func() error {
+		newTblName := fmt.Sprintf("t_%v", rand.Intn(1000))
+		_, err := sqlDB.Exec(fmt.Sprintf(`ALTER TABLE %v.%v.%v RENAME TO %v`, dbName, scName, tblName, newTblName))
+		if err == nil {
+			tblName = newTblName
+			t.Logf("RENAME TABLE TO %v", newTblName)
+		} else if isPQErrWithCode(err, pgcode.UndefinedDatabase, pgcode.UndefinedSchema, pgcode.InvalidSchemaName) {
+			err = nil
+			t.Logf("Parent database or schema is renamed; skipping this table renaming.")
+		}
+		return err
+	})
+
+	// A goroutine that adds columns to `testdb.testsc.t` randomly.
+	go repeatWorkWithInterval("add-column-worker", addColInterval, func() error {
+		newColName := fmt.Sprintf("col_%v", rand.Intn(1000))
+		if _, ok := allColToIndexes[newColName]; ok {
+			return nil
+		}
+		tblName := tblName
+		_, err := sqlDB.Exec(fmt.Sprintf("ALTER TABLE %v.%v.%v ADD COLUMN %v INT DEFAULT %v", dbName, scName, tblName, newColName, rand.Intn(100)))
+		if err == nil {
+			allColToIndexes[newColName] = make(map[string]struct{})
+			t.Logf("ADD COLUMN %v TO TABLE %v", newColName, tblName)
+		} else if isPQErrWithCode(err, pgcode.UndefinedDatabase, pgcode.UndefinedSchema, pgcode.InvalidSchemaName, pgcode.UndefinedTable) {
+			err = nil
+			t.Logf("Parent database or schema or table is renamed; skipping this column addition.")
+		}
+		return err
+	})
+
+	// A goroutine that drops columns from `testdb.testsc.t` randomly.
+	go repeatWorkWithInterval("drop-column-worker", dropColInterval, func() error {
+		// Randomly pick a non-PK column to drop.
+		if len(allColToIndexes) == 1 {
+			return nil
+		}
+		var colName string
+		for col := range allColToIndexes {
+			if col != "col" {
+				colName = col
+				break
+			}
+		}
+
+		tblName := tblName
+		_, err := sqlDB.Exec(fmt.Sprintf("ALTER TABLE %v.%v.%v DROP COLUMN %v;", dbName, scName, tblName, colName))
+		if err == nil {
+			for indexName := range allColToIndexes[colName] {
+				delete(allNonPublicIdxToKeyCols, indexName)
+			}
+			delete(allColToIndexes, colName)
+			t.Logf("DROP COLUMN %v FROM TABLE %v", colName, tblName)
+		} else if isPQErrWithCode(err, pgcode.UndefinedDatabase, pgcode.UndefinedSchema, pgcode.InvalidSchemaName, pgcode.UndefinedTable) {
+			err = nil
+			t.Logf("Parent database or schema or table is renamed; skipping this column removal.")
+		}
+		return err
+	})
+
+	// A goroutine that creates secondary index on a randomly selected column.
+	go repeatWorkWithInterval("create-index-worker", createIdxInterval, func() error {
+		newIndexName := fmt.Sprintf("idx_%v", rand.Intn(1000))
+		if _, ok := allNonPublicIdxToKeyCols[newIndexName]; ok {
+			return nil
+		}
+
+		// Randomly pick a non-PK column to create an index on.
+		if len(allColToIndexes) == 1 {
+			return nil
+		}
+		var colName string
+		for col := range allColToIndexes {
+			if col != "col" {
+				colName = col
+				break
+			}
+		}
+
+		tblName := tblName
+		_, err := sqlDB.Exec(fmt.Sprintf("CREATE INDEX %v ON %v.%v.%v (%v);", newIndexName, dbName, scName, tblName, colName))
+		if err == nil {
+			allNonPublicIdxToKeyCols[newIndexName] = map[string]struct{}{colName: {}}
+			allColToIndexes[colName][newIndexName] = struct{}{}
+			t.Logf("CREATE INDEX %v ON TABLE %v(%v)", newIndexName, tblName, colName)
+		} else if isPQErrWithCode(err, pgcode.UndefinedDatabase, pgcode.UndefinedSchema, pgcode.InvalidSchemaName, pgcode.UndefinedTable, pgcode.UndefinedColumn) {
+			err = nil
+			t.Logf("Parent database or schema or table is renamed or column is dropped; skipping this index creation.")
+		}
+		return err
+	})
+
+	// A goroutine that drops a secondary index randomly.
+	go repeatWorkWithInterval("drop-index-worker", dropIdxInterval, func() error {
+		// Randomly pick a non-public index to drop.
+		if len(allNonPublicIdxToKeyCols) == 0 {
+			return nil
+		}
+		var indexName string
+		var indexKeyCols map[string]struct{}
+		for idx, idxCols := range allNonPublicIdxToKeyCols {
+			indexName = idx
+			indexKeyCols = idxCols
+			break
+		}
+
+		tblName := tblName
+		_, err := sqlDB.Exec(fmt.Sprintf("DROP INDEX %v.%v.%v@%v;", dbName, scName, tblName, indexName))
+		if err == nil {
+			for indexKeyCol := range indexKeyCols {
+				delete(allColToIndexes[indexKeyCol], indexName)
+			}
+			delete(allNonPublicIdxToKeyCols, indexName)
+			t.Logf("DROP INDEX %v FROM TABLE %v", indexName, tblName)
+		} else if isPQErrWithCode(err, pgcode.UndefinedDatabase, pgcode.UndefinedSchema, pgcode.InvalidSchemaName, pgcode.UndefinedTable, pgcode.UndefinedObject) {
+			err = nil
+			t.Logf("Parent database or schema or table is renamed; skipping this index removal.")
+		}
+		return err
+	})
+
+	select {
+	case workerErr := <-workerErrChan:
+		t.Logf("main: a worker error %q is signaled; Inform all workers to stop.", workerErr.Error())
+		cancel()
+		wg.Wait()
+		t.Logf("main: all workers have stopped their work; Test Failure!")
+		t.Fatalf(workerErr.Error())
+	case <-time.After(testDuration):
+		t.Logf("main: time's up! Inform all workers to stop.")
+		cancel()
+		wg.Wait()
+		t.Logf("main: all workers have stopped. Validating descriptors states...")
+		validate()
+		t.Logf("main: validation succeeded! Test success!")
+	}
+}
+
+func isPQErrWithCode(err error, codes ...pgcode.Code) bool {
+	if pqErr := (*pq.Error)(nil); errors.As(err, &pqErr) {
+		for _, code := range codes {
+			if pgcode.MakeCode(string(pqErr.Code)) == code {
+				return true
+			}
+		}
+	}
+	return false
+}