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, #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.

Author: Uzair5162

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

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

@@ -278,6 +278,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/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
 }

pkg/sql/colfetcher/colbatch_scan.go

@@ -257,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
 }

pkg/sql/conn_executor_exec.go

@@ -3352,6 +3352,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_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
 	}
 

pkg/sql/distsql_running.go

@@ -28,6 +28,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/rpc/rpcbase"
 	"github.com/cockroachdb/cockroach/pkg/server/telemetry"
 	"github.com/cockroachdb/cockroach/pkg/settings"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/colflow"
 	"github.com/cockroachdb/cockroach/pkg/sql/distsql"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
@@ -47,10 +49,12 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondatapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlerrors"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqltelemetry"
+	"github.com/cockroachdb/cockroach/pkg/sql/stats"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/buildutil"
 	"github.com/cockroachdb/cockroach/pkg/util/errorutil/unimplemented"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/humanizeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/interval"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
@@ -1114,6 +1118,10 @@ type DistSQLReceiver struct {
 
 	stats topLevelQueryStats
 
+	// scanStageEstimateMap maps stage IDs for logical scans to their
+	// corresponding scanStageEstimate.
+	scanStageEstimateMap map[int32]scanStageEstimate
+
 	// isTenantExplainAnalyze is used to indicate that network egress should be
 	// collected in order to estimate RU consumption for a tenant that is running
 	// a query with EXPLAIN ANALYZE.
@@ -1342,12 +1350,45 @@ func (c *CallbackResultWriter) Err() error {
 	return c.err
 }
 
+// scanStageEstimate holds the optimizer's row count estimate and table
+// statistics metadata for a logical scan. It accumulates actual rows read
+// across all distributed processors in the stage and is used for logging when
+// the estimate is significantly off.
+type scanStageEstimate struct {
+	estimatedRowCount uint64
+	statsCreatedAt    time.Time
+	tableID           descpb.ID
+	tableName         string
+	indexName         string
+
+	rowsRead uint64
+}
+
+var misestimateLogLimiter = log.Every(10 * time.Second)
+
+func (s *scanStageEstimate) logMisestimate(ctx context.Context, refresher *stats.Refresher) {
+	var suffix string
+	if !s.statsCreatedAt.IsZero() {
+		timeSinceStats := timeutil.Since(s.statsCreatedAt)
+		suffix = fmt.Sprintf("; table stats collected %s ago",
+			humanizeutil.LongDuration(timeSinceStats))
+		staleness, err := refresher.EstimateStaleness(ctx, s.tableID)
+		if err == nil {
+			suffix += fmt.Sprintf(" (estimated %.0f%% stale)", staleness*100)
+		}
+	}
+	log.Dev.Warningf(ctx, "inaccurate estimate for scan on table %q (index %q): estimated=%d actual=%d%s",
+		s.tableName, s.indexName, s.estimatedRowCount, s.rowsRead, suffix)
+}
+
 var _ execinfra.RowReceiver = &DistSQLReceiver{}
 var _ execinfra.BatchReceiver = &DistSQLReceiver{}
 
 var receiverSyncPool = sync.Pool{
 	New: func() interface{} {
-		return &DistSQLReceiver{}
+		return &DistSQLReceiver{
+			scanStageEstimateMap: make(map[int32]scanStageEstimate),
+		}
 	},
 }
 
@@ -1391,13 +1432,14 @@ func MakeDistSQLReceiver(
 		batchWriter:  batchWriter,
 		// At the time of writing, there is only one concurrent goroutine that
 		// might send at most one error.
-		concurrentErrorCh: make(chan error, 1),
-		cleanup:           cleanup,
-		rangeCache:        rangeCache,
-		txn:               txn,
-		clockUpdater:      clockUpdater,
-		stmtType:          stmtType,
-		tracing:           tracing,
+		concurrentErrorCh:    make(chan error, 1),
+		cleanup:              cleanup,
+		rangeCache:           rangeCache,
+		txn:                  txn,
+		clockUpdater:         clockUpdater,
+		stmtType:             stmtType,
+		tracing:              tracing,
+		scanStageEstimateMap: r.scanStageEstimateMap,
 	}
 	return r
 }
