backupccl: fix codec selection during restore

Previously, the codec selection logic only looked at a
single manifest to determine whether the backup data requires
a system or tenant codec to be read. In older release branches
if the full backup was empty (but a subsequent incremental
was not) then we could incorrectly use a system codec for
a backup that actually required a tenant codec.

To fix this we now process all the manifests in the backup
chain and use the first non-empty manifest.

Fixes: #115773
Release note(bug fix): an empty full backup, followed by
non-empty incrementals taken inside an application tenant
may be unrestoreable because of the use of an incorrect
SQL codec

pkg/ccl/backupccl/backup_test.go

@@ -3444,7 +3444,7 @@ func TestRestoreAsOfSystemTime(t *testing.T) {
 
 	const numAccounts = 10
 	ctx := context.Background()
-	tc, sqlDB, _, cleanupFn := backupRestoreTestSetup(t, singleNode, numAccounts, InitManualReplication)
+	_, sqlDB, _, cleanupFn := backupRestoreTestSetup(t, singleNode, numAccounts, InitManualReplication)
 	defer cleanupFn()
 	const dir = "nodelocal://1/"
 
@@ -3691,10 +3691,6 @@ func TestRestoreAsOfSystemTime(t *testing.T) {
 	t.Run("backup-create-drop-backup", func(t *testing.T) {
 		var tsBefore string
 		backupPath := "nodelocal://1/create_and_drop"
-		if !tc.ApplicationLayer(0).Codec().ForSystemTenant() {
-			// TODO(adityamaru): Unskip once #115640 is resolved.
-			skip.WithIssue(t, 115640)
-		}
 		sqlDB.Exec(t, "CREATE DATABASE create_and_drop")
 		sqlDB.Exec(t, "CREATE DATABASE create_and_drop_restore")
 		sqlDB.Exec(t, `BACKUP DATABASE create_and_drop INTO $1 WITH revision_history`, backupPath)
@@ -3735,6 +3731,55 @@ func TestRestoreAsOfSystemTime(t *testing.T) {
 	})
 }
 
+func TestEmptyBackupsInChain(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	const numAccounts = 10
+	_, sqlDB, _, cleanupFn := backupRestoreTestSetupWithParams(t, singleNode, numAccounts,
+		InitManualReplication, base.TestClusterArgs{
+			ServerArgs: base.TestServerArgs{
+				DefaultTestTenant: base.TestTenantAlwaysEnabled,
+			},
+		})
+	defer cleanupFn()
+	var tsBefore string
+	backupPath := "nodelocal://1/create_and_drop"
+
+	t.Run("empty-full-non-empty-inc", func(t *testing.T) {
+		sqlDB.Exec(t, "CREATE DATABASE create_and_drop")
+		sqlDB.Exec(t, "CREATE DATABASE create_and_drop_restore")
+		sqlDB.Exec(t, `BACKUP DATABASE create_and_drop INTO $1 WITH revision_history`, backupPath)
+		sqlDB.Exec(t, "CREATE TABLE create_and_drop.a (k int, v string)")
+		sqlDB.Exec(t, "INSERT INTO create_and_drop.a VALUES (1, 'foo')")
+		sqlDB.QueryRow(t, "SELECT cluster_logical_timestamp()").Scan(&tsBefore)
+		sqlDB.Exec(t, "DROP TABLE create_and_drop.a")
+		sqlDB.Exec(t, `BACKUP DATABASE create_and_drop INTO LATEST IN $1 WITH revision_history`, backupPath)
+		restoreQuery := fmt.Sprintf(
+			"RESTORE create_and_drop.* FROM LATEST IN $1 AS OF SYSTEM TIME %s WITH into_db='create_and_drop_restore'", tsBefore)
+		sqlDB.Exec(t, restoreQuery, backupPath)
+
+		restoredTableQuery := "SELECT * FROM create_and_drop_restore.a"
+		backedUpTableQuery := fmt.Sprintf("SELECT * FROM create_and_drop.a AS OF SYSTEM TIME %s", tsBefore)
+		sqlDB.CheckQueryResults(t, backedUpTableQuery, sqlDB.QueryStr(t, restoredTableQuery))
+	})
+
+	t.Run("non-empty-full-empty-inc", func(t *testing.T) {
+		sqlDB.Exec(t, "CREATE DATABASE create_and_drop2")
+		sqlDB.Exec(t, "CREATE DATABASE create_and_drop_restore2")
+		sqlDB.Exec(t, "CREATE TABLE create_and_drop2.a (k int, v string)")
+		sqlDB.Exec(t, "INSERT INTO create_and_drop2.a VALUES (1, 'foo')")
+		sqlDB.Exec(t, "DROP TABLE create_and_drop2.a")
+		sqlDB.Exec(t, `BACKUP DATABASE create_and_drop2 INTO $1 WITH revision_history`, backupPath)
+		sqlDB.QueryRow(t, "SELECT cluster_logical_timestamp()").Scan(&tsBefore)
+		sqlDB.Exec(t, `BACKUP DATABASE create_and_drop2 INTO LATEST IN $1 WITH revision_history`, backupPath)
+		restoreQuery := fmt.Sprintf(
+			"RESTORE create_and_drop2.* FROM LATEST IN $1 AS OF SYSTEM TIME %s WITH into_db='create_and_drop_restore2'", tsBefore)
+		sqlDB.Exec(t, restoreQuery, backupPath)
+		sqlDB.CheckQueryResults(t, `SELECT table_name FROM [SHOW TABLES]`, [][]string{{"bank"}})
+	})
+}
+
 func TestRestoreAsOfSystemTimeGCBounds(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/ccl/backupccl/backupinfo/BUILD.bazel

@@ -74,6 +74,7 @@ go_test(
         "//pkg/ccl/backupccl/backuppb",
         "//pkg/ccl/backupccl/backuptestutils",
         "//pkg/cloud",
+        "//pkg/keys",
         "//pkg/multitenant/mtinfopb",
         "//pkg/roachpb",
         "//pkg/security/securityassets",

pkg/ccl/backupccl/backupinfo/manifest_handling.go

@@ -1557,17 +1557,26 @@ func GetBackupManifests(
 	return manifests, memMu.total, nil
 }
 
-// MakeBackupCodec returns the codec that was used to encode the keys in the backup.
-func MakeBackupCodec(manifest backuppb.BackupManifest) (keys.SQLCodec, error) {
+// MakeBackupCodec returns the codec that was used to encode the keys in the
+// backup. We iterate over all the provided manifests and use the first
+// non-empty manifest to determine the codec. If all manifests are empty we
+// default to the system codec.
+func MakeBackupCodec(manifests []backuppb.BackupManifest) (keys.SQLCodec, error) {
 	backupCodec := keys.SystemSQLCodec
-	if len(manifest.Spans) != 0 && !manifest.HasTenants() {
-		// If there are no tenant targets, then the entire keyspace covered by
-		// Spans must lie in 1 tenant.
-		_, backupTenantID, err := keys.DecodeTenantPrefix(manifest.Spans[0].Key)
-		if err != nil {
-			return backupCodec, err
+	for _, manifest := range manifests {
+		if len(manifest.Spans) == 0 {
+			continue
+		}
+
+		if !manifest.HasTenants() {
+			// If there are no tenant targets, then the entire keyspace covered by
+			// Spans must lie in 1 tenant.
+			_, backupTenantID, err := keys.DecodeTenantPrefix(manifest.Spans[0].Key)
+			if err != nil {
+				return backupCodec, err
+			}
+			backupCodec = keys.MakeSQLCodec(backupTenantID)
 		}
-		backupCodec = keys.MakeSQLCodec(backupTenantID)
 	}
 	return backupCodec, nil
 }

pkg/ccl/backupccl/backupinfo/manifest_handling_test.go

@@ -19,6 +19,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/backupccl/backupinfo"
 	"github.com/cockroachdb/cockroach/pkg/ccl/backupccl/backuppb"
 	"github.com/cockroachdb/cockroach/pkg/cloud"
+	"github.com/cockroachdb/cockroach/pkg/keys"
+	"github.com/cockroachdb/cockroach/pkg/multitenant/mtinfopb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
@@ -292,3 +294,86 @@ func mustCreateFileIterFactory(
 		return it
 	}
 }
+
+func TestMakeBackupCodec(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	tenID, err := roachpb.MakeTenantID(10)
+	require.NoError(t, err)
+	tenSpan := keys.MakeTenantSpan(tenID)
+	for _, tc := range []struct {
+		name          string
+		manifests     []backuppb.BackupManifest
+		expectedCodec keys.SQLCodec
+	}{
+		{
+			name: "full",
+			manifests: []backuppb.BackupManifest{
+				{Spans: []roachpb.Span{{Key: roachpb.Key("/Table/123")}}},
+			},
+			expectedCodec: keys.SystemSQLCodec,
+		},
+		{
+			name: "full-backup-tenant",
+			manifests: []backuppb.BackupManifest{
+				{Spans: []roachpb.Span{tenSpan}},
+			},
+			expectedCodec: keys.MakeSQLCodec(tenID),
+		},
+		{
+			name: "full-backup-of-tenant",
+			manifests: []backuppb.BackupManifest{
+				{
+					Spans:   []roachpb.Span{tenSpan},
+					Tenants: []mtinfopb.TenantInfoWithUsage{{SQLInfo: mtinfopb.SQLInfo{ID: 10}}},
+				},
+			},
+			expectedCodec: keys.SystemSQLCodec,
+		},
+		{
+			name: "empty-full-backup",
+			manifests: []backuppb.BackupManifest{
+				{Spans: []roachpb.Span{}},
+				{Spans: []roachpb.Span{{Key: roachpb.Key("/Table/123")}}},
+			},
+			expectedCodec: keys.SystemSQLCodec,
+		},
+		{
+			name: "empty-full-backup-tenant",
+			manifests: []backuppb.BackupManifest{
+				{Spans: []roachpb.Span{}},
+				{Spans: []roachpb.Span{tenSpan}},
+			},
+			expectedCodec: keys.MakeSQLCodec(tenID),
+		},
+		{
+			name: "empty-full-backup-of-tenant",
+			manifests: []backuppb.BackupManifest{
+				{
+					Spans:   []roachpb.Span{},
+					Tenants: []mtinfopb.TenantInfoWithUsage{{SQLInfo: mtinfopb.SQLInfo{ID: 10}}},
+				},
+				{
+					Spans:   []roachpb.Span{tenSpan},
+					Tenants: []mtinfopb.TenantInfoWithUsage{{SQLInfo: mtinfopb.SQLInfo{ID: 10}}},
+				},
+			},
+			expectedCodec: keys.SystemSQLCodec,
+		},
+		{
+			name: "all-empty",
+			manifests: []backuppb.BackupManifest{
+				{Spans: []roachpb.Span{{}}},
+				{Spans: []roachpb.Span{{}}},
+				{Spans: []roachpb.Span{{}}},
+			},
+			expectedCodec: keys.SystemSQLCodec,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			c, err := backupinfo.MakeBackupCodec(tc.manifests)
+			require.NoError(t, err)
+			require.Equal(t, tc.expectedCodec, c)
+		})
+	}
+}

pkg/ccl/backupccl/restore_job.go

@@ -1644,7 +1644,7 @@ func (r *restoreResumer) doResume(ctx context.Context, execCtx interface{}) erro
 	defer func() {
 		mem.Shrink(ctx, memSize)
 	}()
-	backupCodec, err := backupinfo.MakeBackupCodec(latestBackupManifest)
+	backupCodec, err := backupinfo.MakeBackupCodec(backupManifests)
 	if err != nil {
 		return err
 	}