sql/rowexec: pace index backfill processing with elastic AC

Add admission control pacing to index backfill operations in the
buildIndexEntries loop. Each index entry processing goroutine creates
its own pacer for CPU load management, similar to the existing
pacing in import and bulk SST operations.

Release note: None
Epic: none

Author: dt

pkg/sql/rowexec/BUILD.bazel

@@ -46,6 +46,7 @@ go_library(
         "//pkg/jobs/jobspb",
         "//pkg/keys",
         "//pkg/kv",
+        "//pkg/kv/bulk",
         "//pkg/kv/kvclient/kvstreamer",
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver/kvserverbase",

pkg/sql/rowexec/indexbackfiller.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/kv/bulk"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -78,6 +79,13 @@ var indexBackfillIngestConcurrency = settings.RegisterIntSetting(
 	settings.PositiveInt, /* validateFn */
 )
 
+var indexBackfillElasticCPUControlEnabled = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"bulkio.index_backfill.elastic_control.enabled",
+	"determines whether index backfill operations integrate with elastic CPU control",
+	false, // TODO(dt): enable this by default after more benchmarking.
+)
+
 func newIndexBackfiller(
 	ctx context.Context,
 	flowCtx *execinfra.FlowCtx,
@@ -329,9 +337,16 @@ func (ib *indexBackfiller) ingestIndexEntries(
 	g.GoCtx(func(ctx context.Context) error {
 		defer close(stopProgress)
 
+		// Create a pacer for admission control for index entry processing.
+		pacer := bulk.NewCPUPacer(ctx, ib.flowCtx.Cfg.DB.KV(), indexBackfillElasticCPUControlEnabled)
+		defer pacer.Close()
+
 		var vectorInputEntry rowenc.IndexEntry
 		for indexBatch := range indexEntryCh {
 			for _, indexEntry := range indexBatch.indexEntries {
+				// Pace the admission control before processing each index entry.
+				pacer.Pace(ctx)
+
 				// If there is at least one vector index being written, we need to check to see
 				// if this IndexEntry is going to a vector index and then re-encode it for that
 				// index if so.