sql: unblock user table changes from long-running transactions

Currently, the `CREATE USER` and the `GRANT role` operations use a
bumped descriptor to invalidate caches of the user table.
This blocked those operations until long-running transactions using the
old table version committed.
This change adds special handling of version bumps to wait for
visibility of the change rather than convergence across the cluster. It
no longer creates (blocking) jobs for the version bumps.

Epic: CRDB-49398
Fixes: #138691

Release note (sql change): The `CREATE USER` and `GRANT role` operations
now wait for full-cluster visibility of the new user table version
rather than blocking on convergence.

Author: bghal

pkg/bench/rttanalysis/testdata/benchmark_expectations

@@ -1,7 +1,7 @@
 exp,benchmark
-14,AlterRole/alter_role_with_1_option
-17,AlterRole/alter_role_with_2_options
-23,AlterRole/alter_role_with_3_options
+12,AlterRole/alter_role_with_1_option
+15,AlterRole/alter_role_with_2_options
+19,AlterRole/alter_role_with_3_options
 13,AlterTableAddCheckConstraint/alter_table_add_1_check_constraint
 13,AlterTableAddCheckConstraint/alter_table_add_2_check_constraints
 13,AlterTableAddCheckConstraint/alter_table_add_3_check_constraints
@@ -15,9 +15,9 @@ exp,benchmark
 14,AlterTableConfigureZone/alter_table_configure_zone_5_replicas
 14,AlterTableConfigureZone/alter_table_configure_zone_7_replicas_
 14,AlterTableConfigureZone/alter_table_configure_zone_ranges
-21,AlterTableDropColumn/alter_table_drop_1_column
-21,AlterTableDropColumn/alter_table_drop_2_columns
-21,AlterTableDropColumn/alter_table_drop_3_columns
+22,AlterTableDropColumn/alter_table_drop_1_column
+22,AlterTableDropColumn/alter_table_drop_2_columns
+22,AlterTableDropColumn/alter_table_drop_3_columns
 13,AlterTableDropConstraint/alter_table_drop_1_check_constraint
 13,AlterTableDropConstraint/alter_table_drop_2_check_constraints
 13,AlterTableDropConstraint/alter_table_drop_3_check_constraints
@@ -28,20 +28,20 @@ exp,benchmark
 11,AlterTableUnsplit/alter_table_unsplit_at_2_values
 14,AlterTableUnsplit/alter_table_unsplit_at_3_values
 2-4,Audit/select_from_an_audit_table
-20,CreateRole/create_role_with_1_option
-23,CreateRole/create_role_with_2_options
-26,CreateRole/create_role_with_3_options
-21,CreateRole/create_role_with_no_options
+13,CreateRole/create_role_with_1_option
+16,CreateRole/create_role_with_2_options
+19,CreateRole/create_role_with_3_options
+14,CreateRole/create_role_with_no_options
 14,"Discard/DISCARD_ALL,_1_tables_in_1_db"
 19,"Discard/DISCARD_ALL,_2_tables_in_2_dbs"
 0,"Discard/DISCARD_ALL,_no_tables"
 15,DropDatabase/drop_database_0_tables
 16,DropDatabase/drop_database_1_table
 16,DropDatabase/drop_database_2_tables
 16,DropDatabase/drop_database_3_tables
-29-30,DropRole/drop_1_role
-37,DropRole/drop_2_roles
-45-49,DropRole/drop_3_roles
+22,DropRole/drop_1_role
+30,DropRole/drop_2_roles
+39,DropRole/drop_3_roles
 13,DropSequence/drop_1_sequence
 15,DropSequence/drop_2_sequences
 17,DropSequence/drop_3_sequences
@@ -59,8 +59,8 @@ exp,benchmark
 8,Grant/grant_all_on_1_table
 8,Grant/grant_all_on_2_tables
 8,Grant/grant_all_on_3_tables
-15,GrantRole/grant_1_role
-19,GrantRole/grant_2_roles
+12,GrantRole/grant_1_role
+16,GrantRole/grant_2_roles
 12-15,Jobs/cancel_job
 3,Jobs/crdb_internal.system_jobs
 3-5,Jobs/jobs_page_default
