importer: checkpoint SST metadata in distributed merge for retry correctness

[mw5h]

Summary

• Revert entry count tracking that was superseded by the manifest-based approach.
• Rename IndexBackfillSSTManifest to BulkSSTManifest to reflect shared usage across index backfill and import.
• Extract BulkSink interface into pkg/sql/bulksst for shared bulk write abstraction between index backfill and import.
• Refactor index backfill sink to use shared SSTFilesToManifests conversion.
• Move sstIndexBackfillSink from rowexec to bulksst.SSTSink for reuse by the import processor.
• Refactor import processor to use BulkSink directly, replacing ingestHelper/legacyImportBulkAdder/mergeImportBulkAdder. Introduce importProgressTracker with OnFlush-based manifest accumulation and checkpoint SST metadata incrementally in job progress, fixing metadata loss on retry.

Fixes: #161490

[trunk-io]

😎 Merged successfully - details.

[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.}

[cockroach-teamcity]

This change is 

[mw5h]

@mw5h partially reviewed 28 files and all commit messages, and made 7 comments.
Reviewable status: complete! 0 of 0 LGTMs obtained.

---

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

		// All SST manifests are now sent via the OnFlush callback synchronized
		// with flush events, eliminating the need for final manifest emission.

No need to reference what was here before. This sort of information belongs in the commit message.

---

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

	}

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

No need to reference what was here before. This sort of information belongs in the commit message.

---

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

}

type importProgressTracker struct {

There should be concise comments describing the purpose of each entry in the struct and any invariants we need to maintain.

---

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

	unflushedRows     []int64
	unflushedFraction []float32
}

There should probably a unit test of the importProgressTracker.

---

pkg/sql/importer/import_processor_planning.go line 208 at r6 (raw file):

	// manifestBuf accumulates SST file metadata as processors complete, following
	// the index backfiller pattern. It maintains a complete cumulative list of all
	// manifests rather than deltas, eliminating timing windows for orphaned SSTs.

We also talk about the timing window and orphaned SSTs below. Both comments are not necessary. This one seem extraneous.

---

pkg/sql/importer/import_processor_planning.go line 259 at r6 (raw file):

	}

	// Initialize manifest buffer with checkpointed manifests from prior runs,

The comment above says the same thing.

---

pkg/sql/importer/import_processor_test.go line 799 at r6 (raw file):

	require.Equal(t, int64(csv1.numRows)-resumePos, importSummary.Rows)

	// DEBUG: Show what rows are actually in the table

Remove this debugging code.

[mw5h]

@mw5h partially reviewed 28 files and all commit messages.
Reviewable status: complete! 0 of 0 LGTMs obtained.

[spilchen]

nice to see code reuse between import and index backfill

@spilchen partially reviewed 28 files and all commit messages, and made 8 comments.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on andrew-r-thomas, DrewKimball, jeffswenson, msbutler, and mw5h).

---

pkg/sql/importer/import_processor_planning.go line 273 at r12 (raw file):

The buffer deduplicates and

I couldn't see where the buffer deduplicates manifests. It seems like it's a straight up append.

---

pkg/sql/bulksst/sink.go line 1 at r12 (raw file):

// Copyright 2025 The Cockroach Authors.

super nit: 2026

---

pkg/sql/execinfrapb/data.proto line 337 at r12 (raw file):

    // during distributed merge. This metadata is accumulated and persisted in
    // job progress for retry safety.
    repeated jobs.jobspb.BulkSSTManifest sst_metadata = 11 [(gogoproto.customname) = "SSTMetadata", (gogoproto.nullable) = false];

I think the pattern is to use the progress_details field instead of adding this. It's an Any field, so you put whatever struct in it. That's how the index backfiller flows progress back to the coord.

---

