workload/schemachange: refactor invalid object validation into shared helper

Extract duplicate validation logic from schemachange tests into a shared
ValidateInvalidObjects function in the workload package.

Release note: none

Author: rafiss

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

@@ -26,7 +26,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/randutil"
 	"github.com/cockroachdb/cockroach/pkg/workload"
 	"github.com/cockroachdb/cockroach/pkg/workload/histogram"
-	_ "github.com/cockroachdb/cockroach/pkg/workload/schemachange"
+	"github.com/cockroachdb/cockroach/pkg/workload/schemachange"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/errgroup"
 )
@@ -58,7 +58,8 @@ func TestWorkload(t *testing.T) {
 		workload.Opser
 		workload.Flagser
 	})
-	tdb := sqlutils.MakeSQLRunner(tc.ServerConn(0))
+	db := tc.ServerConn(0)
+	tdb := sqlutils.MakeSQLRunner(db)
 	reg := histogram.NewRegistry(20*time.Second, m.Name)
 	tdb.Exec(t, "CREATE USER testuser")
 	tdb.Exec(t, "CREATE DATABASE schemachange")
@@ -72,36 +73,20 @@ func TestWorkload(t *testing.T) {
 		require.NoError(t, os.WriteFile(fmt.Sprintf("%s/%s.rows", dir, name), []byte(sqlutils.MatrixToStr(mat)), 0666))
 	}
 
