catalog: fix tracking of id and name state

This allows us to re-enable the COMMENT ON statements in the
schemachange workload.

The previous code only updated the entries that were already in the
byIDState map. However, some descriptor IDs may not be in that map, so
instead we should add everything we just read into the map.

Release note (bug fix): Fixed a bug where COMMENT statements could
fail with an "unexpected value" error if multiple COMMENT statements
were running concurrently.

pkg/bench/rttanalysis/testdata/benchmark_expectations

@@ -114,5 +114,5 @@ exp,benchmark
 3,UDFResolution/select_from_udf
 1,VirtualTableQueries/select_crdb_internal.invalid_objects_with_1_fk
 1,VirtualTableQueries/select_crdb_internal.tables_with_1_fk
-11,VirtualTableQueries/virtual_table_cache_with_point_lookups
+5,VirtualTableQueries/virtual_table_cache_with_point_lookups
 15,VirtualTableQueries/virtual_table_cache_with_schema_change

pkg/sql/catalog/internal/catkv/catalog_reader_cached.go

@@ -200,15 +200,27 @@ func (c *cachedCatalogReader) ScanAll(ctx context.Context, txn *kv.Txn) (nstree.
 	c.hasScanAll = true
 	c.hasScanNamespaceForDatabases = true
 	c.hasScanAllComments = true
-	for id, s := range c.byIDState {
-		s.hasScanNamespaceForDatabaseEntries = true
-		s.hasScanNamespaceForDatabaseSchemas = true
-		s.hasGetDescriptorEntries = true
-		c.byIDState[id] = s
-	}
-	for ni, s := range c.byNameState {
-		s.hasGetNamespaceEntries = true
-		c.byNameState[ni] = s
+	if err := read.ForEachDescriptor(func(desc catalog.Descriptor) error {
+		// We must update the byID and byName states for each descriptor that
+		// was read.
+		var idState byIDStateValue
+		var nameState byNameStateValue
+		idState.hasScanNamespaceForDatabaseEntries = true
+		idState.hasScanNamespaceForDatabaseSchemas = true
+		idState.hasGetDescriptorEntries = true
+		nameState.hasGetNamespaceEntries = true
+		c.setByIDState(desc.GetID(), idState)
+		ni := descpb.NameInfo{
+			ParentID: desc.GetParentID(),
+			Name:     desc.GetName(),
+		}
+		if typ := desc.DescriptorType(); typ != catalog.Database && typ != catalog.Schema {
+			ni.ParentSchemaID = desc.GetParentSchemaID()
+		}
+		c.setByNameState(ni, nameState)
+		return nil
+	}); err != nil {
+		return nstree.Catalog{}, err
 	}
 	return read, nil
 }

pkg/sql/catalog/internal/catkv/testdata/testdata_app

@@ -405,6 +405,15 @@ is_desc_id_known_to_not_exist id=123
 ----
 true
 
+# Make sure scan_all properly updated the cache.
+is_id_in_cache id=107
+----
+true
+
+is_id_in_cache id=108
+----
+true
+
 # Get* queries involving IDs or names which don't exist after a
 # ScanAll should bypass storage in the cached CatalogReader.
 get_by_ids id=456

pkg/sql/catalog/internal/catkv/testdata/testdata_system

@@ -438,6 +438,15 @@ is_desc_id_known_to_not_exist id=123
 ----
 true
 
+# Make sure scan_all properly updated the cache.
+is_id_in_cache id=107
+----
+true
+
+is_id_in_cache id=108
+----
+true
+
 # Get* queries involving IDs or names which don't exist after a
 # ScanAll should bypass storage in the cached CatalogReader.
 get_by_ids id=456

pkg/workload/schemachange/operation_generator.go

@@ -2802,7 +2802,12 @@ func (og *operationGenerator) commentOn(ctx context.Context, tx pgx.Tx) (*opStmt
 	}
 
 	stmt := makeOpStmt(OpStmtDDL)
-	stmt.sql = fmt.Sprintf(`COMMENT ON %s IS 'comment from the RSW'`, picked)
+	comment := "comment from the RSW"
+	if og.params.rng.Float64() < 0.3 {
+		// Delete the comment with some probability.
+		comment = ""
+	}
+	stmt.sql = fmt.Sprintf(`COMMENT ON %s IS '%s'`, picked, comment)
 	return stmt, nil
 }
 

pkg/workload/schemachange/optype.go

@@ -279,7 +279,7 @@ var opWeights = []int{
 	alterTableSetColumnDefault:        1,
 	alterTableSetColumnNotNull:        1,
 	alterTypeDropValue:                0, // Disabled and tracked with #114844, #113859, and #115612.
-	commentOn:                         0, // Disabled and tracked with #116795.
+	commentOn:                         1,
 	createFunction:                    1,
 	createIndex:                       1,
 	createSchema:                      1,