Merge #57491

57491: workload/schemachange: complete error screening r=ajwerner a=jayshrivastava

**workload/schemachange: complete error screening**

This change adds error screening for the following ops:
- createIndex
- setColumnNotNull 
- dropConstraint
- dropIndex
- setColumnDefault
- dropColumnStored
- setColumnType
- renameIndex
 
Since all ops now have error screening, this change removes the overhead due to supporting ops that screen for errors and do not.

Closes: #56119
Release note: None

**workload/schemachange: refactor type resolution**

Previously, the type resolver used to generate types.T structs
from type name strings did not have logic for checking if schema names
were present. This was causing enums to only exist in the public schema.

Furthermore, the use of tree.ResolvableTypeReference in functions
such as randType would make it difficult to inspect the type to see
if a schema name were present.

This change updates the type resolver to correctly handle type names
which are prefixed with schemas. Furthermore, the type resolver will
now fetch OIDs which can be used to check for type equivalence when
screening for type-related errors. This change also refactors the
operation generator to use a helper function for type resolution and
the tree.TypeName struct to represent type names.

Release note: None

**workload/schemachange: cleanup error states on op generation failure**

Previously, if an op func failed, then the op state would not
get reset on subsequent iterations of the loop inside randOp.
This meant that any expected execution errors or commit errors
from the failed op generation would linger and potentially affect
the next generated op.

This change adds some simple mechanisms to ensure the expected
errors get reset on op generation failure.

Release note: None

**workload/schemachange: allow validate after write**

Transactions do not support schema changes after writes in the
same transaction. Previously, a FeatureNotSupported error was
anticipated when a validate operation followed an insertRow operation,
but this error would never occur. This change removes that behavior
because a validate should be allowed to occur after an insert
in a transaction without expecting any errors.

Release note: None

 
**workload/schemachange: use regex to fetch tables**

Previously, the clause `LIKE 'table%'` was used to fetch tables.
This pattern was capabable of fetching system tables, such as
information_schema.tables. Generally, this workload should not
alter the system tables, and should only test using the tables it
creates. Tables created by the workload can be fetched using
the clause `~ 'table[0-9]+'`, which is the word 'table' followed
by a sequence number.

Release note: None
 
**workload/schemachange: use nested transactions for op gen and err screening**

Previously, it was possible for transactions to become aborted
while generating operations or screening for errors. When a
transaction became aborted and an operation generation function
returned an error, a new operation generation function would be
called with the same transaction. Because the transaction was
aborted, all subsequent attempts to use it would fail, so this
situation resulted in an infinite loop.

This change updates the operation generator to use savepoints
before each call to an operation generation function. If
the operation generation function returns an error, the
transaction is rolled back to the savepoint so the transaction
can be reused.

If there are any errors caused by using savepoints, the workload
will terminate with a fatal error.

Closes: #56120

Release note: None


Co-authored-by: Jayant Shrivastava <jayants@cockroachlabs.com>

pkg/sql/sem/tree/type_name.go

@@ -54,6 +54,12 @@ func (t *TypeName) String() string {
 	return AsString(t)
 }
 
+// SQLString implements the ResolvableTypeReference interface.
+func (t *TypeName) SQLString() string {
+	// FmtBareIdentifiers prevents the TypeName string from being wrapped in quotations.
+	return AsStringWithFlags(t, FmtBareIdentifiers)
+}
+
 // FQString renders the type name in full, not omitting the prefix
 // schema and catalog names. Suitable for logging, etc.
 func (t *TypeName) FQString() string {
@@ -82,6 +88,17 @@ func MakeUnqualifiedTypeName(typ Name) TypeName {
 	}}
 }
 
