backup: restore using mvcc at-now addsstable

Release note (sql change): Restored data now appears to have been written at the time it was restored, rather than the time at which it was backed up, when reading the lower-level write timestamps from the rows themselves. This affects various internal operations and the result of crdb_internal_mvcc_timestamp.

pkg/ccl/backupccl/restore_data_processor.go

@@ -9,12 +9,14 @@
 package backupccl
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/storageccl"
 	"github.com/cockroachdb/cockroach/pkg/cloud"
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
+	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv/bulk"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings"
@@ -105,7 +107,7 @@ var restoreAtNow = settings.RegisterBoolSetting(
 	settings.TenantWritable,
 	"bulkio.restore_at_current_time.enabled",
 	"write restored data at the current timestamp",
-	false,
+	true,
 )
 
 func newRestoreDataProcessor(
@@ -387,6 +389,16 @@ func (rd *restoreDataProcessor) processRestoreSpanEntry(
 		)
 	}
 
+	// If the system tenant is restoring a guest tenant span, we don't want to
+	// forward all the restored data to now, as there may be importing tables in
+	// that span, that depend on the difference in timestamps on restored existing
+	// vs importing keys to rollback.
+	if writeAtBatchTS && kr.fromSystemTenant &&
+		(bytes.HasPrefix(entry.Span.Key, keys.TenantPrefix) || bytes.HasPrefix(entry.Span.EndKey, keys.TenantPrefix)) {
+		log.Warningf(ctx, "restoring span %s at its original timestamps because it is a tenant span", entry.Span)
+		writeAtBatchTS = false
+	}
+
 	// "disallowing" shadowing of anything older than logical=1 is i.e. allow all
 	// shadowing. We must allow shadowing in case the RESTORE has to retry any
 	// ingestions, but setting a (permissive) disallow like this serves to force

pkg/ccl/backupccl/restore_data_processor_test.go

@@ -402,6 +402,11 @@ func runTestIngest(t *testing.T, init func(*cluster.Settings)) {
 				t.Fatalf("%+v", err)
 			}
 			kvs := clientKVsToEngineKVs(clientKVs)
+			for i := range kvs {
+				if i < len(expectedKVs) {
+					expectedKVs[i].Key.Timestamp = kvs[i].Key.Timestamp
+				}
+			}
 
 			if !reflect.DeepEqual(kvs, expectedKVs) {
 				for i := 0; i < len(kvs) || i < len(expectedKVs); i++ {