@@ -1416,6 +1458,9 @@ func (r *DistSQLReceiver) resetForLocalRerun(stats topLevelQueryStats) {
 	r.closed = false
 	r.stats = stats
 	r.egressCounter = nil
+	for k := range r.scanStageEstimateMap {
+		delete(r.scanStageEstimateMap, k)
+	}
 	if r.progressAtomic != nil {
 		atomic.StoreUint64(r.progressAtomic, math.Float64bits(0))
 	}
@@ -1424,7 +1469,12 @@ func (r *DistSQLReceiver) resetForLocalRerun(stats topLevelQueryStats) {
 // Release releases this DistSQLReceiver back to the pool.
 func (r *DistSQLReceiver) Release() {
 	r.cleanup()
-	*r = DistSQLReceiver{}
+	for k := range r.scanStageEstimateMap {
+		delete(r.scanStageEstimateMap, k)
+	}
+	*r = DistSQLReceiver{
+		scanStageEstimateMap: r.scanStageEstimateMap,
+	}
 	receiverSyncPool.Put(r)
 }
 
@@ -1433,14 +1483,15 @@ func (r *DistSQLReceiver) Release() {
 func (r *DistSQLReceiver) clone() *DistSQLReceiver {
 	ret := receiverSyncPool.Get().(*DistSQLReceiver)
 	*ret = DistSQLReceiver{
-		ctx:               r.ctx,
-		concurrentErrorCh: make(chan error, 1),
-		cleanup:           func() {},
-		rangeCache:        r.rangeCache,
-		txn:               r.txn,
-		clockUpdater:      r.clockUpdater,
-		stmtType:          tree.Rows,
-		tracing:           r.tracing,
+		ctx:                  r.ctx,
+		concurrentErrorCh:    make(chan error, 1),
+		cleanup:              func() {},
+		rangeCache:           r.rangeCache,
+		txn:                  r.txn,
+		clockUpdater:         r.clockUpdater,
+		stmtType:             tree.Rows,
+		tracing:              r.tracing,
+		scanStageEstimateMap: ret.scanStageEstimateMap,
 	}
 	return ret
 }
@@ -1544,6 +1595,12 @@ func (r *DistSQLReceiver) pushMeta(meta *execinfrapb.ProducerMetadata) execinfra
 		r.stats.bytesRead += meta.Metrics.BytesRead
 		r.stats.rowsRead += meta.Metrics.RowsRead
 		r.stats.rowsWritten += meta.Metrics.RowsWritten
+
+		if sm, ok := r.scanStageEstimateMap[meta.Metrics.StageID]; ok {
+			sm.rowsRead += uint64(meta.Metrics.RowsRead)
+			r.scanStageEstimateMap[meta.Metrics.StageID] = sm
+		}
+
 		if r.progressAtomic != nil && r.expectedRowsRead != 0 {
 			progress := float64(r.stats.rowsRead) / float64(r.expectedRowsRead)
 			atomic.StoreUint64(r.progressAtomic, math.Float64bits(progress))
@@ -1772,6 +1829,64 @@ func (r *DistSQLReceiver) ProducerDone() {
 	r.closed = true
 }
 
+func (r *DistSQLReceiver) makeScanEstimates(physPlan *PhysicalPlan, st *cluster.Settings) {
+	if !execinfra.LogScanRowCountMisestimate.Get(&st.SV) {
+		return
+	}
+
+	for _, p := range physPlan.Processors {
+		if p.Spec.Core.TableReader == nil {
+			continue
+		}
+		stageID := p.Spec.StageID
+		if _, exists := r.scanStageEstimateMap[stageID]; !exists {
+			r.scanStageEstimateMap[stageID] = scanStageEstimate{
+				estimatedRowCount: p.Spec.EstimatedRowCount,
+				statsCreatedAt:    p.Spec.StatsCreatedAt,
+				tableID:           p.Spec.Core.TableReader.FetchSpec.TableID,
+				tableName:         p.Spec.Core.TableReader.FetchSpec.TableName,
+				indexName:         p.Spec.Core.TableReader.FetchSpec.IndexName,
+			}
+		}
+	}
+}
+
+func (r *DistSQLReceiver) maybeLogMisestimates(ctx context.Context, planner *planner) {
+	if !execinfra.LogScanRowCountMisestimate.Get(&planner.ExecCfg().Settings.SV) {
+		return
+	}
+
+	checkedLimiter := false
+	for _, s := range r.scanStageEstimateMap {
+		actualRowCount := s.rowsRead
+		estimatedRowCount := s.estimatedRowCount
+		if estimatedRowCount == 0 {
+			continue
+		}
+
+		// Note: This is the same inaccuracy criteria as in explain/emit.go.
+		const inaccurateFactor = 2
+		const inaccurateAdditive = 100
+		inaccurateEstimate := actualRowCount*inaccurateFactor+inaccurateAdditive < estimatedRowCount ||
+			estimatedRowCount*inaccurateFactor+inaccurateAdditive < actualRowCount
+		if !inaccurateEstimate {
+			continue
+		}
+		if isSystemTable, err := planner.IsSystemTable(ctx, int64(s.tableID)); err != nil || isSystemTable {
+			continue
+		}
+
+		// Log all or none of the misestimated scans in the query.
+		if !checkedLimiter {
+			if !misestimateLogLimiter.ShouldLog() {
+				return
+			}
+			checkedLimiter = true
+		}
+		s.logMisestimate(ctx, planner.ExecCfg().StatsRefresher)
+	}
+}
+
 // getFinishedSetupFn returns a function to be passed into
 // DistSQLPlanner.PlanAndRun or DistSQLPlanner.Run when running an "outer" plan
 // that might create "inner" plans (e.g. apply join iterations). The returned
@@ -1959,6 +2074,7 @@ func (dsp *DistSQLPlanner) planAndRunSubquery(
 	// receiver, and use it and serialize the results of the subquery. The type
 	// of the results stored in the container depends on the type of the subquery.
 	subqueryRecv := recv.clone()
+	subqueryRecv.makeScanEstimates(subqueryPhysPlan, dsp.st)
 	defer subqueryRecv.Release()
 	defer recv.stats.add(&subqueryRecv.stats)
 	var typs []*types.T
@@ -2084,6 +2200,7 @@ func (dsp *DistSQLPlanner) planAndRunSubquery(
 		// with many duplicate elements.
 		subqueryResultMemAcc.Shrink(ctx, alreadyAccountedFor-actualSize)
 	}
+	subqueryRecv.maybeLogMisestimates(ctx, planner)
 	return nil
 }
 
@@ -2131,6 +2248,7 @@ func (dsp *DistSQLPlanner) PlanAndRun(
 	} else {
 		finalizePlanWithRowCount(ctx, planCtx, physPlan, planCtx.planner.curPlan.mainRowCount)
 		recv.expectedRowsRead = int64(physPlan.TotalEstimatedScannedRows)
+		recv.makeScanEstimates(physPlan, dsp.st)
 		dsp.Run(ctx, planCtx, txn, physPlan, recv, &extEvalCtx.Context, finishedSetupFn)
 	}
 	if planCtx.isLocal {
@@ -2203,6 +2321,7 @@ func (dsp *DistSQLPlanner) PlanAndRun(
 		}
 		finalizePlanWithRowCount(ctx, localPlanCtx, localPhysPlan, localPlanCtx.planner.curPlan.mainRowCount)
 		recv.expectedRowsRead = int64(localPhysPlan.TotalEstimatedScannedRows)
+		recv.makeScanEstimates(localPhysPlan, dsp.st)
 		// We already called finishedSetupFn in the previous call to Run, since we
 		// only got here if we got a distributed error, not an error during setup.
 		dsp.Run(ctx, localPlanCtx, txn, localPhysPlan, recv, &extEvalCtx.Context, nil /* finishedSetupFn */)

pkg/sql/distsql_spec_exec_factory.go

@@ -362,6 +362,7 @@ func (e *distSQLSpecExecFactory) ConstructScan(
 			reverse:           params.Reverse,
 			parallelize:       params.Parallelize,
 			estimatedRowCount: params.EstimatedRowCount,
+			statsCreatedAt:    params.StatsCreatedAt,
 			reqOrdering:       ReqOrdering(reqOrdering),
 		},
 	)

pkg/sql/execinfra/utils.go

@@ -60,3 +60,15 @@ var IncludeRUEstimateInExplainAnalyze = settings.RegisterBoolSetting(
 	true,
 	settings.WithName("sql.explain_analyze.include_ru_estimation.enabled"),
 )
+
+// LogScanRowCountMisestimate controls whether we log a warning when the
+// actual row count observed by a scan differs significantly from the optimizer
+// estimate.
+var LogScanRowCountMisestimate = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"sql.log.scan_row_count_misestimate.enabled",
+	"when set to true, log a warning when a scan's actual row count differs "+
+		"significantly from the optimizer's estimate",
+	false,
+	settings.WithPublic,
+)

pkg/sql/execinfrapb/data.proto

@@ -339,6 +339,13 @@ message RemoteProducerMetadata {
     optional int64 rows_read = 2 [(gogoproto.nullable) = false];
     // Total number of rows modified while executing a statement.
     optional int64 rows_written = 3 [(gogoproto.nullable) = false];
+    // Stage identifier that produced these metrics. This is used to aggregate
+    // row counts across distributed table reader processors for misestimate
+    // logging. Note that other disk-reading processors (index and lookup joins,
+    // inverted joins, zigzag joins) do not set this field. This is safe because
+    // we enumerate physical plan stages starting at 1, so the default value of
+    // 0 isn't conflated with any actual plan stage.
+    optional int32 stage_id = 4 [(gogoproto.nullable) = false, (gogoproto.customname) = "StageID"];
   }
   oneof value {
     RangeInfos range_info = 1;