+// MakeSchemaQualifiedTypeName returns a new type name.
+func MakeSchemaQualifiedTypeName(schema, typ string) TypeName {
+	return TypeName{objName{
+		ObjectNamePrefix: ObjectNamePrefix{
+			ExplicitSchema: true,
+			SchemaName:     Name(schema),
+		},
+		ObjectName: Name(typ),
+	}}
+}
+
 // MakeNewQualifiedTypeName creates a fully qualified type name.
 func MakeNewQualifiedTypeName(db, schema, typ string) TypeName {
 	return TypeName{objName{
@@ -228,7 +245,8 @@ func (node *ArrayTypeReference) SQLString() string {
 
 // SQLString implements the ResolvableTypeReference interface.
 func (name *UnresolvedObjectName) SQLString() string {
-	return name.String()
+	// FmtBareIdentifiers prevents the TypeName string from being wrapped in quotations.
+	return AsStringWithFlags(name, FmtBareIdentifiers)
 }
 
 // IsReferenceSerialType returns whether the input reference is a known

pkg/workload/schemachange/BUILD.bazel

@@ -15,6 +15,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/security",
+        "//pkg/sql/catalog/colinfo",
         "//pkg/sql/parser",
         "//pkg/sql/pgwire/pgcode",
         "//pkg/sql/pgwire/pgerror",

pkg/workload/schemachange/error_code_set.go

@@ -23,6 +23,12 @@ func makeExpectedErrorSet() errorCodeSet {
 	return errorCodeSet(map[pgcode.Code]bool{})
 }
 
+func (set errorCodeSet) merge(otherSet errorCodeSet) {
+	for code := range otherSet {
+		set[code] = true
+	}
+}
+
 func (set errorCodeSet) add(code pgcode.Code) {
 	set[code] = true
 }

pkg/workload/schemachange/error_screening.go

@@ -55,16 +55,17 @@ func columnExistsOnTable(tx *pgx.Tx, tableName *tree.TableName, columnName strin
    )`, tableName.Schema(), tableName.Object(), columnName)
 }
 
-func typeExists(tx *pgx.Tx, typ tree.ResolvableTypeReference) (bool, error) {
-	if !strings.Contains(typ.SQLString(), "enum") {
+func typeExists(tx *pgx.Tx, typ *tree.TypeName) (bool, error) {
+	if !strings.Contains(typ.Object(), "enum") {
 		return true, nil
 	}
 
 	return scanBool(tx, `SELECT EXISTS (
-	SELECT typname
-		FROM pg_catalog.pg_type
-   WHERE typname = $1
-	)`, typ.SQLString())
+	SELECT ns.nspname, t.typname
+  FROM pg_catalog.pg_namespace AS ns
+  JOIN pg_catalog.pg_type AS t ON t.typnamespace = ns.oid
+ WHERE ns.nspname = $1 AND t.typname = $2
+	)`, typ.Schema(), typ.Object())
 }
 
 func tableHasRows(tx *pgx.Tx, tableName *tree.TableName) (bool, error) {
@@ -321,3 +322,138 @@ func scanStringArrayRows(tx *pgx.Tx, query string, args ...interface{}) ([][]str
 
 	return results, err
 }
+
+func indexExists(tx *pgx.Tx, tableName *tree.TableName, indexName string) (bool, error) {
+	return scanBool(tx, `SELECT EXISTS(
+			SELECT *
+			  FROM information_schema.statistics
+			 WHERE table_schema = $1
+			   AND table_name = $2
+			   AND index_name = $3
+  )`, tableName.Schema(), tableName.Object(), indexName)
+}
+
+func columnsStoredInPrimaryIdx(
+	tx *pgx.Tx, tableName *tree.TableName, columnNames tree.NameList,
+) (bool, error) {
+	columnsMap := map[string]bool{}
+	for _, name := range columnNames {
+		columnsMap[string(name)] = true
+	}
+
+	primaryColumns, err := scanStringArray(tx, `
+	SELECT array_agg(column_name)
+	  FROM (
+	        SELECT DISTINCT column_name
+	          FROM information_schema.statistics
+	         WHERE index_name = 'primary'
+	           AND table_schema = $1
+	           AND table_name = $2
+	       );
+	`, tableName.Schema(), tableName.Object())
+
+	if err != nil {
+		return false, err
+	}
+
+	for _, primaryColumn := range primaryColumns {
+		if _, exists := columnsMap[primaryColumn]; exists {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func scanStringArray(tx *pgx.Tx, query string, args ...interface{}) (b []string, err error) {
+	err = tx.QueryRow(query, args...).Scan(&b)
+	return b, err
+}
+
+// canApplyUniqueConstraint checks if the rows in a table are unique with respect
+// to the specified columns such that a unique constraint can successfully be applied.
+func canApplyUniqueConstraint(
+	tx *pgx.Tx, tableName *tree.TableName, columns []string,
+) (bool, error) {
+	columnNames := strings.Join(columns, ", ")
+
+	// If a row contains NULL in each of the columns relevant to a unique constraint,
+	// then the row will always be unique to other rows with respect to the constraint
+	// (even if there is another row with NULL values in each of the relevant columns).
+	// To account for this, the whereNotNullClause below is constructed to ignore rows
+	// with with NULL values in each of the relevant columns. Then, uniqueness can be
+	// verified easily using a SELECT DISTINCT statement.
+	whereNotNullClause := strings.Builder{}
+	for idx, column := range columns {
+		whereNotNullClause.WriteString(fmt.Sprintf("%s IS NOT NULL ", column))
+		if idx != len(columns)-1 {
+			whereNotNullClause.WriteString("OR ")
+		}
+	}
+
+	return scanBool(tx,
+		fmt.Sprintf(`
+		SELECT (
+	       SELECT count(*)
+	         FROM (
+	               SELECT DISTINCT %s
+	                 FROM %s
+	                WHERE %s
+	              )
+	      )
+	      = (
+	        SELECT count(*)
+	          FROM %s
+	         WHERE %s
+	       );
+	`, columnNames, tableName.String(), whereNotNullClause.String(), tableName.String(), whereNotNullClause.String()))
+
+}
+
+func columnContainsNull(tx *pgx.Tx, tableName *tree.TableName, columnName string) (bool, error) {
+	return scanBool(tx, fmt.Sprintf(`SELECT EXISTS (
+		SELECT %s
+		  FROM %s
+	   WHERE %s IS NULL
+	)`, columnName, tableName.String(), columnName))
+}
+
+func constraintIsPrimary(
+	tx *pgx.Tx, tableName *tree.TableName, constraintName string,
+) (bool, error) {
+	return scanBool(tx, fmt.Sprintf(`
+	SELECT EXISTS(
+	        SELECT *
+	          FROM pg_catalog.pg_constraint
+	         WHERE conrelid = '%s'::REGCLASS::INT
+	           AND conname = '%s'
+	           AND (contype = 'p')
+	       );
+	`, tableName.String(), constraintName))
+}
+
+func constraintIsUnique(
+	tx *pgx.Tx, tableName *tree.TableName, constraintName string,
+) (bool, error) {
+	return scanBool(tx, fmt.Sprintf(`
+	SELECT EXISTS(
+	        SELECT *
+	          FROM pg_catalog.pg_constraint
+	         WHERE conrelid = '%s'::REGCLASS::INT
+	           AND conname = '%s'
+	           AND (contype = 'u')
+	       );
+	`, tableName.String(), constraintName))
+}
+
+func columnIsComputed(tx *pgx.Tx, tableName *tree.TableName, columnName string) (bool, error) {
+	return scanBool(tx, `
+     SELECT (
+			 SELECT is_generated
+				 FROM information_schema.columns
+				WHERE table_schema = $1
+				  AND table_name = $2
+				  AND column_name = $3
+            )
+	         = 'YES'
+`, tableName.Schema(), tableName.Object(), columnName)
+}