@@ -79,19 +79,19 @@ exp,benchmark
 4,ORMQueries/django_column_introspection_1_table
 4,ORMQueries/django_column_introspection_4_tables
 4,ORMQueries/django_column_introspection_8_tables
-6,ORMQueries/django_comment_introspection_with_comments
-6,ORMQueries/django_table_introspection_1_table
-6,ORMQueries/django_table_introspection_8_tables
+5,ORMQueries/django_comment_introspection_with_comments
+5,ORMQueries/django_table_introspection_1_table
+5,ORMQueries/django_table_introspection_8_tables
 0,ORMQueries/has_column_privilege_using_attnum
 0,ORMQueries/has_column_privilege_using_column_name
 0,ORMQueries/has_schema_privilege
 0,ORMQueries/has_sequence_privilege
 0,ORMQueries/has_table_privilege
-5,ORMQueries/hasura_column_descriptions
+6,ORMQueries/hasura_column_descriptions
 20,ORMQueries/hasura_column_descriptions_8_tables
 5,ORMQueries/hasura_column_descriptions_modified
 4,ORMQueries/information_schema._pg_index_position
-4,ORMQueries/introspection_description_join
+5,ORMQueries/introspection_description_join
 4,ORMQueries/liquibase_migrations
 8,ORMQueries/liquibase_migrations_on_multiple_dbs
 5,ORMQueries/npgsql_fields
@@ -105,14 +105,14 @@ exp,benchmark
 3,ORMQueries/pg_namespace
 4,ORMQueries/pg_type
 6,ORMQueries/prisma_column_descriptions
-3,ORMQueries/prisma_column_descriptions_updated
+4,ORMQueries/prisma_column_descriptions_updated
 5,ORMQueries/prisma_types_16
 5,ORMQueries/prisma_types_4
 8,Revoke/revoke_all_on_1_table
 8,Revoke/revoke_all_on_2_tables
 8,Revoke/revoke_all_on_3_tables
-13,RevokeRole/revoke_1_role
-15,RevokeRole/revoke_2_roles
+10,RevokeRole/revoke_1_role
+12,RevokeRole/revoke_2_roles
 4,ShowGrants/grant_2_roles
 4,ShowGrants/grant_3_roles
 4,ShowGrants/grant_4_roles
@@ -129,7 +129,7 @@ exp,benchmark
 0,UDFResolution/select_from_udf
 2,UseManyRoles/use_2_roles
 2,UseManyRoles/use_50_roles
-1,VirtualTableQueries/select_crdb_internal.invalid_objects_with_1_fk
+2,VirtualTableQueries/select_crdb_internal.invalid_objects_with_1_fk
 1,VirtualTableQueries/select_crdb_internal.tables_with_1_fk
 0,VirtualTableQueries/show_create_all_routines
 0,VirtualTableQueries/show_create_all_triggers

pkg/cli/testdata/doctor/test_examine_cluster

@@ -3,5 +3,5 @@ debug doctor examine cluster
 debug doctor examine cluster
 Examining 69 descriptors and 68 namespace entries...
   ParentID 100, ParentSchemaID 101: relation "foo" (105): expected matching namespace entry, found none
-Examining 13 jobs...
+Examining 12 jobs...
 ERROR: validation failed

pkg/cli/testdata/doctor/test_examine_cluster_dropped

@@ -2,5 +2,5 @@ debug doctor examine cluster
 ----
 debug doctor examine cluster
 Examining 68 descriptors and 68 namespace entries...
-Examining 11 jobs...
+Examining 10 jobs...
 No problems found!

pkg/sql/catalog/descs/collection.go

@@ -267,6 +267,11 @@ func (tc *Collection) IsNewUncommittedDescriptor(id descpb.ID) bool {
 	return false
 }
 
+// IsVersionBumpOfUncommittedDescriptor returns true if the descriptor is only having its version bumped (without mutations) in this transaction.
+func (tc *Collection) IsVersionBumpOfUncommittedDescriptor(id descpb.ID) bool {
+	return tc.uncommitted.versionBumpOnly[id]
+}
+
 // HasUncommittedNewOrDroppedDescriptors returns true if the collection contains
 // any uncommitted descriptors that are newly created or dropped.
 func (tc *Collection) HasUncommittedNewOrDroppedDescriptors() bool {
@@ -330,9 +335,30 @@ func (tc *Collection) AddUncommittedDescriptor(
 			desc.DescriptorType(), desc.GetName(), desc.GetID())
 	}
 	tc.markAsShadowedName(desc.GetID())