-	// Reusable validation function.
 	findInvalidObjects := func() {
 		t.Helper()
-		var (
-			id           int
-			databaseName string
-			schemaName   string
-			objName      string
-			objError     string
-		)
-		numInvalidObjects := 0
-		rows, err := tdb.DB.QueryContext(ctx, `SELECT id, database_name, schema_name, obj_name, error FROM "".crdb_internal.invalid_objects`)
+		invalidObjects, err := schemachange.ValidateInvalidObjects(ctx, db)
 		if err != nil {
 			t.Fatal(err)
 		}
-		for rows.Next() {
-			numInvalidObjects++
-			if err := rows.Scan(&id, &databaseName, &schemaName, &objName, &objError); err != nil {
-				t.Fatal(err)
-			}
+		for _, obj := range invalidObjects {
 			t.Logf(
-				"invalid object found: id: %d, database_name: %s, schema_name: %s, obj_name: %s, error: %s",
-				id, databaseName, schemaName, objName, objError,
+				"invalid object found: id: %d, database_name: %s, schema_name: %s, obj_name: %s, error: %v",
+				obj.ID, obj.DatabaseName, obj.SchemaName, obj.ObjName, obj.Error,
 			)
 		}
-		if err := rows.Err(); err != nil {
-			t.Fatal(err)
-		}
-		if numInvalidObjects > 0 {
-			t.Errorf("found %d invalid objects", numInvalidObjects)
+		if len(invalidObjects) > 0 {
+			t.Errorf("found %d invalid objects", len(invalidObjects))
 		}
 	}
 
@@ -142,7 +127,6 @@ func TestWorkload(t *testing.T) {
 		t.Logf("running DROP with use_declarative_schema_changer = %s", schemaChangerSetting)
 		tdb.Exec(t, "SET use_declarative_schema_changer = $1", schemaChangerSetting)
 		tdb.Exec(t, "DROP DATABASE schemachange CASCADE")
-		tdb.Exec(t, "RESET use_declarative_schema_changer")
 		findInvalidObjects()
 	}()
 

pkg/cmd/roachtest/tests/BUILD.bazel

@@ -304,6 +304,7 @@ go_library(
         "//pkg/workload/debug",
         "//pkg/workload/histogram",
         "//pkg/workload/histogram/exporter",
+        "//pkg/workload/schemachange",
         "//pkg/workload/tpcc",
         "//pkg/workload/tpcds",
         "//pkg/workload/tpch",

pkg/cmd/roachtest/tests/schemachange_random_load.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	gosql "database/sql"
 	"fmt"
+	"math/rand/v2"
 	"path/filepath"
 
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/cluster"
@@ -18,6 +19,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/spec"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/test"
 	"github.com/cockroachdb/cockroach/pkg/roachprod/install"
+	"github.com/cockroachdb/cockroach/pkg/workload/schemachange"
 )
 
 const (
@@ -57,33 +59,18 @@ func runSchemaChangeRandomLoad(
 	ctx context.Context, t test.Test, c cluster.Cluster, maxOps, concurrency int,
 ) {
 	validate := func(db *gosql.DB) {
-		var (
-			id           int
-			databaseName string
-			schemaName   string
-			objName      string
-			objError     string
-		)
-		numInvalidObjects := 0
-		rows, err := db.QueryContext(ctx, `SELECT id, database_name, schema_name, obj_name, error FROM crdb_internal.invalid_objects`)
+		invalidObjects, err := schemachange.ValidateInvalidObjects(ctx, db)
 		if err != nil {
 			t.Fatal(err)
 		}
-		for rows.Next() {
-			numInvalidObjects++
-			if err := rows.Scan(&id, &databaseName, &schemaName, &objName, &objError); err != nil {
-				t.Fatal(err)
-			}
+		for _, obj := range invalidObjects {
 			t.L().Errorf(
-				"invalid object found: id: %d, database_name: %s, schema_name: %s, obj_name: %s, error: %s",
-				id, databaseName, schemaName, objName, objError,
+				"invalid object found: id: %d, database_name: %s, schema_name: %s, obj_name: %s, error: %v",
+				obj.ID, obj.DatabaseName, obj.SchemaName, obj.ObjName, obj.Error,
 			)
 		}
-		if err := rows.Err(); err != nil {
-			t.Fatal(err)
-		}
-		if numInvalidObjects > 0 {
-			t.Fatalf("found %d invalid objects", numInvalidObjects)
+		if len(invalidObjects) > 0 {
+			t.Fatalf("found %d invalid objects", len(invalidObjects))
 		}
 	}
 	loadNode := c.Node(1)
@@ -142,8 +129,16 @@ func runSchemaChangeRandomLoad(
 
 	t.Status("performing validation after workload")
 	validate(db)
-	t.Status("dropping database")
-	_, err = db.ExecContext(ctx, `USE defaultdb; DROP DATABASE schemachange CASCADE;`)
+	schemaChangerSetting := "on"
+	if rand.Float32() < 0.5 {
+		schemaChangerSetting = "off"
+	}
+	t.Status(fmt.Sprintf("dropping database with use_declarative_schema_changer = %s", schemaChangerSetting))
+	_, err = db.ExecContext(ctx, "SET use_declarative_schema_changer = $1", schemaChangerSetting)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = db.ExecContext(ctx, "DROP DATABASE schemachange CASCADE")
 	if err != nil {
 		t.Fatal(err)
 	}

pkg/workload/schemachange/BUILD.bazel

@@ -14,6 +14,7 @@ go_library(
         "schemachange.go",
         "tracing.go",
         "type_resolver.go",
+        "validation.go",
         "watch_dog.go",
         "workload_result.go",
         ":gen-optype-stringer",  # keep

pkg/workload/schemachange/validation.go

@@ -0,0 +1,51 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package schemachange
+
+import (
+	"context"
+	gosql "database/sql"
+
+	"github.com/cockroachdb/errors"
+)
+
+// InvalidObject represents an invalid database object found during validation.
+type InvalidObject struct {
+	ID           int
+	DatabaseName string
+	SchemaName   string
+	ObjName      string
+	Error        string
+}
+
+// ValidateInvalidObjects checks for invalid objects in the database by querying
+// crdb_internal.invalid_objects. It returns a slice of InvalidObject structs
+// representing any invalid objects found, or an error if the query fails.
+//
+// This function is useful for validating that schema change operations haven't
+// left the database in an inconsistent state with orphaned or invalid objects.
+func ValidateInvalidObjects(ctx context.Context, db *gosql.DB) ([]InvalidObject, error) {
+	query := `SELECT id, database_name, schema_name, obj_name, error FROM crdb_internal.invalid_objects`
+	rows, err := db.QueryContext(ctx, query)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to query invalid objects")
+	}
+
+	var invalidObjects []InvalidObject
+	for rows.Next() {
+		var obj InvalidObject
+		if err := rows.Scan(&obj.ID, &obj.DatabaseName, &obj.SchemaName, &obj.ObjName, &obj.Error); err != nil {
+			return nil, errors.Wrapf(err, "failed to scan invalid object row")
+		}
+		invalidObjects = append(invalidObjects, obj)
+	}
+
+	if err := rows.Err(); err != nil {
+		return nil, errors.Wrapf(err, "error iterating invalid objects")
+	}
+
+	return invalidObjects, nil
+}