catalog: restrict scope of descriptor validation on write

Previously, we performed descriptor validation at txn commit time for all
so-called "uncommitted descriptors" which includes descriptors which are
only read and remain unchanged. This commit excludes these from validation.

Release note: None

pkg/bench/rttanalysis/testdata/benchmark_expectations

@@ -1,57 +1,57 @@
 exp,benchmark
-19,AlterRole/alter_role_with_1_option
-20,AlterRole/alter_role_with_2_options
-25,AlterRole/alter_role_with_3_options
-17,AlterTableAddCheckConstraint/alter_table_add_1_check_constraint
-17,AlterTableAddCheckConstraint/alter_table_add_2_check_constraints
-17,AlterTableAddCheckConstraint/alter_table_add_3_check_constraints
-17,AlterTableAddColumn/alter_table_add_1_column
-17,AlterTableAddColumn/alter_table_add_2_columns
-17,AlterTableAddColumn/alter_table_add_3_columns
-22,AlterTableAddForeignKey/alter_table_add_1_foreign_key
-27,AlterTableAddForeignKey/alter_table_add_2_foreign_keys
-32,AlterTableAddForeignKey/alter_table_add_3_foreign_keys
-22,AlterTableAddForeignKey/alter_table_add_foreign_key_with_3_columns
-21,AlterTableConfigureZone/alter_table_configure_zone_5_replicas
-21,AlterTableConfigureZone/alter_table_configure_zone_7_replicas_
-21,AlterTableConfigureZone/alter_table_configure_zone_ranges
-21-22,AlterTableDropColumn/alter_table_drop_1_column
-25,AlterTableDropColumn/alter_table_drop_2_columns
-29,AlterTableDropColumn/alter_table_drop_3_columns
-18,AlterTableDropConstraint/alter_table_drop_1_check_constraint
-19,AlterTableDropConstraint/alter_table_drop_2_check_constraints
-20,AlterTableDropConstraint/alter_table_drop_3_check_constraints
-12-13,AlterTableSplit/alter_table_split_at_1_value
-18-19,AlterTableSplit/alter_table_split_at_2_values
-24-25,AlterTableSplit/alter_table_split_at_3_values
-8,AlterTableUnsplit/alter_table_unsplit_at_1_value
-10,AlterTableUnsplit/alter_table_unsplit_at_2_values
-12,AlterTableUnsplit/alter_table_unsplit_at_3_values
-21,CreateRole/create_role_with_1_option
-23,CreateRole/create_role_with_2_options
-24,CreateRole/create_role_with_3_options
-22,CreateRole/create_role_with_no_options
-22,DropDatabase/drop_database_0_tables
-31,DropDatabase/drop_database_1_table
-40,DropDatabase/drop_database_2_tables
-49,DropDatabase/drop_database_3_tables
-29,DropRole/drop_1_role
-36,DropRole/drop_2_roles
-43,DropRole/drop_3_roles
-19,DropSequence/drop_1_sequence
-26,DropSequence/drop_2_sequences
-33,DropSequence/drop_3_sequences
-22,DropTable/drop_1_table
-32,DropTable/drop_2_tables
-42,DropTable/drop_3_tables
-22,DropView/drop_1_view
-30,DropView/drop_2_views
-38,DropView/drop_3_views
-18,Grant/grant_all_on_1_table
-20,Grant/grant_all_on_2_tables
-22,Grant/grant_all_on_3_tables
-21,GrantRole/grant_1_role
-24,GrantRole/grant_2_roles
+17,AlterRole/alter_role_with_1_option
+18,AlterRole/alter_role_with_2_options
+23,AlterRole/alter_role_with_3_options
+16,AlterTableAddCheckConstraint/alter_table_add_1_check_constraint
+16,AlterTableAddCheckConstraint/alter_table_add_2_check_constraints
+16,AlterTableAddCheckConstraint/alter_table_add_3_check_constraints
+16,AlterTableAddColumn/alter_table_add_1_column
+16,AlterTableAddColumn/alter_table_add_2_columns
+16,AlterTableAddColumn/alter_table_add_3_columns
+21,AlterTableAddForeignKey/alter_table_add_1_foreign_key
+26,AlterTableAddForeignKey/alter_table_add_2_foreign_keys
+31,AlterTableAddForeignKey/alter_table_add_3_foreign_keys
+21,AlterTableAddForeignKey/alter_table_add_foreign_key_with_3_columns
+20,AlterTableConfigureZone/alter_table_configure_zone_5_replicas
+20,AlterTableConfigureZone/alter_table_configure_zone_7_replicas_
+20,AlterTableConfigureZone/alter_table_configure_zone_ranges
+20,AlterTableDropColumn/alter_table_drop_1_column
+24,AlterTableDropColumn/alter_table_drop_2_columns
+28,AlterTableDropColumn/alter_table_drop_3_columns
+17,AlterTableDropConstraint/alter_table_drop_1_check_constraint
+18,AlterTableDropConstraint/alter_table_drop_2_check_constraints
+19,AlterTableDropConstraint/alter_table_drop_3_check_constraints
+11,AlterTableSplit/alter_table_split_at_1_value
+17,AlterTableSplit/alter_table_split_at_2_values
+23,AlterTableSplit/alter_table_split_at_3_values
+7,AlterTableUnsplit/alter_table_unsplit_at_1_value
+9,AlterTableUnsplit/alter_table_unsplit_at_2_values
+11,AlterTableUnsplit/alter_table_unsplit_at_3_values
+19,CreateRole/create_role_with_1_option
+21,CreateRole/create_role_with_2_options
+22,CreateRole/create_role_with_3_options
+20,CreateRole/create_role_with_no_options
+20,DropDatabase/drop_database_0_tables
+29,DropDatabase/drop_database_1_table
+38,DropDatabase/drop_database_2_tables
+47,DropDatabase/drop_database_3_tables
+27,DropRole/drop_1_role
+34,DropRole/drop_2_roles
+41,DropRole/drop_3_roles
+18,DropSequence/drop_1_sequence
+25,DropSequence/drop_2_sequences
+32,DropSequence/drop_3_sequences
+21,DropTable/drop_1_table
+31,DropTable/drop_2_tables
+41,DropTable/drop_3_tables
+21,DropView/drop_1_view
+29,DropView/drop_2_views
+37,DropView/drop_3_views
+17,Grant/grant_all_on_1_table
+19,Grant/grant_all_on_2_tables
+21,Grant/grant_all_on_3_tables
+19,GrantRole/grant_1_role
+22,GrantRole/grant_2_roles
 2,ORMQueries/activerecord_type_introspection_query
 2,ORMQueries/django_table_introspection_1_table
 2,ORMQueries/django_table_introspection_4_tables
