Merge #154370

154370: sql: log when optimizer estimates for scans are inaccurate r=Uzair5162 a=Uzair5162

#### sql: estimate table statistic staleness in stats.Refresher
This commit adds an `EstimateStaleness()` method to the table statistic
`Refresher`, which estimates the current fraction of stale rows in a
given table with the formula:

`cur_fraction_stale = (time_since_last_refresh /
avg_time_between_refreshes) * target_fraction_stale_rows`

Although this isn’t used anywhere yet, it will be useful for logging
when scans are misestimated (see #153748).

Part of: #153748, #153873

Release note: None

#### sql: plumb StageID into table reader processors

This commit plumbs `StageID` from the general `ProcessorSpec` into
table reader processors. Specifically, `StageID` is plumbed into
`tableReader` for the row-based flow and `colBatchScanBase` (via
`ColBatchScan` and `ColBatchDirectScan`) for the vectorized flow.
Although `stageID` isn't used in these processors yet, it will
be useful for aggregating row counts from metrics across distributed
table readers for misestimate logging.

Part of: #153748

Release note: None

#### sql: log when optimizer estimates for scans are inaccurate
This commit logs a warning on the gateway node when the estimated row
count for a logical scan is inaccurate. The `DistSQLReceiver` on the
gateway node now maintains the row count estimate and metadata for each
logical scan stage. Table reader metrics are extended to include their
StageID, which we use to aggregate the emitted row counts from all table
readers processors corresponding to the same plan stage at the receiver.

An estimate is considered inaccurate if it is off by at least a factor
of 2 and a fixed offset of 100, matching the logic in the warning from
`EXPLAIN ANALYZE`. The log message includes the table and index being
scanned, the estimated and actual row counts, the time since the last
table stats collection, and the table's estimated staleness.

This log is gated behind a new cluster setting,
`sql.log.scan_row_count_misestimate.enabled` (default off). Logging only
happens for user tables and is rate limited to log misestimates from at
most 1 query every 10 seconds.

Fixes: #153748
Fixes: #153873

Release note (sql change): Added a default-off cluster setting
(`sql.log.scan_row_count_misestimate.enabled`) that enables logging a
warning on the gateway node when optimizer estimates for scans are
inaccurate. The log message includes the table and index being scanned,
the estimated and actual row counts, the time since the last table stats
collection, and the table's estimated staleness.

Co-authored-by: Uzair Ahmad <uzair.ahmad@cockroachlabs.com>

Author: craig[bot]

docs/generated/settings/settings-for-tenants.txt

@@ -325,6 +325,7 @@ sql.insights.execution_insights_capacity	integer	1000	the size of the per-node s
 sql.insights.high_retry_count.threshold	integer	10	the number of retries a slow statement must have undergone for its high retry count to be highlighted as a potential problem	application
 sql.insights.latency_threshold	duration	100ms	amount of time after which an executing statement is considered slow. Use 0 to disable.	application
 sql.log.redact_names.enabled	boolean	false	if set, schema object identifers are redacted in SQL statements that appear in event logs	application
+sql.log.scan_row_count_misestimate.enabled	boolean	false	when set to true, log a warning when a scan's actual row count differs significantly from the optimizer's estimate	application
 sql.log.slow_query.experimental_full_table_scans.enabled	boolean	false	when set to true, statements that perform a full table/index scan will be logged to the slow query log even if they do not meet the latency threshold. Must have the slow query log enabled for this setting to have any effect.	application
 sql.log.slow_query.internal_queries.enabled	boolean	false	when set to true, internal queries which exceed the slow query log threshold are logged to a separate log. Must have the slow query log enabled for this setting to have any effect.	application
 sql.log.slow_query.latency_threshold	duration	0s	when set to non-zero, log statements whose service latency exceeds the threshold to a secondary logger on each node	application

docs/generated/settings/settings.html

@@ -280,6 +280,7 @@
 <tr><td><div id="setting-sql-insights-high-retry-count-threshold" class="anchored"><code>sql.insights.high_retry_count.threshold</code></div></td><td>integer</td><td><code>10</code></td><td>the number of retries a slow statement must have undergone for its high retry count to be highlighted as a potential problem</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-insights-latency-threshold" class="anchored"><code>sql.insights.latency_threshold</code></div></td><td>duration</td><td><code>100ms</code></td><td>amount of time after which an executing statement is considered slow. Use 0 to disable.</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-log-redact-names-enabled" class="anchored"><code>sql.log.redact_names.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>if set, schema object identifers are redacted in SQL statements that appear in event logs</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
+<tr><td><div id="setting-sql-log-scan-row-count-misestimate-enabled" class="anchored"><code>sql.log.scan_row_count_misestimate.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>when set to true, log a warning when a scan&#39;s actual row count differs significantly from the optimizer&#39;s estimate</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-log-slow-query-experimental-full-table-scans-enabled" class="anchored"><code>sql.log.slow_query.experimental_full_table_scans.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>when set to true, statements that perform a full table/index scan will be logged to the slow query log even if they do not meet the latency threshold. Must have the slow query log enabled for this setting to have any effect.</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-log-slow-query-internal-queries-enabled" class="anchored"><code>sql.log.slow_query.internal_queries.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>when set to true, internal queries which exceed the slow query log threshold are logged to a separate log. Must have the slow query log enabled for this setting to have any effect.</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-log-slow-query-latency-threshold" class="anchored"><code>sql.log.slow_query.latency_threshold</code></div></td><td>duration</td><td><code>0s</code></td><td>when set to non-zero, log statements whose service latency exceeds the threshold to a secondary logger on each node</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>

pkg/sql/colexec/colbuilder/execplan.go

@@ -566,6 +566,7 @@ func (r opResult) createAndWrapRowSource(
 	core *execinfrapb.ProcessorCoreUnion,
 	post *execinfrapb.PostProcessSpec,
 	processorID int32,
+	stageID int32,
 	factory coldata.ColumnFactory,
 	causeToWrap error,
 ) error {
@@ -592,7 +593,7 @@ func (r opResult) createAndWrapRowSource(
 			// here because when wrapping the processor, the materializer will
 			// be its output, and it will be set up in wrapRowSources.
 			proc, err := args.ProcessorConstructor(
-				ctx, flowCtx, processorID, core, post, inputs, args.LocalProcessors,
+				ctx, flowCtx, processorID, stageID, core, post, inputs, args.LocalProcessors,
 			)
 			if err != nil {
 				return nil, err
@@ -819,7 +820,7 @@ func NewColOperator(
 		post = &newPosts[1]
 		err = result.createAndWrapRowSource(
 			ctx, flowCtx, args, inputs, inputTypes, core,
-			wrappingPost, spec.ProcessorID, factory, err,
+			wrappingPost, spec.ProcessorID, spec.StageID, factory, err,
 		)
 	} else {
 		switch {
@@ -969,7 +970,8 @@ func NewColOperator(
 				if canUseDirectScan() {
 					scanOp, resultTypes, err = colfetcher.NewColBatchDirectScan(
 						ctx, colmem.NewAllocator(ctx, accounts[0], factory), accounts[1],
-						flowCtx, spec.ProcessorID, core.TableReader, post, args.TypeResolver,
+						flowCtx, spec.ProcessorID, spec.StageID, core.TableReader, post,
+						args.TypeResolver,
 					)
 					if err != nil {
 						return r, err
@@ -979,7 +981,8 @@ func NewColOperator(
 			if scanOp == nil {
 				scanOp, resultTypes, err = colfetcher.NewColBatchScan(
 					ctx, colmem.NewAllocator(ctx, accounts[0], factory), accounts[1],
-					flowCtx, spec.ProcessorID, core.TableReader, post, estimatedRowCount, args.TypeResolver,
+					flowCtx, spec.ProcessorID, spec.StageID, core.TableReader, post,
+					estimatedRowCount, args.TypeResolver,
 				)
 				if err != nil {
 					return r, err
@@ -1027,7 +1030,7 @@ func NewColOperator(
 			result.ColumnTypes = spec.Input[0].ColumnTypes
 			result.Root = inputs[0].Root
 			if err := result.planAndMaybeWrapFilter(
-				ctx, flowCtx, args, spec.ProcessorID, core.Filterer.Filter, factory,
+				ctx, flowCtx, args, spec.ProcessorID, spec.StageID, core.Filterer.Filter, factory,
 			); err != nil {
 				return r, err
 			}
@@ -1309,7 +1312,7 @@ func NewColOperator(
 
 			if !core.HashJoiner.OnExpr.Empty() && core.HashJoiner.Type == descpb.InnerJoin {
 				if err = result.planAndMaybeWrapFilter(
-					ctx, flowCtx, args, spec.ProcessorID, core.HashJoiner.OnExpr, factory,
+					ctx, flowCtx, args, spec.ProcessorID, spec.StageID, core.HashJoiner.OnExpr, factory,
 				); err != nil {
 					return r, err
 				}
@@ -1354,7 +1357,7 @@ func NewColOperator(
 
 			if onExpr != nil {
 				if err = result.planAndMaybeWrapFilter(
-					ctx, flowCtx, args, spec.ProcessorID, *onExpr, factory,
+					ctx, flowCtx, args, spec.ProcessorID, spec.StageID, *onExpr, factory,
 				); err != nil {
 					return r, err
 				}
@@ -1797,7 +1800,8 @@ func NewColOperator(
 	}
 	err = ppr.planPostProcessSpec(ctx, flowCtx, args, post, factory, &r.Releasables, args.Spec.EstimatedRowCount)
 	if err != nil {
-		err = result.wrapPostProcessSpec(ctx, flowCtx, args, post, spec.ProcessorID, factory, err)
+		err = result.wrapPostProcessSpec(ctx, flowCtx, args, post,
+			spec.ProcessorID, spec.StageID, factory, err)
 	} else {
 		// The result can be updated with the post process result.
 		r.Root = ppr.Op
@@ -1845,7 +1849,8 @@ func NewColOperator(
 				post.RenderExprs[i].LocalExpr = tree.NewTypedOrdinalReference(i, args.Spec.ResultTypes[i])
 			}
 		}
-		if err = result.wrapPostProcessSpec(ctx, flowCtx, args, post, spec.ProcessorID, factory, errWrappedCast); err != nil {
+		if err = result.wrapPostProcessSpec(ctx, flowCtx, args, post,
+			spec.ProcessorID, spec.StageID, factory, errWrappedCast); err != nil {
 			return r, err
 		}
 	} else if numMismatchedTypes > 0 {
@@ -1909,6 +1914,7 @@ func (r opResult) planAndMaybeWrapFilter(
 	flowCtx *execinfra.FlowCtx,
 	args *colexecargs.NewColOperatorArgs,
 	processorID int32,
+	stageID int32,
 	filter execinfrapb.Expression,
 	factory coldata.ColumnFactory,
 ) error {
@@ -1928,7 +1934,7 @@ func (r opResult) planAndMaybeWrapFilter(
 		return r.createAndWrapRowSource(
 			ctx, flowCtx, args, []colexecargs.OpWithMetaInfo{inputToMaterializer},
 			[][]*types.T{r.ColumnTypes}, filtererCore, &execinfrapb.PostProcessSpec{},
-			processorID, factory, err,
+			processorID, stageID, factory, err,
 		)
 	}
 	return nil
@@ -1945,6 +1951,7 @@ func (r opResult) wrapPostProcessSpec(
 	args *colexecargs.NewColOperatorArgs,
 	post *execinfrapb.PostProcessSpec,
 	processorID int32,
+	stageID int32,
 	factory coldata.ColumnFactory,
 	causeToWrap error,
 ) error {
@@ -1956,7 +1963,8 @@ func (r opResult) wrapPostProcessSpec(
 	// createAndWrapRowSource updates r.ColumnTypes accordingly.
 	return r.createAndWrapRowSource(
 		ctx, flowCtx, args, []colexecargs.OpWithMetaInfo{inputToMaterializer},
-		[][]*types.T{r.ColumnTypes}, noopCore, post, processorID, factory, causeToWrap,
+		[][]*types.T{r.ColumnTypes}, noopCore, post, processorID, stageID, factory,
+		causeToWrap,
 	)
 }
 

pkg/sql/colexec/values_test.go

@@ -150,7 +150,8 @@ func BenchmarkValues(b *testing.B) {
 					var core execinfrapb.ProcessorCoreUnion
 					core.Values = spec
 					proc, err := rowexec.NewProcessor(
-						ctx, &flowCtx, 0 /* processorID */, &core, &post, nil /* inputs */, nil, /* localProcessors */
+						ctx, &flowCtx, 0 /* processorID */, 0 /* stageID */, &core, &post,
+						nil /* inputs */, nil, /* localProcessors */
 					)
 					if err != nil {
 						b.Fatal(err)

pkg/sql/colfetcher/colbatch_direct_scan.go

@@ -127,6 +127,7 @@ func (s *ColBatchDirectScan) DrainMeta() []execinfrapb.ProducerMetadata {
 	meta.Metrics = execinfrapb.GetMetricsMeta()
 	meta.Metrics.BytesRead = s.GetBytesRead()
 	meta.Metrics.RowsRead = s.GetRowsRead()
+	meta.Metrics.StageID = s.stageID
 	trailingMeta = append(trailingMeta, *meta)
 	return trailingMeta
 }
@@ -181,12 +182,13 @@ func NewColBatchDirectScan(
 	kvFetcherMemAcc *mon.BoundAccount,
 	flowCtx *execinfra.FlowCtx,
 	processorID int32,
+	stageID int32,
 	spec *execinfrapb.TableReaderSpec,
 	post *execinfrapb.PostProcessSpec,
 	typeResolver *descs.DistSQLTypeResolver,
 ) (*ColBatchDirectScan, []*types.T, error) {
 	base, bsHeader, tableArgs, err := newColBatchScanBase(
-		ctx, kvFetcherMemAcc, flowCtx, processorID, spec, post, typeResolver,
+		ctx, kvFetcherMemAcc, flowCtx, processorID, stageID, spec, post, typeResolver,
 	)
 	if err != nil {
 		return nil, nil, err

pkg/sql/colfetcher/colbatch_scan.go

@@ -38,6 +38,7 @@ type colBatchScanBase struct {
 
 	flowCtx                *execinfra.FlowCtx
 	processorID            int32
+	stageID                int32
 	limitHint              rowinfra.RowLimit
 	batchBytesLimit        rowinfra.BytesLimit
 	parallelize            bool
@@ -122,6 +123,7 @@ func newColBatchScanBase(
 	kvFetcherMemAcc *mon.BoundAccount,
 	flowCtx *execinfra.FlowCtx,
 	processorID int32,
+	stageID int32,
 	spec *execinfrapb.TableReaderSpec,
 	post *execinfrapb.PostProcessSpec,
 	typeResolver *descs.DistSQLTypeResolver,
@@ -186,6 +188,7 @@ func newColBatchScanBase(
 		SpansWithCopy:          s.SpansWithCopy,
 		flowCtx:                flowCtx,
 		processorID:            processorID,
+		stageID:                stageID,
 		limitHint:              limitHint,
 		batchBytesLimit:        batchBytesLimit,
 		parallelize:            spec.Parallelize,
@@ -254,6 +257,7 @@ func (s *ColBatchScan) DrainMeta() []execinfrapb.ProducerMetadata {
 	meta.Metrics = execinfrapb.GetMetricsMeta()
 	meta.Metrics.BytesRead = s.GetBytesRead()
 	meta.Metrics.RowsRead = s.GetRowsRead()
+	meta.Metrics.StageID = s.stageID
 	trailingMeta = append(trailingMeta, *meta)
 	return trailingMeta
 }
@@ -312,13 +316,14 @@ func NewColBatchScan(
 	kvFetcherMemAcc *mon.BoundAccount,
 	flowCtx *execinfra.FlowCtx,
 	processorID int32,
+	stageID int32,
 	spec *execinfrapb.TableReaderSpec,
 	post *execinfrapb.PostProcessSpec,
 	estimatedRowCount uint64,
 	typeResolver *descs.DistSQLTypeResolver,
 ) (*ColBatchScan, []*types.T, error) {
 	base, bsHeader, tableArgs, err := newColBatchScanBase(
-		ctx, kvFetcherMemAcc, flowCtx, processorID, spec, post, typeResolver,
+		ctx, kvFetcherMemAcc, flowCtx, processorID, stageID, spec, post, typeResolver,
 	)
 	if err != nil {
 		return nil, nil, err

pkg/sql/conn_executor_exec.go

@@ -3360,6 +3360,10 @@ func (ex *connExecutor) execWithDistSQLEngine(
 		}
 		err = ex.server.cfg.DistSQLPlanner.PlanAndRunAll(ctx, evalCtx, planCtx, planner, recv, evalCtxFactory)
 	}
+
+	if err == nil && res.Err() == nil {
+		recv.maybeLogMisestimates(ctx, planner)
+	}
 	return recv.stats, err
 }
 

pkg/sql/distsql/columnar_utils_test.go

@@ -115,7 +115,7 @@ func verifyColOperator(t *testing.T, args verifyColOperatorArgs) error {
 	}
 
 	proc, err := rowexec.NewProcessor(
-		ctx, flowCtx, 0, &args.pspec.Core, &args.pspec.Post, inputsProc, nil,
+		ctx, flowCtx, 0, 0, &args.pspec.Core, &args.pspec.Post, inputsProc, nil,
 	)
 	if err != nil {
 		return err

pkg/sql/distsql_physical_planner.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"reflect"
 	"sort"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/cloud"
@@ -2245,6 +2246,7 @@ func (dsp *DistSQLPlanner) createTableReaders(
 			reverse:             n.reverse,
 			parallelize:         n.parallelize,
 			estimatedRowCount:   n.estimatedRowCount,
+			statsCreatedAt:      n.statsCreatedAt,
 			reqOrdering:         n.reqOrdering,
 			finalizeLastStageCb: planCtx.associateWithPlanNode(n),
 		},
@@ -2263,6 +2265,7 @@ type tableReaderPlanningInfo struct {
 	reverse             bool
 	parallelize         bool
 	estimatedRowCount   uint64
+	statsCreatedAt      time.Time
 	reqOrdering         ReqOrdering
 	finalizeLastStageCb func(*physicalplan.PhysicalPlan) // will be nil in the spec factory
 }
@@ -2498,6 +2501,7 @@ func (dsp *DistSQLPlanner) planTableReaders(
 
 		corePlacement[i].SQLInstanceID = sp.SQLInstanceID
 		corePlacement[i].EstimatedRowCount = info.estimatedRowCount
+		corePlacement[i].StatsCreatedAt = info.statsCreatedAt
 		corePlacement[i].Core.TableReader = tr
 	}