importer: fix SST metadata loss on retry in distributed merge

[mw5h]

Before this change, IMPORT with distributed merge emitted SST metadata as a third column in row results (row[2]). This caused a critical bug on job retry: when the import job resumed after a node failure, processorOutput was reinitialized as an empty array. Only incomplete files (ResumePos < MaxInt64) were reprocessed, so SST metadata from completed files was lost. The merge phase then ran with an incomplete SST list, producing wrong data and fingerprint mismatches.

This commit changes SST metadata to flow through the progress/metadata channel instead of row results, following the same pattern as index backfill. Import processors now emit SST metadata via BulkProcessorProgress.SSTMetadata. The coordinator accumulates this metadata via the metaFn callback and periodically persists it to ImportProgress.CompletedMapWork. On retry, the SST metadata is restored from job progress, ensuring the merge phase has the complete SST list.

To avoid code duplication, we reuse jobspb.IndexBackfillSSTManifest (from index backfill) for SST metadata storage. Conversion functions in sst_conversion.go translate between bulksst.SSTFiles (used during processing) and IndexBackfillSSTManifest (used for persistence).

Changes:

• Added sst_metadata field (IndexBackfillSSTManifest) to BulkProcessorProgress
• Added completed_map_work field (IndexBackfillSSTManifest) to ImportProgress
• Created sst_conversion.go with conversion helpers
• Modified import_processor.go to emit SST metadata via progress channel from within the worker goroutine before the channel closes
• Updated import_processor_planning.go to accumulate SST metadata from progress metadata, persist to CompletedMapWork, and restore on retry
• Removed row[2] SST handling from rowResultWriter callback

Fixes: #161490

Epic: None

Release note: None

[blathers-crl]

It looks like your PR touches production code but doesn't add or edit any test code. Did you consider adding tests to your PR?

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is dev-inf.}

[cockroach-teamcity]

This change is 

[spilchen]

Is there any test that we can add for this that runs in the CI? Potentially even taking an existing IMPORT resume test and modifying it so that it works for distributed merge.

@spilchen reviewed 1 file and made 5 comments.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andrew-r-thomas, @mw5h, and @ZhouXing19).

---

pkg/jobs/jobspb/jobs.proto line 791 at r1 (raw file):

  // finished during the map phase of distributed merge so that the files
  // are not lost on retry.
  repeated IndexBackfillSSTManifest completed_map_work = 9 [(gogoproto.nullable) = false];

Can we consider renaming IndexBackfillSSTManifest to something more generic? It's a bit of a code smell to have something in an import struct refer to an backfill struct. I do like the reuse though.

---

pkg/sql/importer/import_processor_planning.go line 263 at r1 (raw file):

		}
		res.Add(counts)
		if len(row) == 3 {

Is SST info still being sent? I think it's done in Next().

    // When using distributed merge, the processor will emit the SSTs and their
    // start and end keys.
    var fileDatums rowenc.EncDatumRow
    if idp.files != nil {
        bytes, err := protoutil.Marshal(idp.files)
        if err != nil {
            idp.MoveToDraining(err)
            return nil, idp.DrainHelper()
        }
        sstInfo := tree.NewDBytes(tree.DBytes(bytes))
        fileDatums = rowenc.EncDatumRow{
            rowenc.DatumToEncDatumUnsafe(types.Bytes, sstInfo),
        }
    }

do we still need this?

---

pkg/sql/importer/import_processor.go line 190 at r1 (raw file):

		// When using distributed merge, emit SST metadata via progress metadata
		// before closing the channel. This ensures the metadata is properly
		// checkpointed and survives job retries.

This only works if the processor completes their work successfully. Isn't it possible for retry to begin in the middle of a file (using ResumePos)? If that happens, won't it be missing the SST metadata. Perhaps we should have the SST metadata update tied with the update to ResumePos? That way they stay in lock step.

---

pkg/sql/importer/sst_conversion.go line 1 at r1 (raw file):

// Copyright 2026 The Cockroach Authors.

This helper file is good. I have very similar logic in rowexec/indexbackfiller_sst_sink.go (see collectNewManifests). I wonder if we can combine the two so we have this mapping once? Maybe put in a common package that both importer and index backfill can use.

[mw5h]

Yikes. Bad squash.

@mw5h made 1 comment.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andrew-r-thomas and @ZhouXing19).

[github-actions]

Potential Bug(s) Detected

The three-stage Claude Code analysis has identified potential bug(s) in this PR that may warrant investigation.

Next Steps:
Please review the detailed findings in the workflow run.

Note: When viewing the workflow output, scroll to the bottom to find the Final Analysis Summary.

After you review the findings, please tag the issue as follows:

• If the detected issue is real or was helpful in any way, please tag the issue with O-AI-Review-Real-Issue-Found
• If the detected issue was not helpful in any way, please tag the issue with O-AI-Review-Not-Helpful

[mw5h]

@mw5h made 4 comments.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andrew-r-thomas, @spilchen, and @ZhouXing19).

---

pkg/jobs/jobspb/jobs.proto line 791 at r1 (raw file):

Previously, spilchen wrote…

Can we consider renaming IndexBackfillSSTManifest to something more generic? It's a bit of a code smell to have something in an import struct refer to an backfill struct. I do like the reuse though.

