Merge #144011

144011: roachtest: compact backups in backup restore roundtrip r=kev-cao a=msbutler

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

Co-authored-by: Michael Butler <[email protected]>

Author: craig[bot]

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(), sp.mock, 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,29 +2890,47 @@ 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
 		cache []*gosql.DB
 	}
 }
 
-// newCommonTestUtils creates a connection to each node (given that the nodes list is not empty)
-// and puts these connections in a cache for reuse. The caller should remember to close all connections
-// once done with them to prevent any goroutine leaks (CloseConnections).
+type commonTestOption func(*CommonTestUtils)
+
+func withMock(mock bool) commonTestOption {
+	return func(c *CommonTestUtils) {
+		c.mock = mock
+	}
+}
+
+func withOnlineRestore(or bool) commonTestOption {
+	return func(c *CommonTestUtils) {
+		c.onlineRestore = or
+	}
+}
+
+func withCompaction(c bool) commonTestOption {
+	return func(cu *CommonTestUtils) {
+		cu.compactionEnabled = c
+	}
+}
+
+// Change the function signature
 func newCommonTestUtils(
 	ctx context.Context,
 	t test.Test,
 	c cluster.Cluster,
 	connectFunc func(int) (*gosql.DB, error),
 	nodes option.NodeListOption,
-	mock bool,
-	onlineRestore bool,
+	opts ...commonTestOption,
 ) (*CommonTestUtils, error) {
 	cc := make([]*gosql.DB, len(nodes))
 	for _, node := range nodes {
@@ -2870,13 +2946,16 @@ func newCommonTestUtils(
 	}
 
 	u := &CommonTestUtils{
-		t:             t,
-		cluster:       c,
-		roachNodes:    nodes,
-		mock:          mock,
-		onlineRestore: onlineRestore,
+		t:          t,
+		cluster:    c,
+		roachNodes: nodes,
 	}
 	u.connCache.cache = cc
+
+	// Apply all options
+	for _, opt := range opts {
+		opt(u)
+	}
 	return u, nil
 }
 
@@ -2887,7 +2966,7 @@ func (mvb *mixedVersionBackup) CommonTestUtils(
 	mvb.utilsOnce.Do(func() {
 		connectFunc := func(node int) (*gosql.DB, error) { return h.Connect(node), nil }
 		mvb.commonTestUtils, err = newCommonTestUtils(
-			ctx, mvb.t, mvb.cluster, connectFunc, mvb.roachNodes, false, false,
+			ctx, mvb.t, mvb.cluster, connectFunc, mvb.roachNodes,
 		)
 	})
 	return mvb.commonTestUtils, err

pkg/cmd/roachtest/tests/mixed_version_job_compatibility_in_declarative_schema_changer.go

@@ -86,7 +86,7 @@ func executeSupportedDDLs(
 	// here because these connnections are managed by the mixedversion
 	// framework, which already closes them at the end of the test.
 	testUtils, err := newCommonTestUtils(
-		ctx, t, c, connectFunc, helper.DefaultService().Descriptor.Nodes, false, false,
+		ctx, t, c, connectFunc, helper.DefaultService().Descriptor.Nodes,
 	)
 	if err != nil {
 		return err