sql: use ExecutorConfig in inspect logger

The inspect resumer needs to be able to log issues. The logger can use
the `ExecutorConfig` embedded in the context without changing its
behavior.

Part of: #155472
Epic: CRDB-55075

Release note: None

Author: bghal

pkg/sql/inspect/index_consistency_check.go

@@ -16,6 +16,7 @@ import (
 	"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"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catenumpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo"
@@ -79,7 +80,7 @@ const (
 type indexConsistencyCheck struct {
 	indexConsistencyCheckApplicability
 
-	flowCtx *execinfra.FlowCtx
+	execCfg *sql.ExecutorConfig
 	indexID descpb.IndexID
 	// tableVersion is the descriptor version recorded when the check was planned.
 	// It is used to detect concurrent schema changes for non-AS OF inspections.
@@ -255,7 +256,7 @@ func (c *indexConsistencyCheck) Start(
 		}
 	}
 
-	if indexConsistencyHashEnabled.Get(&c.flowCtx.Cfg.Settings.SV) && len(allColNames) > 0 {
+	if indexConsistencyHashEnabled.Get(&c.execCfg.Settings.SV) && len(allColNames) > 0 {
 		// The hash precheck uses crdb_internal.datums_to_bytes, which depends on
 		// keyside.Encode. Skip if any column type isn’t encodable (i.e. TSQUERY, etc.).
 		if !allColumnsDatumsToBytesCompatible(c.columns) {
@@ -298,8 +299,8 @@ func (c *indexConsistencyCheck) Start(
 	c.lastQueryPlaceholders = queryArgs
 
 	// Execute the query with AS OF SYSTEM TIME embedded in the SQL
-	qos := getInspectQoS(&c.flowCtx.Cfg.Settings.SV)
-	it, err := c.flowCtx.Cfg.DB.Executor().QueryIteratorEx(
+	qos := getInspectQoS(&c.execCfg.Settings.SV)
+	it, err := c.execCfg.DistSQLSrv.DB.Executor().QueryIteratorEx(
 		ctx, "inspect-index-consistency-check", nil, /* txn */
 		sessiondata.InternalExecutorOverride{
 			User:             username.NodeUserName(),
@@ -457,7 +458,7 @@ func (c *indexConsistencyCheck) Rows() uint64 {
 // eligible for consistency checking. If the index is valid, it stores the
 // descriptor and index metadata in the indexConsistencyCheck struct.
 func (c *indexConsistencyCheck) loadCatalogInfo(ctx context.Context) error {
-	return c.flowCtx.Cfg.DB.DescsTxn(ctx, func(ctx context.Context, txn descs.Txn) error {
+	return c.execCfg.DistSQLSrv.DB.DescsTxn(ctx, func(ctx context.Context, txn descs.Txn) error {
 		if !c.asOf.IsEmpty() {
 			if err := txn.KV().SetFixedTimestamp(ctx, c.asOf); err != nil {
 				return err
@@ -813,8 +814,8 @@ func (c *indexConsistencyCheck) computeHashAndRowCount(
 	query := buildIndexHashQuery(c.tableDesc.GetID(), index, columnNames, predicate)
 	queryWithAsOf := fmt.Sprintf("SELECT * FROM (%s) AS OF SYSTEM TIME %s", query, c.asOf.AsOfSystemTime())
 
-	qos := getInspectQoS(&c.flowCtx.Cfg.Settings.SV)
-	row, err := c.flowCtx.Cfg.DB.Executor().QueryRowEx(
+	qos := getInspectQoS(&c.execCfg.Settings.SV)
+	row, err := c.execCfg.DistSQLSrv.DB.Executor().QueryRowEx(
 		ctx, "inspect-index-consistency-hash", nil, /* txn */
 		sessiondata.InternalExecutorOverride{
 			User:             username.NodeUserName(),

pkg/sql/inspect/inspect_processor.go

@@ -234,16 +234,15 @@ func getProcessorConcurrency(flowCtx *execinfra.FlowCtx) int {
 }
 
 // getInspectLogger returns a logger bundle for the inspect processor.
-func getInspectLogger(flowCtx *execinfra.FlowCtx, jobID jobspb.JobID) *inspectLoggerBundle {
-	execCfg := flowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig)
+func getInspectLogger(execCfg *sql.ExecutorConfig, jobID jobspb.JobID) *inspectLoggerBundle {
 	metrics := execCfg.JobRegistry.MetricsStruct().Inspect.(*InspectMetrics)
 
 	loggers := []inspectLogger{
 		&logSink{},
 		&tableSink{
-			db:    flowCtx.Cfg.DB,
+			db:    execCfg.DistSQLSrv.DB,
 			jobID: jobID,
-			sv:    &flowCtx.Cfg.Settings.SV,
+			sv:    &execCfg.Settings.SV,
 		},
 		&metricsLogger{
 			issuesFoundCtr: metrics.IssuesFound,
@@ -415,15 +414,14 @@ func (p *inspectProcessor) pushProgressMeta(
 func newInspectProcessor(
 	ctx context.Context, flowCtx *execinfra.FlowCtx, processorID int32, spec execinfrapb.InspectSpec,
 ) (execinfra.Processor, error) {
-	checkFactories, err := buildInspectCheckFactories(flowCtx, spec)
+	execCfg := flowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig)
+
+	checkFactories, err := buildInspectCheckFactories(execCfg, spec)
 	if err != nil {
 		return nil, err
 	}
-	var spansProcessedCtr *metric.Counter
-	if execCfg, ok := flowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig); ok {
-		inspectMetrics := execCfg.JobRegistry.MetricsStruct().Inspect.(*InspectMetrics)
-		spansProcessedCtr = inspectMetrics.SpansProcessed
-	}
+	inspectMetrics := execCfg.JobRegistry.MetricsStruct().Inspect.(*InspectMetrics)
+	spansProcessedCtr := inspectMetrics.SpansProcessed
 
 	return &inspectProcessor{
 		spec:              spec,
@@ -432,7 +430,7 @@ func newInspectProcessor(
 		checkFactories:    checkFactories,
 		cfg:               flowCtx.Cfg,
 		spanSrc:           newSliceSpanSource(spec.Spans),
-		loggerBundle:      getInspectLogger(flowCtx, spec.JobID),
+		loggerBundle:      getInspectLogger(execCfg, spec.JobID),
 		concurrency:       getProcessorConcurrency(flowCtx),
 		clock:             flowCtx.Cfg.DB.KV().Clock(),
 		spansProcessedCtr: spansProcessedCtr,
@@ -446,7 +444,7 @@ func newInspectProcessor(
 // This indirection ensures that each check instance is freshly created per span,
 // avoiding shared state across concurrent workers.
 func buildInspectCheckFactories(
-	flowCtx *execinfra.FlowCtx, spec execinfrapb.InspectSpec,
+	execCfg *sql.ExecutorConfig, spec execinfrapb.InspectSpec,
 ) ([]inspectCheckFactory, error) {
 	checkFactories := make([]inspectCheckFactory, 0, len(spec.InspectDetails.Checks))
 	for _, specCheck := range spec.InspectDetails.Checks {
@@ -460,7 +458,7 @@ func buildInspectCheckFactories(
 					indexConsistencyCheckApplicability: indexConsistencyCheckApplicability{
 						tableID: tableID,
 					},
-					flowCtx:      flowCtx,
+					execCfg:      execCfg,
 					indexID:      indexID,
 					tableVersion: tableVersion,
 					asOf:         asOf,