roachtest: compact backups in backup restore roundtrip

This patch enables the conventional restore roundtrip tests to run compacted
backups. Once the limit on inc layers to online restore from is lifted, we'll
enable compacted backups to run on online restore as well.

Epic: none

Release note: none

Author: msbutler

pkg/cmd/roachtest/tests/backup_restore_roundtrip.go

@@ -122,7 +122,8 @@ func backupRestoreRoundTrip(
 
 			return conn, err
 		}
-		testUtils, err := newCommonTestUtils(ctx, t, c, connectFunc, c.CRDBNodes(), withMock(sp.mock), withOnlineRestore(sp.onlineRestore))
+		// TODO (msbutler): enable compaction for online restore test once inc layer limit is increased.
+		testUtils, err := newCommonTestUtils(ctx, t, c, connectFunc, c.CRDBNodes(), withMock(sp.mock), withOnlineRestore(sp.onlineRestore), withCompaction(!sp.onlineRestore))
 		if err != nil {
 			return err
 		}

pkg/cmd/roachtest/tests/mixed_version_backup.go

@@ -179,7 +179,7 @@ var (
 	}
 
 	possibleNumIncrementalBackups = []int{
-		1,
+		2,
 		3,
 	}
 
@@ -410,6 +410,14 @@ type (
 		incNum     int
 	}
 