I added a new commit to this branch to handle this.

---

pkg/sql/importer/import_processor.go line 190 at r1 (raw file):

Previously, spilchen wrote…

This only works if the processor completes their work successfully. Isn't it possible for retry to begin in the middle of a file (using ResumePos)? If that happens, won't it be missing the SST metadata. Perhaps we should have the SST metadata update tied with the update to ResumePos? That way they stay in lock step.

Good point. I think this new version resolves this by updating both at once.

---

pkg/sql/importer/import_processor_planning.go line 263 at r1 (raw file):

Previously, spilchen wrote…

Is SST info still being sent? I think it's done in Next().

    // When using distributed merge, the processor will emit the SSTs and their
    // start and end keys.
    var fileDatums rowenc.EncDatumRow
    if idp.files != nil {
        bytes, err := protoutil.Marshal(idp.files)
        if err != nil {
            idp.MoveToDraining(err)
            return nil, idp.DrainHelper()
        }
        sstInfo := tree.NewDBytes(tree.DBytes(bytes))
        fileDatums = rowenc.EncDatumRow{
            rowenc.DatumToEncDatumUnsafe(types.Bytes, sstInfo),
        }
    }

do we still need this?

Nope.

---

pkg/sql/importer/sst_conversion.go line 1 at r1 (raw file):

Previously, spilchen wrote…

This helper file is good. I have very similar logic in rowexec/indexbackfiller_sst_sink.go (see collectNewManifests). I wonder if we can combine the two so we have this mapping once? Maybe put in a common package that both importer and index backfill can use.

Pulled this out into a shared helper.

[github-actions]

Potential Bug(s) Detected

The three-stage Claude Code analysis has identified potential bug(s) in this PR that may warrant investigation.

Next Steps:
Please review the detailed findings in the workflow run.

Note: When viewing the workflow output, scroll to the bottom to find the Final Analysis Summary.

After you review the findings, please tag the issue as follows:

• If the detected issue is real or was helpful in any way, please tag the issue with O-AI-Review-Real-Issue-Found
• If the detected issue was not helpful in any way, please tag the issue with O-AI-Review-Not-Helpful

[github-actions]

Potential Bug(s) Detected

The three-stage Claude Code analysis has identified potential bug(s) in this PR that may warrant investigation.

Next Steps:
Please review the detailed findings in the workflow run.

Note: When viewing the workflow output, scroll to the bottom to find the Final Analysis Summary.

After you review the findings, please tag the issue as follows:

• If the detected issue is real or was helpful in any way, please tag the issue with O-AI-Review-Real-Issue-Found
• If the detected issue was not helpful in any way, please tag the issue with O-AI-Review-Not-Helpful

[spilchen]

Regarding the tests for this, can we consider updating TestCSVImportCanBeResumed so that it runs in distributed merge and non-distributed merge? That would give us a bit of test coverage at least.

@spilchen reviewed 22 files and all commit messages, made 2 comments, and resolved 4 discussions.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andrew-r-thomas, @mw5h, and @ZhouXing19).

---

pkg/sql/importer/import_processor.go line 252 at r6 (raw file):

	}

	// SST metadata is now emitted via the progress channel (see lines 188-199)

nit: we should probably leave out line numbers in the comment as this could get stale in a hurry

[mw5h]

Yeah, that's a good idea. I tried to get an explicit unit test together and just kinda failed. It's hard to wire up just part of the chain.

@mw5h made 2 comments.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @andrew-r-thomas, @spilchen, and @ZhouXing19).

---

pkg/sql/importer/import_processor.go line 252 at r6 (raw file):

Previously, spilchen wrote…

nit: we should probably leave out line numbers in the comment as this could get stale in a hurry

Yup. Claude loves to do this in comments and, in general, over-comments. I thought I had got them all but I was scrambling at the end of the day yesterday and missed a spot.

[spilchen]

thanks for revising.

@spilchen reviewed 12 files and all commit messages, and made 1 comment.
Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @andrew-r-thomas and @ZhouXing19).

[github-actions]

Potential Bug(s) Detected

The three-stage Claude Code analysis has identified potential bug(s) in this PR that may warrant investigation.

Next Steps:
Please review the detailed findings in the workflow run.

Note: When viewing the workflow output, scroll to the bottom to find the Final Analysis Summary.

After you review the findings, please tag the issue as follows:

• If the detected issue is real or was helpful in any way, please tag the issue with O-AI-Review-Real-Issue-Found
• If the detected issue was not helpful in any way, please tag the issue with O-AI-Review-Not-Helpful

[blathers-crl]

Your pull request contains more than 1000 changes. It is strongly encouraged to split big PRs into smaller chunks.

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is dev-inf.}

[github-actions]

Potential Bug(s) Detected

The three-stage Claude Code analysis has identified potential bug(s) in this PR that may warrant investigation.

Next Steps:
Please review the detailed findings in the workflow run.

Note: When viewing the workflow output, scroll to the bottom to find the Final Analysis Summary.

After you review the findings, please tag the issue as follows:

• If the detected issue is real or was helpful in any way, please tag the issue with O-AI-Review-Real-Issue-Found
• If the detected issue was not helpful in any way, please tag the issue with O-AI-Review-Not-Helpful