@@ -64,19 +64,19 @@ exp,benchmark
 2,ORMQueries/pg_class
 2,ORMQueries/pg_namespace
 2,ORMQueries/pg_type
-18,Revoke/revoke_all_on_1_table
-20,Revoke/revoke_all_on_2_tables
-22,Revoke/revoke_all_on_3_tables
-20,RevokeRole/revoke_1_role
-22,RevokeRole/revoke_2_roles
+17,Revoke/revoke_all_on_1_table
+19,Revoke/revoke_all_on_2_tables
+21,Revoke/revoke_all_on_3_tables
+18,RevokeRole/revoke_1_role
+20,RevokeRole/revoke_2_roles
 1,SystemDatabaseQueries/select_system.users_with_empty_database_Name
 1,SystemDatabaseQueries/select_system.users_with_schema_Name
 2,SystemDatabaseQueries/select_system.users_without_schema_Name
-26,Truncate/truncate_1_column_0_rows
-26,Truncate/truncate_1_column_1_row
-26,Truncate/truncate_1_column_2_rows
-26,Truncate/truncate_2_column_0_rows
-26,Truncate/truncate_2_column_1_rows
-26,Truncate/truncate_2_column_2_rows
+25,Truncate/truncate_1_column_0_rows
+25,Truncate/truncate_1_column_1_row
+25,Truncate/truncate_1_column_2_rows
+25,Truncate/truncate_2_column_0_rows
+25,Truncate/truncate_2_column_1_rows
+25,Truncate/truncate_2_column_2_rows
 1,VirtualTableQueries/select_crdb_internal.invalid_objects_with_1_fk
 1,VirtualTableQueries/select_crdb_internal.tables_with_1_fk

