backupccl: reduce over-splitting during RESTORE

When constructing the spans to restore, we can produce fewer, wider spans
that each contain more files to avoid splitting as much.

Previously we split at the bounds of every file in the base backup, then
added incremental layer files to any span they overlapped. This changes
that slightly to keep appending files that would create a new span at the
right-hand-side of the span covering to instead extend current rightmost
entry in the covering if the total size of what has been added to it is
still below our target size per span (384mb).

Release justification: bug fix.

Release note (bug fix): reduce the amount that RESTORE over-splits.

pkg/ccl/backupccl/bench_covering_test.go

@@ -58,7 +58,7 @@ func BenchmarkRestoreEntryCover(b *testing.B) {
 								if err := checkCoverage(ctx, backups[numBackups-1].Spans, backups); err != nil {
 									b.Fatal(err)
 								}
-								cov := makeSimpleImportSpans(backups[numBackups-1].Spans, backups, nil, nil)
+								cov := makeSimpleImportSpans(backups[numBackups-1].Spans, backups, nil, nil, 0)
 								b.ReportMetric(float64(len(cov)), "coverSize")
 							}
 						})

pkg/ccl/backupccl/restore_job.go

@@ -275,7 +275,7 @@ func restore(
 	highWaterMark := job.Progress().Details.(*jobspb.Progress_Restore).Restore.HighWater
 
 	importSpans := makeSimpleImportSpans(dataToRestore.getSpans(), backupManifests, backupLocalityMap,
-		highWaterMark)
+		highWaterMark, targetRestoreSpanSize)
 
 	if len(importSpans) == 0 {
 		// There are no files to restore.

pkg/ccl/backupccl/restore_span_covering.go

@@ -30,6 +30,21 @@ func (ie intervalSpan) Range() interval.Range {
 	return interval.Range{Start: []byte(ie.Key), End: []byte(ie.EndKey)}
 }
 
+// targetRestoreSpanSize defines a minimum size for the sum of sizes of files in
+// the initial level (base backup or first inc covering some span) of a restore
+// span. As each restore span is explicitly split, scattered, processed, and
+// explicitly flushed, large numbers of tiny restore spans are costly, both in
+// terms of overhead associated with the splits, scatters and flushes and then
+// in subsequently merging the tiny ranges. Thus making sure this is at least in
+// the neighborhood of a typical range size minimizes that overhead. A restore
+// span may well grow beyond this size when later incremental layer files are
+// added to it, but that's fine: we will split as we fill if needed. 384mb is
+// big enough to avoid the worst of tiny-span overhead, while small enough to be
+// a granular unit of work distribution and progress tracking. If progress were
+// tracked within restore spans, this could become dynamic and much larger (e.g.
+// totalSize/numNodes*someConstant).
+const targetRestoreSpanSize = 384 << 20
+
 // makeSimpleImportSpans partitions the spans of requiredSpans into a covering
 // of RestoreSpanEntry's which each have all overlapping files from the passed
 // backups assigned to them. The spans of requiredSpans are trimmed/removed
@@ -51,11 +66,17 @@ func (ie intervalSpan) Range() interval.Range {
 //  [f, i): 3, 5, 6, 8
 //  [l, m): 9
 // This example is tested in TestRestoreEntryCoverExample.
+//
+// If targetSize > 0, then spans which would be added to the right-hand side of
+// the first level are instead used to extend the current rightmost span in
+// if its current data size plus that of the new span is less than the target
+// size.
 func makeSimpleImportSpans(
 	requiredSpans []roachpb.Span,
 	backups []backuppb.BackupManifest,
 	backupLocalityMap map[int]storeByLocalityKV,
 	lowWaterMark roachpb.Key,
+	targetSize int64,
 ) []execinfrapb.RestoreSpanEntry {
 	if len(backups) < 1 {
 		return nil
@@ -78,20 +99,61 @@ func makeSimpleImportSpans(
 
 		for layer := range backups {
 			covPos := spanCoverStart
+
+			// lastCovSpanSize is the size of files added to the right-most span of
+			// the cover so far.
+			var lastCovSpanSize int64
+
 			// TODO(dt): binary search to the first file in required span?
 			for _, f := range backups[layer].Files {
 				if sp := span.Intersect(f.Span); sp.Valid() {
 					fileSpec := execinfrapb.RestoreFileSpec{Path: f.Path, Dir: backups[layer].Dir}
 					if dir, ok := backupLocalityMap[layer][f.LocalityKV]; ok {
 						fileSpec = execinfrapb.RestoreFileSpec{Path: f.Path, Dir: dir}
 					}
+
+					// Lookup the size of the file being added; if the backup didn't
+					// record a file size, just assume it is 16mb for estimating.
+					sz := f.EntryCounts.DataSize
+					if sz == 0 {
+						sz = 16 << 20
+					}
+
 					if len(cover) == spanCoverStart {
 						cover = append(cover, makeEntry(span.Key, sp.EndKey, fileSpec))
+						lastCovSpanSize = sz
 					} else {
-						// Add each file to every matching partition in the cover.
+						// If this file extends beyond the end of the last partition of the
+						// cover, either append a new partition for the uncovered span or
+						// grow the last one if size allows.
+						if covEnd := cover[len(cover)-1].Span.EndKey; sp.EndKey.Compare(covEnd) > 0 {
+							// If adding the item size to the current rightmost span size will
+							// exceed the target size, make a new span, otherwise extend the
+							// rightmost span to include the item.
+							if lastCovSpanSize+sz > targetSize {
+								cover = append(cover, makeEntry(covEnd, sp.EndKey, fileSpec))
+								lastCovSpanSize = sz
+							} else {
+								cover[len(cover)-1].Span.EndKey = sp.EndKey
+								cover[len(cover)-1].Files = append(cover[len(cover)-1].Files, fileSpec)
+								lastCovSpanSize += sz
+							}
+						}
+						// Now ensure the file is included in any partition in the existing
+						// cover which overlaps.
 						for i := covPos; i < len(cover) && cover[i].Span.Key.Compare(sp.EndKey) < 0; i++ {
+							// If file overlaps, it needs to be in this partition.
 							if cover[i].Span.Overlaps(sp) {
-								cover[i].Files = append(cover[i].Files, fileSpec)
+								// If this is the last partition, we might have added it above.
+								if i == len(cover)-1 {
+									if last := len(cover[i].Files) - 1; last < 0 || cover[i].Files[last] != fileSpec {
+										cover[i].Files = append(cover[i].Files, fileSpec)
+										lastCovSpanSize += sz
+									}
+								} else {
+									// If it isn't the last partition, we always need to add it.
+									cover[i].Files = append(cover[i].Files, fileSpec)
+								}
 							}
 							// If partition i of the cover ends before this file starts, we
 							// know it also ends before any remaining files start too, as the
@@ -101,11 +163,6 @@ func makeSimpleImportSpans(
 								covPos = i + 1
 							}
 						}
-						// If this file extends beyond the end of the last partition of the
-						// cover, append a new partition for the uncovered span.
-						if covEnd := cover[len(cover)-1].Span.EndKey; sp.EndKey.Compare(covEnd) > 0 {
-							cover = append(cover, makeEntry(covEnd, sp.EndKey, fileSpec))
-						}
 					}
 				} else if span.EndKey.Compare(f.Span.Key) <= 0 {
 					// If this file starts after the needed span ends, then all the files
@@ -121,6 +178,7 @@ func makeSimpleImportSpans(
 
 func makeEntry(start, end roachpb.Key, f execinfrapb.RestoreFileSpec) execinfrapb.RestoreSpanEntry {
 	return execinfrapb.RestoreSpanEntry{
-		Span: roachpb.Span{Key: start, EndKey: end}, Files: []execinfrapb.RestoreFileSpec{f},
+		Span:  roachpb.Span{Key: start, EndKey: end},
+		Files: []execinfrapb.RestoreFileSpec{f},
 	}
 }

pkg/ccl/backupccl/restore_span_covering_test.go

@@ -62,6 +62,7 @@ func MockBackupChain(length, spans, baseFiles int, r *rand.Rand) []backuppb.Back
 			backups[i].Files[f].Span.Key = encoding.EncodeVarintAscending(k, int64(start))
 			backups[i].Files[f].Span.EndKey = encoding.EncodeVarintAscending(k, int64(end))
 			backups[i].Files[f].Path = fmt.Sprintf("12345-b%d-f%d.sst", i, f)
+			backups[i].Files[f].EntryCounts.DataSize = 1 << 20
 		}
 		// A non-nil Dir more accurately models the footprint of produced coverings.
 		backups[i].Dir = cloudpb.ExternalStorage{S3Config: &cloudpb.ExternalStorage_S3{}}
@@ -82,7 +83,10 @@ func MockBackupChain(length, spans, baseFiles int, r *rand.Rand) []backuppb.Back
 // key when walking files in order by start key in the backups. This check is
 // thus sensitive to ordering; the coverage correctness check however is not.
 func checkRestoreCovering(
-	backups []backuppb.BackupManifest, spans roachpb.Spans, cov []execinfrapb.RestoreSpanEntry,
+	backups []backuppb.BackupManifest,
+	spans roachpb.Spans,
+	cov []execinfrapb.RestoreSpanEntry,
+	merged bool,
 ) error {
 	var expectedPartitions int
 	required := make(map[string]*roachpb.SpanGroup)
@@ -110,15 +114,17 @@ func checkRestoreCovering(
 	}
 	for name, uncovered := range required {
 		for _, missing := range uncovered.Slice() {
-			return errors.Errorf("file %s is was supposed to cover span %s", name, missing)
+			return errors.Errorf("file %s was supposed to cover span %s", name, missing)
 		}
 	}
-	if got := len(cov); got != expectedPartitions {
-		return errors.Errorf("expected at least %d partitions, got %d", expectedPartitions, got)
+	if got := len(cov); got != expectedPartitions && !merged {
+		return errors.Errorf("expected %d partitions, got %d", expectedPartitions, got)
 	}
 	return nil
 }
 
+const noSpanTargetSize = 0
+
 func TestRestoreEntryCoverExample(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
@@ -144,14 +150,30 @@ func TestRestoreEntryCoverExample(t *testing.T) {
 		{Files: []backuppb.BackupManifest_File{f("a", "h", "6"), f("j", "k", "7")}},
 		{Files: []backuppb.BackupManifest_File{f("h", "i", "8"), f("l", "m", "9")}},
 	}
-	cover := makeSimpleImportSpans(spans, backups, nil, nil)
+
+	// Pretend every span has 1MB.
+	for i := range backups {
+		for j := range backups[i].Files {
+			backups[i].Files[j].EntryCounts.DataSize = 1 << 20
+		}
+	}
+
+	cover := makeSimpleImportSpans(spans, backups, nil, nil, noSpanTargetSize)
 	require.Equal(t, []execinfrapb.RestoreSpanEntry{
 		{Span: sp("a", "c"), Files: paths("1", "4", "6")},
 		{Span: sp("c", "e"), Files: paths("2", "4", "6")},
 		{Span: sp("e", "f"), Files: paths("6")},
 		{Span: sp("f", "i"), Files: paths("3", "5", "6", "8")},
 		{Span: sp("l", "m"), Files: paths("9")},
 	}, cover)
+
+	coverSized := makeSimpleImportSpans(spans, backups, nil, nil, 2<<20)
+	require.Equal(t, []execinfrapb.RestoreSpanEntry{
+		{Span: sp("a", "f"), Files: paths("1", "2", "4", "6")},
+		{Span: sp("f", "i"), Files: paths("3", "5", "6", "8")},
+		{Span: sp("l", "m"), Files: paths("9")},
+	}, coverSized)
+
 }
 
 func TestRestoreEntryCover(t *testing.T) {
@@ -162,13 +184,14 @@ func TestRestoreEntryCover(t *testing.T) {
 		for _, spans := range []int{1, 2, 3, 5, 9, 11, 12} {
 			for _, files := range []int{0, 1, 2, 3, 4, 10, 12, 50} {
 				backups := MockBackupChain(numBackups, spans, files, r)
-
-				t.Run(fmt.Sprintf("numBackups=%d, numSpans=%d, numFiles=%d", numBackups, spans, files), func(t *testing.T) {
-					cover := makeSimpleImportSpans(backups[numBackups-1].Spans, backups, nil, nil)
-					if err := checkRestoreCovering(backups, backups[numBackups-1].Spans, cover); err != nil {
-						t.Fatal(err)
-					}
-				})
+				for _, target := range []int64{0, 1, 4, 100, 1000} {
+					t.Run(fmt.Sprintf("numBackups=%d, numSpans=%d, numFiles=%d, merge=%d", numBackups, spans, files, target), func(t *testing.T) {
+						cover := makeSimpleImportSpans(backups[numBackups-1].Spans, backups, nil, nil, target<<20)
+						if err := checkRestoreCovering(backups, backups[numBackups-1].Spans, cover, target != noSpanTargetSize); err != nil {
+							t.Fatal(err)
+						}
+					})
+				}
 			}
 		}
 	}

pkg/ccl/backupccl/targets.go

@@ -495,6 +495,7 @@ func MakeBackupTableEntry(
 		backupManifests,
 		nil,           /*backupLocalityInfo*/
 		roachpb.Key{}, /*lowWaterMark*/
+		targetRestoreSpanSize,
 	)
 
 	lastSchemaChangeTime := findLastSchemaChangeTime(backupManifests, tbDesc, endTime)

pkg/ccl/backupccl/testdata/backup-restore/column-families

@@ -50,12 +50,3 @@ query-sql
 SELECT max(length(start_key)) FROM [SHOW RANGES FROM TABLE orig.cfs];
 ----
 2
-
-# Regression test for #67488.
-# This can return up to 6 if RESTORE improperly splits the ranges in the middle
-# of a SQL row since column family keys are longer. E.g. Keys are expected to be
-# of the form '/1' (row with PK 1), and not '/3/1/1' (row with PK 3, and CF 1).
-query-sql
-SELECT max(length(start_key)) FROM [SHOW RANGES FROM TABLE r1.cfs];
-----
-2