sql/import: pass deep copy of BulkOpSummary to progress

Fixes: #153480

Prior to this change, we passed the `BulkOpSummary` map directly from
`accumulatedBulkSummary` to `prog.Summary` in #152745. This caused a
`concurrent map iteration and map write` panic because the map was being
updated via `metaFn` while simultaneously being marshaled into a protobuf
during `jobs.Update` calls from `FractionProgressed`.

This change creates a deep copy of the map when assigning from
`accumulatedBulkSummary` to `prog.Summary`, eliminating the concurrency
issue by ensuring each goroutine operates on separate map instances.

Release Notes: None

Author: ZhouXing19

pkg/kv/kvpb/api.go

@@ -2133,6 +2133,21 @@ func (b *BulkOpSummary) Add(other BulkOpSummary) {
 	}
 }
 
+// DeepCopy returns a deep copy of the original BulkOpSummary.
+func (b *BulkOpSummary) DeepCopy() BulkOpSummary {
+	cpy := BulkOpSummary{
+		DataSize:    b.DataSize,
+		SSTDataSize: b.SSTDataSize,
+	}
+	if b.EntryCounts != nil {
+		cpy.EntryCounts = make(map[uint64]int64, len(b.EntryCounts))
+		for k, v := range b.EntryCounts {
+			cpy.EntryCounts[k] = v
+		}
+	}
+	return cpy
+}
+
 // MustSetValue is like SetValue, except it resets the enum and panics if the
 // provided value is not a valid variant type.
 func (e *RangeFeedEvent) MustSetValue(value interface{}) {

pkg/kv/kvpb/api.proto

@@ -1935,6 +1935,7 @@ message BulkOpSummary {
   // generation logic is also available in the BulkOpSummaryID helper. It does
   // not take MVCC range tombstones into account.
   map<uint64, int64> entry_counts = 5;
+  // Please update BulkOpSummary.DeepCopy() if new fields are added.
 }
 
 // ExportResponse is the response to an Export() operation.

pkg/sql/importer/import_processor_planning.go

@@ -176,7 +176,7 @@ func distImport(
 			}
 
 			accumulatedBulkSummary.Lock()
-			prog.Summary = accumulatedBulkSummary.BulkOpSummary
+			prog.Summary = accumulatedBulkSummary.BulkOpSummary.DeepCopy()
 			accumulatedBulkSummary.Unlock()
 			return overall / float32(len(from))
 		},