workload/schemachange: remove FK violation prediction for INSERT operations

Previously, the random schema workload attempted to predict whether INSERT
operations would violate foreign key constraints by analyzing the database
state and constraint definitions. This prediction logic involved:
- Querying the database to check if FK violations would occur
- Checking for ongoing FK constraint mutations
- Distinguishing between expected and potential FK violations

This commit removes all FK violation prediction logic and makes
ForeignKeyViolation unconditionally a potential execution and commit error
for INSERT operations. This aligns with the existing pattern used for
adding FK constraints, where FK violations are always treated as potential
errors due to the asynchronous validation job.

This change simplifies the code by removing ~200 lines of complex
prediction logic and deflakes tests that depend on accurate error handling
during concurrent schema changes.

Resolves: #155686

Epic: None

Release note: None

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: rafiss

pkg/workload/schemachange/error_screening.go

@@ -8,7 +8,6 @@ package schemachange
 import (
 	"context"
 	"fmt"
-	"regexp"
 	"slices"
 	"strings"
 
@@ -789,36 +788,6 @@ func (og *operationGenerator) scanStringArrayNullableRows(
 	return results, nil
 }
 
-func (og *operationGenerator) scanStringArrayRows(
-	ctx context.Context, tx pgx.Tx, query string, args ...interface{},
-) ([][]string, error) {
-	rows, err := tx.Query(ctx, query, args...)
-	if err != nil {
-		return nil, errors.Wrapf(err, "scanStringArrayRows: %q %q", query, args)
-	}
-	defer rows.Close()
-
-	var results [][]string
-	for rows.Next() {
-		var columnNames []string
-		err := rows.Scan(&columnNames)
-		if err != nil {
-			return nil, errors.Wrapf(err, "scan: %q, args %q, scanArgs %q", query, columnNames, args)
-		}
-		results = append(results, columnNames)
-	}
-
-	if rows.Err() != nil {
-		return nil, rows.Err()
-	}
-
-	og.LogQueryResults(
-		query,
-		results,
-		args...)
-	return results, nil
-}
-
 func (og *operationGenerator) indexExists(
 	ctx context.Context, tx pgx.Tx, tableName *tree.TableName, indexName string,
 ) (bool, error) {
@@ -1051,188 +1020,6 @@ func (og *operationGenerator) constraintExists(
 	 )`, string(tableName), string(constraintName))
 }
 
-var (
-	// regexpUnknownSchemaErr matches unknown schema errors with
-	// a descriptor ID, which will have the form: unknown schema "[123]"
-	regexpUnknownSchemaErr = regexp.MustCompile(`unknown schema "\[\d+]"`)
-)
-
-// checkAndAdjustForUnknownSchemaErrors in certain contexts we will attempt to
-// bind descriptors without leasing them, since we are using crdb_internal tables,
-// so it's possible for said descriptor to be dropped before we bind it. This
-// method will allow for "unknown schema [xx]" in those contexts.
-func (og *operationGenerator) checkAndAdjustForUnknownSchemaErrors(err error) error {
-	if pgErr := new(pgconn.PgError); errors.As(err, &pgErr) &&
-		pgcode.MakeCode(pgErr.Code) == pgcode.InvalidSchemaName {
-		if regexpUnknownSchemaErr.MatchString(pgErr.Message) {
-			og.LogMessage(fmt.Sprintf("Rolling back due to unknown schema error %v",
-				err))
-			// Force a rollback and log inside the operation generator.
-			return errors.Mark(err, errRunInTxnRbkSentinel)
-		}
-	}
-	return err
-}
-
-// violatesFkConstraints checks if the rows to be inserted will result in a foreign key violation.
-func (og *operationGenerator) violatesFkConstraints(
-	ctx context.Context,
-	tx pgx.Tx,
-	tableName *tree.TableName,
-	nonGeneratedColNames []tree.Name,
-	rows [][]string,
-) (expectedViolation, potentialViolation bool, err error) {
-	// TODO(annie): readd the join on active constraints once #120702 is resolved.
-	//
-	// N.B. We add random noise to column names that makes it hard to just directly call on these names. This is
-	// not the case with table/schema names; thus, only column names are quote_ident'ed to ensure that they get
-	// referenced properly.
-	fkConstraints, err := og.scanStringArrayRows(ctx, tx, fmt.Sprintf(`
-		  SELECT array[
-			  parent.table_schema,
-			  parent.table_name,
-			  parent.column_name,
-			  parent.is_generated,
-			  child.column_name,
-			  child.is_generated
-		  ]
-		  FROM (
-		        SELECT conname, conkey, confkey, conrelid, confrelid
-		          FROM pg_constraint
-		         WHERE contype = 'f'
-		           AND conrelid = '%s'::REGCLASS::INT8
-		       ) AS con
-		  JOIN (
-		        SELECT column_name, ordinal_position, column_default, is_generated
-		          FROM information_schema.columns
-		         WHERE table_schema = '%s'
-		           AND table_name = '%s'
-		       ) AS child ON conkey[1] = child.ordinal_position
-		  JOIN (
-		        SELECT pc.oid,
-		               cols.table_schema,
-		               cols.table_name,
-		               cols.column_name,
-		               cols.ordinal_position,
-		               cols.is_generated
-		          FROM pg_class AS pc
-		          JOIN pg_namespace AS pn ON pc.relnamespace = pn.oid
-		          JOIN information_schema.columns AS cols ON (pc.relname = cols.table_name AND pn.nspname = cols.table_schema)
-		       ) AS parent ON (
-		                       con.confkey[1] = parent.ordinal_position
-		                       AND con.confrelid = parent.oid
-		                      )
-		 WHERE child.column_name != 'rowid';
-`, tableName.String(), tableName.Schema(), tableName.Object()))
-	if err != nil {
-		return false, false, og.checkAndAdjustForUnknownSchemaErrors(err)
-	}
-
-	// Maps a non-generated column name to its index. This way, the value of a
-	// column in a row can be looked up using row[colToIndexMap["columnName"]] = "valueForColumn"
-	nonGeneratedColumnNameToIndexMap := map[tree.Name]int{}
-
-	for i, name := range nonGeneratedColNames {
-		nonGeneratedColumnNameToIndexMap[name] = i
-	}
-
-	for _, row := range rows {
-		for _, constraint := range fkConstraints {
-			parentTableSchema := tree.Name(constraint[0])
-			parentTableName := tree.Name(constraint[1])
-			parentColumnName := tree.Name(constraint[2])
-			parentIsGenerated := constraint[3] == "ALWAYS"
-			childColumnName := tree.Name(constraint[4])
-			childIsGenerated := constraint[5] == "ALWAYS"
-
-			// If self referential, there cannot be a violation.
-			parentAndChildAreSame := parentTableSchema == tableName.SchemaName && parentTableName == tableName.ObjectName
-			if parentAndChildAreSame && parentColumnName == childColumnName {
-				continue
-			}
-
-			// If the foreign key involves any generated columns, skip detailed FK checking
-			// and conservatively assume a violation may occur. This avoids the complexity
-			// of reasoning about values that are automatically computed by the database.
-			if parentIsGenerated || childIsGenerated {
-				return false, true, nil
-			}
-
-			violation, err := og.violatesFkConstraintsHelper(
-				ctx, tx, nonGeneratedColumnNameToIndexMap, parentTableSchema, parentTableName, parentColumnName, childColumnName, tableName, parentAndChildAreSame, row, rows,
-			)
-			if err != nil {
-				return false, false, err
-			}
-			if violation {
-				return true, false, nil
-			}
-		}
-	}
-
-	return false, false, nil
-}
-
-// violatesFkConstraintsHelper checks if inserting a single row would violate a
-// foreign key constraint between the given child and parent columns.
-//
-// This function assumes that neither the parent nor child columns are generated.
-// The caller must ensure this; generated columns are not supported.
-func (og *operationGenerator) violatesFkConstraintsHelper(
-	ctx context.Context,
-	tx pgx.Tx,
-	nonGeneratedColumnNameToIndexMap map[tree.Name]int,
-	parentTableSchema, parentTableName, parentColumn, childColumn tree.Name,
-	childTableName *tree.TableName,
-	parentAndChildAreSameTable bool,
-	rowToInsert []string,
-	allRows [][]string,
-) (bool, error) {
-
-	childIndex, ok := nonGeneratedColumnNameToIndexMap[childColumn]
-	if !ok {
-		return false, errors.Newf("child column %s does not exist in table %s", childColumn, childTableName)
-	}
-	childValue := rowToInsert[childIndex]
-	// If the value to insert in the child column is NULL and the column default is NULL, then it is not possible to have a fk violation.
-	if childValue == "NULL" {
-		return false, nil
-	}
-	// If the parent and child are the same table, then any rows in an existing
-	// insert may satisfy the same constraint.
-	var parentAndChildSameQueryColumns []string
-	if parentAndChildAreSameTable {
-		for _, otherRow := range allRows {
-			parentIdx, ok := nonGeneratedColumnNameToIndexMap[parentColumn]
-			if !ok {
-				return false, errors.Newf("parent column %s does not exist in table %s", parentColumn, childTableName)
-			}
-			parentValueInSameInsert := otherRow[parentIdx]
-
-			// Skip over NULL values.
-			if parentValueInSameInsert == "NULL" {
-				continue
-			}
-			parentAndChildSameQueryColumns = append(parentAndChildSameQueryColumns,
-				fmt.Sprintf("%s = %s", parentValueInSameInsert, childValue))
-		}
-	}
-	checkSharedParentChildRows := ""
-	if len(parentAndChildSameQueryColumns) > 0 {
-		// Check if none of the rows being inserted satisfy the foreign key constraint,
-		// since the foreign key constraints refers to the same table. So, anything in
-		// the insert batch can satisfy the constraint.
-		checkSharedParentChildRows = fmt.Sprintf("NOT (true = ANY (ARRAY [%s])) AND",
-			strings.Join(parentAndChildSameQueryColumns, ","))
-	}
-	q := fmt.Sprintf(`
-	    SELECT %s count(*) = 0 from %s.%s
-	    WHERE %s = (%s)
-	`,
-		checkSharedParentChildRows, parentTableSchema.String(), parentTableName.String(), parentColumn.String(), childValue)
-	return og.scanBool(ctx, tx, q)
-}
-
 func (og *operationGenerator) columnIsInAddingOrDroppingIndex(
 	ctx context.Context, tx pgx.Tx, tableName *tree.TableName, columnName tree.Name,
 ) (bool, error) {
@@ -1766,35 +1553,6 @@ func (og *operationGenerator) tableHasUniqueConstraintMutation(
 		);`, tableName)
 }
 