pkg/sql/importer/import_processor.go line 427 at r12 (raw file):

		if err := pkSink.Flush(ctx); err != nil {
			if errors.HasType(err, (*kvserverbase.DuplicateKeyError)(nil)) {
				return errors.Wrapf(err, "duplicate key in primary index")

nit: will it always be a duplicate in the primary? For the distributed merge path, there is only one sink, which I guess handles both PK and secondary indexes? If so, the error message should be a bit more generic.

---

pkg/sql/importer/import_processor.go line 635 at r12 (raw file):

		// When the PK adder flushes, everything written has been flushed, so we set
		// pkFlushedRow to writtenRow. Additionally if the indexAdder is empty then we

the old code had an optimization where the PK adder's SetOnFlush callback would also advance the index adder's flushed row position if the index adder's buffer was empty. I think this allowed the resume position to advance with PK flushes when the index adder was idle. Without this, the resume position could stay back, which could mean restarting from an early spot on retry for tables with large PKs and small secondary indexes. Is this a problem?

---

pkg/sql/importer/import_processor_planning.go line 447 at r12 (raw file):

			var distributedMergeFilePrefix string
			if details.UseDistributedMerge {
				distributedMergeFilePrefix = bulkutil.NewDistMergePaths(job.ID()).MapPath()

nit: doesn't the file prefix stay the same for each loop iteration? We could compute this once outside the loop.

---

pkg/sql/importer/import_processor_test.go line 1012 at r12 (raw file):

		// fall back to the original server factory (if provided).
		if es.Provider != cloudpb.ExternalStorageProvider_http {
			if len(originalFactory) != 1 && originalFactory[0] != nil {

was this suppose to be len(originalFactory) == 1? Otherwise, don't we risk an 'index out of range' error here?

[msbutler]

I would strongly suggest storing this field in a transparently sharded job info key, via jobs.WriteChunkedFileToJobInfo api. It seems like this repeated proto field could get arbitrarily large. Big keys in the job system can cause all sorts of availability problems.

At some point, import should probably migrate other fields in this progress proto to info keys, but for now, this seems like a good opportunity to dip import's toes into the info key framework, as you don't need to worry about backwards compatibility here. cc @dt

[mw5h]

Good suggestion. For this PR the manifests are stored in the existing CompletedMapWork repeated field in import progress, matching the index backfiller pattern (which also uses a repeated proto field in SchemaChangeDetails). The list is bounded by the number of SST files produced in a single import attempt.

Filed #164675 to track migrating both import and index backfill to WriteChunkedFileToJobInfo.

[mw5h]

@msbutler let me know if this does or does not meet with your approval.

[msbutler]

i think the issue captures the problem, thanks! i imagine that if you did this work for 26.2, distributed merge wouldn't require a migration, as customers can't run it in 26.1?

[mw5h]

That's correct.

[mw5h]

Sorry to be pushy @msbutler, but am I okay to push this pr with the understanding that we address #164675?

[msbutler]

yeah, sgtm

[mw5h]

@spilchen re: the SetOnFlush optimization where the PK flush advanced the index flushed row when the index adder was empty —

This is now covered. I added IsEmpty() to the BulkSink interface and use it at reporting time in formatProgress: when computing the min flushed position across sinks, if a sink's buffer is empty it uses the current unflushedRows position instead of its (possibly stale) flushedRows. This means an empty sink never holds back the resume position, matching the old behavior without coupling the sinks' OnFlush callbacks together.

The TestImportProgressTracker test now exercises this case explicitly.

[mw5h]

This is RFAL.

[mw5h]

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

---

pkg/sql/execinfrapb/data.proto line 337 at r12 (raw file):

Previously, spilchen wrote…

I think the pattern is to use the progress_details field instead of adding this. It's an Any field, so you put whatever struct in it. That's how the index backfiller flows progress back to the coord.

Done.

---

pkg/sql/importer/import_processor.go line 635 at r12 (raw file):

Previously, spilchen wrote…

the old code had an optimization where the PK adder's SetOnFlush callback would also advance the index adder's flushed row position if the index adder's buffer was empty. I think this allowed the resume position to advance with PK flushes when the index adder was idle. Without this, the resume position could stay back, which could mean restarting from an early spot on retry for tables with large PKs and small secondary indexes. Is this a problem?

Done.

---

pkg/sql/importer/import_processor_planning.go line 273 at r12 (raw file):

Previously, spilchen wrote…

The buffer deduplicates and

I couldn't see where the buffer deduplicates manifests. It seems like it's a straight up append.

Done.

---

pkg/sql/importer/import_processor_test.go line 1012 at r12 (raw file):

Previously, spilchen wrote…

was this suppose to be len(originalFactory) == 1? Otherwise, don't we risk an 'index out of range' error here?

Done.

[spilchen]

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

---

pkg/sql/importer/import_processor.go line 502 at r23 (raw file):

				if err := sink.Add(ctx, kv.Key, kv.Value.RawBytes); err != nil {
					if errors.HasType(err, (*kvserverbase.DuplicateKeyError)(nil)) {
						return errors.Wrapf(err, "duplicate key in %s", errTarget)

nit: will this be correct for the distributed merge path? I think it'll say "duplicate key in primary index" since there is only one sink. But could the duplicate be in the secondary?

[mw5h]

@mw5h partially reviewed 28 files and all commit messages, made 1 comment, and resolved 1 discussion.
Reviewable status: complete! 0 of 0 LGTMs obtained (and 1 stale) (waiting on andrew-r-thomas, DrewKimball, dt, jeffswenson, and spilchen).

---

pkg/sql/bulksst/sink.go line 1 at r12 (raw file):

Previously, spilchen wrote…

super nit: 2026

Done.

[msbutler]

sorry, had not realized you were waiting on me. lgtm!