catalog/lease: update locked timestamps when updates are received

fqazi · fqazi · commit 2c7f8316a005 · 2025-10-07T15:09:03.000-04:00
Previously, the lease manager would increase its locked read time stamp was when a check point occurred on the range feed. This was not adequate because this could cause schema changes could be slowed down if the old version of descriptors was held. While the current design does not intentionally reserve, previous versions, a later update will start holding old versions intentionally to serve queries. To address this, this patch has schema changes write the list of descriptors modified to a special key `/Table/Descriptors/2/<First ID in Schema Change`, which will be used by the lease manager range feed to to update timestamps. Additionally, this patch also deflakes and fixes `TestLeaseManagerLockedTimestampBasic`, which was not setup correctly. Informs: #153618 Fixes: #153826 Release note: None
diff --git a/pkg/backup/backup_test.go b/pkg/backup/backup_test.go
@@ -6687,7 +6687,6 @@ INSERT INTO baz.bar VALUES (110, 'a'), (210, 'b'), (310, 'c'), (410, 'd'), (510,
 	tenant10.Exec(t, `BACKUP DATABASE baz INTO 'userfile://defaultdb.myfililes/test4' with revision_history`)
 	expected = nil
 	for _, resume := range []exportResumePoint{
-		{mkSpan(id2, "/Tenant/10/Table/3", "/Tenant/10/Table/4"), withoutTS},
 		{mkSpan(id2, "/Tenant/10/Table/:id/1", "/Tenant/10/Table/:id/2"), withoutTS},
 		{mkSpan(id2, "/Tenant/10/Table/:id/1/210", "/Tenant/10/Table/:id/2"), withoutTS},
 		// We have two entries for 210 because of history and super small table
diff --git a/pkg/backup/targets.go b/pkg/backup/targets.go
@@ -185,7 +185,7 @@ func getAllDescChanges(
 	startTime, endTime hlc.Timestamp,
 	priorIDs map[descpb.ID]descpb.ID,
 ) ([]backuppb.BackupManifest_DescriptorRevision, error) {
-	startKey := codec.TablePrefix(keys.DescriptorTableID)
+	startKey := codec.IndexPrefix(keys.DescriptorTableID, keys.DescriptorTablePrimaryKeyIndexID)
 	endKey := startKey.PrefixEnd()
 
 	g := ctxgroup.WithContext(ctx)
diff --git a/pkg/ccl/changefeedccl/schemafeed/schema_feed.go b/pkg/ccl/changefeedccl/schemafeed/schema_feed.go
@@ -831,7 +831,7 @@ func (tf *schemaFeed) fetchDescriptorVersions(
 	}
 	codec := tf.leaseMgr.Codec()
 	start := crtime.NowMono()
-	span := roachpb.Span{Key: codec.TablePrefix(keys.DescriptorTableID)}
+	span := roachpb.Span{Key: codec.IndexPrefix(keys.DescriptorTableID, keys.DescriptorTablePrimaryKeyIndexID)}
 	span.EndKey = span.Key.PrefixEnd()
 
 	tf.mu.Lock()
diff --git a/pkg/ccl/changefeedccl/schemafeed/schematestutils/schema_test_utils.go b/pkg/ccl/changefeedccl/schemafeed/schematestutils/schema_test_utils.go
@@ -148,7 +148,7 @@ func FetchDescVersionModificationTime(
 	db := serverutils.OpenDBConn(
 		t, s.SQLAddr(), dbName, false, s.AppStopper())
 
-	tblKey := s.Codec().TablePrefix(keys.DescriptorTableID)
+	tblKey := s.Codec().IndexPrefix(keys.DescriptorTableID, keys.DescriptorTablePrimaryKeyIndexID)
 	header := kvpb.RequestHeader{
 		Key:    tblKey,
 		EndKey: tblKey.PrefixEnd(),
diff --git a/pkg/config/system.go b/pkg/config/system.go
@@ -211,9 +211,9 @@ func (s *SystemConfig) GetLargestObjectID(
 	// Search for the descriptor table entries within the SystemConfig. lowIndex
 	// (in s.Values) is the first and highIndex one past the last KV pair in the
 	// descriptor table.
-	lowBound := keys.SystemSQLCodec.TablePrefix(keys.DescriptorTableID)
+	lowBound := keys.SystemSQLCodec.IndexPrefix(keys.DescriptorTableID, keys.DescriptorTablePrimaryKeyIndexID)
 	lowIndex := s.getIndexBound(lowBound)
-	highBound := keys.SystemSQLCodec.TablePrefix(keys.DescriptorTableID + 1)
+	highBound := keys.SystemSQLCodec.IndexPrefix(keys.DescriptorTableID+1, keys.DescriptorTablePrimaryKeyIndexID)
 	highIndex := s.getIndexBound(highBound)
 	if lowIndex == highIndex {
 		return 0, fmt.Errorf("descriptor table not found in system config of %d values", len(s.Values))
diff --git a/pkg/keys/constants.go b/pkg/keys/constants.go
@@ -426,9 +426,14 @@ const (
 	ZonesTableConfigColumnID = 2
 	ZonesTableConfigColFamID = 2
 
-	DescriptorTablePrimaryKeyIndexID         = 1
-	DescriptorTableDescriptorColID           = 2
-	DescriptorTableDescriptorColFamID        = 2
+	DescriptorTablePrimaryKeyIndexID  = 1
+	DescriptorTableDescriptorColID    = 2
+	DescriptorTableDescriptorColFamID = 2
+	// DescriptorTableDescriptorUpdateIndexID is not a real index. It is a special
+	// ID used to construct index entries that inform the lease subsystem of
+	// descriptor updates within a transaction. The value for such an entry is a
+	// descpb.DescriptorUpdates message.
+	DescriptorTableDescriptorUpdateIndexID   = 2
 	TenantsTablePrimaryKeyIndexID            = 1
 	SpanConfigurationsTablePrimaryKeyIndexID = 1
 	CommentsTablePrimaryKeyIndexID           = 1
diff --git a/pkg/keys/sql.go b/pkg/keys/sql.go
@@ -209,6 +209,12 @@ func (e sqlEncoder) DescMetadataPrefix() roachpb.Key {
 	return e.IndexPrefix(DescriptorTableID, DescriptorTablePrimaryKeyIndexID)
 }
 
+// DescUpdatePrefix returns the key prefix for all descriptors in the
+// system.descriptor table.
+func (e sqlEncoder) DescUpdatePrefix() roachpb.Key {
+	return e.IndexPrefix(DescriptorTableID, DescriptorTableDescriptorUpdateIndexID)
+}
+
 // DescMetadataKey returns the key for the descriptor in the system.descriptor
 // table.
 func (e sqlEncoder) DescMetadataKey(descID uint32) roachpb.Key {
@@ -217,6 +223,14 @@ func (e sqlEncoder) DescMetadataKey(descID uint32) roachpb.Key {
 	return MakeFamilyKey(k, DescriptorTableDescriptorColFamID)
 }
 
+// DescMetadataUpdateKey returns the key for the descriptor in the system.descriptor
+// table.
+func (e sqlEncoder) DescMetadataUpdateKey(descID uint32) roachpb.Key {
+	k := e.DescUpdatePrefix()
+	k = encoding.EncodeUvarintAscending(k, uint64(descID))
+	return MakeFamilyKey(k, 0)
+}
+
 // TenantMetadataKey returns the key for the tenant metadata in the
 // system.tenants table.
 func (e sqlEncoder) TenantMetadataKey(tenID roachpb.TenantID) roachpb.Key {
@@ -307,6 +321,19 @@ func (d sqlDecoder) DecodeDescMetadataID(key roachpb.Key) (uint32, error) {
 	return uint32(id), nil
 }
 
+// DecodeDescUpdateKey decodes a descriptor ID from a descriptor metadata key.
+func (d sqlDecoder) DecodeDescUpdateKey(key roachpb.Key) (bool, error) {
+	// Extract table and index ID from key.
+	_, tableID, indexID, err := d.DecodeIndexPrefix(key)
+	if err != nil {
+		return false, err
+	}
+	if tableID != DescriptorTableID {
+		return false, errors.Errorf("key is not a descriptor table entry: %v", key)
+	}
+	return indexID == DescriptorTableDescriptorUpdateIndexID, nil
+}
+
 // DecodeTenantMetadataID decodes a tenant ID from a tenant metadata key.
 func (d sqlDecoder) DecodeTenantMetadataID(key roachpb.Key) (roachpb.TenantID, error) {
 	// Extract table and index ID from key.
diff --git a/pkg/server/diagnostics/reporter.go b/pkg/server/diagnostics/reporter.go
@@ -449,7 +449,7 @@ func (r *Reporter) populateSQLInfo(uptime int64, sql *diagnosticspb.SQLInstanceI
 // type fields. Check out `schematelemetry` package for a better data source for
 // collecting redacted schema information.
 func (r *Reporter) collectSchemaInfo(ctx context.Context) ([]descpb.TableDescriptor, error) {
-	startKey := keys.MakeSQLCodec(r.TenantID).TablePrefix(keys.DescriptorTableID)
+	startKey := keys.MakeSQLCodec(r.TenantID).IndexPrefix(keys.DescriptorTableID, keys.DescriptorTablePrimaryKeyIndexID)
 	endKey := startKey.PrefixEnd()
 	kvs, err := r.DB.Scan(ctx, startKey, endKey, 0)
 	if err != nil {
diff --git a/pkg/spanconfig/spanconfigsqlwatcher/sqlwatcher.go b/pkg/spanconfig/spanconfigsqlwatcher/sqlwatcher.go
@@ -195,6 +195,15 @@ func (s *SQLWatcher) watchForDescriptorUpdates(
 			// Event for a tombstone on a tombstone -- nothing for us to do here.
 			return
 		}
+		// Skip over any modifications to the descriptor update tracking key, this
+		// is transaction information for the lease manager only.
+		if isUpdateKey, err := s.codec.DecodeDescUpdateKey(ev.Key); isUpdateKey || err != nil {
+			if err != nil {
+				log.Dev.Warningf(ctx, "failed to decode descriptor update key: %v", err)
+			}
+			return
+		}
+
 		value := ev.Value
 		if !ev.Value.IsPresent() {
 			// The descriptor was deleted.
diff --git a/pkg/sql/catalog/catalogkeys/keys.go b/pkg/sql/catalog/catalogkeys/keys.go
@@ -287,6 +287,11 @@ func MakeDescMetadataKey(codec keys.SQLCodec, descID descpb.ID) roachpb.Key {
 	return codec.DescMetadataKey(uint32(descID))
 }
 
+// MakeDescUpdateKey returns the key for the descriptor.
+func MakeDescUpdateKey(codec keys.SQLCodec, id descpb.ID) roachpb.Key {
+	return codec.DescMetadataUpdateKey(uint32(id))
+}
+
 // CommentsMetadataPrefix returns the key prefix for all comments in the
 // system.comments table.
 func CommentsMetadataPrefix(codec keys.SQLCodec) roachpb.Key {
diff --git a/pkg/sql/catalog/descpb/structured.proto b/pkg/sql/catalog/descpb/structured.proto
@@ -1937,3 +1937,14 @@ message Descriptor {
     FunctionDescriptor function = 5;
   }
 }
+
+// DescriptorUpdates tracks updates to descriptors within a transaction, which
+// will be written to a special key.
+message DescriptorUpdates {
+  option (gogoproto.equal) = true;
+  // Needed for the descriptorProto interface.
+  option (gogoproto.goproto_getters) = true;
+
+  repeated uint32 descriptor_ids = 1 [(gogoproto.customname) = "DescriptorIDs", (gogoproto.casttype) = "ID"];
+  repeated uint32 descriptor_versions = 2  [(gogoproto.customname) = "DescriptorVersions", (gogoproto.casttype) = "DescriptorVersion"];
+}
diff --git a/pkg/sql/catalog/descs/collection.go b/pkg/sql/catalog/descs/collection.go
@@ -436,6 +436,36 @@ func (tc *Collection) maybeMarkVersionBumpOnly(
 			isVersionBumpOnly
 }
 
+// EmitDescriptorUpdatesKey writes a special key that helps the lease manager
+// know which descriptors were updated by this transaction. This allows it to
+// transactionally make these descriptors available.
+func (tc *Collection) EmitDescriptorUpdatesKey(ctx context.Context, txn *kv.Txn) error {
+	// This key is only emitted for 25.4 and above. Additionally, PCR
+	// reader catalog can update a large number of descriptors in a transaction,
+	// and are exempt from this logic.
+	if !tc.settings.Version.IsActive(ctx, clusterversion.V25_4) ||
+		tc.readerCatalogSetup ||
+		tc.uncommitted.uncommitted.Len() == 0 ||
+		!lease.LockedLeaseTimestamp.Get(&tc.settings.SV) {
+		return nil
+	}
+	updates := &descpb.DescriptorUpdates{}
+	descUpdateID := descpb.InvalidID
+	// Add all the descriptors that have been modified in this transaction.
+	if err := tc.uncommitted.iterateUncommittedByID(func(desc catalog.Descriptor) error {
+		updates.DescriptorIDs = append(updates.DescriptorIDs, desc.GetID())
+		updates.DescriptorVersions = append(updates.DescriptorVersions, desc.GetVersion())
+		if descUpdateID < desc.GetID() {
+			descUpdateID = desc.GetID()
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+	descUpdateKey := catalogkeys.MakeDescUpdateKey(tc.codec(), descUpdateID)
+	return txn.Put(ctx, descUpdateKey, updates)
+}
+
 // WriteDescToBatch calls MaybeIncrementVersion, adds the descriptor to the
 // collection as an uncommitted descriptor, and writes it into b.
 func (tc *Collection) WriteDescToBatch(
diff --git a/pkg/sql/catalog/lease/BUILD.bazel b/pkg/sql/catalog/lease/BUILD.bazel
@@ -69,6 +69,7 @@ go_library(
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_logtags//:logtags",
         "@com_github_cockroachdb_redact//:redact",
+        "@com_github_google_btree//:btree",
     ],
 )
 
diff --git a/pkg/sql/catalog/lease/lease.go b/pkg/sql/catalog/lease/lease.go
diff --git a/pkg/sql/catalog/lease/lease_internal_test.go b/pkg/sql/catalog/lease/lease_internal_test.go
diff --git a/pkg/sql/catalog/lease/testutils.go b/pkg/sql/catalog/lease/testutils.go
diff --git a/pkg/sql/conn_executor_exec.go b/pkg/sql/conn_executor_exec.go
diff --git a/pkg/sql/internal.go b/pkg/sql/internal.go
diff --git a/pkg/sql/vecindex/manager_test.go b/pkg/sql/vecindex/manager_test.go

Original file line number	Diff line number	Diff line change
`@@ -831,7 +831,7 @@ func (tf *schemaFeed) fetchDescriptorVersions(`
`831`	`831`	`}`
`832`	`832`	`codec := tf.leaseMgr.Codec()`
`833`	`833`	`start := crtime.NowMono()`
`834`		`- span := roachpb.Span{Key: codec.TablePrefix(keys.DescriptorTableID)}`
	`834`	`+ span := roachpb.Span{Key: codec.IndexPrefix(keys.DescriptorTableID, keys.DescriptorTablePrimaryKeyIndexID)}`
`835`	`835`	`span.EndKey = span.Key.PrefixEnd()`
`836`	`836`
`837`	`837`	`tf.mu.Lock()`