-// tableHasForeignKeyMutation determines if a table has any foreign key constraint
-// mutation ongoing. This means either being added or dropped.
-func (og *operationGenerator) tableHasForeignKeyMutation(
-	ctx context.Context, tx pgx.Tx, tableName *tree.TableName,
-) (bool, error) {
-	return og.scanBool(ctx, tx, `
-		WITH table_desc AS (
-			SELECT crdb_internal.pb_to_json(
-				'desc',
-				descriptor,
-				false
-			)->'table' as d
-			FROM system.descriptor
-			WHERE id = $1::REGCLASS
-		)
-		SELECT EXISTS (
-			SELECT * FROM (
-			SELECT jsonb_array_elements(
-				CASE WHEN d->'mutations' IS NULL
-				THEN '[]'::JSONB
-				ELSE d->'mutations'
-				END
-			) as m
-			FROM table_desc)
-			WHERE (m->>'direction')::STRING IN ('ADD', 'DROP')
-			AND (m->'constraint'->>'foreign_key') IS NOT NULL
-		);`, tableName)
-}
-
 // getTableForeignKeyReferences returns a list of tables that reference
 // the specified table via foreign key references.
 func (og *operationGenerator) getTableForeignKeyReferences(

pkg/workload/schemachange/operation_generator.go

@@ -3134,9 +3134,6 @@ func (og *operationGenerator) insertRow(ctx context.Context, tx pgx.Tx) (stmt *o
 	// we will need to evaluate generated expressions below.
 	hasUniqueConstraints := false
 	hasUniqueConstraintsMutations := false
-	hasForeignKeyMutations := false
-	expectedFkViolation := false
-	potentialFkViolation := false
 	if !anyInvalidInserts {
 		// Verify if the new row may violate unique constraints by checking the
 		// constraints in the database.
@@ -3151,31 +3148,17 @@ func (og *operationGenerator) insertRow(ctx context.Context, tx pgx.Tx) (stmt *o
 		if err != nil {
 			return nil, err
 		}
-		// Verify if the new row will violate fk constraints by checking the constraints and rows
-		// in the database.
-		expectedFkViolation, potentialFkViolation, err = og.violatesFkConstraints(ctx, tx, tableName, nonGeneratedColNames, rows)
-		if err != nil {
-			return nil, err
-		}
-		// Determine if a foreign key mutation exists.
-		hasForeignKeyMutations, err = og.tableHasForeignKeyMutation(ctx, tx, tableName)
-		if err != nil {
-			return nil, err
-		}
 	}
 
 	stmt.potentialExecErrors.addAll(codesWithConditions{
 		{code: pgcode.UniqueViolation, condition: hasUniqueConstraints || hasUniqueConstraintsMutations},
-		{code: pgcode.ForeignKeyViolation, condition: expectedFkViolation || potentialFkViolation || hasForeignKeyMutations},
+		{code: pgcode.ForeignKeyViolation, condition: true},
 		{code: pgcode.NotNullViolation, condition: true},
 		{code: pgcode.CheckViolation, condition: true},
 		{code: pgcode.InsufficientPrivilege, condition: true}, // For RLS violations
 	})
-	og.expectedCommitErrors.addAll(codesWithConditions{
-		{code: pgcode.ForeignKeyViolation, condition: expectedFkViolation && !hasForeignKeyMutations},
-	})
 	og.potentialCommitErrors.addAll(codesWithConditions{
-		{code: pgcode.ForeignKeyViolation, condition: potentialFkViolation || hasForeignKeyMutations},
+		{code: pgcode.ForeignKeyViolation, condition: true},
 	})
 
 	var formattedRows []string