sql: avoid unnecessary version bumps for UDTs during ALTER TABLE

Previously, in the legacy schema changer, we unconditionally updated
backreferences from tables to user-defined types (UDTs) during ALTER
TABLE operations. This happened even when the backreferences remained
unchanged.

While this behavior was functionally harmless, it caused the version of
the referenced types to be incremented unnecessarily.

This change avoids version bumps unless the backreference actually
changes.

Fixes: #144293

Epic: none
Release note: none

Author: spilchen

pkg/backup/restore_job.go

@@ -1401,7 +1401,7 @@ func createImportingDescriptors(
 					if err != nil {
 						return err
 					}
-					typDesc.AddReferencingDescriptorID(table.GetID())
+					_ = typDesc.AddReferencingDescriptorID(table.GetID())
 					if err := descsCol.WriteDescToBatch(
 						ctx, kvTrace, typDesc, b,
 					); err != nil {
@@ -3119,7 +3119,7 @@ func (r *restoreResumer) removeExistingTypeBackReferences(
 				}
 				existing := desc.(*typedesc.Mutable)
 				existing.MaybeIncrementVersion()
-				existing.RemoveReferencingDescriptorID(tbl.ID)
+				_ = existing.RemoveReferencingDescriptorID(tbl.ID)
 			}
 		}
 	}

pkg/sql/alter_table_locality.go

