Merge pull request #155123 from Uzair5162/blathers/backport-release-25.4-154370

release-25.4: sql: log when optimizer estimates for scans are inaccurate

Author: Uzair5162

docs/generated/eventlog.md

@@ -948,6 +948,37 @@ and the cluster setting `sql.log.all_statements.enabled` is set.
 | `BulkJobId` | The job id for bulk job (IMPORT/BACKUP/RESTORE). | no |
 | `StmtPosInTxn` | The statement's index in the transaction, starting at 1. | no |
 
+### `scan_row_count_misestimate`
+
+An event of type `scan_row_count_misestimate` is recorded when the optimizer's row count estimate
+for a logical scan differs significantly from the actual number of rows read,
+and cluster setting `sql.log.scan_row_count_misestimate.enabled` is set.
+
+
+| Field | Description | Sensitive |
+|--|--|--|
+| `TableName` | The fully qualified name of the table being scanned. | no |
+| `IndexName` | The name of the index being scanned. | no |
+| `EstimatedRowCount` | The optimizer's estimated row count for the scan. | no |
+| `ActualRowCount` | The actual number of rows read by all processors performing the scan. | no |
+| `NanosSinceStatsCollected` | Time in nanoseconds that have passed since full stats were collected on the table. | no |
+| `EstimatedStaleness` | Estimated fraction of stale rows in the table based on the time since stats were last collected. | no |
+
+
+#### Common fields
+
+| Field | Description | Sensitive |
+|--|--|--|
+| `Timestamp` | The timestamp of the event. Expressed as nanoseconds since the Unix epoch. | no |
+| `EventType` | The type of the event. | no |
+| `Statement` | A normalized copy of the SQL statement that triggered the event. The statement string contains a mix of sensitive and non-sensitive details (it is redactable). | partially |
+| `Tag` | The statement tag. This is separate from the statement string, since the statement string can contain sensitive information. The tag is guaranteed not to. | no |
+| `User` | The user account that triggered the event. The special usernames `root` and `node` are not considered sensitive. | depends |
+| `DescriptorID` | The primary object descriptor affected by the operation. Set to zero for operations that don't affect descriptors. | no |
+| `ApplicationName` | The application name for the session where the event was emitted. This is included in the event to ease filtering of logging output by application. | no |
+| `PlaceholderValues` | The mapping of SQL placeholders to their values, for prepared statements. | yes |
+| `TxnReadTimestamp` | The current read timestamp of the transaction that triggered the event, if in a transaction. | no |
+
 ## SQL Logical Schema Changes
 
 Events in this category pertain to DDL (Data Definition Language)

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

@@ -321,6 +321,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

@@ -276,6 +276,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

@@ -3359,6 +3359,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"
@@ -2258,6 +2259,7 @@ func (dsp *DistSQLPlanner) createTableReaders(
 			reverse:             n.reverse,
 			parallelize:         n.parallelize,
 			estimatedRowCount:   n.estimatedRowCount,
+			statsCreatedAt:      n.statsCreatedAt,
 			reqOrdering:         n.reqOrdering,
 			finalizeLastStageCb: planCtx.associateWithPlanNode(n),
 		},
@@ -2276,6 +2278,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
 }
@@ -2511,6 +2514,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
 	}