Merge #143857 #143859 #143865 #143867 #143868

143857: explain: show "execution time" on EXPLAIN ANALYZE output r=yuzefovich a=yuzefovich

We've been tracking "execution time" statistic for most operators (except for vectorized KV-reading ones which get separate "KV time" stat) for many years now, but for some reason we only reported it on the DistSQL diagram.  This commit fixes that oversight so that the stat is now included on `EXPLAIN ANALYZE` output when available.

Fixes: #143443.

Release note (sql change): New statistic `execution time` is now reported on EXPLAIN ANALYZE output for most operators. It was previously only available on the DistSQL diagrams (included in `EXPLAIN ANALYZE (DISTSQL)` output).

143859: explain: clarify "network usage" line of EXPLAIN ANALYZE r=yuzefovich a=yuzefovich

In "network statistics" (messages and bytes sent) we only include the network traffic between DistSQL components completely ignoring any traffic from KV layer to SQL layer. In order to clarify this a bit, this commit renames `network usage` top-level information to include "DistSQL" prefix.

Epic: None

Release note (sql change): `network usage` top-level statistic in EXPLAIN ANALYZE output has been renamed to `DistSQL network usage` to clarify that it only tracks network traffic between DistSQL components and completely ignores traffic from KV layer to DistSQL components.

143865: kv/bulk: close bulk-adder-monitor on error r=yuzefovich a=yuzefovich

Previously, we would have left the `bulk-adder-monitor` not closed if we hit an error in `MakeBulkAdder`, and this is now fixed. (The only error possible is actually not enough memory budget to reserve the buffer size.)

Fixes: #143862.

Release note: None

143867: explain: harden RU reporting for read-only queries r=yuzefovich a=yuzefovich

In a recent change 365dfa4 we made it possible to no longer report RU estimate in `planNodeToRowSource` if it wraps a non-mutation planNode. That change was a de-flaking fix that was later super-seded by 66065df, so we can revert the former. I think it makes the code more bullet-proof even though AFAICT we report consumed RUs for only for mutation planNodes.

Also this commit makes some minor fixes to `TestEstimateQueryRUConsumption` which is skipped unconditionally.

Epic: None
Release note: None

143868: memo: check `insert_fast_path` for staleness r=yuzefovich a=yuzefovich

Since we consult the value of `insert_fast_path` session variable in `CanUseUniqueChecksForInsertFastPath` (which is called during the exploration), we need to check the value for memo staleness, and this has been missing.

Epic: None

Release note: None

Co-authored-by: Yahor Yuzefovich <[email protected]>

Author: craig[bot]

docs/generated/eventlog.md

@@ -3222,10 +3222,10 @@ Fields in this struct should be updated in sync with apps_stats.proto.
 
 | Field | Description | Sensitive |
 |--|--|--|
-| `NetworkBytes` | NetworkBytes collects the number of bytes sent over the network. | no |
+| `NetworkBytes` | NetworkBytes collects the number of bytes sent over the network by DistSQL components. | no |
 | `MaxMemUsage` | MaxMemUsage collects the maximum memory usage that occurred on a node. | no |
 | `ContentionTime` | ContentionTime collects the time in seconds statements in the transaction spent contending. | no |
-| `NetworkMessages` | NetworkMessages collects the number of messages that were sent over the network. | no |
+| `NetworkMessages` | NetworkMessages collects the number of messages that were sent over the network by DistSQL components. | no |
 | `MaxDiskUsage` | MaxDiskUsage collects the maximum temporary disk usage that occurred. This is set in cases where a query had to spill to disk, e.g. when performing a large sort where not all of the tuples fit in memory. | no |
 | `CPUSQLNanos` | CPUSQLNanos collects the CPU time spent executing SQL operations in nanoseconds. Currently, it is only collected for statements without mutations that have a vectorized plan. | no |
 | `MVCCIteratorStats` | Internal storage iteration statistics. | yes |
@@ -3273,13 +3273,13 @@ contains common SQL event/execution details.
 | `ZigZagJoinCount` | The number of zig zag joins in the query plan. | no |
 | `ContentionNanos` | The duration of time in nanoseconds that the query experienced contention. | no |
 | `Regions` | The regions of the nodes where SQL processors ran. | no |