+
+	// It's the responsibility of the caller to restore the flag (see MaybeMarkVersionBump)
+	tc.uncommitted.versionBumpOnly[desc.GetID()] = false
+
 	return tc.uncommitted.upsert(ctx, desc)
 }
 
+// MaybeMarkVersionBump provides a defer-friendly function that updates the
+// version bump only flag for the descriptor so it reflects previous mutations
+// to the descriptor along with the current mutation.
+// The returned function is to be called after AddUncommittedDescriptor or
+// functions that call it (such as WriteDescToBatch).
+func (tc *Collection) MaybeMarkVersionBump(
+	desc catalog.MutableDescriptor, isVersionBump bool,
+) func() {
+	prev, ok := tc.uncommitted.versionBumpOnly[desc.GetID()]
+
+	return func() {
+		tc.uncommitted.versionBumpOnly[desc.GetID()] =
+			(!ok || prev) && // if the flag isn't set or it was previously set up
+				isVersionBump
+	}
+}
+
 // WriteDescToBatch calls MaybeIncrementVersion, adds the descriptor to the
 // collection as an uncommitted descriptor, and writes it into b.
 func (tc *Collection) WriteDescToBatch(

pkg/sql/catalog/descs/uncommitted_descriptors.go

@@ -42,14 +42,21 @@ type uncommittedDescriptors struct {
 	uncommitted       nstree.NameMap
 	original, mutable nstree.IDMap
 
+	// versionBumpOnly stores descriptors that have been modified only by
+	// changing their version counter, without any other mutations. These bumps
+	// are used to trigger cache invalidation and are tracked to avoid the
+	// typical block on version convergence behavior.
+	versionBumpOnly map[descpb.ID]bool
+
 	// memAcc is the actual account of an injected, upstream monitor
 	// to track memory usage of uncommittedDescriptors.
 	memAcc mon.BoundAccount
 }
 
 func makeUncommittedDescriptors(monitor *mon.BytesMonitor) uncommittedDescriptors {
 	return uncommittedDescriptors{
-		memAcc: monitor.MakeBoundAccount(),
+		versionBumpOnly: map[descpb.ID]bool{},
+		memAcc:          monitor.MakeBoundAccount(),
 	}
 }
 
@@ -61,10 +68,11 @@ func (ud *uncommittedDescriptors) reset(ctx context.Context) {
 	ud.memAcc.Clear(ctx)
 	old := *ud
 	*ud = uncommittedDescriptors{
-		original:    old.original,
-		uncommitted: old.uncommitted,
-		mutable:     old.mutable,
-		memAcc:      old.memAcc,
+		original:        old.original,
+		uncommitted:     old.uncommitted,
+		mutable:         old.mutable,
+		versionBumpOnly: map[descpb.ID]bool{},
+		memAcc:          old.memAcc,
 	}
 }
 

pkg/sql/catalog/lease/lease.go

@@ -265,6 +265,46 @@ SELECT DISTINCT session_id FROM system.lease WHERE desc_id=%d AND crdb_internal.
 	return sessionIDs, nil
 }
 
+// countSessionsHoldingStaleDescriptor finds sessionIDs that are holding a lease
+// on the previous version of desc but not the current version.
+func countSessionsHoldingStaleDescriptor(
+	ctx context.Context, txn isql.Txn, desc catalog.Descriptor, region string,
+) (int, error) {
+	b := strings.Builder{}
+
+	// Counts sessions that have previous version of the descriptor but not the current version
+	b.WriteString(fmt.Sprintf(`
+		SELECT count(DISTINCT l1.session_id)
+		FROM system.lease l1 
+		WHERE l1.desc_id = %d 
+		AND l1.version < %d 
+		AND crdb_internal.sql_liveness_is_alive(l1.session_id)
+		AND NOT EXISTS (
+			SELECT 1 FROM system.lease l2 
+			WHERE l2.desc_id = l1.desc_id 
+			AND l2.session_id = l1.session_id 
+			AND l2.version = %d
+		`, desc.GetID(), desc.GetVersion(), desc.GetVersion()))
+	if region != "" {
+		b.WriteString(fmt.Sprintf(" AND l2.crdb_region='%s'", region))
+	}
+	b.WriteString(")")
+	if region != "" {
+		b.WriteString(fmt.Sprintf(" AND l1.crdb_region='%s'", region))
+	}
+
+	rows, err := txn.QueryBuffered(ctx, "count-sessions-holding-stale-descriptor", txn.KV(), b.String())
+	if err != nil {
+		return 0, err
+	}
+
+	if len(rows) == 0 {
+		return 0, nil
+	}
+
+	return int(tree.MustBeDInt(rows[0][0])), nil
+}
+
 // WaitForInitialVersion waits for a lease to be acquired on a newly created
 // object on any nodes that have already leased the schema out. This ensures
 // that their leaseGeneration is incremented before the user commit completes,
