importer: checkpoint SST metadata in distributed merge for retry correctness

pkg/cli/testdata/table_dump_column_parsing

@@ -14,7 +14,7 @@ progress  \x10effdc3929b9f8a030d0000803fa8019984a0c1e09cd9e3486a0d1204000000002a
 ----
 {"Progress":null,"modified_micros":1733902763383997,"Details":{"table_metadata_cache":{}},"trace_id":0}
 {"Progress":null,"modified_micros":1733902768876277,"Details":{"AutoSpanConfigReconciliation":{"checkpoint":{"wall_time":1733902763154984000}}},"trace_id":0}
-{"Progress":{"fraction_completed":1},"modified_micros":1733902763884271,"Details":{"import":{"read_progress":[0],"span_progress":null,"resume_pos":[0],"sequence_details":[{}],"summary":{}}},"trace_id":5244271230238327321}
+{"Progress":{"fraction_completed":1},"modified_micros":1733902763884271,"Details":{"import":{"read_progress":[0],"span_progress":null,"resume_pos":[0],"sequence_details":[{}],"summary":{},"completed_map_work":null}},"trace_id":5244271230238327321}
 
 
 system.tenants.txt

pkg/jobs/jobspb/jobs.proto

@@ -793,6 +793,11 @@ message ImportProgress {
   // distributed-merge SSTs for this import. These prefixes are used to clean
   // up job-scoped files on completion or cancellation.
   repeated string sst_storage_prefixes = 8 [(gogoproto.customname) = "SSTStoragePrefixes"];
+
+  // CompletedMapWork stores SST file metadata from processors that have
+  // finished during the map phase of distributed merge so that the files
+  // are not lost on retry.
+  repeated BulkSSTManifest completed_map_work = 9 [(gogoproto.nullable) = false];
 }
 
 // TypeSchemaChangeDetails is the job detail information for a type schema change job.
