backup: fail compaction jobs upon encountering range keys

Previously if a compaction job encountered a range key, it would
erroneously continue the job and succeed. As the current implementation
of the backup compaction iterator skips range keys, if the compaction
job encounters an SST file that contains range keys, it should fail the
job.

Epic: None

Release note: Compaction jobs now properly fail upon encountering range
keys.

Author: kev-cao

pkg/backup/compaction_job.go

@@ -534,7 +534,7 @@ func (c compactionChain) createCompactionManifest(
 	// range keys.
 	lastBackup := c.lastBackup()
 	if len(lastBackup.Tenants) != 0 {
-		return nil, errors.New("backup compactions does not yet support range keys")
+		return nil, errors.New("backup compactions does not support range keys")
 	}
 	if err := checkCoverage(ctx, c.lastBackup().Spans, c.backupChain); err != nil {
 		return nil, err

pkg/backup/compaction_processor.go

@@ -367,6 +367,11 @@ func openSSTs(
 	storeFiles := make([]storageccl.StoreFile, 0, len(entry.Files))
 	for idx := range entry.Files {
 		file := entry.Files[idx]
+		if file.HasRangeKeys {
+			// TODO (kev-cao): Come back and update this to range keys when
+			// SSTSinkKeyWriter has been updated to support range keys.
+			return mergedSST{}, errors.New("backup compactions does not support range keys")
+		}
 		dir, err := execCfg.DistSQLSrv.ExternalStorage(ctx, file.Dir)
 		if err != nil {
 			return mergedSST{}, err
@@ -375,8 +380,6 @@ func openSSTs(
 		storeFiles = append(storeFiles, storageccl.StoreFile{Store: dir, FilePath: file.Path})
 	}
 	iterOpts := storage.IterOptions{
-		// TODO (kev-cao): Come back and update this to range keys when
-		// SSTSinkKeyWriter has been updated to support range keys.
 		KeyTypes:   storage.IterKeyTypePointsOnly,
 		LowerBound: keys.LocalMax,
 		UpperBound: keys.MaxKey,

pkg/backup/datadriven_test.go

@@ -397,6 +397,10 @@ func (d *datadrivenTestState) getSQLDBForVC(
 //   - expect-pausepoint: expects the backup job to end up in a paused state because
 //     of a pausepoint error.
 //
+//   - aost: expects a tag referencing a previously saved cluster timestamp
+//     using `save-cluster-ts`. It then runs the backup as of the saved cluster
+//     timestamp.
+//
 //   - "restore" [args]
 //     Executes restore specific operations.
 //
@@ -410,13 +414,29 @@ func (d *datadrivenTestState) getSQLDBForVC(
 //   - aost: expects a tag referencing a previously saved cluster timestamp
 //     using `save-cluster-ts`. It then runs the restore as of the saved cluster
 //     timestamp.
+
+//   - "compact" [args]
+//     Executes compaction specific operations.
+//
+//     Supported arguments:
+//
+//   - tag=<tag>: tag the compaction job to reference it in the future.
+//
+//   - expect-pausepoint: expects the restore job to end up in a paused state because
+//     of a pausepoint error.
+//
+//   - start: expects a tag referencing a previously saved cluster timestamp
+//     using `save-cluster-ts`.
+//
+//   - end: expects a tag referencing a previously saved cluster timestamp
+//     using `save-cluster-ts`.
 //
 //   - "schema" [args]
 //     Executes schema change specific operations.
 //
 //     Supported arguments:
 //
-//   - tag=<tag>: tag the schema change job to reference it in the future.
+// - tag=<tag>: tag the schema change job to reference it in the future.
 //
 //   - expect-pausepoint: expects the schema change job to end up in a paused state because
 //     of a pausepoint error.
@@ -426,9 +446,9 @@ func (d *datadrivenTestState) getSQLDBForVC(
 //
 //     Supported arguments:
 //
-//   - type: kv request type. Currently, only DeleteRange is supported
+// - type: kv request type. Currently, only DeleteRange is supported
 //
-//   - target: SQL target. Currently, only table names are supported.
+// - target: SQL target. Currently, only table names are supported.
 //
 //   - "corrupt-backup" uri=<collectionUri>
 //     Finds the latest backup in the provided collection uri an flips a bit in one SST in the backup
@@ -797,6 +817,18 @@ func runTestDataDriven(t *testing.T, testFilePathFromWorkspace string) {
 			return ""
 
 		case "backup":
+			if d.HasArg("aost") {
+				var aost string
+				d.ScanArgs(t, "aost", &aost)
+				var ts string
+				var ok bool
+				if ts, ok = ds.clusterTimestamps[aost]; !ok {
+					t.Fatalf("no cluster timestamp found for %s", aost)
+				}
+				// Replace the ts tag with the actual timestamp.
+				d.Input = strings.Replace(d.Input, aost,
+					fmt.Sprintf("'%s'", ts), 1)
+			}
 			return execWithTagAndPausePoint(jobspb.TypeBackup)
 
 		case "import":
@@ -818,6 +850,29 @@ func runTestDataDriven(t *testing.T, testFilePathFromWorkspace string) {
 			}
 			return execWithTagAndPausePoint(jobspb.TypeRestore)
 
+		case "compact":
+			if !d.HasArg("start") || !d.HasArg("end") {
+				t.Fatalf("must specify start and end for compaction")
+			}
+			var start, end string
+			d.ScanArgs(t, "start", &start)
+			d.ScanArgs(t, "end", &end)
+			var startTs, endTs string
+			var ok bool
+			if startTs, ok = ds.clusterTimestamps[start]; !ok {
+				t.Fatalf("no cluster timestamp found for %s", start)
+			}
+			if endTs, ok = ds.clusterTimestamps[end]; !ok {
+				t.Fatalf("no cluster timestamp found for %s", end)
+			}
+			// Replace the ts tag with the actual timestamp.
+			// ds.clusterTimestamps stores a stringified nanosecond epoch.
+			d.Input = strings.Replace(d.Input, start,
+				fmt.Sprintf("'%s'::DECIMAL", startTs), 1)
+			d.Input = strings.Replace(d.Input, end,
+				fmt.Sprintf("'%s'::DECIMAL", endTs), 1)
+
+			return execWithTagAndPausePoint(jobspb.TypeBackup)
 		case "new-schema-change":
 			return execWithTagAndPausePoint(jobspb.TypeNewSchemaChange)
 

pkg/backup/testdata/backup-restore/rangekeys

@@ -29,8 +29,11 @@ exec-sql
 INSERT INTO foo VALUES (3,'z');
 ----
 
-exec-sql
-BACKUP INTO 'nodelocal://1/test-root/';
+save-cluster-ts tag=start
+----
+
+backup aost=start
+BACKUP INTO 'nodelocal://1/test-root/' AS OF SYSTEM TIME start;
 ----
 
 exec-sql
@@ -61,15 +64,27 @@ exec-sql
 INSERT INTO foo VALUES (4,'a'),(5,'b');
 ----
 
+# Create two incremental backups to test compaction
+exec-sql
+BACKUP INTO LATEST IN 'nodelocal://1/test-root/';
+----
+
+exec-sql
+INSERT INTO foo VALUES (6,'a'),(7,'b');
+----
+
 kv request=DeleteRange target=baz
 ----
 
 exec-sql
 INSERT INTO baz VALUES (33,'zz');
 ----
 
-exec-sql
-BACKUP INTO LATEST IN 'nodelocal://1/test-root/';
+save-cluster-ts tag=end
+----
+
+backup aost=end
+BACKUP INTO LATEST IN 'nodelocal://1/test-root/' AS OF SYSTEM TIME end;
 ----
 
 exec-sql
@@ -84,9 +99,25 @@ regex matches error
 query-sql
 SELECT count(*) from orig1.foo
 ----
-3
+5
 
 query-sql
 SELECT count(*) from orig1.baz
 ----
 1
+
+let $backup_path
+SHOW BACKUPS IN 'nodelocal://1/test-root/';
+----
+
+compact start=start end=end tag=comp_job
+SELECT crdb_internal.backup_compaction('BACKUP INTO LATEST IN ''nodelocal://1/test-root/''', '$backup_path', start, end);
+----
+
+job tag=comp_job wait-for-state=failed
+----
+
+query-sql regex=(compactions does not support range keys)
+SELECT error FROM [SHOW JOBS] WHERE job_type = 'BACKUP' AND status = 'failed' ORDER BY created DESC LIMIT 1;
+----
+true