Merge #153961 #154866 #154977

craig[bot] · fqazi · msbutler · craig[bot] · commit 67f33c276e39 · 2025-10-07T21:00:54.000Z
153961: catalog/lease: update locked timestamps when updates are received r=fqazi a=fqazi 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 Fixes: #154075 Release note: None 154866: roachtest: deflake c2c/disconnect r=jeffswenson a=msbutler Previously, the c2c/disconnect roachtest was disconnecting two source nodes, instead of a pair of source/destination nodes. This patch fixes this bug. Informs: #152248 Informs: #150474 Release note: none 154977: roachtest: deflake backup/mvcc-range-tombstones r=kev-cao a=msbutler Informs #154841 Release note: none Co-authored-by: Faizan Qazi <faizan@cockroachlabs.com> Co-authored-by: Michael Butler <butler@cockroachlabs.com>
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/cmd/roachtest/tests/backup.go b/pkg/cmd/roachtest/tests/backup.go
@@ -756,7 +756,7 @@ func runBackupMVCCRangeTombstones(
 			`IMPORT INTO orders CSV DATA ('%s') WITH delimiter='|', detached`,
 			strings.Join(files, "', '")),
 		).Scan(&jobID))
-		waitForState(jobID, jobs.StatePaused, "", 30*time.Minute)
+		waitForState(jobID, jobs.StatePaused, "", time.Hour)
 
 		t.Status("canceling import")
 		_, err = conn.ExecContext(ctx, fmt.Sprintf(`CANCEL JOB %s`, jobID))
diff --git a/pkg/cmd/roachtest/tests/cluster_to_cluster.go b/pkg/cmd/roachtest/tests/cluster_to_cluster.go
@@ -2057,17 +2057,19 @@ func registerClusterReplicationDisconnect(r registry.Registry) {
 		var dstNode int
 		srcTenantSQL.QueryRow(t, `select split_part(consumer, '[', 1) from crdb_internal.cluster_replication_node_streams order by random() limit 1`).Scan(&dstNode)
 
+		roachprodDstNode := dstNode + sp.srcNodes
+
 		disconnectDuration := sp.additionalDuration
 		rd.t.L().Printf("Disconnecting Src %d, Dest %d for %.2f minutes", srcNode,
-			dstNode, disconnectDuration.Minutes())
+			roachprodDstNode, disconnectDuration.Minutes())
 
 		// Normally, the blackholeFailer is accessed through the failer interface,
 		// at least in the failover tests. Because this test shouldn't use all the
 		// failer interface calls (e.g. Setup(), and Ready()), we use the
 		// blakholeFailer struct directly. In other words, in this test, we
 		// shouldn't treat the blackholeFailer as an abstracted api.
 		blackholeFailer := &blackholeFailer{t: rd.t, c: rd.c, input: true, output: true}
-		blackholeFailer.FailPartial(ctx, srcNode, []int{dstNode})
+		blackholeFailer.FailPartial(ctx, srcNode, []int{roachprodDstNode})
 
 		time.Sleep(disconnectDuration)
 		// Calling this will log the latest topology.
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()`