pkg/sql/catalog/catalog.go

@@ -29,12 +29,12 @@ type MutableDescriptor interface {
 	// SetDrainingNames sets the draining names for the descriptor.
 	SetDrainingNames([]descpb.NameInfo)
 
-	// Accessors for the original state of the descriptor prior to the mutations.
-	OriginalName() string
-	OriginalID() descpb.ID
-	OriginalVersion() descpb.DescriptorVersion
 	// ImmutableCopy returns an immutable copy of this descriptor.
 	ImmutableCopy() Descriptor
+	// ImmutableCopyOfOriginalVersion returns an immutable copy of the original
+	// version of this descriptor.
+	ImmutableCopyOfOriginalVersion() Descriptor
+
 	// IsNew returns whether the descriptor was created in this transaction.
 	IsNew() bool
 

pkg/sql/catalog/dbdesc/database_desc.go

@@ -311,37 +311,22 @@ func (desc *Mutable) MaybeIncrementVersion() {
 	desc.ModificationTime = hlc.Timestamp{}
 }
 
-// OriginalName implements the MutableDescriptor interface.
-func (desc *Mutable) OriginalName() string {
-	if desc.ClusterVersion == nil {
-		return ""
-	}
-	return desc.ClusterVersion.Name
-}
-
-// OriginalID implements the MutableDescriptor interface.
-func (desc *Mutable) OriginalID() descpb.ID {
-	if desc.ClusterVersion == nil {
-		return descpb.InvalidID
-	}
-	return desc.ClusterVersion.ID
-}
-
-// OriginalVersion implements the MutableDescriptor interface.
-func (desc *Mutable) OriginalVersion() descpb.DescriptorVersion {
-	if desc.ClusterVersion == nil {
-		return 0
-	}
-	return desc.ClusterVersion.Version
-}
-
 // ImmutableCopy implements the MutableDescriptor interface.
 func (desc *Mutable) ImmutableCopy() catalog.Descriptor {
-	imm := NewBuilder(desc.DatabaseDesc()).BuildImmutableDatabase()
+	imm := NewBuilder(desc.DatabaseDesc()).BuildImmutable()
 	imm.(*immutable).isUncommittedVersion = desc.IsUncommittedVersion()
 	return imm
 }
 
+// ImmutableCopyOfOriginalVersion implements the MutableDescriptor interface.
+func (desc *Mutable) ImmutableCopyOfOriginalVersion() catalog.Descriptor {
+	if desc.ClusterVersion == nil {
+		empty := descpb.DatabaseDescriptor{}
+		return NewBuilder(&empty).BuildImmutable()
+	}
+	return NewBuilder(desc.ClusterVersion.DatabaseDesc()).BuildImmutable()
+}
+
 // IsNew implements the MutableDescriptor interface.
 func (desc *Mutable) IsNew() bool {
 	return desc.ClusterVersion == nil

pkg/sql/catalog/descs/collection_test.go

@@ -239,7 +239,7 @@ func TestAddUncommittedDescriptorAndMutableResolution(t *testing.T) {
 
 			immByName, err := descriptors.GetImmutableDatabaseByName(ctx, txn, "db", flags)
 			require.NoError(t, err)
-			require.Equal(t, mut.OriginalVersion(), immByName.GetVersion())
+			require.Equal(t, mut.ImmutableCopyOfOriginalVersion().GetVersion(), immByName.GetVersion())
 
 			_, immByID, err := descriptors.GetImmutableDatabaseByID(ctx, txn, db.GetID(), flags)
 			require.NoError(t, err)

pkg/sql/catalog/descs/kv_descriptors.go

@@ -325,7 +325,8 @@ func (kd *kvDescriptors) getDescriptorsWithNewVersion() []lease.IDVersion {
 	_ = kd.uncommittedDescriptors.IterateByID(func(entry catalog.NameEntry) error {
 		desc := entry.(*uncommittedDescriptor)
 		if mut := desc.mutable; !mut.IsNew() && mut.IsUncommittedVersion() {
-			descs = append(descs, lease.NewIDVersionPrev(mut.OriginalName(), mut.OriginalID(), mut.OriginalVersion()))
+			orig := mut.ImmutableCopyOfOriginalVersion()
+			descs = append(descs, lease.NewIDVersionPrev(orig.GetName(), orig.GetID(), orig.GetVersion()))
 		}
 		return nil
 	})
@@ -347,7 +348,13 @@ func (kd *kvDescriptors) getUncommittedTables() (tables []catalog.TableDescripto
 func (kd *kvDescriptors) validateUncommittedDescriptors(ctx context.Context, txn *kv.Txn) error {
 	descs := make([]catalog.Descriptor, 0, kd.uncommittedDescriptors.Len())
 	_ = kd.uncommittedDescriptors.IterateByID(func(descriptor catalog.NameEntry) error {
-		descs = append(descs, descriptor.(*uncommittedDescriptor).immutable)
+		ud := descriptor.(*uncommittedDescriptor)
+		newVersion := ud.mutable.ImmutableCopy()
+		oldVersion := ud.mutable.ImmutableCopyOfOriginalVersion()
+		if !oldVersion.DescriptorProto().Equal(newVersion.DescriptorProto()) {
+			// Only validate descriptors which have changed.
+			descs = append(descs, ud.immutable)
+		}
 		return nil
 	})
 	if len(descs) == 0 {
@@ -388,7 +395,7 @@ func (kd *kvDescriptors) addUncommittedDescriptor(
 	desc catalog.MutableDescriptor,
 ) (*uncommittedDescriptor, error) {
 	version := desc.GetVersion()
-	origVersion := desc.OriginalVersion()
+	origVersion := desc.ImmutableCopyOfOriginalVersion().GetVersion()
 	if version != origVersion && version != origVersion+1 {
 		return nil, errors.AssertionFailedf(
 			"descriptor %d version %d not compatible with cluster version %d",

pkg/sql/catalog/hydratedtables/hydratedcache.go

@@ -41,7 +41,7 @@ import (
 // The cache's contract is a bit tricky. In order to use a hydrated type, the
 // caller needs to have a lease on the relevant type descriptor. The way that
 // this is made to work is that the user provides a handle to a leased
-// ImmutableCopy and then the cache will call through to type resolver for each
+// descriptor and then the cache will call through to type resolver for each
 // of the referenced types which ensures that user always uses properly leased
 // descriptors. While all of the types will need to be resolved, they should
 // already be cached so, in this way, this cache prevents the need to copy
@@ -132,7 +132,7 @@ type cachedType struct {
 	typDesc catalog.TypeDescriptor
 }
 
-// GetHydratedTableDescriptor returns an ImmutableCopy with the types
+// GetHydratedTableDescriptor returns a table descriptor with the types
 // hydrated. It may use a cached copy but all of the relevant type descriptors
 // will be retrieved via the TypeDescriptorResolver. Note that if the table
 // descriptor is modified, nil will be returned. If any of the types used by

pkg/sql/catalog/schemadesc/schema_desc.go

@@ -65,7 +65,7 @@ func formatSafeMessage(typeName string, desc catalog.SchemaDescriptor) string {
 //
 // Note: Today this isn't actually ever mutated but rather exists for a future
 // where we anticipate having a mutable copy of Schema descriptors. There's a
-// large amount of space to question this `Mutable|ImmutableCopy` version of each
+// large amount of space to question this `Mutable|immutable` version of each
 // descriptor type. Maybe it makes no sense but we're running with it for the
 // moment. This is an intermediate state on the road to descriptors being
 // handled outside of the catalog entirely as interfaces.
@@ -217,37 +217,22 @@ func (desc *Mutable) MaybeIncrementVersion() {
 	desc.ModificationTime = hlc.Timestamp{}
 }
 
-// OriginalName implements the MutableDescriptor interface.
-func (desc *Mutable) OriginalName() string {
-	if desc.ClusterVersion == nil {
-		return ""
-	}
-	return desc.ClusterVersion.Name
-}
-
-// OriginalID implements the MutableDescriptor interface.
-func (desc *Mutable) OriginalID() descpb.ID {
-	if desc.ClusterVersion == nil {
-		return descpb.InvalidID
-	}
-	return desc.ClusterVersion.ID
-}
-
-// OriginalVersion implements the MutableDescriptor interface.
-func (desc *Mutable) OriginalVersion() descpb.DescriptorVersion {
-	if desc.ClusterVersion == nil {
-		return 0
-	}
-	return desc.ClusterVersion.Version
-}
-
 // ImmutableCopy implements the MutableDescriptor interface.
 func (desc *Mutable) ImmutableCopy() catalog.Descriptor {
 	imm := NewBuilder(desc.SchemaDesc()).BuildImmutable()
 	imm.(*immutable).isUncommittedVersion = desc.IsUncommittedVersion()
 	return imm
 }
 
+// ImmutableCopyOfOriginalVersion implements the MutableDescriptor interface.
+func (desc *Mutable) ImmutableCopyOfOriginalVersion() catalog.Descriptor {
+	if desc.ClusterVersion == nil {
+		empty := descpb.SchemaDescriptor{}
+		return NewBuilder(&empty).BuildImmutable()
+	}
+	return NewBuilder(desc.ClusterVersion.SchemaDesc()).BuildImmutable()
+}
+
 // IsNew implements the MutableDescriptor interface.
 func (desc *Mutable) IsNew() bool {
 	return desc.ClusterVersion == nil

pkg/sql/catalog/tabledesc/structured.go

@@ -963,21 +963,6 @@ func (desc *Mutable) MaybeIncrementVersion() {
 	desc.ModificationTime = hlc.Timestamp{}
 }
 
-// OriginalName implements the MutableDescriptor interface.
-func (desc *Mutable) OriginalName() string {
-	return desc.ClusterVersion.Name
-}
-
-// OriginalID implements the MutableDescriptor interface.
-func (desc *Mutable) OriginalID() descpb.ID {
-	return desc.ClusterVersion.ID
-}
-
-// OriginalVersion implements the MutableDescriptor interface.
-func (desc *Mutable) OriginalVersion() descpb.DescriptorVersion {
-	return desc.ClusterVersion.Version
-}
-
 // FormatTableLocalityConfig formats the table locality.
 func FormatTableLocalityConfig(c *descpb.TableDescriptor_LocalityConfig, f *tree.FmtCtx) error {
 	switch v := c.Locality.(type) {
@@ -2113,7 +2098,6 @@ func (desc *Mutable) addMutation(m descpb.DescriptorMutation) {
 func (desc *wrapper) MakeFirstMutationPublic(
 	includeConstraints catalog.MutationPublicationFilter,
 ) (catalog.TableDescriptor, error) {
-	// Clone the ImmutableTable descriptor because we want to create an ImmutableCopy one.
 	table := NewBuilder(desc.TableDesc()).BuildExistingMutableTable()
 	mutationID := desc.Mutations[0].MutationID
 	i := 0
@@ -2138,9 +2122,9 @@ func (desc *wrapper) MakeFirstMutationPublic(
 	return table, nil
 }
 
-// MakePublic creates a Mutable from the immutable by making the it public.
+// MakePublic returns this table descriptor as a Mutable with the state set to
+// PUBLIC.
 func (desc *wrapper) MakePublic() catalog.TableDescriptor {
-	// Clone the ImmutableTable descriptor because we want to create an ImmutableCopy one.
 	table := NewBuilder(desc.TableDesc()).BuildExistingMutableTable()
 	table.State = descpb.DescriptorState_PUBLIC
 	table.Version++

pkg/sql/catalog/tabledesc/table_desc.go

@@ -121,6 +121,11 @@ func (desc *Mutable) ImmutableCopy() catalog.Descriptor {
 	return NewBuilder(desc.TableDesc()).BuildImmutable()
 }
 
+// ImmutableCopyOfOriginalVersion implements the MutableDescriptor interface.
+func (desc *Mutable) ImmutableCopyOfOriginalVersion() catalog.Descriptor {
+	return NewBuilder(&desc.ClusterVersion).BuildImmutable()
+}
+
 // IsUncommittedVersion implements the Descriptor interface.
 func (desc *Mutable) IsUncommittedVersion() bool {
 	return desc.IsNew() || desc.GetVersion() != desc.ClusterVersion.GetVersion()