backupccl,importccl: only reset ModificationTime on tables after upgrade

In cockroachdb#54296 we began resetting the version and modification time for all newly
added descriptors. This is fine for the descriptors which did not exist or
have those fields in 20.1 (schema, type, db). For tables however, this is not
okay because code in release 20.1 is missing a check added in the 20.2 cycle
to allow a missing modification time for version 1. This breaks restore and
import in the mixed version state.

This PR gates the resetting of modification time on the cluster version. I
have verified that this fixes the jobs/mixed-version roachtest which previously
failed deterministically due to this bug.

No release note because we haven't shipped this bug.

Fixes cockroachdb#54319.

Release note: None

pkg/ccl/backupccl/restore_job.go

@@ -685,6 +685,15 @@ type restoreResumer struct {
 	settings *cluster.Settings
 	execCfg  *sql.ExecutorConfig
 
+	// canResetTableModificationTime is true if the cluster version is new enough.
+	// In release-20.1, any decoded table descriptor needed a populated
+	// modification time regardless of its version number. In 20.2 and later we've
+	// relaxed this requirement to allow version 1 descriptors to be serialized
+	// with a zero modification time.
+	//
+	// TODO(ajwerner): Remove in 21.1.
+	canResetTableModificationTime bool
+
 	testingKnobs struct {
 		// beforePublishingDescriptors is called right before publishing
 		// descriptors, after any data has been restored.
@@ -850,7 +859,10 @@ func createImportingDescriptors(
 
 	// Assign new IDs and privileges to the tables, and update all references to
 	// use the new IDs.
-	if err := RewriteTableDescs(mutableTables, details.DescriptorRewrites, details.OverrideDB); err != nil {
+	if err := RewriteTableDescs(
+		mutableTables, details.DescriptorRewrites, details.OverrideDB,
+		r.canResetTableModificationTime,
+	); err != nil {
 		return nil, nil, nil, err
 	}
 	tableDescs := make([]*descpb.TableDescriptor, len(mutableTables))
@@ -1052,6 +1064,8 @@ func (r *restoreResumer) Resume(
 ) error {
 	details := r.job.Details().(jobspb.RestoreDetails)
 	p := phs.(sql.PlanHookState)
+	r.canResetTableModificationTime = p.ExecCfg().Settings.Version.IsActive(
+		ctx, clusterversion.VersionLeasedDatabaseDescriptors)
 
 	backupManifests, latestBackupManifest, sqlDescs, err := loadBackupSQLDescs(
 		ctx, p, details, details.Encryption,

pkg/ccl/backupccl/restore_planning.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/storageccl"
 	"github.com/cockroachdb/cockroach/pkg/ccl/utilccl"
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
@@ -835,9 +836,17 @@ func maybeRewriteSchemaID(curSchemaID descpb.ID, descriptorRewrites DescRewriteM
 
 // RewriteTableDescs mutates tables to match the ID and privilege specified
 // in descriptorRewrites, as well as adjusting cross-table references to use the
-// new IDs. overrideDB can be specified to set database names in views.
+// new IDs. overrideDB can be specified to set database names in views. The
+// canResetModTime parameter is set based on the cluster version. It is unsafe
+// to reset the mod time in a mixed version state as 20.1 nodes expect the mod
+// time to be set on all descriptors which are deserialized, even at version 1.
+//
+// TODO(ajwerner): Remove canResetModTime in 21.1.
 func RewriteTableDescs(
-	tables []*tabledesc.Mutable, descriptorRewrites DescRewriteMap, overrideDB string,
+	tables []*tabledesc.Mutable,
+	descriptorRewrites DescRewriteMap,
+	overrideDB string,
+	canResetModTime bool,
 ) error {
 	for _, table := range tables {
 		tableRewrite, ok := descriptorRewrites[table.ID]
@@ -846,7 +855,9 @@ func RewriteTableDescs(
 		}
 		// Reset the version and modification time on this new descriptor.
 		table.Version = 1
-		table.ModificationTime = hlc.Timestamp{}
+		if canResetModTime {
+			table.ModificationTime = hlc.Timestamp{}
+		}
 
 		if table.IsView() && overrideDB != "" {
 			// restore checks that all dependencies are also being restored, but if
@@ -1496,7 +1507,13 @@ func doRestorePlan(
 
 	// We attempt to rewrite ID's in the collected type and table descriptors
 	// to catch errors during this process here, rather than in the job itself.
-	if err := RewriteTableDescs(tables, descriptorRewrites, intoDB); err != nil {
+	//
+	// TODO(ajwerner): Remove this version check in 21.1.
+	canResetModTime := p.ExecCfg().Settings.Version.IsActive(
+		ctx, clusterversion.VersionLeasedDatabaseDescriptors)
+	if err := RewriteTableDescs(
+		tables, descriptorRewrites, intoDB, canResetModTime,
+	); err != nil {
 		return err
 	}
 	if err := rewriteDatabaseDescs(databases, descriptorRewrites); err != nil {

pkg/ccl/importccl/import_stmt.go

@@ -976,8 +976,12 @@ func prepareNewTableDescsForIngestion(
 		}
 		seqVals[id] = tableDesc.SeqVal
 	}
-
-	if err := backupccl.RewriteTableDescs(newMutableTableDescriptors, tableRewrites, ""); err != nil {
+	// TODO(ajwerner): Remove this in 21.1.
+	canResetModTime := p.ExecCfg().Settings.Version.IsActive(
+		ctx, clusterversion.VersionLeasedDatabaseDescriptors)
+	if err := backupccl.RewriteTableDescs(
+		newMutableTableDescriptors, tableRewrites, "", canResetModTime,
+	); err != nil {
 		return nil, err
 	}