Merge #42118 #42121

42118: Revert "importccl: small refactor of unnecessary loop" r=dt a=danhhz

This reverts commit 2e24bf9.

This refactor had an unintentional change of behavior. Before there were
numCPU workers all collaborating to produce the rows between BatchBegin
and BatchEnd. After there are numCPU _each_ producing the rows between
BatchBegin and BatchEnd. This means we're importing numCPU duplicates of
every kv.

I am incredibly surprised that no tests broke, this points to a big hole
in our test coverage. I think it was "working" because we require
workload implementations to be totally deterministic and AddSSTable has
to be resilient to replaying an exact request?

Release note: none

42121: sql: fix assignment of dropped table job status r=lucy-zhang a=lucy-zhang

We were mistakenly mutating a copy of a slice element in a loop while updating
the status for a job for a table drop, causing an incorrect job status to be
reported.

Fixes #39347.

Release note (bug fix): A bug was fixed that caused jobs for dropping tables to
report an inaccurate status.

Co-authored-by: Daniel Harrison <[email protected]>
Co-authored-by: Lucy Zhang <[email protected]>

pkg/ccl/importccl/read_import_workload.go

@@ -102,6 +102,8 @@ func (w *workloadReader) readFiles(
 	_ roachpb.IOFileFormat,
 	_ cloud.ExternalStorageFactory,
 ) error {
+
+	wcs := make([]*WorkloadKVConverter, 0, len(dataFiles))
 	for fileID, fileName := range dataFiles {
 		file, err := url.Parse(fileName)
 		if err != nil {
@@ -138,10 +140,15 @@ func (w *workloadReader) readFiles(
 			return errors.Wrapf(err, `unknown table %s for generator %s`, conf.Table, meta.Name)
 		}
 
+		wc := NewWorkloadKVConverter(
+			fileID, w.table, t.InitialRows, int(conf.BatchBegin), int(conf.BatchEnd), w.kvCh)
+		wcs = append(wcs, wc)
+	}
+
+	for _, wc := range wcs {
 		if err := ctxgroup.GroupWorkers(ctx, runtime.NumCPU(), func(ctx context.Context) error {
 			evalCtx := w.evalCtx.Copy()
-			return NewWorkloadKVConverter(
-				fileID, w.table, t.InitialRows, int(conf.BatchBegin), int(conf.BatchEnd), w.kvCh).Worker(ctx, evalCtx)
+			return wc.Worker(ctx, evalCtx)
 		}); err != nil {
 			return err
 		}

pkg/sql/table.go

@@ -798,8 +798,8 @@ func (p *planner) createDropTablesJob(
 	if !drainNames {
 		detailStatus = jobspb.Status_WAIT_FOR_GC_INTERVAL
 	}
-	for _, droppedDetail := range droppedDetails {
-		droppedDetail.Status = detailStatus
+	for i := range droppedDetails {
+		droppedDetails[i].Status = detailStatus
 	}
 
 	runningStatus := RunningStatusDrainingNames