Merge #88085

88085: workload/schemachanger: re-enable the schema changer workload r=fqazi a=fqazi

Previously the schema changer workload was disabled because of two intermittent issues:

1.  The workload itself had an issue due to a recent refactor that did not properly detect/handle potential execution errors.  While the set of errors was generated it was never used properly in some cases after a recent refactor
2. Some scenarios involving inserts could fail binding OID's inside the resolveOID code path, specifically this code path does not properly handle dropped descriptors. These changes enforces the old behaviour where the resolution should just *fail*.

Co-authored-by: Faizan Qazi <[email protected]>

pkg/ccl/testccl/workload/schemachange/BUILD.bazel

@@ -3,11 +3,12 @@ load("@io_bazel_rules_go//go:def.bzl", "go_test")
 
 go_test(
     name = "schemachange_test",
+    size = "large",
     srcs = [
         "main_test.go",
         "schema_change_external_test.go",
     ],
-    args = ["-test.timeout=295s"],
+    args = ["-test.timeout=895s"],
     data = [
         "//c-deps:libgeos",
     ],

pkg/ccl/testccl/workload/schemachange/schema_change_external_test.go

@@ -34,7 +34,7 @@ import (
 func TestWorkload(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer utilccl.TestingEnableEnterprise()()
-	skip.WithIssue(t, 78478)
+	skip.UnderRace(t, "test connections can be too slow under race option.")
 
 	dir := t.TempDir()
 	ctx := context.Background()
@@ -90,7 +90,7 @@ func TestWorkload(t *testing.T) {
 	ql, err := wl.Ops(ctx, []string{pgURL.String()}, reg)
 	require.NoError(t, err)
 
-	const N = 100
+	const N = 800
 	workerFn := func(ctx context.Context, fn func(ctx context.Context) error) func() error {
 		return func() error {
 			for i := 0; i < N; i++ {

pkg/cmd/roachtest/tests/schemachange_random_load.go

@@ -45,9 +45,6 @@ func registerSchemaChangeRandomLoad(r registry.Registry) {
 			spec.Geo(),
 			spec.Zones(geoZonesStr),
 		),
-		// This is set while development is still happening on the workload and we
-		// fix (or bypass) minor schema change bugs that are discovered.
-		NonReleaseBlocker: true,
 		Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
 			maxOps := 5000
 			concurrency := 20

pkg/sql/resolve_oid.go

@@ -15,6 +15,7 @@ import (
 	"fmt"
 
 	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -77,7 +78,12 @@ func resolveOID(
 	results, err := ie.QueryRowEx(ctx, "queryOid", txn,
 		sessiondata.NoSessionDataOverride, q, toResolve)
 	if err != nil {
-		if errors.HasType(err, (*tree.MultipleResultsError)(nil)) {
+		if catalog.HasInactiveDescriptorError(err) {
+			// Descriptor is either dropped or offline, so
+			// the OID does not exist.
+			return nil, true, pgerror.Newf(info.errType,
+				"%s %s does not exist", info.objName, toResolve)
+		} else if errors.HasType(err, (*tree.MultipleResultsError)(nil)) {
 			return nil, false, pgerror.Newf(pgcode.AmbiguousAlias,
 				"more than one %s named %s", info.objName, toResolve)
 		}

pkg/workload/schemachange/error_screening.go

@@ -560,14 +560,20 @@ SELECT count(*) > 0
 	return nil
 }
 
+func getValidGenerationErrors() errorCodeSet {
+	return errorCodeSet{
+		pgcode.NumericValueOutOfRange:    true,
+		pgcode.FloatingPointException:    true,
+		pgcode.InvalidTextRepresentation: true,
+	}
+}
+
 // isValidGenerationError these codes can be observed when evaluating values
 // for generated expressions. These are errors are not ignored, but added into
 // the expected set of errors.
 func isValidGenerationError(code string) bool {
 	pgCode := pgcode.MakeCode(code)
-	return pgCode == pgcode.NumericValueOutOfRange ||
-		pgCode == pgcode.FloatingPointException ||
-		pgCode == pgcode.InvalidTextRepresentation
+	return getValidGenerationErrors().contains(pgCode)
 }
 
 // validateGeneratedExpressionsForInsert goes through generated expressions and

pkg/workload/schemachange/operation_generator.go

@@ -54,7 +54,6 @@ type operationGeneratorParams struct {
 // The OperationBuilder has the sole responsibility of generating ops
 type operationGenerator struct {
 	params                *operationGeneratorParams
-	potentialExecErrors   errorCodeSet
 	expectedCommitErrors  errorCodeSet
 	potentialCommitErrors errorCodeSet
 
@@ -103,7 +102,6 @@ func makeOperationGenerator(params *operationGeneratorParams) *operationGenerato
 	return &operationGenerator{
 		params:                        params,
 		expectedCommitErrors:          makeExpectedErrorSet(),
-		potentialExecErrors:           makeExpectedErrorSet(),
 		potentialCommitErrors:         makeExpectedErrorSet(),
 		candidateExpectedCommitErrors: makeExpectedErrorSet(),
 	}
@@ -112,7 +110,6 @@ func makeOperationGenerator(params *operationGeneratorParams) *operationGenerato
 // Reset internal state used per operation within a transaction
 func (og *operationGenerator) resetOpState() {
 	og.candidateExpectedCommitErrors.reset()
-	og.potentialExecErrors.reset()
 	og.opGenLog = strings.Builder{}
 }
 
@@ -881,7 +878,7 @@ func (og *operationGenerator) addForeignKeyConstraint(
 	// perfectly if an error is expected. We can confirm post transaction with a time
 	// travel query.
 	_ = rowsSatisfyConstraint
-	og.potentialExecErrors.add(pgcode.ForeignKeyViolation)
+	stmt.potentialExecErrors.add(pgcode.ForeignKeyViolation)
 	og.potentialCommitErrors.add(pgcode.ForeignKeyViolation)
 
 	// It's possible for the table to be dropped concurrently, while we are running
@@ -1338,6 +1335,14 @@ func (og *operationGenerator) createTableAs(ctx context.Context, tx pgx.Tx) (*op
 		{code: pgcode.DuplicateColumn, condition: duplicateColumns},
 	}.add(opStmt.expectedExecErrors)
 
+	// Confirm the select itself doesn't run into any column generation errors,
+	// by executing it independently first until we add validation when adding
+	// generated columns. See issue: #81698?, which will allow us to remove this
+	// logic in the future.
+	if opStmt.expectedExecErrors.empty() {
+		opStmt.potentialExecErrors.merge(getValidGenerationErrors())
+	}
+
 	opStmt.sql = fmt.Sprintf(`CREATE TABLE %s AS %s FETCH FIRST %d ROWS ONLY`,
 		destTableName, selectStatement.String(), MaxRowsToConsume)
 	return opStmt, nil
@@ -2404,7 +2409,7 @@ func (og *operationGenerator) insertRow(ctx context.Context, tx pgx.Tx) (stmt *o
 			return nil, err
 		}
 		// We may have errors that are possible, but not guaranteed.
-		potentialErrors.add(og.potentialExecErrors)
+		potentialErrors.add(stmt.potentialExecErrors)
 		if invalidInsert {
 			generatedErrors.add(stmt.expectedExecErrors)
 			// We will be pessimistic and assume that other column related errors can

pkg/workload/schemachange/schemachange.go

@@ -313,17 +313,12 @@ func (w *schemaChangeWorker) recordInHist(elapsed time.Duration, bin histBin) {
 
 func (w *schemaChangeWorker) getErrorState() string {
 	return fmt.Sprintf("Dumping state before death:\n"+
-		"Expected errors: %s"+
-		"Potential exec errors: %s"+
 		"Expected commit errors: %s"+
 		"Potential commit errors: %s"+
 		"==========================="+
 		"Executed queries for generating errors: %s"+
 		"==========================="+
 		"Previous statements %s",
-		"",
-		//	w.opGen.expectedExecErrors.String(),
-		w.opGen.potentialExecErrors.String(),
 		w.opGen.expectedCommitErrors.String(),
 		w.opGen.potentialCommitErrors.String(),
 		w.opGen.GetOpGenLog(),

pkg/workload/schemachange/watch_dog.go

@@ -93,7 +93,15 @@ func (w *schemaChangeWatchDog) watchLoop(ctx context.Context) {
 			close(responseChannel)
 			return
 		case <-ctx.Done():
-			panic("dumping stacks, we failed to terminate threads on time.")
+			// Give the connections a small amount of time to clean up, if they fail
+			// to do so, we will dump stacks.
+			select {
+			case <-w.cmdChannel:
+				return
+			case <-time.After(time.Second * 4):
+				panic("dumping stacks, we failed to terminate threads on time.")
+
+			}
 		case <-time.After(time.Second):
 			// If the connection is making progress, the watch dog timer can be reset
 			// again.