backup: hooked up compaction processor logic to job

Author: kev-cao

Diff for: pkg/backup/backup_job.go

@@ -246,7 +246,7 @@ func backup(
 	progCh := make(chan *execinfrapb.RemoteProducerMetadata_BulkProcessorProgress)
 	checkpointLoop := func(ctx context.Context) error {
 		// When a processor is done exporting a span, it will send a progress update
-		// to progCh.
+		// to .
 		defer close(requestFinishedCh)
 		defer close(perNodeProgressCh)
 		var numBackedUpFiles int64

Diff for: pkg/backup/compaction.go

@@ -36,6 +36,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/util"
+	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/retry"
@@ -341,7 +342,9 @@ func (b *backupResumer) ResumeCompaction(
 	// dying), so if we receive a retryable error, re-plan and retry the backup.
 	// TODO (kev-cao): Add progress tracking to compactions.
 	for r := retry.StartWithCtx(ctx, retryOpts); r.Next(); {
-		if err = compactChain.Compact(ctx, execCtx, updatedDetails, backupManifest, defaultStore, kmsEnv); err == nil {
+		if err = doCompaction(
+			ctx, execCtx, b.job.ID(), updatedDetails, compactChain, backupManifest, defaultStore, kmsEnv,
+		); err == nil {
 			break
 		}
 
@@ -728,7 +731,7 @@ func concludeBackupCompaction(
 func processProgress(
 	ctx context.Context,
 	manifest *backuppb.BackupManifest,
-	progCh <-chan execinfrapb.RemoteProducerMetadata_BulkProcessorProgress,
+	progCh <-chan *execinfrapb.RemoteProducerMetadata_BulkProcessorProgress,
 ) error {
 	// When a processor is done exporting a span, it will send a progress update
 	// to progCh.
@@ -772,6 +775,37 @@ func compactionJobDescription(details jobspb.BackupDetails) (string, error) {
 	return fmtCtx.CloseAndGetString(), nil
 }
 
+func doCompaction(
+	ctx context.Context,
+	execCtx sql.JobExecContext,
+	jobID jobspb.JobID,
+	details jobspb.BackupDetails,
+	compactChain compactionChain,
+	manifest *backuppb.BackupManifest,
+	defaultStore cloud.ExternalStorage,
+	kmsEnv cloud.KMSEnv,
+) error {
+	progCh := make(chan *execinfrapb.RemoteProducerMetadata_BulkProcessorProgress)
+	runDistCompaction := func(ctx context.Context) error {
+		return runCompactionPlan(
+			ctx, execCtx, jobID, details, compactChain, manifest, defaultStore, kmsEnv, progCh,
+		)
+	}
+	checkpointLoop := func(ctx context.Context) error {
+		return processProgress(ctx, manifest, progCh)
+	}
+
+	if err := ctxgroup.GoAndWait(
+		ctx, runDistCompaction, checkpointLoop,
+	); err != nil {
+		return err
+	}
+
+	return concludeBackupCompaction(
+		ctx, execCtx, defaultStore, details.EncryptionOptions, kmsEnv, manifest,
+	)
+}
+
 func init() {
 	builtins.StartCompactionJob = StartCompactionJob
 }

Diff for: pkg/backup/compaction_processor.go

@@ -177,6 +177,9 @@ func runCompactBackups(
 		return errors.Wrapf(err, "export configuration")
 	}
 	defaultStore, err := execCfg.DistSQLSrv.ExternalStorage(ctx, defaultConf)
+	if err != nil {
+		return errors.Wrapf(err, "external storage")
+	}
 
 	compactChain, encryption, err := compactionChainFromSpec(ctx, execCfg, spec, user)
 	if err != nil {
@@ -186,6 +189,7 @@ func runCompactBackups(
 	intersectSpanCh := make(chan execinfrapb.RestoreSpanEntry, 1000)
 	tasks := []func(context.Context) error{
 		func(ctx context.Context) error {
+			defer close(intersectSpanCh)
 			return genIntersectingSpansForCompaction(ctx, execCfg, spec, compactChain, intersectSpanCh)
 		},
 		func(ctx context.Context) error {
@@ -284,9 +288,7 @@ func genIntersectingSpansForCompaction(
 	compactChain compactionChain,
 	intersectSpanCh chan execinfrapb.RestoreSpanEntry,
 ) error {
-	user := spec.User()
-
-	backupLocalityMap, err := makeBackupLocalityMap(compactChain.compactedLocalityInfo, user)
+	backupLocalityMap, err := makeBackupLocalityMap(compactChain.compactedLocalityInfo, spec.User())
 	if err != nil {
 		return err
 	}
@@ -299,7 +301,6 @@ func genIntersectingSpansForCompaction(
 	}
 	defer introducedSpanFrontier.Release()
 
-	entryCh := make(chan execinfrapb.RestoreSpanEntry, 1000)
 	targetSize := targetRestoreSpanSize.Get(&execCfg.Settings.SV)
 	maxFiles := maxFileCount.Get(&execCfg.Settings.SV)
 
@@ -311,12 +312,12 @@ func genIntersectingSpansForCompaction(
 		targetSize,
 		maxFiles,
 	)
+	if err != nil {
+		return err
+	}
 
-	genSpan := func(ctx context.Context) error {
+	genSpan := func(ctx context.Context, entryCh chan execinfrapb.RestoreSpanEntry) error {
 		defer close(entryCh)
-		if err != nil {
-			return err
-		}
 		return errors.Wrapf(generateAndSendImportSpans(
 			ctx,
 			spec.Spans,
@@ -329,7 +330,7 @@ func genIntersectingSpansForCompaction(
 		), "generate and send import spans")
 	}
 
-	filterIntersectingSpans := func(ctx context.Context) error {
+	filterIntersectingSpans := func(ctx context.Context, entryCh chan execinfrapb.RestoreSpanEntry) error {
 		for {
 			select {
 			case <-ctx.Done():
@@ -348,12 +349,13 @@ func genIntersectingSpansForCompaction(
 		}
 	}
 
+	entryCh := make(chan execinfrapb.RestoreSpanEntry, 1000)
 	tasks := []func(context.Context) error{
 		func(ctx context.Context) error {
-			return genSpan(ctx)
+			return genSpan(ctx, entryCh)
 		},
 		func(ctx context.Context) error {
-			return filterIntersectingSpans(ctx)
+			return filterIntersectingSpans(ctx, entryCh)
 		},
 	}
 	return ctxgroup.GoAndWait(ctx, tasks...)

Diff for: pkg/backup/compaction_processor_planning.go

@@ -32,27 +32,34 @@ import (
 // in the backup chain that need to be compacted. It sends updates from the
 // BulkProcessor to the provided progress channel. It is the caller's
 // responsibility to close the progress channel.
-func (c *compactionChain) runCompactionPlan(
+func runCompactionPlan(
 	ctx context.Context,
 	execCtx sql.JobExecContext,
 	jobID jobspb.JobID,
-	manifest *backuppb.BackupManifest,
 	details jobspb.BackupDetails,
+	compactChain compactionChain,
+	manifest *backuppb.BackupManifest,
 	defaultStore cloud.ExternalStorage,
 	kmsEnv cloud.KMSEnv,
 	progCh chan *execinfrapb.RemoteProducerMetadata_BulkProcessorProgress,
 ) error {
+	defer close(progCh)
 	log.Infof(
 		ctx, "planning compaction of %d backups: %s",
-		len(c.chainToCompact), util.Map(c.chainToCompact, func(m backuppb.BackupManifest) string {
+		len(compactChain.chainToCompact),
+		util.Map(compactChain.chainToCompact, func(m backuppb.BackupManifest) string {
 			return m.ID.String()
 		}),
 	)
-	backupLocalityMap, err := makeBackupLocalityMap(c.compactedLocalityInfo, execCtx.User())
+	backupLocalityMap, err := makeBackupLocalityMap(
+		compactChain.compactedLocalityInfo, execCtx.User(),
+	)
 	if err != nil {
 		return err
 	}
-	introducedSpanFrontier, err := createIntroducedSpanFrontier(c.backupChain, manifest.EndTime)
+	introducedSpanFrontier, err := createIntroducedSpanFrontier(
+		compactChain.backupChain, manifest.EndTime,
+	)
 	if err != nil {
 		return err
 	}
@@ -72,18 +79,17 @@ func (c *compactionChain) runCompactionPlan(
 	}
 
 	spansToCompact, err := getSpansToCompact(
-		ctx, execCtx, manifest, c.chainToCompact, details, defaultStore, kmsEnv,
+		ctx, execCtx, manifest, compactChain.chainToCompact, details, defaultStore, kmsEnv,
 	)
 	if err != nil {
 		return err
 	}
 	genSpan := func(ctx context.Context, spanCh chan execinfrapb.RestoreSpanEntry) error {
-		defer close(spanCh)
 		return errors.Wrap(generateAndSendImportSpans(
 			ctx,
 			spansToCompact,
-			c.chainToCompact,
-			c.compactedIterFactory,
+			compactChain.chainToCompact,
+			compactChain.compactedIterFactory,
 			backupLocalityMap,
 			filter,
 			fsc,
@@ -105,9 +111,10 @@ func (c *compactionChain) runCompactionPlan(
 		}
 		return nil
 	}
+	rowResultWriter := sql.NewRowResultWriter(nil)
 	recv := sql.MakeDistSQLReceiver(
 		ctx,
-		sql.NewMetadataCallbackWriter(nil, metaFn),
+		sql.NewMetadataCallbackWriter(rowResultWriter, metaFn),
 		tree.Rows,
 		nil, /* rangeCache */
 		nil, /* txn */
@@ -125,7 +132,7 @@ func (c *compactionChain) runCompactionPlan(
 	return nil
 }
 
-// createCompactionPlan creates an unfinalized physical plan that will
+// createCompactionPlan creates an un-finalized physical plan that will
 // distribute spans from a generator across the cluster for compaction.
 func createCompactionPlan(
 	ctx context.Context,
@@ -184,6 +191,7 @@ func countRestoreSpanEntries(
 			return nil
 		},
 		func(ctx context.Context) error {
+			defer close(countSpansCh)
 			return genSpan(ctx, countSpansCh)
 		},
 	}
@@ -213,13 +221,14 @@ func createCompactionCorePlacements(
 	spanEntryCh := make(chan execinfrapb.RestoreSpanEntry, 1000)
 	var tasks []func(ctx context.Context) error
 	tasks = append(tasks, func(ctx context.Context) error {
+		defer close(spanEntryCh)
 		return genSpan(ctx, spanEntryCh)
 	})
 	tasks = append(tasks, func(ctx context.Context) error {
 		numEntriesPerNode := numEntries / numNodes
 		leftoverEntries := numEntries % numNodes
 		getTargetNumEntries := func(nodeIdx int) int {
-			if nodeIdx <= leftoverEntries {
+			if nodeIdx < leftoverEntries {
 				// This more evenly distributes the leftover entries across the nodes
 				// after doing integer division to assign the entries to the nodes.
 				return numEntriesPerNode + 1
@@ -231,10 +240,12 @@ func createCompactionCorePlacements(
 		targetNumEntries := getTargetNumEntries(currNode)
 
 		for entry := range spanEntryCh {
+			currEntries = append(currEntries, entry)
 			if len(currEntries) == targetNumEntries {
 				corePlacements[currNode].SQLInstanceID = sqlInstanceIDs[currNode]
 				corePlacements[currNode].Core.CompactBackups = &execinfrapb.CompactBackupsSpec{
 					JobID:       int64(jobID),
+					DefaultURI:  details.URI,
 					Destination: details.Destination,
 					Encryption:  details.EncryptionOptions,
 					StartTime:   details.StartTime,
@@ -253,7 +264,6 @@ func createCompactionCorePlacements(
 			if currNode == numNodes {
 				return nil
 			}
-			currEntries = append(currEntries, entry)
 		}
 		return nil
 	})

Diff for: pkg/sql/execinfrapb/flow_diagram.go

@@ -771,6 +771,29 @@ func (i *IngestStoppedSpec) summary() (string, []string) {
 	return "IngestStoppedSpec", []string{detail}
 }
 
+// summary implements the diagramCellType interface.
+func (m *CompactBackupsSpec) summary() (string, []string) {
+	var spanStr strings.Builder
+	if len(m.Spans) > 0 {
+		spanStr.WriteString(fmt.Sprintf("Spans [%d]: ", len(m.AssignedSpans)))
+		const limit = 3
+		for i := 0; i < len(m.AssignedSpans) && i < limit; i++ {
+			if i > 0 {
+				spanStr.WriteString(", ")
+			}
+			spanStr.WriteString(m.AssignedSpans[i].String())
+		}
+		if len(m.Spans) > limit {
+			spanStr.WriteString("...")
+		}
+	}
+
+	details := []string{
+		spanStr.String(),
+	}
+	return "CompactBackupsSpec", details
+}
+
 type diagramCell struct {
 	Title   string   `json:"title"`
 	Details []string `json:"details"`

Diff for: pkg/sql/rowexec/processors.go

@@ -412,7 +412,7 @@ func NewProcessor(
 		if NewCompactBackupsProcessor == nil {
 			return nil, errors.New("CompactBackups processor unimplemented")
 		}
-		return NewCompactBackupsProcessor(ctx, flowCtx, processorID, *core.CompactBackups, post, inputs[0])
+		return NewCompactBackupsProcessor(ctx, flowCtx, processorID, *core.CompactBackups, post)
 	}
 	return nil, errors.Errorf("unsupported processor core %q", core)
 }