upgrades: log and record progress for FirstUpgrade job

The Precondition step does not record job progress, as jobs are not used
for the Precondition step. Instead, we just log there.

For the Upgrade itself, a job is used, and we log progress as well as
store it in the job metadata. This is the phase that looks at all
descriptors, so it's more useful to have finer-grained progress here.

This also fixes a mistake in 0cc42a8 (where the upgrade was changed
to rewrite all descriptors), to make sure we actually consider all
descriptors. As part of this, we need to make sure to skip virtual and
syntehtic descriptors now.

Release note: None

Author: rafiss

pkg/sql/catalog/lease/descriptor_set.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/redact"
 )
 
 // descriptorSet maintains an ordered set of descriptorVersionState objects
@@ -43,7 +44,7 @@ func (l *descriptorSet) String() string {
 func (l *descriptorSet) insert(s *descriptorVersionState) {
 	i, match := l.findIndex(s.GetVersion())
 	if match {
-		panic("unable to insert duplicate lease")
+		panic(redact.Sprintf("unable to insert duplicate lease for id=%d", s.GetID()))
 	}
 	if i == len(l.data) {
 		l.data = append(l.data, s)

pkg/upgrade/upgrades/first_upgrade.go

@@ -11,13 +11,15 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
+	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catalogkeys"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/nstree"
+	"github.com/cockroachdb/cockroach/pkg/sql/isql"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/upgrade"
 	"github.com/cockroachdb/cockroach/pkg/util/envutil"
@@ -56,9 +58,29 @@ func FirstUpgradeFromRelease(
 	}
 	var descsToUpdate catalog.DescriptorIDSet
 	if err := all.ForEachDescriptor(func(desc catalog.Descriptor) error {
+		// SKip virtual and synthetic descriptors.
+		switch d := desc.(type) {
+		case catalog.SchemaDescriptor:
+			switch d.SchemaKind() {
+			case catalog.SchemaPublic, catalog.SchemaVirtual, catalog.SchemaTemporary:
+				return nil
+			}
+		case catalog.TableDescriptor:
+			if d.IsVirtualTable() {
+				return nil
+			}
+		}
 		changes := desc.GetPostDeserializationChanges()
 		if !changes.HasChanges() || (changes.Len() == 1 && changes.Contains(catalog.SetModTimeToMVCCTimestamp)) {
-			return nil
+			// In the upgrade to 25.4 (in between 25.3 and 25.4), we do a one-time
+			// rewrite of all descriptors in order to upgrade them to use the new type
+			// serialization format. Skip the unconditional rewrite if this is a
+			// database descriptor, as those never reference types.
+			// See https://github.com/cockroachdb/cockroach/issues/152629.
+			duringUpgradeTo25_4 := d.Settings.Version.IsActive(ctx, clusterversion.V25_3) && !d.Settings.Version.IsActive(ctx, clusterversion.V25_4)
+			if !duringUpgradeTo25_4 || desc.DescriptorType() == catalog.Database {
+				return nil
+			}
 		}
 		descsToUpdate.Add(desc.GetID())
 		return nil
@@ -82,6 +104,10 @@ func upgradeDescriptors(
 	repairBatchTimeLimit := 1 * time.Minute
 	currentIdx := 0
 	idsToRewrite := ids.Ordered()
+	totalDescs := len(idsToRewrite)
+	every := log.Every(time.Minute)
+	log.Dev.Infof(ctx, "upgrading format of %d descriptors for first upgrade from release", totalDescs)
+
 	for currentIdx <= len(idsToRewrite) {
 		descBatch := idsToRewrite[currentIdx:min(currentIdx+batchSize, len(idsToRewrite))]
 		err := timeutil.RunWithTimeout(ctx, "repair-post-deserialization", repairBatchTimeLimit, func(ctx context.Context) error {
@@ -109,11 +135,6 @@ func upgradeDescriptors(
 						if d.Settings.Version.IsActive(ctx, clusterversion.V25_4) {
 							continue
 						}
-						// Skip the unconditional rewrite if this is a database descriptor,
-						// as those never reference types.
-						if mut.DescriptorType() == catalog.Database {
-							continue
-						}
 					}
 					key := catalogkeys.MakeDescMetadataKey(d.Codec, mut.GetID())
 					b.CPut(key, mut.DescriptorProto(), mut.GetRawBytesInStorage())
@@ -135,6 +156,19 @@ func upgradeDescriptors(
 			return err
 		}
 		currentIdx += batchSize
+
+		if every.ShouldLog() {
+			completedDescs := min(currentIdx, totalDescs)
+			log.Dev.Infof(ctx, "upgraded %d of %d descriptors so far", completedDescs, totalDescs)
+			if jobID := d.OptionalJobID; jobID != 0 {
+				frac := float32(completedDescs) / float32(totalDescs)
+				if err := d.DB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+					return jobs.ProgressStorage(jobID).Set(ctx, txn, float64(frac), txn.KV().ReadTimestamp())
+				}); err != nil {
+					log.Dev.Warningf(ctx, "failed to update progress for job %d: %v", jobID, err)
+				}
+			}
+		}
 	}
 	return nil
 }
@@ -171,16 +205,16 @@ func FirstUpgradeFromReleasePrecondition(
 	// there are no corruptions now. Otherwise, we retry and do everything
 	// without an AOST clause henceforth.
 	withAOST := firstUpgradePreconditionUsesAOST
-	diagnose := func(tbl redact.SafeString) (hasRows bool, err error) {
+	diagnose := func(tbl redact.SafeString) (count int, err error) {
 		withAOST := withAOST
 		for {
 			q := fmt.Sprintf("SELECT count(*) FROM \"\".crdb_internal.%s", tbl)
 			if withAOST {
 				q = q + " AS OF SYSTEM TIME '-10s'"
 			}
 			row, err := d.InternalExecutor.QueryRow(ctx, redact.Sprintf("query-%s", tbl), nil /* txn */, q)
-			if err == nil && row[0].String() != "0" {
-				hasRows = true
+			if err == nil {
+				count = int(tree.MustBeDInt(row[0]))
 			}
 			// In tests like "declarative_schema_changer/job-compatibility-mixed-version", its
 			// possible to hit BatchTimestampBeforeGCError, because the GC interval is
@@ -191,24 +225,25 @@ func FirstUpgradeFromReleasePrecondition(
 				withAOST = false
 				continue
 			}
-			return hasRows, err
+			return count, err
 		}
 	}
 	// Check for possibility of time travel.
-	if hasRows, err := diagnose("databases"); err != nil {
+	if count, err := diagnose("databases"); err != nil {
 		return err
-	} else if !hasRows {
+	} else if count == 0 {
 		// We're looking back in time to before the cluster was bootstrapped
 		// and no databases exist at that point. Disable time-travel henceforth.
 		withAOST = false
 	}
 	// Check for repairable catalog corruptions.
-	if hasRows, err := diagnose("kv_repairable_catalog_corruptions"); err != nil {
+	if totalCorruptions, err := diagnose("kv_repairable_catalog_corruptions"); err != nil {
 		return err
-	} else if hasRows {
+	} else if totalCorruptions > 0 {
 		// Attempt to repair catalog corruptions in batches.
-		log.Dev.Info(ctx, "auto-repairing catalog corruptions detected during upgrade attempt")
-		var n int
+		log.Dev.Infof(ctx, "beginning auto-repair of %d catalog corruptions detected during upgrade attempt", totalCorruptions)
+
+		var repairedCount int
 		const repairQuery = `
 SELECT
 	count(*)
@@ -256,7 +291,7 @@ WHERE
 				if kv.IsAutoRetryLimitExhaustedError(err) ||
 					errors.HasType(err, (*timeutil.TimeoutError)(nil)) {
 					batchSize = max(batchSize/2, 1)
-					log.Dev.Infof(ctx, "reducing batch size of invalid_object repair query to %d (hipri=%t)",
+					log.Dev.Infof(ctx, "reducing batch size of repair query to %d (hipri=%t)",
 						batchSize,
 						batchSize <= HighPriBatchSize)
 					continue
@@ -267,21 +302,19 @@ WHERE
 			if rowsUpdated == 0 {
 				break
 			}
-			n += int(rowsUpdated)
+			repairedCount += int(rowsUpdated)
 			log.Dev.Infof(ctx, "repaired %d catalog corruptions", rowsUpdated)
 		}
-		if n == 0 {
-			log.Dev.Info(ctx, "no catalog corruptions found to repair during upgrade attempt")
-		} else {
+		if repairedCount > 0 {
 			// Repairs have actually been performed: stop all time travel henceforth.
 			withAOST = false
-			log.Dev.Infof(ctx, "%d catalog corruptions have been repaired in total", n)
+			log.Dev.Infof(ctx, "%d catalog corruptions have been repaired in total", repairedCount)
 		}
 	}
 	// Check for all known catalog corruptions.
-	if hasRows, err := diagnose("invalid_objects"); err != nil {
+	if invalidObjectCount, err := diagnose("invalid_objects"); err != nil {
 		return err
-	} else if !hasRows {
+	} else if invalidObjectCount == 0 {
 		return nil
 	}
 	if !withAOST {
@@ -291,9 +324,9 @@ WHERE
 	// Re-run the diagnosis without the clause, because we might not be seeing
 	// repairs which might have taken place recently.
 	withAOST = false
-	if hasRows, err := diagnose("invalid_objects"); err != nil {
+	if invalidObjectCount, err := diagnose("invalid_objects"); err != nil {
 		return err
-	} else if !hasRows {
+	} else if invalidObjectCount == 0 {
 		return nil
 	}
 	return errors.AssertionFailedf("\"\".crdb_internal.invalid_objects is not empty")

pkg/upgrade/upgrades/first_upgrade_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/desctestutils"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/funcdesc"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/lease"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/tabledesc"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
@@ -140,8 +141,14 @@ func TestFirstUpgrade(t *testing.T) {
 	// the only post-deserialization change should be SetModTimeToMVCCTimestamp.
 	require.False(t, readDescFromStorage().GetModificationTime().IsEmpty())
 	changes = readDescFromStorage().GetPostDeserializationChanges()
-	require.Equal(t, changes.Len(), 1)
-	require.True(t, changes.Contains(catalog.SetModTimeToMVCCTimestamp))
+	if v1.Equal(clusterversion.V25_4.Version()) {
+		// In 25.4, we do a one-time rewrite of all descriptors, so there should be
+		// no changes here. In later versions, there should be one change.
+		require.Equal(t, 0, changes.Len())
+	} else {
+		require.Equal(t, 1, changes.Len())
+		require.True(t, changes.Contains(catalog.SetModTimeToMVCCTimestamp))
+	}
 }
 
 // TestFirstUpgradeRepair tests the correct repair behavior of upgrade
@@ -163,6 +170,7 @@ func TestFirstUpgradeRepair(t *testing.T) {
 	// also holds a lease on the system database descriptor, which we will wait to
 	// be released. Reducing the lease duration makes this part of the test speed
 	// up.
+	lease.LeaseDuration.Override(ctx, &settings.SV, time.Second*30)
 	require.NoError(t, clusterversion.Initialize(ctx, v0, &settings.SV))
 	upgradePausePoint := make(chan struct{})
 	upgradeResumePoint := make(chan struct{})
@@ -375,6 +383,7 @@ func TestFirstUpgradeRepairBatchSize(t *testing.T) {
 	// also holds a lease on the system database descriptor, which we will wait to
 	// be released. Reducing the lease duration makes this part of the test speed
 	// up.
+	lease.LeaseDuration.Override(ctx, &settings.SV, time.Second*30)
 	require.NoError(t, clusterversion.Initialize(ctx, v0, &settings.SV))
 	testServer, sqlDB, _ := serverutils.StartServer(t, base.TestServerArgs{
 		Settings: settings,