-| `NetworkBytesSent` | The number of network bytes sent by nodes for this query. | no |
+| `NetworkBytesSent` | The number of network bytes by DistSQL components. | no |
 | `MaxMemUsage` | The maximum amount of memory usage by nodes for this query. | no |
 | `MaxDiskUsage` | The maximum amount of disk usage by nodes for this query. | no |
 | `KVBytesRead` | The number of bytes read at the KV layer for this query. | no |
 | `KVPairsRead` | The number of key-value pairs read at the KV layer for this query. | no |
 | `KVRowsRead` | The number of rows read at the KV layer for this query. | no |
-| `NetworkMessages` | The number of network messages sent by nodes for this query. | no |
+| `NetworkMessages` | The number of network messages sent by nodes for this query by DistSQL components. | no |
 | `IndexRecommendations` | Generated index recommendations for this query. | no |
 | `ScanCount` | The number of scans in the query plan. | no |
 | `ScanWithStatsCount` | The number of scans using statistics (including forecasted statistics) in the query plan. | no |

pkg/ccl/multitenantccl/tenantcostclient/query_ru_estimate_test.go

@@ -72,7 +72,6 @@ func TestEstimateQueryRUConsumption(t *testing.T) {
 	tenantID := serverutils.TestTenantID()
 	tenant1, tenantDB1 := serverutils.StartTenant(t, s, base.TestTenantArgs{
 		TenantID: tenantID,
-		Settings: st,
 		TestingKnobs: base.TestingKnobs{
 			SQLEvalContext: &eval.TestingKnobs{
 				// We disable the randomization of some batch sizes because with
@@ -196,7 +195,7 @@ func TestEstimateQueryRUConsumption(t *testing.T) {
 	const deltaFraction = 0.25
 	allowedDelta := tenantMeasuredRUs * deltaFraction
 	require.InDeltaf(t, tenantMeasuredRUs, tenantEstimatedRUs, allowedDelta,
-		"estimated RUs (%d) were not within %f RUs of the expected value (%f)",
+		"estimated RUs (%.2f) were not within %.2f RUs of the expected value (%.2f)",
 		tenantEstimatedRUs,
 		allowedDelta,
 		tenantMeasuredRUs,

pkg/kv/bulk/buffering_adder.go

@@ -79,6 +79,9 @@ var _ kvserverbase.BulkAdder = &BufferingAdder{}
 // passed to add into SSTs that are then ingested. rangeCache if set is
 // consulted to avoid generating an SST that will span a range boundary and thus
 // encounter an error and need to be split and retired to be applied.
+//
+// The BulkAdder takes ownership of the memory monitor which must be non-nil. In
+// case of an error, the monitor will be stopped.
 func MakeBulkAdder(
 	ctx context.Context,
 	db *kv.DB,
@@ -88,7 +91,15 @@ func MakeBulkAdder(
 	opts kvserverbase.BulkAdderOptions,
 	bulkMon *mon.BytesMonitor,
 	sendLimiter limit.ConcurrentRequestLimiter,
-) (*BufferingAdder, error) {
+) (_ *BufferingAdder, retErr error) {
+	if bulkMon == nil {
+		return nil, errors.New("bulkMon must be non-nil")
+	}
+	defer func() {
+		if retErr != nil {
+			bulkMon.Stop(ctx)
+		}
+	}()
 	if opts.MinBufferSize == 0 {
 		opts.MinBufferSize = 32 << 20
 	}
@@ -166,11 +177,8 @@ func (b *BufferingAdder) Close(ctx context.Context) {
 		b.sink.mu.Unlock()
 	}
 	b.sink.Close(ctx)
-
-	if b.bulkMon != nil {
-		b.memAcc.Close(ctx)
-		b.bulkMon.Stop(ctx)
-	}
+	b.memAcc.Close(ctx)
+	b.bulkMon.Stop(ctx)
 }
 
 // Add adds a key to the buffer and checks if it needs to flush.

pkg/sql/appstatspb/app_stats.proto

@@ -328,7 +328,9 @@ message ExecStats {
   // run.
   optional int64 count = 1 [(gogoproto.nullable) = false];
 
-  // NetworkBytes collects the number of bytes sent over the network.
+  // NetworkBytes collects the number of bytes sent over the network by DistSQL
+  // components.
+  // TODO(yuzefovich): rename this all the way up to the DB Console.
   optional NumericStat network_bytes = 2 [(gogoproto.nullable) = false];
 
   // MaxMemUsage collects the maximum memory usage that occurred on a node.
@@ -338,7 +340,8 @@ message ExecStats {
   optional NumericStat contention_time = 4 [(gogoproto.nullable) = false];
 
   // NetworkMessages collects the number of messages that were sent over the
-  // network.
+  // network by DistSQL components.
+  // TODO(yuzefovich): rename this all the way up to the DB Console.
   optional NumericStat network_messages = 5 [(gogoproto.nullable) = false];
 
   // MaxDiskUsage collects the maximum temporary disk usage that occurred. This

pkg/sql/exec_log.go

@@ -396,15 +396,15 @@ func (p *planner) maybeLogStatementInternal(
 			SQLInstanceIDs:           queryLevelStats.SQLInstanceIDs,
 			KVNodeIDs:                queryLevelStats.KVNodeIDs,
 			UsedFollowerRead:         queryLevelStats.UsedFollowerRead,
-			NetworkBytesSent:         queryLevelStats.NetworkBytesSent,
+			NetworkBytesSent:         queryLevelStats.DistSQLNetworkBytesSent,
 			MaxMemUsage:              queryLevelStats.MaxMemUsage,
 			MaxDiskUsage:             queryLevelStats.MaxDiskUsage,
 			KVBytesRead:              queryLevelStats.KVBytesRead,
 			KVPairsRead:              queryLevelStats.KVPairsRead,
 			KVRowsRead:               queryLevelStats.KVRowsRead,
 			KvTimeNanos:              queryLevelStats.KVTime.Nanoseconds(),
 			KvGrpcCalls:              queryLevelStats.KVBatchRequestsIssued,
-			NetworkMessages:          queryLevelStats.NetworkMessages,
+			NetworkMessages:          queryLevelStats.DistSQLNetworkMessages,
 			CpuTimeNanos:             queryLevelStats.CPUTime.Nanoseconds(),
 			IndexRecommendations:     indexRecs,
 			// TODO(mgartner): Use a slice of struct{uint64, uint64} instead of
@@ -498,10 +498,10 @@ func (p *planner) logTransaction(
 
 	if txnStats.CollectedExecStats {
 		sampledTxn.SampledExecStats = &eventpb.SampledExecStats{
-			NetworkBytes:    txnStats.ExecStats.NetworkBytesSent,
+			NetworkBytes:    txnStats.ExecStats.DistSQLNetworkBytesSent,
 			MaxMemUsage:     txnStats.ExecStats.MaxMemUsage,
 			ContentionTime:  int64(txnStats.ExecStats.ContentionTime.Seconds()),
-			NetworkMessages: txnStats.ExecStats.NetworkMessages,
+			NetworkMessages: txnStats.ExecStats.DistSQLNetworkMessages,
 			MaxDiskUsage:    txnStats.ExecStats.MaxDiskUsage,
 			CPUSQLNanos:     txnStats.ExecStats.CPUTime.Nanoseconds(),
 			MVCCIteratorStats: eventpb.MVCCIteratorStats{

pkg/sql/execstats/traceanalyzer.go

@@ -95,7 +95,7 @@ func NewFlowsMetadata(flows map[base.SQLInstanceID]*execinfrapb.FlowSpec) *Flows
 // given traces and flow metadata.
 // NOTE: When adding fields to this struct, be sure to update Accumulate.
 type QueryLevelStats struct {
-	NetworkBytesSent                   int64
+	DistSQLNetworkBytesSent            int64
 	MaxMemUsage                        int64
 	MaxDiskUsage                       int64
 	KVBytesRead                        int64
@@ -116,7 +116,7 @@ type QueryLevelStats struct {
 	MvccRangeKeyCount                  int64
 	MvccRangeKeyContainedPoints        int64
 	MvccRangeKeySkippedPoints          int64
-	NetworkMessages                    int64
+	DistSQLNetworkMessages             int64
 	ContentionTime                     time.Duration
 	LockWaitTime                       time.Duration
 	LatchWaitTime                      time.Duration
@@ -156,7 +156,7 @@ func MakeQueryLevelStatsWithErr(stats QueryLevelStats, err error) QueryLevelStat
 
 // Accumulate accumulates other's stats into the receiver.
 func (s *QueryLevelStats) Accumulate(other QueryLevelStats) {
-	s.NetworkBytesSent += other.NetworkBytesSent
+	s.DistSQLNetworkBytesSent += other.DistSQLNetworkBytesSent
 	if other.MaxMemUsage > s.MaxMemUsage {
 		s.MaxMemUsage = other.MaxMemUsage
 	}
@@ -181,7 +181,7 @@ func (s *QueryLevelStats) Accumulate(other QueryLevelStats) {
 	s.MvccRangeKeyCount += other.MvccRangeKeyCount
 	s.MvccRangeKeyContainedPoints += other.MvccRangeKeyContainedPoints
 	s.MvccRangeKeySkippedPoints += other.MvccRangeKeySkippedPoints
-	s.NetworkMessages += other.NetworkMessages
+	s.DistSQLNetworkMessages += other.DistSQLNetworkMessages
 	s.ContentionTime += other.ContentionTime
 	s.LockWaitTime += other.LockWaitTime
 	s.LatchWaitTime += other.LatchWaitTime
@@ -295,8 +295,8 @@ func (a *TraceAnalyzer) ProcessStats() {
 		if stats.stats == nil {
 			continue
 		}
-		s.NetworkBytesSent += getNetworkBytesFromComponentStats(stats.stats)
-		s.NetworkMessages += getNumNetworkMessagesFromComponentsStats(stats.stats)
+		s.DistSQLNetworkBytesSent += getNetworkBytesFromComponentStats(stats.stats)
+		s.DistSQLNetworkMessages += getNumDistSQLNetworkMessagesFromComponentsStats(stats.stats)
 	}
 
 	// Process flowStats.
@@ -350,7 +350,7 @@ func getNetworkBytesFromComponentStats(v *execinfrapb.ComponentStats) int64 {
 	return 0
 }
 
-func getNumNetworkMessagesFromComponentsStats(v *execinfrapb.ComponentStats) int64 {
+func getNumDistSQLNetworkMessagesFromComponentsStats(v *execinfrapb.ComponentStats) int64 {
 	// We expect exactly one of MessagesReceived and MessagesSent to be set. It
 	// may seem like we are double-counting everything (from both the send and
 	// the receive side) but in practice only one side of each stream presents

pkg/sql/execstats/traceanalyzer_test.go

@@ -161,7 +161,7 @@ func TestTraceAnalyzer(t *testing.T) {
 			// The stats don't count the actual bytes, but they are a synthetic value
 			// based on the number of tuples. In this test 21 tuples flow over the
 			// network.
-			require.Equal(t, int64(21*8), queryLevelStats.NetworkBytesSent)
+			require.Equal(t, int64(21*8), queryLevelStats.DistSQLNetworkBytesSent)
 
 			// Soft check that MaxMemUsage is set to a non-zero value. The actual
 			// value differs between test runs due to metamorphic randomization.
@@ -176,7 +176,7 @@ func TestTraceAnalyzer(t *testing.T) {
 
 			// For tests, network messages is a synthetic value based on the number of
 			// network tuples. In this test 21 tuples flow over the network.
-			require.Equal(t, int64(21/2), queryLevelStats.NetworkMessages)
+			require.Equal(t, int64(21/2), queryLevelStats.DistSQLNetworkMessages)
 		})
 	}
 }
@@ -233,14 +233,14 @@ func TestTraceAnalyzerProcessStats(t *testing.T) {
 func TestQueryLevelStatsAccumulate(t *testing.T) {
 	aEvent := kvpb.ContentionEvent{Duration: 7 * time.Second}
 	a := execstats.QueryLevelStats{
-		NetworkBytesSent:                   1,
+		DistSQLNetworkBytesSent:            1,
 		MaxMemUsage:                        2,
 		KVBytesRead:                        3,
 		KVPairsRead:                        4,
 		KVRowsRead:                         4,
 		KVBatchRequestsIssued:              4,
 		KVTime:                             5 * time.Second,
-		NetworkMessages:                    6,
+		DistSQLNetworkMessages:             6,
 		ContentionTime:                     7 * time.Second,
 		LockWaitTime:                       4 * time.Second,
 		LatchWaitTime:                      3 * time.Second,
@@ -269,14 +269,14 @@ func TestQueryLevelStatsAccumulate(t *testing.T) {
 	}
 	bEvent := kvpb.ContentionEvent{Duration: 14 * time.Second}
 	b := execstats.QueryLevelStats{
-		NetworkBytesSent:                   8,
+		DistSQLNetworkBytesSent:            8,
 		MaxMemUsage:                        9,
 		KVBytesRead:                        10,
 		KVPairsRead:                        11,
 		KVRowsRead:                         11,
 		KVBatchRequestsIssued:              11,
 		KVTime:                             12 * time.Second,
-		NetworkMessages:                    13,
+		DistSQLNetworkMessages:             13,
 		ContentionTime:                     14 * time.Second,
 		LockWaitTime:                       10 * time.Second,
 		LatchWaitTime:                      4 * time.Second,
@@ -304,14 +304,14 @@ func TestQueryLevelStatsAccumulate(t *testing.T) {
 		ClientTime:                         2 * time.Second,
 	}
 	expected := execstats.QueryLevelStats{
-		NetworkBytesSent:                   9,
+		DistSQLNetworkBytesSent:            9,
 		MaxMemUsage:                        9,
 		KVBytesRead:                        13,
 		KVPairsRead:                        15,
 		KVRowsRead:                         15,
 		KVBatchRequestsIssued:              15,
 		KVTime:                             17 * time.Second,
-		NetworkMessages:                    19,
+		DistSQLNetworkMessages:             19,
 		ContentionTime:                     21 * time.Second,
 		LockWaitTime:                       14 * time.Second,
 		LatchWaitTime:                      7 * time.Second,

pkg/sql/instrumentation.go

@@ -839,7 +839,7 @@ func (ih *instrumentationHelper) emitExplainAnalyzePlanToOutputBuilder(
 		}
 
 		ob.AddMaxMemUsage(queryStats.MaxMemUsage)
-		ob.AddNetworkStats(queryStats.NetworkMessages, queryStats.NetworkBytesSent)
+		ob.AddDistSQLNetworkStats(queryStats.DistSQLNetworkMessages, queryStats.DistSQLNetworkBytesSent)
 		ob.AddMaxDiskUsage(queryStats.MaxDiskUsage)
 		if len(queryStats.Regions) > 0 {
 			ob.AddRegionsStats(queryStats.Regions)
@@ -1060,6 +1060,10 @@ func (m execNodeTraceMetadata) annotateExplain(
 					nodeStats.InternalStepCount.MaybeAdd(stats.KV.NumInternalSteps)
 					nodeStats.SeekCount.MaybeAdd(stats.KV.NumInterfaceSeeks)
 					nodeStats.InternalSeekCount.MaybeAdd(stats.KV.NumInternalSeeks)
+					// If multiple physical plan stages correspond to a single
+					// operator, we want to aggregate the execution time across
+					// all of them.
+					nodeStats.ExecTime.MaybeAdd(stats.Exec.ExecTime)
 					nodeStats.MaxAllocatedMem.MaybeAdd(stats.Exec.MaxAllocatedMem)
 					nodeStats.MaxAllocatedDisk.MaybeAdd(stats.Exec.MaxAllocatedDisk)
 					if noMutations && !makeDeterministic {

pkg/sql/opt/exec/execbuilder/testdata/call

@@ -139,7 +139,7 @@ distribution: <hidden>
 vectorized: <hidden>
 plan type: custom
 maximum memory usage: <hidden>
-network usage: <hidden>
+DistSQL network usage: <hidden>
 regions: <hidden>
 isolation level: serializable
 priority: normal
@@ -149,6 +149,7 @@ quality of service: regular
   sql nodes: <hidden>
   regions: <hidden>
   actual row count: 0
+  execution time: 0µs
   estimated row count: 0
   procedure: foo(3)
 
@@ -161,7 +162,7 @@ distribution: <hidden>
 vectorized: <hidden>
 plan type: custom
 maximum memory usage: <hidden>
-network usage: <hidden>
+DistSQL network usage: <hidden>
 regions: <hidden>
 isolation level: serializable
 priority: normal
@@ -171,6 +172,7 @@ quality of service: regular
   sql nodes: <hidden>
   regions: <hidden>
   actual row count: 0
+  execution time: 0µs
   estimated row count: 0
   procedure: foo(3)
 ·

pkg/sql/opt/exec/execbuilder/testdata/call_plpgsql

@@ -145,7 +145,7 @@ distribution: <hidden>
 vectorized: <hidden>
 plan type: custom
 maximum memory usage: <hidden>
-network usage: <hidden>
+DistSQL network usage: <hidden>
 regions: <hidden>
 isolation level: serializable
 priority: normal
@@ -155,6 +155,7 @@ quality of service: regular
   sql nodes: <hidden>
   regions: <hidden>
   actual row count: 0
+  execution time: 0µs
   estimated row count: 0
   procedure: foo(3)
 
@@ -167,7 +168,7 @@ distribution: <hidden>
 vectorized: <hidden>
 plan type: custom
 maximum memory usage: <hidden>
-network usage: <hidden>
+DistSQL network usage: <hidden>
 regions: <hidden>
 isolation level: serializable
 priority: normal
@@ -177,6 +178,7 @@ quality of service: regular
   sql nodes: <hidden>
   regions: <hidden>
   actual row count: 0
+  execution time: 0µs
   estimated row count: 0
   procedure: foo(3)
 ·