sql/importer: pace import processing with elastic AC

Add admission control pacing to IMPORT operations in both the
file-to-rows and rows-to-datums pipelines. Each goroutine creates
its own pacer for CPU load management.

Epic: none
Release note: None

Author: dt

pkg/sql/importer/BUILD.bazel

@@ -35,6 +35,7 @@ go_library(
         "//pkg/jobs/jobspb",
         "//pkg/jobs/jobsprofiler",
         "//pkg/kv",
+        "//pkg/kv/bulk",
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver/kvserverbase",
         "//pkg/roachpb",

pkg/sql/importer/read_import_base.go

@@ -21,9 +21,11 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/cloud"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster"
 	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/cockroach/pkg/kv/bulk"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
+	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/tabledesc"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/typedesc"
@@ -44,6 +46,13 @@ import (
 	"github.com/cockroachdb/errors"
 )
 
+var importElasticCPUControlEnabled = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"bulkio.import.elastic_control.enabled",
+	"determines whether import operations integrate with elastic CPU control",
+	false, // TODO(dt): enable this by default after more benchmarking.
+)
+
 func runImport(
 	ctx context.Context,
 	flowCtx *execinfra.FlowCtx,
@@ -567,9 +576,15 @@ func runParallelImport(
 		var span *tracing.Span
 		ctx, span = tracing.ChildSpan(ctx, "import-file-to-rows")
 		defer span.Finish()
+
+		// Create a pacer for admission control for the producer.
+		pacer := bulk.NewCPUPacer(ctx, importCtx.db, importElasticCPUControlEnabled)
+		defer pacer.Close()
+
 		var numSkipped int64
 		var count int64
 		for producer.Scan() {
+			pacer.Pace(ctx)
 			// Skip rows if needed.
 			count++
 			if count <= fileCtx.skip {
@@ -660,6 +675,10 @@ func (p *parallelImporter) importWorker(
 	fileCtx *importFileContext,
 	minEmitted []int64,
 ) error {
+	// Create a pacer for admission control for this worker.
+	pacer := bulk.NewCPUPacer(ctx, importCtx.db, importElasticCPUControlEnabled)
+	defer pacer.Close()
+
 	conv, err := makeDatumConverter(ctx, importCtx, fileCtx, importCtx.db)
 	if err != nil {
 		return err
@@ -680,6 +699,8 @@ func (p *parallelImporter) importWorker(
 		conv.KvBatch.Progress = batch.progress
 		for batchIdx, record := range batch.data {
 			rowNum = batch.startPos + int64(batchIdx)
+			// Pace the admission control before processing each row.
+			pacer.Pace(ctx)
 			if err := consumer.FillDatums(ctx, record, rowNum, conv); err != nil {
 				if err = handleCorruptRow(ctx, fileCtx, err); err != nil {
 					return err

pkg/sql/importer/read_import_workload.go

@@ -17,6 +17,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/cloud"
 	"github.com/cockroachdb/cockroach/pkg/col/coldata"
 	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/cockroach/pkg/kv/bulk"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
@@ -245,6 +246,12 @@ func NewWorkloadKVConverter(
 func (w *WorkloadKVConverter) Worker(
 	ctx context.Context, evalCtx *eval.Context, semaCtx *tree.SemaContext,
 ) error {
+	// Workload needs to pace itself explicitly since it manages its own workers
+	// and loops rather than using the "runParallelImport" helper which the other
+	// formats use and which has pacing built-in.
+	pacer := bulk.NewCPUPacer(ctx, w.db, importElasticCPUControlEnabled)
+	defer pacer.Close()
+
 	conv, err := row.NewDatumRowConverter(
 		ctx, semaCtx, w.tableDesc, nil, /* targetColNames */
 		evalCtx, w.kvCh, nil /* seqChunkProvider */, nil /* metrics */, w.db,
@@ -268,6 +275,7 @@ func (w *WorkloadKVConverter) Worker(
 		a = a.Truncate()
 		w.rows.FillBatch(batchIdx, cb, &a)
 		for rowIdx, numRows := 0, cb.Length(); rowIdx < numRows; rowIdx++ {
+			pacer.Pace(ctx)
 			for colIdx, col := range cb.ColVecs() {
 				// TODO(dan): This does a type switch once per-datum. Reduce this to
 				// a one-time switch per column.