@@ -666,9 +666,10 @@ func setNewLocalityConfig(
 		if err != nil {
 			return err
 		}
-		typ.RemoveReferencingDescriptorID(desc.GetID())
-		if err := descsCol.WriteDescToBatch(ctx, kvTrace, typ, b); err != nil {
-			return err
+		if typ.RemoveReferencingDescriptorID(desc.GetID()) {
+			if err := descsCol.WriteDescToBatch(ctx, kvTrace, typ, b); err != nil {
+				return err
+			}
 		}
 	}
 	desc.LocalityConfig = &config
@@ -678,9 +679,10 @@ func setNewLocalityConfig(
 		if err != nil {
 			return err
 		}
-		typ.AddReferencingDescriptorID(desc.GetID())
-		if err := descsCol.WriteDescToBatch(ctx, kvTrace, typ, b); err != nil {
-			return err
+		if typ.AddReferencingDescriptorID(desc.GetID()) {
+			if err := descsCol.WriteDescToBatch(ctx, kvTrace, typ, b); err != nil {
+				return err
+			}
 		}
 	}
 	return nil

pkg/sql/catalog/typedesc/type_desc.go

@@ -409,25 +409,29 @@ func (desc *Mutable) AddEnumValue(node *tree.AlterTypeAddValue) error {
 }
 
 // AddReferencingDescriptorID adds a new referencing descriptor ID to the
-// TypeDescriptor. It ensures that duplicates are not added.
-func (desc *Mutable) AddReferencingDescriptorID(new descpb.ID) {
+// TypeDescriptor, ensuring no duplicates are added. Returns false if the ID
+// was already present and no changes were made.
+func (desc *Mutable) AddReferencingDescriptorID(new descpb.ID) bool {
 	for _, id := range desc.ReferencingDescriptorIDs {
 		if new == id {
-			return
+			return false
 		}
 	}
 	desc.ReferencingDescriptorIDs = append(desc.ReferencingDescriptorIDs, new)
+	return true
 }
 
 // RemoveReferencingDescriptorID removes the desired referencing descriptor ID
-// from the catalog.TypeDescriptor. It has no effect if the requested ID is not present.
-func (desc *Mutable) RemoveReferencingDescriptorID(remove descpb.ID) {
+// from the catalog.TypeDescriptor. If the ID is not present, the method has no
+// effect and returns false to indicate that no removal occurred.
+func (desc *Mutable) RemoveReferencingDescriptorID(remove descpb.ID) bool {
 	for i, id := range desc.ReferencingDescriptorIDs {
 		if id == remove {
 			desc.ReferencingDescriptorIDs = append(desc.ReferencingDescriptorIDs[:i], desc.ReferencingDescriptorIDs[i+1:]...)
-			return
+			return true
 		}
 	}
+	return false
 }
 
 // SetParentSchemaID sets the SchemaID of the type.

pkg/sql/create_table.go

@@ -520,7 +520,7 @@ func (n *createTableNode) startExec(params runParams) error {
 			if err != nil {
 				return errors.Wrap(err, "error resolving multi-region enum")
 			}
-			typeDesc.AddReferencingDescriptorID(desc.ID)
+			_ = typeDesc.AddReferencingDescriptorID(desc.ID)
 			err = params.p.writeTypeSchemaChange(
 				params.ctx, typeDesc, "add REGIONAL BY TABLE back reference")
 			if err != nil {

pkg/sql/drop_type.go

@@ -170,7 +170,9 @@ func (p *planner) addTypeBackReference(
 		return err
 	}
 
-	mutDesc.AddReferencingDescriptorID(ref)
+	if !mutDesc.AddReferencingDescriptorID(ref) {
+		return nil // no-op
+	}
 	return p.writeTypeSchemaChange(ctx, mutDesc, jobDesc)
 }
 
@@ -182,9 +184,10 @@ func (p *planner) removeTypeBackReferences(
 		if err != nil {
 			return err
 		}
-		mutDesc.RemoveReferencingDescriptorID(ref)
-		if err := p.writeTypeSchemaChange(ctx, mutDesc, jobDesc); err != nil {
-			return err
+		if mutDesc.RemoveReferencingDescriptorID(ref) {
+			if err := p.writeTypeSchemaChange(ctx, mutDesc, jobDesc); err != nil {
+				return err
+			}
 		}
 	}
 	return nil

pkg/sql/logictest/testdata/logic_test/alter_table

@@ -4128,9 +4128,6 @@ alter table roach add column serial_id2 SERIAL
 
 subtest end
 
-statement ok
-set use_declarative_schema_changer = on
-
 # Tests for #131948 where we incorrectly backfilled empty column
 # families for composite datums.
 subtest composite_type_131948
@@ -4454,6 +4451,7 @@ CREATE TABLE alter_table_alter_primary_key_duplicate_storage_params_a (
   b TEXT NOT NULL
 );
 
+skipif config local-legacy-schema-changer
 statement error pgcode 22023 pq: parameter "bucket_count" specified more than once
 ALTER TABLE alter_table_alter_primary_key_duplicate_storage_params_a
   ALTER PRIMARY KEY
@@ -4468,6 +4466,11 @@ subtest end
 # when adding a column with an invalid geometry expression.
 subtest alter_table_add_column_with_invalid_geometry_expression
 
+let $use_decl_sc
+SHOW use_declarative_schema_changer
+
+statement ok
+set use_declarative_schema_changer = on
 
 statement ok
 CREATE TABLE alter_table_add_column_with_invalid_geometry_expression (
@@ -4478,4 +4481,85 @@ INSERT INTO alter_table_add_column_with_invalid_geometry_expression VALUES (1);
 statement error pgcode 22023 pq: failed to construct index entries during backfill: error parsing EWKB: unexpected EOF
 ALTER TABLE alter_table_add_column_with_invalid_geometry_expression ADD COLUMN geom GEOMETRY NULL DEFAULT x'001a'
 
+statement ok
+SET use_declarative_schema_changer = $use_decl_sc;
+
+subtest end
+
+# This is a regression test for #144293 where we were bumping the UDT version of
+# all types unconditionally in the legacy schema changer.
+subtest conditional_bump_udt_version
+
+statement ok
+CREATE TABLE t_conditional_bump_udt_version (
+  id INT PRIMARY KEY
+);
+
+statement ok
+CREATE TYPE e1 AS ENUM ('a', 'b', 'c');
+
+let $e1_version
+SELECT crdb_internal.pb_to_json('descriptor', descriptor) -> 'type' ->> 'version'
+from system.descriptor
+where id = 'e1'::REGTYPE::INT - 100000;
+
+# Add a column using e1. We expect a version bump in the enum.
+statement ok
+ALTER TABLE t_conditional_bump_udt_version ADD COLUMN e1_col e1;
+
+query I
+SELECT 1
+FROM system.descriptor
+WHERE id = 'e1'::REGTYPE::INT - 100000 AND
+  (crdb_internal.pb_to_json('descriptor', descriptor) -> 'type' ->> 'version')::INT > $e1_version;
+----
+1
+
+let $e1_version
+SELECT crdb_internal.pb_to_json('descriptor', descriptor) -> 'type' ->> 'version'
+from system.descriptor
+where id = 'e1'::REGTYPE::INT - 100000;
+
+# Add a regular int column. That should not bump the version.
+statement ok
+ALTER TABLE t_conditional_bump_udt_version ADD COLUMN i_col INT;
+
+query I
+SELECT 1
+FROM system.descriptor
+WHERE id = 'e1'::REGTYPE::INT - 100000 AND
+  (crdb_internal.pb_to_json('descriptor', descriptor) -> 'type' ->> 'version')::INT = $e1_version;
+----
+1
+
+# No bump for drop column.
+statement ok
+ALTER TABLE t_conditional_bump_udt_version DROP COLUMN i_col;
+
+query I
+SELECT 1
+FROM system.descriptor
+WHERE id = 'e1'::REGTYPE::INT - 100000 AND
+  (crdb_internal.pb_to_json('descriptor', descriptor) -> 'type' ->> 'version')::INT = $e1_version;
+----
+1
+
+# Ensure version bump happens when we drop the column using the type.
+statement ok
+ALTER TABLE t_conditional_bump_udt_version DROP COLUMN e1_col;
+
+query I
+SELECT 1
+FROM system.descriptor
+WHERE id = 'e1'::REGTYPE::INT - 100000 AND
+  (crdb_internal.pb_to_json('descriptor', descriptor) -> 'type' ->> 'version')::INT > $e1_version;
+----
+1
+
+statement ok
+DROP TABLE t_conditional_bump_udt_version;
+
+statement ok
+DROP TYPE e1;
+
 subtest end

pkg/sql/schema_changer.go

@@ -1160,10 +1160,11 @@ func (sc *SchemaChanger) dropViewDeps(
 				log.Warningf(ctx, "error resolving type dependency %d", id)
 				continue
 			}
-			typeDesc.RemoveReferencingDescriptorID(viewDesc.GetID())
-			if err := descsCol.WriteDescToBatch(ctx, false /* kvTrace*/, typeDesc, b); err != nil {
-				log.Warningf(ctx, "error removing dependency from type ID %d", id)
-				return err
+			if typeDesc.RemoveReferencingDescriptorID(viewDesc.GetID()) {
+				if err := descsCol.WriteDescToBatch(ctx, false /* kvTrace*/, typeDesc, b); err != nil {
+					log.Warningf(ctx, "error removing dependency from type ID %d", id)
+					return err
+				}
 			}
 		}
 		for i := 0; i < col.NumUsesSequences(); i++ {
@@ -1878,9 +1879,9 @@ func (sc *SchemaChanger) done(ctx context.Context) error {
 					return err
 				}
 				if isAddition {
-					typ.AddReferencingDescriptorID(scTable.ID)
+					_ = typ.AddReferencingDescriptorID(scTable.ID)
 				} else {
-					typ.RemoveReferencingDescriptorID(scTable.ID)
+					_ = typ.RemoveReferencingDescriptorID(scTable.ID)
 				}
 				if err := txn.Descriptors().WriteDescToBatch(ctx, kvTrace, typ, b); err != nil {
 					return err