@@ -511,6 +551,100 @@ func (m *Manager) WaitForOneVersion(
 	return desc, nil
 }
 
+// WaitForNewVersion returns once all leaseholders of any version of the
+// descriptor hold a lease of the current version.
+//
+// The MaxRetries and MaxDuration in retryOpts should not be set.
+func (m *Manager) WaitForNewVersion(
+	ctx context.Context,
+	descriptorId descpb.ID,
+	retryOpts retry.Options,
+	regions regionliveness.CachedDatabaseRegions,
+) (catalog.Descriptor, error) {
+	if retryOpts.MaxRetries != 0 {
+		return nil, errors.New("The MaxRetries option shouldn't be set in WaitForNewVersion")
+	}
+	if retryOpts.MaxDuration != 0 {
+		return nil, errors.New("The MaxDuration option shouldn't be set in WaitForNewVersion")
+	}
+
+	var desc catalog.Descriptor
+
+	var success bool
+	// Block until each leaseholder on the previous version of the descriptor
+	// also holds a lease on the current version of the descriptor (`for all
+	// session: (session in Prev => session in Curr)` for the set theory
+	// enjoyers).
+	for r := retry.Start(retryOpts); r.Next(); {
+		var err error
+		desc, err = m.maybeGetDescriptorWithoutValidation(ctx, descriptorId, true)
+		if err != nil {
+			return nil, err
+		}
+
+		db := m.storage.db
+
+		// Get the sessions with leases in each region.
+		if err := db.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+			prober := regionliveness.NewLivenessProber(db.KV(), m.storage.codec, regions, m.settings)
+			regionMap, err := prober.QueryLiveness(ctx, txn.KV())
+			if err != nil {
+				return err
+			}
+
+			// On single region clusters we can query everything at once.
+			if regionMap == nil {
+				regionMap = regionliveness.LiveRegions{"": struct{}{}}
+			}
+
+			var staleSessionCount int
+			for region := range regionMap {
+				var regionStaleSessionCount int
+				var err error
+				if hasTimeout, timeout := prober.GetProbeTimeout(); hasTimeout {
+					err = timeutil.RunWithTimeout(ctx, "count-sessions-holding-stale-descriptor-by-region", timeout, func(ctx context.Context) (countErr error) {
+						regionStaleSessionCount, countErr = countSessionsHoldingStaleDescriptor(ctx, txn, desc, region)
+						return countErr
+					})
+				} else {
+					regionStaleSessionCount, err = countSessionsHoldingStaleDescriptor(ctx, txn, desc, region)
+				}
+				if err != nil {
+					return handleRegionLivenessErrors(ctx, prober, region, err)
+				}
+
+				staleSessionCount += regionStaleSessionCount
+
+				if regionStaleSessionCount != 0 { // quit early
+					if region == "" {
+						log.Infof(ctx, "%d sessions holding stale descriptor", regionStaleSessionCount)
+					} else {
+						log.Infof(ctx, "Region '%s' has %d sessions holding stale descriptor", region, regionStaleSessionCount)
+					}
+					break
+				}
+			}
+
+			if staleSessionCount == 0 {
+				success = true
+			}
+
+			return nil
+		}); err != nil {
+			return nil, err
+		}
+
+		if success {
+			break
+		}
+	}
+	if !success {
+		return nil, errors.New("Exited lease acquisition loop before success")
+	}
+
+	return desc, nil
+}
+
 // IDVersion represents a descriptor ID, version pair that are
 // meant to map to a single immutable descriptor.
 type IDVersion struct {