@@ -889,7 +894,7 @@ message BackfillProgress {
   // For example, after the map phase it holds all SSTs generated by that phase.
   // As those SSTs are consumed by the first merge iteration, they are replaced
   // by the outputs of that iteration.
-  repeated IndexBackfillSSTManifest sst_manifests = 6 [(gogoproto.nullable) = false, (gogoproto.customname) = "SSTManifests"];
+  repeated BulkSSTManifest sst_manifests = 6 [(gogoproto.nullable) = false, (gogoproto.customname) = "SSTManifests"];
 
   // SSTStoragePrefixes identifies the external storage prefixes used to write
   // distributed-merge SSTs for this backfill. These prefixes are used to clean
@@ -914,8 +919,8 @@ message BackfillProgress {
   repeated int64 merge_iteration_completed_tasks = 10;
 }
 
-// IndexBackfillSSTManifest describes an SST produced by the backfill map stage.
-message IndexBackfillSSTManifest {
+// BulkSSTManifest describes an SST produced by bulk operations (backfill, import).
+message BulkSSTManifest {
   string uri = 1 [(gogoproto.customname) = "URI"];
   roachpb.Span span = 2;
   uint64 file_size = 3;
@@ -984,7 +989,7 @@ message KeyVisualizerProgress {}
 
 message ResumeSpanList {
   repeated roachpb.Span resume_spans = 1 [(gogoproto.nullable) = false];
-  repeated IndexBackfillSSTManifest sst_manifests = 2 [
+  repeated BulkSSTManifest sst_manifests = 2 [
     (gogoproto.nullable) = false,
     (gogoproto.customname) = "SSTManifests"
   ];

pkg/sql/backfill.go

@@ -945,7 +945,7 @@ func (sc *SchemaChanger) distIndexBackfill(
 ) error {
 	// Gather the initial resume spans for the table.
 	var todoSpans []roachpb.Span
-	var resumeManifests []jobspb.IndexBackfillSSTManifest
+	var resumeManifests []jobspb.BulkSSTManifest
 	var mutationIdx int
 	jobDetails := sc.job.Details().(jobspb.SchemaChangeDetails)
 	useDistributedMerge := jobDetails.DistributedMergeMode == jobspb.IndexBackfillDistributedMergeMode_Enabled
@@ -1083,7 +1083,7 @@ func (sc *SchemaChanger) distIndexBackfill(
 			}()
 
 			if useDistributedMerge {
-				var mapProgress execinfrapb.IndexBackfillMapProgress
+				var mapProgress execinfrapb.BulkMapProgress
 				if gogotypes.Is(&meta.BulkProcessorProgress.ProgressDetails, &mapProgress) {
 					if err := gogotypes.UnmarshalAny(&meta.BulkProcessorProgress.ProgressDetails, &mapProgress); err != nil {
 						return err
@@ -1184,7 +1184,7 @@ func (sc *SchemaChanger) distIndexBackfill(
 			if spansToPersist == nil {
 				spansToPersist = todoSpans
 			}
-			var manifestSnapshot []jobspb.IndexBackfillSSTManifest
+			var manifestSnapshot []jobspb.BulkSSTManifest
 			if manifestDirty {
 				manifestSnapshot = manifestBuf.SnapshotAndMarkClean()
 			} else {

pkg/sql/backfill/sst_manifest.go

@@ -18,13 +18,13 @@ import (
 // ingestion pipeline.
 type SSTManifestBuffer struct {
 	syncutil.Mutex
-	manifests []jobspb.IndexBackfillSSTManifest
+	manifests []jobspb.BulkSSTManifest
 	dirty     bool
 }
 
 // NewSSTManifestBuffer constructs a buffer initialized with the provided
 // manifests.
-func NewSSTManifestBuffer(initial []jobspb.IndexBackfillSSTManifest) *SSTManifestBuffer {
+func NewSSTManifestBuffer(initial []jobspb.BulkSSTManifest) *SSTManifestBuffer {
 	buf := &SSTManifestBuffer{}
 	buf.manifests = append(buf.manifests, initial...)
 	if len(initial) > 0 {
@@ -33,22 +33,20 @@ func NewSSTManifestBuffer(initial []jobspb.IndexBackfillSSTManifest) *SSTManifes
 	return buf
 }
 
-func (b *SSTManifestBuffer) snapshotLocked() []jobspb.IndexBackfillSSTManifest {
-	return append([]jobspb.IndexBackfillSSTManifest(nil), b.manifests...)
+func (b *SSTManifestBuffer) snapshotLocked() []jobspb.BulkSSTManifest {
+	return append([]jobspb.BulkSSTManifest(nil), b.manifests...)
 }
 
 // Snapshot returns a copy of the buffered manifests.
-func (b *SSTManifestBuffer) Snapshot() []jobspb.IndexBackfillSSTManifest {
+func (b *SSTManifestBuffer) Snapshot() []jobspb.BulkSSTManifest {
 	b.Lock()
 	defer b.Unlock()
 	return b.snapshotLocked()
 }
 
 // Append adds the provided manifests to the buffer and returns the updated
 // snapshot.
-func (b *SSTManifestBuffer) Append(
-	newManifests []jobspb.IndexBackfillSSTManifest,
-) []jobspb.IndexBackfillSSTManifest {
+func (b *SSTManifestBuffer) Append(newManifests []jobspb.BulkSSTManifest) []jobspb.BulkSSTManifest {
 	if len(newManifests) == 0 {
 		return b.Snapshot()
 	}
@@ -72,7 +70,7 @@ func (b *SSTManifestBuffer) Dirty() bool {
 
 // SnapshotAndMarkClean returns a copy of the buffered manifests and clears the
 // dirty flag. Callers should invoke this once they have persisted the snapshot.
-func (b *SSTManifestBuffer) SnapshotAndMarkClean() []jobspb.IndexBackfillSSTManifest {
+func (b *SSTManifestBuffer) SnapshotAndMarkClean() []jobspb.BulkSSTManifest {
 	if b == nil {
 		return nil
 	}
@@ -86,9 +84,9 @@ func (b *SSTManifestBuffer) SnapshotAndMarkClean() []jobspb.IndexBackfillSSTMani
 // removing tenant prefixes before persisting it in job state. This matches the
 // CompletedSpans handling and keeps job progress tenant-agnostic.
 func StripTenantPrefixFromSSTManifests(
-	codec keys.SQLCodec, manifests []jobspb.IndexBackfillSSTManifest,
-) ([]jobspb.IndexBackfillSSTManifest, error) {
-	ret := make([]jobspb.IndexBackfillSSTManifest, len(manifests))
+	codec keys.SQLCodec, manifests []jobspb.BulkSSTManifest,
+) ([]jobspb.BulkSSTManifest, error) {
+	ret := make([]jobspb.BulkSSTManifest, len(manifests))
 	for i, out := range manifests {
 		ret[i].URI = out.URI
 		ret[i].FileSize = out.FileSize
@@ -117,9 +115,9 @@ func StripTenantPrefixFromSSTManifests(
 // AddTenantPrefixToSSTManifests rehydrates manifests with the executing
 // processor's tenant prefix so they can be used against the local keyspace.
 func AddTenantPrefixToSSTManifests(
-	codec keys.SQLCodec, manifests []jobspb.IndexBackfillSSTManifest,
-) []jobspb.IndexBackfillSSTManifest {
-	ret := make([]jobspb.IndexBackfillSSTManifest, len(manifests))
+	codec keys.SQLCodec, manifests []jobspb.BulkSSTManifest,
+) []jobspb.BulkSSTManifest {
+	ret := make([]jobspb.BulkSSTManifest, len(manifests))
 	for i, out := range manifests {
 		ret[i].URI = out.URI
 		ret[i].FileSize = out.FileSize

pkg/sql/bulksst/BUILD.bazel

@@ -23,6 +23,8 @@ go_library(
     name = "bulksst",
     srcs = [
         "combine_file_info.go",
+        "manifest_conversion.go",
+        "sink.go",
         "sst_file_allocator.go",
         "sst_writer.go",
     ],
@@ -31,10 +33,12 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/cloud",
+        "//pkg/jobs/jobspb",
         "//pkg/keys",
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver/kvserverbase",
         "//pkg/roachpb",
+        "//pkg/security/username",
         "//pkg/settings",
         "//pkg/settings/cluster",
         "//pkg/sql/execinfrapb",
@@ -54,6 +58,7 @@ go_test(
     srcs = [
         "combine_file_info_test.go",
         "main_test.go",
+        "sink_test.go",
         "sst_file_allocator_test.go",
         "sst_writer_test.go",
     ],
@@ -63,6 +68,7 @@ go_test(
         "//pkg/cloud/cloudpb",
         "//pkg/cloud/impl:cloudimpl",
         "//pkg/cloud/nodelocal",
+        "//pkg/jobs/jobspb",
         "//pkg/keys",
         "//pkg/kv/kvpb",
         "//pkg/roachpb",

pkg/sql/bulksst/manifest_conversion.go

@@ -0,0 +1,84 @@
+// Copyright 2026 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package bulksst
+
+import (
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+)
+
+// SSTFileToManifest converts a single bulksst.SSTFileInfo to a
+// jobspb.BulkSSTManifest for persistence in job progress.
+//
+// The writeTS parameter is optional (can be nil). When non-nil, it's stored in
+// the manifest's WriteTimestamp field. This is used by index backfill but not
+// by import operations.
+func SSTFileToManifest(f *SSTFileInfo, writeTS *hlc.Timestamp) jobspb.BulkSSTManifest {
+	span := &roachpb.Span{
+		Key:    append(roachpb.Key(nil), f.StartKey...),
+		EndKey: append(roachpb.Key(nil), f.EndKey...),
+	}
+	manifest := jobspb.BulkSSTManifest{
+		URI:       f.URI,
+		Span:      span,
+		FileSize:  f.FileSize,
+		RowSample: append(roachpb.Key(nil), f.RowSample...),
+		KeyCount:  f.KeyCount,
+	}
+	if writeTS != nil {
+		manifest.WriteTimestamp = writeTS
+	}
+	return manifest
+}
+
+// SSTFilesToManifests converts a slice of bulksst.SSTFileInfo to
+// jobspb.BulkSSTManifest entries for persistence.
+//
+// This is used by both import and index backfill operations to convert their
+// in-memory SST file lists to the format stored in job progress. The writeTS
+// parameter is optional and only used by index backfill.
+func SSTFilesToManifests(files *SSTFiles, writeTS *hlc.Timestamp) []jobspb.BulkSSTManifest {
+	if files == nil || len(files.SST) == 0 {
+		return nil
+	}
+
+	manifests := make([]jobspb.BulkSSTManifest, 0, len(files.SST))
+	for _, f := range files.SST {
+		manifests = append(manifests, SSTFileToManifest(f, writeTS))
+	}
+	return manifests
+}
+
+// ManifestsToSSTFiles converts jobspb.BulkSSTManifest entries back to
+// bulksst.SSTFiles. This is the inverse of SSTFilesToManifests.
+//
+// This is used when restoring SST metadata from job progress during retry or
+// resume operations.
+func ManifestsToSSTFiles(manifests []jobspb.BulkSSTManifest) SSTFiles {
+	if len(manifests) == 0 {
+		return SSTFiles{}
+	}
+
+	result := SSTFiles{
+		SST: make([]*SSTFileInfo, 0, len(manifests)),
+	}
+
+	for _, manifest := range manifests {
+		fileInfo := &SSTFileInfo{
+			URI:       manifest.URI,
+			StartKey:  append(roachpb.Key(nil), manifest.Span.Key...),
+			EndKey:    append(roachpb.Key(nil), manifest.Span.EndKey...),
+			FileSize:  manifest.FileSize,
+			RowSample: append(roachpb.Key(nil), manifest.RowSample...),
+			KeyCount:  manifest.KeyCount,
+		}
+		result.SST = append(result.SST, fileInfo)
+		result.TotalSize += manifest.FileSize
+	}
+
+	return result
+}