backupccl: check revision start time

Record the revision start time, which may not be the same as the start
time, namely in full backups where start time is 0 but revision history
obviously can only be captured since the GC threshold. The backup
records the latest safe time and then restore checks that requested
times are after that.

Fixes #19455.

Release note (enterprise change): Check that backup actually contains requested restore time.

pkg/ccl/backupccl/backup.go

@@ -688,13 +688,17 @@ func backup(
 				StartTime:  span.start,
 				MVCCFilter: roachpb.MVCCFilter(backupDesc.MVCCFilter),
 			}
-			res, pErr := client.SendWrappedWith(gCtx, db.GetSender(), header, req)
+			rawRes, pErr := client.SendWrappedWith(gCtx, db.GetSender(), header, req)
 			if pErr != nil {
 				return pErr.GoError()
 			}
+			res := rawRes.(*roachpb.ExportResponse)
 
 			mu.Lock()
-			for _, file := range res.(*roachpb.ExportResponse).Files {
+			if backupDesc.RevisionStartTime.Less(res.StartTime) {
+				backupDesc.RevisionStartTime = res.StartTime
+			}
+			for _, file := range res.Files {
 				f := BackupDescriptor_File{
 					Span:        file.Span,
 					Path:        file.Path,

pkg/ccl/backupccl/backup.proto

@@ -54,6 +54,9 @@ message BackupDescriptor {
   util.hlc.Timestamp start_time = 1 [(gogoproto.nullable) = false];
   util.hlc.Timestamp end_time = 2 [(gogoproto.nullable) = false];
   MVCCFilter mvcc_filter = 13 [(gogoproto.customname) = "MVCCFilter"];
+  // Even if StartTime is zero, we only get revisions since gc threshold, so
+  // do not allow AS OF SYSTEM TIME before revision_start_time.
+  util.hlc.Timestamp revision_start_time = 17 [(gogoproto.nullable) = false];
 
   // Spans contains the spans requested for backup. The keyranges covered by
   // `files` may be a subset of this if there were ranges with no changes since

pkg/ccl/backupccl/backup_test.go

@@ -36,12 +36,14 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/ccl/backupccl"
 	"github.com/cockroachdb/cockroach/pkg/ccl/utilccl/sampledataccl"
+	"github.com/cockroachdb/cockroach/pkg/internal/client"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/jobs"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
@@ -1917,6 +1919,46 @@ func TestRestoreAsOfSystemTime(t *testing.T) {
 	})
 }
 
+func TestRestoreAsOfSystemTimeGCBounds(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	const numAccounts = 10
+	ctx, tc, sqlDB, _, cleanupFn := backupRestoreTestSetup(t, singleNode, numAccounts, initNone)
+	defer cleanupFn()
+	const dir = "nodelocal:///"
+	preGC := tree.TimestampToDecimal(tc.Server(0).Clock().Now()).String()
+
+	gcr := roachpb.GCRequest{
+		// Bogus span to make it a valid request.
+		Span: roachpb.Span{
+			Key:    keys.MakeTablePrefix(keys.MaxReservedDescID + 1),
+			EndKey: keys.MaxKey,
+		},
+		Threshold: tc.Server(0).Clock().Now(),
+	}
+	if _, err := client.SendWrapped(ctx, tc.Server(0).DistSender(), &gcr); err != nil {
+		t.Fatal(err)
+	}
+
+	postGC := tree.TimestampToDecimal(tc.Server(0).Clock().Now()).String()
+
+	lateFullTableBackup := filepath.Join(dir, "tbl-after-gc")
+	sqlDB.Exec(t, `BACKUP data.bank TO $1 WITH experimental_revision_history`, lateFullTableBackup)
+	sqlDB.Exec(t, `DROP TABLE data.bank`)
+	if _, err := sqlDB.DB.Exec(
+		fmt.Sprintf(`RESTORE data.bank FROM $1 EXPERIMENTAL AS OF SYSTEM TIME %s`, preGC),
+		lateFullTableBackup,
+	); !testutils.IsError(err, `BACKUP only has revision history from`) {
+		t.Fatal(err)
+	}
+	if _, err := sqlDB.DB.Exec(
+		fmt.Sprintf(`RESTORE data.bank FROM $1 EXPERIMENTAL AS OF SYSTEM TIME %s`, postGC),
+		lateFullTableBackup,
+	); err != nil {
+		t.Fatal(err)
+	}
+}
+
 func TestAsOfSystemTimeOnRestoredData(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 

pkg/ccl/backupccl/restore.go

@@ -1202,9 +1202,24 @@ func doRestorePlan(
 
 	if !endTime.IsEmpty() {
 		for _, b := range backupDescs {
-			if b.StartTime.Less(endTime) && endTime.Less(b.EndTime) && b.MVCCFilter != MVCCFilter_All {
-				return errors.Errorf(
-					"incompatible RESTORE timestamp (BACKUP needs option '%s')", backupOptRevisionHistory)
+			// Find the backup that covers the requested time.
+			if b.StartTime.Less(endTime) && !b.EndTime.Less(endTime) {
+				// Ensure that the backup actually has revision history.
+				if b.MVCCFilter != MVCCFilter_All {
+					return errors.Errorf(
+						"incompatible RESTORE timestamp (BACKUP needs option '%s')", backupOptRevisionHistory,
+					)
+				}
+				// Ensure that the revision history actually covers the requested time -
+				// while the BACKUP's start and end might contain the requested time for
+				// example if start time is 0 (full backup), the revision history was
+				// only captured since the GC window. Note that the RevisionStartTime is
+				// the latest for ranges backed up.
+				if !b.RevisionStartTime.Less(endTime) {
+					return errors.Errorf(
+						"incompatible RESTORE timestamp (BACKUP only has revision history from %v)", b.RevisionStartTime,
+					)
+				}
 			}
 		}
 	}