+	compactedBackup struct {
+		collection backupCollection
+		startTime  string
+		endTime    string
+
+		fullSubdir string
+	}
+
 	// labeledNodes allows us to label a set of nodes with the version
 	// they are running, to allow for human-readable backup names
 	labeledNodes struct {
@@ -456,6 +464,7 @@ func tableNamesWithDB(db string, tables []string) []string {
 
 func (fb fullBackup) String() string        { return "full" }
 func (ib incrementalBackup) String() string { return "incremental" }
+func (cb compactedBackup) String() string   { return "compacted" }
 
 func (rh revisionHistory) String() string {
 	return "revision_history"
@@ -1866,6 +1875,14 @@ func (d *BackupRestoreTestDriver) runBackup(
 		collection = b.collection
 		latest = " LATEST IN"
 		l.Printf("creating incremental backup num %d for %s", b.incNum, collection.name)
+	case compactedBackup:
+		collection = b.collection
+
+		// This latest var is only required to comply with this driver. It has no
+		// effect on compacted backup construction, as the full subdir passed to the
+		// cmd determines the full we compact on.
+		latest = " LATEST IN"
+		l.Printf("creating compacted backup for %s from start %s to end %s", collection.name, b.startTime, b.endTime)
 	}
 
 	for _, opt := range collection.options {
@@ -1883,12 +1900,24 @@ func (d *BackupRestoreTestDriver) runBackup(
 		backupTime,
 		strings.Join(options, ", "),
 	)
-	l.Printf("creating %s backup via node %d: %s", bType, node, stmt)
+
 	var jobID int
-	if err := db.QueryRowContext(ctx, stmt).Scan(&jobID); err != nil {
-		return backupCollection{}, "", fmt.Errorf("error while creating %s backup %s: %w", bType, collection.name, err)
+	if compactData, ok := bType.(compactedBackup); ok {
+		backupTime = compactData.endTime
+		if err := db.QueryRowContext(ctx,
+			`SELECT crdb_internal.backup_compaction(
+				$1, 
+				$2,
+				$3::DECIMAL, $4::DECIMAL
+			)`, stmt, compactData.fullSubdir, compactData.startTime, compactData.endTime).Scan(&jobID); err != nil {
+			return backupCollection{}, "", fmt.Errorf("error while creating %s compacted backup %s: %w", bType, collection.name, err)
+		}
+	} else {
+		l.Printf("creating %s backup via node %d: %s", bType, node, stmt)
+		if err := db.QueryRowContext(ctx, stmt).Scan(&jobID); err != nil {
+			return backupCollection{}, "", fmt.Errorf("error while creating %s backup %s: %w", bType, collection.name, err)
+		}
 	}
-
 	backupErr := make(chan error)
 	tasker.Go(func(ctx context.Context, l *logger.Logger) error {
 		defer close(backupErr)
@@ -2024,6 +2053,7 @@ func (d *BackupRestoreTestDriver) createBackupCollection(
 	isMultitenant bool,
 ) (*backupCollection, error) {
 	var collection backupCollection
+	backupEndTimes := make([]string, 0)
 	var latestIncBackupEndTime string
 	var fullBackupEndTime string
 
@@ -2038,14 +2068,12 @@ func (d *BackupRestoreTestDriver) createBackupCollection(
 	}); err != nil {
 		return nil, err
 	}
+	backupEndTimes = append(backupEndTimes, fullBackupEndTime)
 
 	// Create incremental backups.
 	numIncrementals := possibleNumIncrementalBackups[rng.Intn(len(possibleNumIncrementalBackups))]
 	if d.testUtils.mock {
-		numIncrementals = 1
-	}
-	if d.testUtils.onlineRestore {
-		numIncrementals = 0
+		numIncrementals = 2
 	}
 	l.Printf("creating %d incremental backups", numIncrementals)
 	for i := 0; i < numIncrementals; i++ {
@@ -2060,6 +2088,36 @@ func (d *BackupRestoreTestDriver) createBackupCollection(
 		}); err != nil {
 			return nil, err
 		}
+		backupEndTimes = append(backupEndTimes, latestIncBackupEndTime)
+
+		if d.testUtils.compactionEnabled && !collection.withRevisionHistory() && len(backupEndTimes) >= 3 {
+			// Require that endIdx - startIdx >= 2 so at least 2 inc backups are
+			// compacted. If there are 3 backupEndTimes, the start must be the the
+			// 0th. Thn endIdx is always the last index for now, so the compacted
+			// backup gets selected to restore.
+			startIdx := rng.Intn(len(backupEndTimes) - 2)
+			endIdx := len(backupEndTimes) - 1
+
+			var fullPath string
+			_, db := d.testUtils.RandomDB(rng, d.roachNodes)
+			row := db.QueryRowContext(ctx, fmt.Sprintf(`SELECT path 
+        FROM [SHOW BACKUPS IN '%s']
+        ORDER BY path DESC
+        LIMIT 1`, collection.uri()))
+			if err := row.Scan(&fullPath); err != nil {
+				return nil, errors.Wrapf(err, "error while getting full backup path %s", collection.name)
+			}
+			compact := compactedBackup{collection: collection, startTime: backupEndTimes[startIdx], endTime: backupEndTimes[endIdx], fullSubdir: fullPath}
+			if err := d.testUtils.runJobOnOneOf(ctx, l, incBackupSpec.Execute.Nodes, func() error {
+				var err error
+				collection, latestIncBackupEndTime, err = d.runBackup(
+					ctx, l, tasker, rng, incBackupSpec.Plan.Nodes, incBackupSpec.PauseProbability,
+					compact, internalSystemJobs, isMultitenant)
+				return err
+			}); err != nil {
+				return nil, err
+			}
+		}
 	}
 
 	if err := collection.maybeUseRestoreAOST(l, rng, fullBackupEndTime, latestIncBackupEndTime); err != nil {
@@ -2832,11 +2890,12 @@ func prepSchemaChangeWorkload(
 }
 
 type CommonTestUtils struct {
-	t             test.Test
-	cluster       cluster.Cluster
-	roachNodes    option.NodeListOption
-	mock          bool
-	onlineRestore bool
+	t                 test.Test
+	cluster           cluster.Cluster
+	roachNodes        option.NodeListOption
+	mock              bool
+	onlineRestore     bool
+	compactionEnabled bool
 
 	connCache struct {
 		mu    syncutil.Mutex
@@ -2858,6 +2917,12 @@ func withOnlineRestore(or bool) commonTestOption {
 	}
 }
 
+func withCompaction(c bool) commonTestOption {
+	return func(cu *CommonTestUtils) {
+		cu.compactionEnabled = c
+	}
+}
+
 // Change the function signature
 func newCommonTestUtils(
 	ctx context.Context,