Merge #152745 #152985

152745: sql/import: fix progress summary loss during concurrent updates r=ZhouXing19 a=ZhouXing19

fixes #152543
fixes #152519

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.


152985: roachtest: use LeaderLeases for schemachange/bulkingest tests r=spilchen a=spilchen

Changed from MetamorphicLeases to LeaderLeases in
makeSchemaChangeBulkIngestTest to address potential test stability issues identified in issue #152857.

Closes #152857

Release note: none

Epic: None

Co-authored-by: ZhouXing19 <zhouxing@uchicago.edu>
Co-authored-by: Matt Spilchen <matt.spilchen@cockroachlabs.com>

Author: 3 people

pkg/cmd/roachtest/tests/schemachange.go

@@ -359,7 +359,7 @@ func makeSchemaChangeBulkIngestTest(
 		Cluster:          r.MakeClusterSpec(numNodes, spec.WorkloadNode(), spec.SSD(4)),
 		CompatibleClouds: registry.AllExceptAWS,
 		Suites:           registry.Suites(registry.Nightly),
-		Leases:           registry.MetamorphicLeases,
+		Leases:           registry.LeaderLeases,
 		Timeout:          length * 2,
 		PostProcessPerfMetrics: func(test string, histogram *roachtestutil.HistogramMetric) (roachtestutil.AggregatedPerfMetrics, error) {
 			// The histogram tracks the total elapsed time for the CREATE INDEX operation.

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)
+			}
 		})
 	}
 }