sql/import: fix progress summary loss during concurrent updates

Previously, `accumulatedBulkSummary` was reset on each `updateJobProgress`
call, assuming the job progress update would always succeed. However, if
the update failed due to concurrent job state changes and retried, the
reset buffer would cause progress data loss, resulting in incorrect row
counts in progress reporting.

Fix this by removing the reset logic and directly assigning
`accumulatedBulkSummary` to the progress summary instead of adding to it.

Release note (bug fix): Fixed import progress reporting to show correct
row counts when concurrent job state changes occur.

Author: ZhouXing19

pkg/sql/importer/BUILD.bazel

@@ -187,6 +187,7 @@ go_test(
         "//pkg/testutils",
         "//pkg/testutils/datapathutils",
         "//pkg/testutils/jobutils",
+        "//pkg/testutils/kvclientutils",
         "//pkg/testutils/pgurlutils",
         "//pkg/testutils/serverutils",
         "//pkg/testutils/skip",

pkg/sql/importer/import_processor_planning.go

@@ -47,6 +47,10 @@ var replanFrequency = settings.RegisterDurationSetting(
 	settings.PositiveDuration,
 )
 
+// importProgressDebugName is used to mark the transaction for updating
+// import job progress.
+const importProgressDebugName = `import_progress`
+
 // distImport is used by IMPORT to run a DistSQL flow to ingest data by starting
 // reader processes on many nodes that each read and ingest their assigned files
 // and then send back a summary of what they ingested. The combined summary is
@@ -121,8 +125,7 @@ func distImport(
 	}
 
 	// accumulatedBulkSummary accumulates the BulkOpSummary returned from each
-	// processor in their progress updates. It stores stats about the amount of
-	// data written since the last time we update the job progress.
+	// processor in their progress updates. It is used to update the job progress.
 	accumulatedBulkSummary := struct {
 		syncutil.Mutex
 		kvpb.BulkOpSummary
@@ -158,7 +161,7 @@ func distImport(
 	fractionProgress := make([]uint32, len(from))
 
 	updateJobProgress := func() error {
-		return job.NoTxn().FractionProgressed(ctx, func(
+		return job.DebugNameNoTxn(importProgressDebugName).FractionProgressed(ctx, func(
 			ctx context.Context, details jobspb.ProgressDetails,
 		) float32 {
 			var overall float32
@@ -173,8 +176,7 @@ func distImport(
 			}
 
 			accumulatedBulkSummary.Lock()
-			prog.Summary.Add(accumulatedBulkSummary.BulkOpSummary)
-			accumulatedBulkSummary.Reset()
+			prog.Summary = accumulatedBulkSummary.BulkOpSummary
 			accumulatedBulkSummary.Unlock()
 			return overall / float32(len(from))
 		},

pkg/sql/importer/import_stmt_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobstest"
 	"github.com/cockroachdb/cockroach/pkg/keys"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvclient/kvcoord"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase"
@@ -54,6 +55,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/datapathutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/jobutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/kvclientutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/pgurlutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
@@ -3836,6 +3838,8 @@ func TestImportDefaultWithResume(t *testing.T) {
 	defer TestingSetParallelImporterReaderBatchSize(batchSize)()
 	defer row.TestingSetDatumRowConverterBatchSize(2 * batchSize)()
 
+	ctx := context.Background()
+	filterFunc, verifyFunc := kvclientutils.PrefixTransactionRetryFilter(t, importProgressDebugName, 1)
 	s, db, _ := serverutils.StartServer(t,
 		base.TestServerArgs{
 			// Test hangs when run within a test tenant. More investigation
@@ -3846,11 +3850,13 @@ func TestImportDefaultWithResume(t *testing.T) {
 				DistSQL: &execinfra.TestingKnobs{
 					BulkAdderFlushesEveryBatch: true,
 				},
+				KVClient: &kvcoord.ClientTestingKnobs{
+					TransactionRetryFilter: filterFunc,
+				},
 			},
 			SQLMemoryPoolSize: 1 << 30, // 1 GiB
 		})
 	registry := s.JobRegistry().(*jobs.Registry)
-	ctx := context.Background()
 	defer s.Stopper().Stop(ctx)
 
 	sqlDB := sqlutils.MakeSQLRunner(db)
@@ -3971,6 +3977,16 @@ func TestImportDefaultWithResume(t *testing.T) {
 			sqlDB.QueryRow(t, fmt.Sprintf(`SELECT last_value FROM %s`,
 				test.sequence)).Scan(&seqValOnSuccess)
 			require.Equal(t, seqValOnPause, seqValOnSuccess)
+
+			verifyFunc()
+
+			// Verify final summary counts - should contain exactly the expected number of rows
+			// with no duplication despite multiple pause/resume cycles
+			require.Equal(t, 1, len(js.prog.Summary.EntryCounts))
+
+			for _, entry := range js.prog.Summary.EntryCounts {
+				require.Equal(t, int64(expectedNumRows), entry)
+			}
 		})
 	}
 }