From 7fa4d379cd129c34ee113137389e0a8fdab32563 Mon Sep 17 00:00:00 2001
From: Alfonso Subiotto Marques <alfonso@cockroachlabs.com>
Date: Thu, 28 Jan 2021 12:39:50 +0100
Subject: [PATCH 1/4] sql: prescribe stats collection up front

Previously, each processor would make the decision of whether stats should be
collected by checking whether recording was enabled on a span obtained through
the context. This commit centralizes that decision by moving it into the conn
executor's instrumentation helper. Processors now observe a CollectStats bool
set on the FlowCtx in order to determine whether they should collect stats or
not.

Release note: None
---
 pkg/sql/colflow/vectorized_flow.go   |   2 +-
 pkg/sql/conn_executor_exec.go        |   1 +
 pkg/sql/distsql/server.go            |   4 +-
 pkg/sql/distsql_physical_planner.go  |   3 +
 pkg/sql/distsql_running.go           |   8 +-
 pkg/sql/execinfra/BUILD.bazel        |   1 +
 pkg/sql/execinfra/flow_context.go    |   3 +
 pkg/sql/execinfra/stats.go           |  25 ++++
 pkg/sql/execinfrapb/api.pb.go        | 171 ++++++++++++++++-----------
 pkg/sql/execinfrapb/api.proto        |   6 +
 pkg/sql/instrumentation.go           |  14 ++-
 pkg/sql/rowexec/aggregator.go        |   3 +-
 pkg/sql/rowexec/countrows.go         |   3 +-
 pkg/sql/rowexec/distinct.go          |   3 +-
 pkg/sql/rowexec/filterer.go          |   3 +-
 pkg/sql/rowexec/hashjoiner.go        |   3 +-
 pkg/sql/rowexec/inverted_filterer.go |   3 +-
 pkg/sql/rowexec/inverted_joiner.go   |   3 +-
 pkg/sql/rowexec/joinreader.go        |   3 +-
 pkg/sql/rowexec/mergejoiner.go       |   3 +-
 pkg/sql/rowexec/noop.go              |   3 +-
 pkg/sql/rowexec/ordinality.go        |   3 +-
 pkg/sql/rowexec/sorter.go            |   3 +-
 pkg/sql/rowexec/tablereader.go       |   3 +-
 pkg/sql/rowexec/tablereader_test.go  |   3 +
 pkg/sql/rowexec/windower.go          |   3 +-
 26 files changed, 183 insertions(+), 100 deletions(-)
 create mode 100644 pkg/sql/execinfra/stats.go

diff --git a/pkg/sql/colflow/vectorized_flow.go b/pkg/sql/colflow/vectorized_flow.go
index 8f448a09231c..2590a15671ad 100644
--- a/pkg/sql/colflow/vectorized_flow.go
+++ b/pkg/sql/colflow/vectorized_flow.go
@@ -177,7 +177,7 @@ func (f *vectorizedFlow) Setup(
 		log.Infof(ctx, "setting up vectorize flow %s", f.ID.Short())
 	}
 	recordingStats := false
-	if sp := tracing.SpanFromContext(ctx); sp != nil && sp.IsVerbose() {
+	if execinfra.ShouldCollectStats(ctx, &f.FlowCtx) {
 		recordingStats = true
 	}
 	helper := newVectorizedFlowCreatorHelper(f.FlowBase)
diff --git a/pkg/sql/conn_executor_exec.go b/pkg/sql/conn_executor_exec.go
index d9f63af3265a..ddb548114391 100644
--- a/pkg/sql/conn_executor_exec.go
+++ b/pkg/sql/conn_executor_exec.go
@@ -975,6 +975,7 @@ func (ex *connExecutor) execWithDistSQLEngine(
 		planCtx.saveFlows = planCtx.getDefaultSaveFlowsFunc(ctx, planner, planComponentTypeMainQuery)
 	}
 	planCtx.traceMetadata = planner.instrumentation.traceMetadata
+	planCtx.collectExecStats = planner.instrumentation.ShouldCollectExecStats()
 
 	var evalCtxFactory func() *extendedEvalContext
 	if len(planner.curPlan.subqueryPlans) != 0 ||
diff --git a/pkg/sql/distsql/server.go b/pkg/sql/distsql/server.go
index c9219de19f48..160c3b4cf344 100644
--- a/pkg/sql/distsql/server.go
+++ b/pkg/sql/distsql/server.go
@@ -314,7 +314,7 @@ func (ds *ServerImpl) setupFlow(
 
 	// Create the FlowCtx for the flow.
 	flowCtx := ds.NewFlowContext(
-		ctx, req.Flow.FlowID, evalCtx, req.TraceKV, localState, req.Flow.Gateway == roachpb.NodeID(ds.NodeID.SQLInstanceID()),
+		ctx, req.Flow.FlowID, evalCtx, req.TraceKV, req.CollectStats, localState, req.Flow.Gateway == roachpb.NodeID(ds.NodeID.SQLInstanceID()),
 	)
 
 	// req always contains the desired vectorize mode, regardless of whether we
@@ -400,6 +400,7 @@ func (ds *ServerImpl) NewFlowContext(
 	id execinfrapb.FlowID,
 	evalCtx *tree.EvalContext,
 	traceKV bool,
+	collectStats bool,
 	localState LocalState,
 	isGatewayNode bool,
 ) execinfra.FlowCtx {
@@ -411,6 +412,7 @@ func (ds *ServerImpl) NewFlowContext(
 		EvalCtx:        evalCtx,
 		NodeID:         ds.ServerConfig.NodeID,
 		TraceKV:        traceKV,
+		CollectStats:   collectStats,
 		Local:          localState.IsLocal,
 		Gateway:        isGatewayNode,
 	}
diff --git a/pkg/sql/distsql_physical_planner.go b/pkg/sql/distsql_physical_planner.go
index a99ba6745a16..a5c15588db8b 100644
--- a/pkg/sql/distsql_physical_planner.go
+++ b/pkg/sql/distsql_physical_planner.go
@@ -627,6 +627,9 @@ type PlanningCtx struct {
 	// If set, we will record the mapping from planNode to tracing metadata to
 	// later allow associating statistics with the planNode.
 	traceMetadata execNodeTraceMetadata
+
+	// If set, statement execution stats should be collected.
+	collectExecStats bool
 }
 
 var _ physicalplan.ExprContext = &PlanningCtx{}
diff --git a/pkg/sql/distsql_running.go b/pkg/sql/distsql_running.go
index da61cdd68499..1247c3100872 100644
--- a/pkg/sql/distsql_running.go
+++ b/pkg/sql/distsql_running.go
@@ -125,6 +125,7 @@ func (dsp *DistSQLPlanner) setupFlows(
 	recv *DistSQLReceiver,
 	localState distsql.LocalState,
 	vectorizeThresholdMet bool,
+	collectStats bool,
 ) (context.Context, flowinfra.Flow, error) {
 	thisNodeID := dsp.gatewayNodeID
 	_, ok := flows[thisNodeID]
@@ -141,6 +142,7 @@ func (dsp *DistSQLPlanner) setupFlows(
 		Version:           execinfra.Version,
 		EvalContext:       evalCtxProto,
 		TraceKV:           evalCtx.Tracing.KVTracingEnabled(),
+		CollectStats:      collectStats,
 	}
 
 	// Start all the flows except the flow on this node (there is always a flow on
@@ -333,7 +335,9 @@ func (dsp *DistSQLPlanner) Run(
 		localState.IsLocal = true
 	}
 
-	ctx, flow, err := dsp.setupFlows(ctx, evalCtx, leafInputState, flows, recv, localState, vectorizedThresholdMet)
+	ctx, flow, err := dsp.setupFlows(
+		ctx, evalCtx, leafInputState, flows, recv, localState, vectorizedThresholdMet, planCtx.collectExecStats,
+	)
 	if err != nil {
 		recv.SetError(err)
 		return func() {}
@@ -834,6 +838,7 @@ func (dsp *DistSQLPlanner) planAndRunSubquery(
 		subqueryPlanCtx.saveFlows = subqueryPlanCtx.getDefaultSaveFlowsFunc(ctx, planner, planComponentTypeSubquery)
 	}
 	subqueryPlanCtx.traceMetadata = planner.instrumentation.traceMetadata
+	subqueryPlanCtx.collectExecStats = planner.instrumentation.ShouldCollectExecStats()
 	// Don't close the top-level plan from subqueries - someone else will handle
 	// that.
 	subqueryPlanCtx.ignoreClose = true
@@ -1126,6 +1131,7 @@ func (dsp *DistSQLPlanner) planAndRunPostquery(
 		postqueryPlanCtx.saveFlows = postqueryPlanCtx.getDefaultSaveFlowsFunc(ctx, planner, planComponentTypePostquery)
 	}
 	postqueryPlanCtx.traceMetadata = planner.instrumentation.traceMetadata
+	postqueryPlanCtx.collectExecStats = planner.instrumentation.ShouldCollectExecStats()
 
 	postqueryPhysPlan, err := dsp.createPhysPlan(postqueryPlanCtx, postqueryPlan)
 	if err != nil {
diff --git a/pkg/sql/execinfra/BUILD.bazel b/pkg/sql/execinfra/BUILD.bazel
index cf50e75c37b9..8df4b2dfece7 100644
--- a/pkg/sql/execinfra/BUILD.bazel
+++ b/pkg/sql/execinfra/BUILD.bazel
@@ -15,6 +15,7 @@ go_library(
         "readerbase.go",
         "scanbase.go",
         "server_config.go",
+        "stats.go",
         "testutils.go",
         "version.go",
         ":gen-consumerstatus-stringer",  # keep
diff --git a/pkg/sql/execinfra/flow_context.go b/pkg/sql/execinfra/flow_context.go
index 7aa499f5df3f..4a1eaebac855 100644
--- a/pkg/sql/execinfra/flow_context.go
+++ b/pkg/sql/execinfra/flow_context.go
@@ -59,6 +59,9 @@ type FlowCtx struct {
 	// TraceKV is true if KV tracing was requested by the session.
 	TraceKV bool
 
+	// CollectStats is true if execution stats collection was requested.
+	CollectStats bool
+
 	// Local is true if this flow is being run as part of a local-only query.
 	Local bool
 
diff --git a/pkg/sql/execinfra/stats.go b/pkg/sql/execinfra/stats.go
new file mode 100644
index 000000000000..607f7aa186d5
--- /dev/null
+++ b/pkg/sql/execinfra/stats.go
@@ -0,0 +1,25 @@
+// Copyright 2021 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package execinfra
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/util/tracing"
+)
+
+// ShouldCollectStats is a helper function used to determine if a processor
+// should collect stats. The two requirements are that tracing must be enabled
+// (to be able to output the stats somewhere), and that the flowCtx.CollectStats
+// flag was set by the gateway node.
+func ShouldCollectStats(ctx context.Context, flowCtx *FlowCtx) bool {
+	return tracing.SpanFromContext(ctx) != nil && flowCtx.CollectStats
+}
diff --git a/pkg/sql/execinfrapb/api.pb.go b/pkg/sql/execinfrapb/api.pb.go
index 4e02332109be..f7bc2e13c3f3 100644
--- a/pkg/sql/execinfrapb/api.pb.go
+++ b/pkg/sql/execinfrapb/api.pb.go
@@ -52,13 +52,18 @@ type SetupFlowRequest struct {
 	Flow        FlowSpec       `protobuf:"bytes,3,opt,name=flow" json:"flow"`
 	EvalContext EvalContext    `protobuf:"bytes,6,opt,name=evalContext" json:"evalContext"`
 	TraceKV     bool           `protobuf:"varint,8,opt,name=TraceKV" json:"TraceKV"`
+	// CollectStats specifies whether stats collection should be enabled for this
+	// flow. Note that some form of tracing must be enabled for these stats to be
+	// observed, since the resulting stats are added as structured data to a
+	// trace.
+	CollectStats bool `protobuf:"varint,9,opt,name=CollectStats" json:"CollectStats"`
 }
 
 func (m *SetupFlowRequest) Reset()         { *m = SetupFlowRequest{} }
 func (m *SetupFlowRequest) String() string { return proto.CompactTextString(m) }
 func (*SetupFlowRequest) ProtoMessage()    {}
 func (*SetupFlowRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_api_db601d0d8134b9c6, []int{0}
+	return fileDescriptor_api_2d75f74fa50c27f4, []int{0}
 }
 func (m *SetupFlowRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -96,7 +101,7 @@ func (m *FlowSpec) Reset()         { *m = FlowSpec{} }
 func (m *FlowSpec) String() string { return proto.CompactTextString(m) }
 func (*FlowSpec) ProtoMessage()    {}
 func (*FlowSpec) Descriptor() ([]byte, []int) {
-	return fileDescriptor_api_db601d0d8134b9c6, []int{1}
+	return fileDescriptor_api_2d75f74fa50c27f4, []int{1}
 }
 func (m *FlowSpec) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -132,7 +137,7 @@ func (m *EvalContext) Reset()         { *m = EvalContext{} }
 func (m *EvalContext) String() string { return proto.CompactTextString(m) }
 func (*EvalContext) ProtoMessage()    {}
 func (*EvalContext) Descriptor() ([]byte, []int) {
-	return fileDescriptor_api_db601d0d8134b9c6, []int{2}
+	return fileDescriptor_api_2d75f74fa50c27f4, []int{2}
 }
 func (m *EvalContext) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -165,7 +170,7 @@ func (m *SimpleResponse) Reset()         { *m = SimpleResponse{} }
 func (m *SimpleResponse) String() string { return proto.CompactTextString(m) }
 func (*SimpleResponse) ProtoMessage()    {}
 func (*SimpleResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_api_db601d0d8134b9c6, []int{3}
+	return fileDescriptor_api_2d75f74fa50c27f4, []int{3}
 }
 func (m *SimpleResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -208,7 +213,7 @@ func (m *ConsumerSignal) Reset()         { *m = ConsumerSignal{} }
 func (m *ConsumerSignal) String() string { return proto.CompactTextString(m) }
 func (*ConsumerSignal) ProtoMessage()    {}
 func (*ConsumerSignal) Descriptor() ([]byte, []int) {
-	return fileDescriptor_api_db601d0d8134b9c6, []int{4}
+	return fileDescriptor_api_2d75f74fa50c27f4, []int{4}
 }
 func (m *ConsumerSignal) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -240,7 +245,7 @@ func (m *DrainRequest) Reset()         { *m = DrainRequest{} }
 func (m *DrainRequest) String() string { return proto.CompactTextString(m) }
 func (*DrainRequest) ProtoMessage()    {}
 func (*DrainRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_api_db601d0d8134b9c6, []int{5}
+	return fileDescriptor_api_2d75f74fa50c27f4, []int{5}
 }
 func (m *DrainRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -291,7 +296,7 @@ func (m *ConsumerHandshake) Reset()         { *m = ConsumerHandshake{} }
 func (m *ConsumerHandshake) String() string { return proto.CompactTextString(m) }
 func (*ConsumerHandshake) ProtoMessage()    {}
 func (*ConsumerHandshake) Descriptor() ([]byte, []int) {
-	return fileDescriptor_api_db601d0d8134b9c6, []int{6}
+	return fileDescriptor_api_2d75f74fa50c27f4, []int{6}
 }
 func (m *ConsumerHandshake) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -611,6 +616,14 @@ func (m *SetupFlowRequest) MarshalTo(dAtA []byte) (int, error) {
 		dAtA[i] = 0
 	}
 	i++
+	dAtA[i] = 0x48
+	i++
+	if m.CollectStats {
+		dAtA[i] = 1
+	} else {
+		dAtA[i] = 0
+	}
+	i++
 	return i, nil
 }
 
@@ -848,6 +861,7 @@ func (m *SetupFlowRequest) Size() (n int) {
 		n += 1 + l + sovApi(uint64(l))
 	}
 	n += 2
+	n += 2
 	return n
 }
 
@@ -1141,6 +1155,26 @@ func (m *SetupFlowRequest) Unmarshal(dAtA []byte) error {
 				}
 			}
 			m.TraceKV = bool(v != 0)
+		case 9:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field CollectStats", wireType)
+			}
+			var v int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowApi
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				v |= (int(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			m.CollectStats = bool(v != 0)
 		default:
 			iNdEx = preIndex
 			skippy, err := skipApi(dAtA[iNdEx:])
@@ -1938,68 +1972,69 @@ var (
 	ErrIntOverflowApi   = fmt.Errorf("proto: integer overflow")
 )
 
-func init() { proto.RegisterFile("sql/execinfrapb/api.proto", fileDescriptor_api_db601d0d8134b9c6) }
+func init() { proto.RegisterFile("sql/execinfrapb/api.proto", fileDescriptor_api_2d75f74fa50c27f4) }
 
-var fileDescriptor_api_db601d0d8134b9c6 = []byte{
-	// 950 bytes of a gzipped FileDescriptorProto
+var fileDescriptor_api_2d75f74fa50c27f4 = []byte{
+	// 966 bytes of a gzipped FileDescriptorProto
 	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x55, 0xcf, 0x72, 0xdb, 0x44,
-	0x18, 0xb7, 0xe4, 0xb5, 0xad, 0xac, 0x1d, 0x57, 0xd9, 0xe9, 0x30, 0xc2, 0x07, 0xdb, 0xa3, 0x29,
-	0x60, 0x60, 0x90, 0x8b, 0x0b, 0x17, 0xe0, 0x82, 0x6b, 0xa0, 0x56, 0x9b, 0x0e, 0x95, 0xd3, 0x0e,
+	0x18, 0xb7, 0xe4, 0x4d, 0x2c, 0xaf, 0x1d, 0x57, 0xd9, 0xe9, 0x30, 0xc2, 0x07, 0xdb, 0xa3, 0x29,
+	0x60, 0x60, 0x90, 0x8b, 0x0b, 0x17, 0xe0, 0x82, 0x63, 0xa0, 0x56, 0x9b, 0x0e, 0x95, 0xd3, 0x0e,
 	0xc3, 0x01, 0xcd, 0x5a, 0x5a, 0xcb, 0x9a, 0xc8, 0xbb, 0x8a, 0x76, 0xd5, 0xb8, 0x6f, 0xc0, 0xb1,
-	0xc3, 0x13, 0x70, 0xe0, 0x45, 0xb8, 0xe5, 0xd8, 0x03, 0x87, 0x0e, 0x87, 0x00, 0xce, 0x5b, 0x70,
-	0x62, 0x24, 0x4b, 0xb1, 0xea, 0xe0, 0x10, 0x6e, 0xda, 0xfd, 0x7e, 0xbf, 0xdf, 0xb7, 0xdf, 0x5f,
-	0xc1, 0xb7, 0xf9, 0x49, 0xd0, 0x27, 0x4b, 0xe2, 0xf8, 0x74, 0x16, 0xe1, 0x70, 0xda, 0xc7, 0xa1,
-	0x6f, 0x84, 0x11, 0x13, 0x0c, 0x69, 0x0e, 0x73, 0x8e, 0x23, 0x86, 0x9d, 0xb9, 0xc1, 0x4f, 0x02,
-	0xc3, 0xf5, 0xb9, 0xe0, 0x27, 0x41, 0x14, 0xd3, 0xd6, 0x6d, 0x8f, 0x79, 0x2c, 0x05, 0xf5, 0x93,
-	0xaf, 0x35, 0xbe, 0xd5, 0xf1, 0x18, 0xf3, 0x02, 0xd2, 0x4f, 0x4f, 0xd3, 0x78, 0xd6, 0x17, 0xfe,
-	0x82, 0x70, 0x81, 0x17, 0x61, 0x06, 0x40, 0xa9, 0x58, 0x38, 0xed, 0xbb, 0x58, 0xe0, 0xec, 0xae,
-	0xb5, 0xed, 0xbf, 0x60, 0xeb, 0x6e, 0xdb, 0xc2, 0x88, 0x39, 0x84, 0x73, 0x16, 0xf1, 0x0c, 0x71,
-	0x27, 0x41, 0x70, 0xc2, 0xb9, 0xcf, 0x68, 0x42, 0x0c, 0xa7, 0xf9, 0xc9, 0xde, 0xe8, 0xe8, 0xbf,
-	0xc9, 0x50, 0x9d, 0x10, 0x11, 0x87, 0x5f, 0x07, 0xec, 0xd4, 0x22, 0x27, 0x31, 0xe1, 0x02, 0x7d,
-	0x01, 0xc1, 0x2c, 0x60, 0xa7, 0x5a, 0xb9, 0x2b, 0xf5, 0xea, 0x03, 0xdd, 0xd8, 0x15, 0xac, 0x91,
-	0x90, 0x26, 0x21, 0x71, 0x86, 0xe0, 0xec, 0xbc, 0x53, 0xb2, 0x52, 0x16, 0xba, 0x0b, 0x6b, 0xcf,
-	0x49, 0x94, 0x38, 0xd2, 0x2a, 0x5d, 0xa9, 0xb7, 0x3f, 0x7c, 0x2b, 0x31, 0xfe, 0x7d, 0xde, 0x69,
-	0x8e, 0x7c, 0x2e, 0x26, 0x4f, 0x1e, 0x3d, 0x5b, 0x5b, 0xad, 0x1c, 0x86, 0x0e, 0x61, 0x9d, 0x3c,
-	0xc7, 0xc1, 0x7d, 0x46, 0x05, 0x59, 0x0a, 0xad, 0x9a, 0xba, 0x7d, 0x67, 0xb7, 0xdb, 0xaf, 0x36,
-	0xe0, 0xcc, 0x73, 0x91, 0x8f, 0x9e, 0xc2, 0xdb, 0x01, 0xc1, 0x33, 0x5b, 0x2c, 0xa9, 0xed, 0xd3,
-	0x30, 0x16, 0x36, 0x17, 0x58, 0x10, 0xad, 0x96, 0xea, 0xde, 0x29, 0xe8, 0x66, 0x49, 0x37, 0x1e,
-	0x11, 0x3c, 0x3b, 0x5a, 0xd2, 0x71, 0x02, 0x9e, 0x24, 0x58, 0xeb, 0x20, 0xd8, 0xbe, 0x42, 0x6d,
-	0x58, 0x3b, 0x8a, 0xb0, 0x43, 0x1e, 0x3e, 0xd3, 0x94, 0xae, 0xd4, 0x53, 0x32, 0xd7, 0xf9, 0xa5,
-	0x09, 0x14, 0x49, 0x95, 0x4d, 0xa0, 0xc8, 0x6a, 0x59, 0xbf, 0x90, 0xa0, 0x92, 0x27, 0x07, 0x7d,
-	0x0c, 0x6b, 0x49, 0x62, 0x6c, 0xdf, 0xd5, 0xa4, 0xae, 0xd4, 0x6b, 0x0c, 0xb5, 0x84, 0xf8, 0xfb,
-	0x79, 0xa7, 0x9a, 0x40, 0xc6, 0xa3, 0xd5, 0xe5, 0x97, 0x55, 0x4d, 0x80, 0x63, 0x17, 0x1d, 0x42,
-	0xb8, 0x29, 0xa8, 0x26, 0x77, 0xcb, 0xbd, 0xfa, 0xe0, 0xbd, 0xdd, 0x09, 0xf9, 0x36, 0xc7, 0x16,
-	0x8a, 0x51, 0x10, 0x40, 0x4f, 0x61, 0xcd, 0xc3, 0x82, 0x9c, 0xe2, 0x17, 0x69, 0x4d, 0x2b, 0xc3,
-	0xcf, 0xb3, 0x92, 0xdc, 0xf3, 0x7c, 0x31, 0x8f, 0xa7, 0x86, 0xc3, 0x16, 0xfd, 0x4b, 0x75, 0x77,
-	0xba, 0xf9, 0xee, 0x87, 0xc7, 0x5e, 0x3f, 0x4f, 0xd3, 0x63, 0xe6, 0x92, 0xf1, 0xc8, 0xca, 0xb5,
-	0xf4, 0x9f, 0x64, 0x58, 0x2f, 0xd4, 0x02, 0x7d, 0x02, 0x11, 0x17, 0x0b, 0x71, 0x94, 0xf7, 0xf6,
-	0x63, 0x4c, 0x19, 0x4f, 0x63, 0x2e, 0x67, 0x8f, 0xfa, 0x17, 0x3b, 0x1a, 0xc0, 0x03, 0xb1, 0xa4,
-	0x5b, 0x24, 0xb9, 0x40, 0xba, 0x6a, 0x46, 0x4f, 0x60, 0xa3, 0xd8, 0xcc, 0xda, 0xad, 0xb4, 0xb4,
-	0xbd, 0xad, 0x0c, 0xbd, 0xd1, 0xfd, 0xc6, 0x64, 0x7d, 0x1a, 0x61, 0x81, 0xf3, 0xae, 0xe1, 0x9b,
-	0x2b, 0x13, 0x28, 0x65, 0x15, 0x98, 0x40, 0x01, 0x6a, 0xc5, 0x04, 0x4a, 0x45, 0xad, 0x9a, 0x40,
-	0xa9, 0xaa, 0x35, 0x13, 0x28, 0x35, 0x55, 0x31, 0x81, 0xa2, 0xa8, 0x7b, 0x26, 0x50, 0xf6, 0x54,
-	0x68, 0x02, 0x05, 0xaa, 0x75, 0x13, 0x28, 0x75, 0xb5, 0x61, 0x02, 0xa5, 0xa1, 0xee, 0x9b, 0x40,
-	0xd9, 0x57, 0x9b, 0x26, 0x50, 0x9a, 0xea, 0x2d, 0xfd, 0x1b, 0xd8, 0x9c, 0xf8, 0x8b, 0x30, 0x20,
-	0x16, 0xe1, 0x21, 0xa3, 0x9c, 0xa0, 0x4f, 0x61, 0x85, 0x44, 0x11, 0x8b, 0xd2, 0x4c, 0xd4, 0x07,
-	0x9d, 0x6b, 0x1a, 0x3b, 0x81, 0x59, 0x6b, 0xb4, 0xfe, 0xa3, 0x0c, 0x9b, 0xf7, 0x19, 0xe5, 0xf1,
-	0x82, 0x44, 0x13, 0xdf, 0xa3, 0x38, 0x40, 0x0f, 0xe1, 0xbe, 0x1b, 0x61, 0x9f, 0xda, 0xd1, 0x7a,
-	0x52, 0x33, 0xc5, 0x77, 0x77, 0x2b, 0x8e, 0x12, 0x78, 0x36, 0xd7, 0x56, 0xc3, 0x2d, 0x9c, 0xd0,
-	0x77, 0x10, 0xf1, 0x64, 0xf2, 0xed, 0xb4, 0x39, 0x73, 0x45, 0x39, 0x55, 0xfc, 0x60, 0xb7, 0xe2,
-	0xf6, 0xb6, 0xb0, 0x54, 0xbe, 0xbd, 0x3f, 0xc6, 0x70, 0x6f, 0x8e, 0xa9, 0xcb, 0xe7, 0xf8, 0x98,
-	0x64, 0x4b, 0xe4, 0xc3, 0xdd, 0x82, 0x79, 0x8c, 0x0f, 0x72, 0x8a, 0xb5, 0x61, 0x7f, 0x06, 0xce,
-	0x7e, 0xee, 0x48, 0x7a, 0x13, 0x36, 0x8a, 0x81, 0xe8, 0xbf, 0xc8, 0xf0, 0xe0, 0x0a, 0x0d, 0xdd,
-	0x83, 0xc8, 0xc9, 0x2e, 0x6d, 0xee, 0xcc, 0x89, 0x1b, 0x07, 0x64, 0x3d, 0x72, 0xf9, 0xac, 0x1e,
-	0xe4, 0xf6, 0x49, 0x6e, 0x46, 0x3f, 0xc0, 0xd6, 0x15, 0x92, 0xed, 0x12, 0xec, 0x06, 0x3e, 0x25,
-	0x59, 0x36, 0x5a, 0xc6, 0x7a, 0x7d, 0x1b, 0xf9, 0xfa, 0x36, 0x2e, 0xdb, 0x71, 0x08, 0x5e, 0xfe,
-	0xd1, 0x91, 0x2c, 0x6d, 0x5b, 0x78, 0x94, 0x29, 0x14, 0xb7, 0x61, 0xf9, 0x66, 0xdb, 0xf0, 0x01,
-	0xbc, 0xbd, 0xf0, 0xa9, 0x8d, 0x1d, 0x87, 0x84, 0x82, 0xb8, 0x76, 0x4e, 0x07, 0xd7, 0xd2, 0xd1,
-	0xc2, 0xa7, 0x5f, 0x66, 0x94, 0xec, 0x6e, 0xf0, 0xab, 0x0c, 0x6b, 0x19, 0x0c, 0xcd, 0x61, 0xdd,
-	0x8a, 0xe9, 0xe4, 0x05, 0x75, 0x92, 0x4a, 0xa1, 0xde, 0x7f, 0xd7, 0x63, 0xdd, 0x73, 0xad, 0xf7,
-	0xaf, 0x5d, 0x3b, 0x6e, 0xec, 0x90, 0xe8, 0x90, 0x70, 0x8e, 0x3d, 0xa2, 0x97, 0x7a, 0xd2, 0x5d,
-	0x09, 0x39, 0x70, 0xef, 0xb2, 0x47, 0xd0, 0xff, 0x68, 0xa4, 0xd6, 0x35, 0x6f, 0x7a, 0x73, 0xa2,
-	0xf4, 0x12, 0xf2, 0x20, 0x4c, 0xf7, 0xab, 0x88, 0x08, 0x5e, 0xa0, 0x9b, 0xbf, 0xb1, 0x75, 0xe3,
-	0xc0, 0xd7, 0xd1, 0x0c, 0x3f, 0x3a, 0xfb, 0xab, 0x5d, 0x3a, 0x5b, 0xb5, 0xa5, 0x57, 0xab, 0xb6,
-	0xf4, 0x7a, 0xd5, 0x96, 0xfe, 0x5c, 0xb5, 0xa5, 0x97, 0x17, 0xed, 0xd2, 0xab, 0x8b, 0x76, 0xe9,
-	0xf5, 0x45, 0xbb, 0xf4, 0x7d, 0xbd, 0xf0, 0x1b, 0xfe, 0x27, 0x00, 0x00, 0xff, 0xff, 0x1d, 0x6c,
-	0xd4, 0x10, 0x34, 0x08, 0x00, 0x00,
+	0xc3, 0x13, 0x70, 0xe0, 0x45, 0xb8, 0xe5, 0xd8, 0x63, 0x87, 0x43, 0x00, 0xe7, 0x2d, 0x38, 0x30,
+	0x8c, 0x64, 0x29, 0x56, 0x1c, 0x1c, 0xc2, 0x4d, 0xbb, 0xdf, 0xef, 0xf7, 0xfb, 0xf6, 0xfb, 0x2b,
+	0xf8, 0x36, 0x3f, 0x09, 0x7a, 0x64, 0x41, 0x1c, 0x9f, 0x4e, 0x23, 0x1c, 0x4e, 0x7a, 0x38, 0xf4,
+	0x8d, 0x30, 0x62, 0x82, 0x21, 0xcd, 0x61, 0xce, 0x71, 0xc4, 0xb0, 0x33, 0x33, 0xf8, 0x49, 0x60,
+	0xb8, 0x3e, 0x17, 0xfc, 0x24, 0x88, 0x62, 0xda, 0xbc, 0xeb, 0x31, 0x8f, 0xa5, 0xa0, 0x5e, 0xf2,
+	0xb5, 0xc2, 0x37, 0xdb, 0x1e, 0x63, 0x5e, 0x40, 0x7a, 0xe9, 0x69, 0x12, 0x4f, 0x7b, 0xc2, 0x9f,
+	0x13, 0x2e, 0xf0, 0x3c, 0xcc, 0x00, 0x28, 0x15, 0x0b, 0x27, 0x3d, 0x17, 0x0b, 0x9c, 0xdd, 0x35,
+	0x37, 0xfd, 0x17, 0x6c, 0x9d, 0x4d, 0x5b, 0x18, 0x31, 0x87, 0x70, 0xce, 0x22, 0x9e, 0x21, 0xee,
+	0x25, 0x08, 0x4e, 0x38, 0xf7, 0x19, 0x4d, 0x88, 0xe1, 0x24, 0x3f, 0xd9, 0x6b, 0x1d, 0xfd, 0x6f,
+	0x19, 0xaa, 0x63, 0x22, 0xe2, 0xf0, 0xeb, 0x80, 0x9d, 0x5a, 0xe4, 0x24, 0x26, 0x5c, 0xa0, 0x2f,
+	0x20, 0x98, 0x06, 0xec, 0x54, 0x2b, 0x77, 0xa4, 0x6e, 0xad, 0xaf, 0x1b, 0xdb, 0x82, 0x35, 0x12,
+	0xd2, 0x38, 0x24, 0xce, 0x00, 0x9c, 0x9d, 0xb7, 0x4b, 0x56, 0xca, 0x42, 0xf7, 0x61, 0xe5, 0x05,
+	0x89, 0x12, 0x47, 0xda, 0x4e, 0x47, 0xea, 0xee, 0x0d, 0xde, 0x4a, 0x8c, 0x7f, 0x9d, 0xb7, 0x1b,
+	0x43, 0x9f, 0x8b, 0xf1, 0xd3, 0xc7, 0xcf, 0x57, 0x56, 0x2b, 0x87, 0xa1, 0x43, 0x58, 0x23, 0x2f,
+	0x70, 0x70, 0xc0, 0xa8, 0x20, 0x0b, 0xa1, 0xed, 0xa6, 0x6e, 0xdf, 0xd9, 0xee, 0xf6, 0xab, 0x35,
+	0x38, 0xf3, 0x5c, 0xe4, 0xa3, 0x67, 0xf0, 0x6e, 0x40, 0xf0, 0xd4, 0x16, 0x0b, 0x6a, 0xfb, 0x34,
+	0x8c, 0x85, 0xcd, 0x05, 0x16, 0x44, 0xab, 0xa4, 0xba, 0xf7, 0x0a, 0xba, 0x59, 0xd2, 0x8d, 0xc7,
+	0x04, 0x4f, 0x8f, 0x16, 0x74, 0x94, 0x80, 0xc7, 0x09, 0xd6, 0xda, 0x0f, 0x36, 0xaf, 0x50, 0x0b,
+	0x56, 0x8e, 0x22, 0xec, 0x90, 0x47, 0xcf, 0x35, 0xa5, 0x23, 0x75, 0x95, 0xcc, 0x75, 0x7e, 0x89,
+	0xba, 0xb0, 0x7e, 0xc0, 0x82, 0x80, 0x38, 0x29, 0x9e, 0x6b, 0xd5, 0x02, 0xe8, 0x8a, 0xc5, 0x04,
+	0x8a, 0xa4, 0xca, 0x26, 0x50, 0x64, 0xb5, 0xac, 0x5f, 0x48, 0x50, 0xc9, 0xd3, 0x88, 0x3e, 0x86,
+	0x95, 0x24, 0x85, 0xb6, 0xef, 0x6a, 0x52, 0x47, 0xea, 0xd6, 0x07, 0x5a, 0xc2, 0xfe, 0xed, 0xbc,
+	0xbd, 0x9b, 0x40, 0x46, 0xc3, 0xe5, 0xe5, 0x97, 0xb5, 0x9b, 0x00, 0x47, 0x2e, 0x3a, 0x84, 0x70,
+	0x5d, 0x7a, 0x4d, 0xee, 0x94, 0xbb, 0xb5, 0xfe, 0x7b, 0xdb, 0x53, 0xf7, 0x6d, 0x8e, 0x2d, 0x94,
+	0xad, 0x20, 0x80, 0x9e, 0xc1, 0x8a, 0x87, 0x05, 0x39, 0xc5, 0x2f, 0xd3, 0xea, 0xef, 0x0c, 0x3e,
+	0xcf, 0x8a, 0xf7, 0xc0, 0xf3, 0xc5, 0x2c, 0x9e, 0x18, 0x0e, 0x9b, 0xf7, 0x2e, 0xd5, 0xdd, 0xc9,
+	0xfa, 0xbb, 0x17, 0x1e, 0x7b, 0xbd, 0x3c, 0xa1, 0x4f, 0x98, 0x4b, 0x46, 0x43, 0x2b, 0xd7, 0xd2,
+	0x7f, 0x92, 0x61, 0xad, 0x50, 0x35, 0xf4, 0x09, 0x44, 0x5c, 0xcc, 0xc5, 0x51, 0x3e, 0x05, 0x4f,
+	0x30, 0x65, 0x3c, 0x8d, 0xb9, 0x9c, 0x3d, 0xea, 0x5f, 0xec, 0xa8, 0x0f, 0xf7, 0xc5, 0x82, 0x6e,
+	0x90, 0xe4, 0x02, 0xe9, 0xba, 0x19, 0x3d, 0x85, 0xf5, 0x62, 0xdb, 0x6b, 0x77, 0xd2, 0x26, 0xe8,
+	0x6e, 0x64, 0xe8, 0xca, 0x9c, 0x18, 0xe3, 0xd5, 0x69, 0x88, 0x05, 0xce, 0xfb, 0x8b, 0xaf, 0xaf,
+	0x4c, 0xa0, 0x94, 0x55, 0x60, 0x02, 0x05, 0xa8, 0x3b, 0x26, 0x50, 0x76, 0xd4, 0x5d, 0x13, 0x28,
+	0xbb, 0x6a, 0xc5, 0x04, 0x4a, 0x45, 0x55, 0x4c, 0xa0, 0x28, 0x6a, 0xd5, 0x04, 0x4a, 0x55, 0x85,
+	0x26, 0x50, 0xa0, 0x5a, 0x33, 0x81, 0x52, 0x53, 0xeb, 0x26, 0x50, 0xea, 0xea, 0x9e, 0x09, 0x94,
+	0x3d, 0xb5, 0x61, 0x02, 0xa5, 0xa1, 0xde, 0xd1, 0xbf, 0x81, 0x8d, 0xb1, 0x3f, 0x0f, 0x03, 0x62,
+	0x11, 0x1e, 0x32, 0xca, 0x09, 0xfa, 0x14, 0xee, 0x90, 0x28, 0x62, 0x51, 0x9a, 0x89, 0x5a, 0xbf,
+	0x7d, 0xc3, 0x08, 0x24, 0x30, 0x6b, 0x85, 0xd6, 0x7f, 0x94, 0x61, 0xe3, 0x80, 0x51, 0x1e, 0xcf,
+	0x49, 0x34, 0xf6, 0x3d, 0x8a, 0x03, 0xf4, 0x08, 0xee, 0xb9, 0x11, 0xf6, 0xa9, 0x1d, 0xad, 0x66,
+	0x3a, 0x53, 0x7c, 0x77, 0xbb, 0xe2, 0x30, 0x81, 0x67, 0x1b, 0xc0, 0xaa, 0xbb, 0x85, 0x13, 0xfa,
+	0x0e, 0x22, 0x9e, 0xec, 0x08, 0x3b, 0x6d, 0xce, 0x5c, 0x51, 0x4e, 0x15, 0x3f, 0xd8, 0xae, 0xb8,
+	0xb9, 0x57, 0x2c, 0x95, 0x6f, 0x6e, 0x9a, 0x11, 0xac, 0xce, 0x30, 0x75, 0xf9, 0x0c, 0x1f, 0x93,
+	0x6c, 0xdd, 0x7c, 0xb8, 0x5d, 0x30, 0x8f, 0xf1, 0x61, 0x4e, 0xb1, 0xd6, 0xec, 0xcf, 0xc0, 0xd9,
+	0xcf, 0x6d, 0x49, 0x6f, 0xc0, 0x7a, 0x31, 0x10, 0xfd, 0x17, 0x19, 0xee, 0x5f, 0xa3, 0xa1, 0x07,
+	0x10, 0x39, 0xd9, 0xa5, 0xcd, 0x9d, 0x19, 0x71, 0xe3, 0x80, 0xac, 0x46, 0x2e, 0x1f, 0xd8, 0xfd,
+	0xdc, 0x3e, 0xce, 0xcd, 0xe8, 0x07, 0xd8, 0xbc, 0x46, 0xb2, 0x5d, 0x82, 0xdd, 0xc0, 0xa7, 0x24,
+	0xcb, 0x46, 0xd3, 0x58, 0x2d, 0x7a, 0x23, 0x5f, 0xf4, 0xc6, 0x65, 0x3b, 0x0e, 0xc0, 0xab, 0xdf,
+	0xdb, 0x92, 0xa5, 0x6d, 0x0a, 0x0f, 0x33, 0x85, 0xe2, 0xde, 0x2c, 0xdf, 0x6e, 0x6f, 0x3e, 0x84,
+	0x77, 0xe7, 0x3e, 0xb5, 0xb1, 0xe3, 0x90, 0x50, 0x10, 0xd7, 0xce, 0xe9, 0xe0, 0x46, 0x3a, 0x9a,
+	0xfb, 0xf4, 0xcb, 0x8c, 0x92, 0xdd, 0xf5, 0x7f, 0x95, 0x61, 0x25, 0x83, 0xa1, 0x19, 0xac, 0x59,
+	0x31, 0x1d, 0xbf, 0xa4, 0x4e, 0x52, 0x29, 0xd4, 0xfd, 0xef, 0x7a, 0xac, 0x7a, 0xae, 0xf9, 0xfe,
+	0x8d, 0x6b, 0xc7, 0x8d, 0x1d, 0x12, 0x1d, 0x12, 0xce, 0xb1, 0x47, 0xf4, 0x52, 0x57, 0xba, 0x2f,
+	0x21, 0x07, 0x56, 0x2f, 0x7b, 0x04, 0xfd, 0x8f, 0x46, 0x6a, 0xde, 0xf0, 0xa6, 0xab, 0x13, 0xa5,
+	0x97, 0x90, 0x07, 0x61, 0xba, 0x5f, 0x45, 0x44, 0xf0, 0x1c, 0xdd, 0xfe, 0x8d, 0xcd, 0x5b, 0x07,
+	0xbe, 0x8a, 0x66, 0xf0, 0xd1, 0xd9, 0x9f, 0xad, 0xd2, 0xd9, 0xb2, 0x25, 0xbd, 0x5e, 0xb6, 0xa4,
+	0x37, 0xcb, 0x96, 0xf4, 0xc7, 0xb2, 0x25, 0xbd, 0xba, 0x68, 0x95, 0x5e, 0x5f, 0xb4, 0x4a, 0x6f,
+	0x2e, 0x5a, 0xa5, 0xef, 0x6b, 0x85, 0x1f, 0xf6, 0x3f, 0x01, 0x00, 0x00, 0xff, 0xff, 0x9f, 0xf9,
+	0xeb, 0x68, 0x5e, 0x08, 0x00, 0x00,
 }
diff --git a/pkg/sql/execinfrapb/api.proto b/pkg/sql/execinfrapb/api.proto
index ca87fe8c386c..aeea1ff33e86 100644
--- a/pkg/sql/execinfrapb/api.proto
+++ b/pkg/sql/execinfrapb/api.proto
@@ -45,6 +45,12 @@ message SetupFlowRequest {
   optional EvalContext evalContext = 6 [(gogoproto.nullable) = false];
 
   optional bool TraceKV = 8 [(gogoproto.nullable) = false];
+
+  // CollectStats specifies whether stats collection should be enabled for this
+  // flow. Note that some form of tracing must be enabled for these stats to be
+  // observed, since the resulting stats are added as structured data to a
+  // trace.
+  optional bool CollectStats = 9 [(gogoproto.nullable) = false];
 }
 
 // FlowSpec describes a "flow" which is a subgraph of a distributed SQL
diff --git a/pkg/sql/instrumentation.go b/pkg/sql/instrumentation.go
index 9cb951fae0fa..7240b2ab5019 100644
--- a/pkg/sql/instrumentation.go
+++ b/pkg/sql/instrumentation.go
@@ -63,6 +63,10 @@ type instrumentationHelper struct {
 	// statement; it triggers saving of extra information like the plan string.
 	collectBundle bool
 
+	// collectExecStats is set when we are collecting execution statistics for a
+	// statement.
+	collectExecStats bool
+
 	// discardRows is set if we want to discard any results rather than sending
 	// them back to the client. Used for testing/benchmarking. Note that the
 	// resulting schema or the plan are not affected.
@@ -145,9 +149,11 @@ func (ih *instrumentationHelper) Setup(
 	ih.savePlanForStats = appStats.shouldSaveLogicalPlanDescription(fingerprint, implicitTxn)
 
 	if !ih.collectBundle && ih.withStatementTrace == nil && ih.outputMode == unmodifiedOutput {
+		// TODO(asubiotto): Create a span for stat collection in future commit.
 		return ctx, false
 	}
 
+	ih.collectExecStats = true
 	ih.traceMetadata = make(execNodeTraceMetadata)
 	ih.origCtx = ctx
 	ih.evalCtx = p.EvalContext()
@@ -264,7 +270,7 @@ func (ih *instrumentationHelper) ShouldDiscardRows() bool {
 // ShouldSaveFlows is true if we should save the flow specifications (to be able
 // to generate diagrams).
 func (ih *instrumentationHelper) ShouldSaveFlows() bool {
-	return ih.collectBundle || ih.outputMode == explainAnalyzeDistSQLOutput
+	return ih.collectBundle || ih.outputMode == explainAnalyzeDistSQLOutput || ih.collectExecStats
 }
 
 // ShouldUseJobForCreateStats indicates if we should run CREATE STATISTICS as a
@@ -282,6 +288,12 @@ func (ih *instrumentationHelper) ShouldBuildExplainPlan() bool {
 		ih.outputMode == explainAnalyzeDistSQLOutput
 }
 
+// ShouldCollectExecStats returns true if we should collect statement execution
+// statistics.
+func (ih *instrumentationHelper) ShouldCollectExecStats() bool {
+	return ih.collectExecStats
+}
+
 // RecordExplainPlan records the explain.Plan for this query.
 func (ih *instrumentationHelper) RecordExplainPlan(explainPlan *explain.Plan) {
 	ih.explainPlan = explainPlan
diff --git a/pkg/sql/rowexec/aggregator.go b/pkg/sql/rowexec/aggregator.go
index b75006847913..8b1c01923129 100644
--- a/pkg/sql/rowexec/aggregator.go
+++ b/pkg/sql/rowexec/aggregator.go
@@ -24,7 +24,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
 	"github.com/cockroachdb/cockroach/pkg/util/optional"
 	"github.com/cockroachdb/cockroach/pkg/util/stringarena"
-	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 )
 
@@ -94,7 +93,7 @@ func (ag *aggregatorBase) init(
 ) error {
 	ctx := flowCtx.EvalCtx.Ctx()
 	memMonitor := execinfra.NewMonitor(ctx, flowCtx.EvalCtx.Mon, "aggregator-mem")
-	if sp := tracing.SpanFromContext(ctx); sp != nil && sp.IsVerbose() {
+	if execinfra.ShouldCollectStats(ctx, flowCtx) {
 		input = newInputStatCollector(input)
 		ag.ExecStatsForTrace = ag.execStatsForTrace
 	}
diff --git a/pkg/sql/rowexec/countrows.go b/pkg/sql/rowexec/countrows.go
index 2c33eee4fc9e..58a5eaf5add0 100644
--- a/pkg/sql/rowexec/countrows.go
+++ b/pkg/sql/rowexec/countrows.go
@@ -18,7 +18,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
-	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 )
 
 // countAggregator is a simple processor that counts the number of rows it
@@ -45,7 +44,7 @@ func newCountAggregator(
 	ag := &countAggregator{}
 	ag.input = input
 
-	if sp := tracing.SpanFromContext(flowCtx.EvalCtx.Ctx()); sp != nil && sp.IsVerbose() {
+	if execinfra.ShouldCollectStats(flowCtx.EvalCtx.Ctx(), flowCtx) {
 		ag.input = newInputStatCollector(input)
 		ag.ExecStatsForTrace = ag.execStatsForTrace
 	}
diff --git a/pkg/sql/rowexec/distinct.go b/pkg/sql/rowexec/distinct.go
index b6129d476146..88ab096f2d7e 100644
--- a/pkg/sql/rowexec/distinct.go
+++ b/pkg/sql/rowexec/distinct.go
@@ -23,7 +23,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
 	"github.com/cockroachdb/cockroach/pkg/util/optional"
 	"github.com/cockroachdb/cockroach/pkg/util/stringarena"
-	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 )
 
@@ -133,7 +132,7 @@ func newDistinct(
 	// So we have to set up the account here.
 	d.arena = stringarena.Make(&d.memAcc)
 
-	if sp := tracing.SpanFromContext(ctx); sp != nil && sp.IsVerbose() {
+	if execinfra.ShouldCollectStats(ctx, flowCtx) {
 		d.input = newInputStatCollector(d.input)
 		d.ExecStatsForTrace = d.execStatsForTrace
 	}
diff --git a/pkg/sql/rowexec/filterer.go b/pkg/sql/rowexec/filterer.go
index 74b203e69fd9..77f834e32b38 100644
--- a/pkg/sql/rowexec/filterer.go
+++ b/pkg/sql/rowexec/filterer.go
@@ -16,7 +16,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
-	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 )
 
@@ -63,7 +62,7 @@ func newFiltererProcessor(
 	}
 
 	ctx := flowCtx.EvalCtx.Ctx()
-	if sp := tracing.SpanFromContext(ctx); sp != nil && sp.IsVerbose() {
+	if execinfra.ShouldCollectStats(ctx, flowCtx) {
 		f.input = newInputStatCollector(f.input)
 		f.ExecStatsForTrace = f.execStatsForTrace
 	}
diff --git a/pkg/sql/rowexec/hashjoiner.go b/pkg/sql/rowexec/hashjoiner.go
index 2dc3dd6cb8e8..84219b74cde0 100644
--- a/pkg/sql/rowexec/hashjoiner.go
+++ b/pkg/sql/rowexec/hashjoiner.go
@@ -22,7 +22,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
 	"github.com/cockroachdb/cockroach/pkg/util/optional"
-	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 )
 
@@ -142,7 +141,7 @@ func newHashJoiner(
 	)
 
 	// If the trace is recording, instrument the hashJoiner to collect stats.
-	if sp := tracing.SpanFromContext(ctx); sp != nil && sp.IsVerbose() {
+	if execinfra.ShouldCollectStats(ctx, flowCtx) {
 		h.leftSource = newInputStatCollector(h.leftSource)
 		h.rightSource = newInputStatCollector(h.rightSource)
 		h.ExecStatsForTrace = h.execStatsForTrace
diff --git a/pkg/sql/rowexec/inverted_filterer.go b/pkg/sql/rowexec/inverted_filterer.go
index d71e80403b49..8a18270b2b27 100644
--- a/pkg/sql/rowexec/inverted_filterer.go
+++ b/pkg/sql/rowexec/inverted_filterer.go
@@ -24,7 +24,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
 	"github.com/cockroachdb/cockroach/pkg/util/optional"
-	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 )
 
@@ -124,7 +123,7 @@ func newInvertedFilterer(
 		ifr.diskMonitor,
 	)
 
-	if sp := tracing.SpanFromContext(flowCtx.EvalCtx.Ctx()); sp != nil && sp.IsVerbose() {
+	if execinfra.ShouldCollectStats(flowCtx.EvalCtx.Ctx(), flowCtx) {
 		ifr.input = newInputStatCollector(ifr.input)
 		ifr.ExecStatsForTrace = ifr.execStatsForTrace
 	}
diff --git a/pkg/sql/rowexec/inverted_joiner.go b/pkg/sql/rowexec/inverted_joiner.go
index c74eea616bd0..0f1c6ecc6b7d 100644
--- a/pkg/sql/rowexec/inverted_joiner.go
+++ b/pkg/sql/rowexec/inverted_joiner.go
@@ -32,7 +32,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
 	"github.com/cockroachdb/cockroach/pkg/util/optional"
-	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 )
 
@@ -311,7 +310,7 @@ func newInvertedJoiner(
 	}
 
 	collectingStats := false
-	if sp := tracing.SpanFromContext(flowCtx.EvalCtx.Ctx()); sp != nil && sp.IsVerbose() {
+	if execinfra.ShouldCollectStats(flowCtx.EvalCtx.Ctx(), flowCtx) {
 		collectingStats = true
 	}
 	if collectingStats {
diff --git a/pkg/sql/rowexec/joinreader.go b/pkg/sql/rowexec/joinreader.go
index 5ee27332ccc4..9ae6eaa36e73 100644
--- a/pkg/sql/rowexec/joinreader.go
+++ b/pkg/sql/rowexec/joinreader.go
@@ -33,7 +33,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
 	"github.com/cockroachdb/cockroach/pkg/util/optional"
-	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 )
 
@@ -273,7 +272,7 @@ func newJoinReader(
 	}
 
 	collectingStats := false
-	if sp := tracing.SpanFromContext(flowCtx.EvalCtx.Ctx()); sp != nil && sp.IsVerbose() {
+	if execinfra.ShouldCollectStats(flowCtx.EvalCtx.Ctx(), flowCtx) {
 		collectingStats = true
 	}
 
diff --git a/pkg/sql/rowexec/mergejoiner.go b/pkg/sql/rowexec/mergejoiner.go
index 0064afb648c6..cadfc282241e 100644
--- a/pkg/sql/rowexec/mergejoiner.go
+++ b/pkg/sql/rowexec/mergejoiner.go
@@ -20,7 +20,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util"
 	"github.com/cockroachdb/cockroach/pkg/util/cancelchecker"
 	"github.com/cockroachdb/cockroach/pkg/util/optional"
-	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 )
 
@@ -75,7 +74,7 @@ func newMergeJoiner(
 		trackMatchedRight: shouldEmitUnmatchedRow(rightSide, spec.Type) || spec.Type == descpb.RightSemiJoin,
 	}
 
-	if sp := tracing.SpanFromContext(flowCtx.EvalCtx.Ctx()); sp != nil && sp.IsVerbose() {
+	if execinfra.ShouldCollectStats(flowCtx.EvalCtx.Ctx(), flowCtx) {
 		m.leftSource = newInputStatCollector(m.leftSource)
 		m.rightSource = newInputStatCollector(m.rightSource)
 		m.ExecStatsForTrace = m.execStatsForTrace
diff --git a/pkg/sql/rowexec/noop.go b/pkg/sql/rowexec/noop.go
index 9573a9d16ec0..e0c0e0b2bfaa 100644
--- a/pkg/sql/rowexec/noop.go
+++ b/pkg/sql/rowexec/noop.go
@@ -16,7 +16,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
-	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 )
 
@@ -56,7 +55,7 @@ func newNoopProcessor(
 		return nil, err
 	}
 	ctx := flowCtx.EvalCtx.Ctx()
-	if sp := tracing.SpanFromContext(ctx); sp != nil && sp.IsVerbose() {
+	if execinfra.ShouldCollectStats(ctx, flowCtx) {
 		n.input = newInputStatCollector(n.input)
 		n.ExecStatsForTrace = n.execStatsForTrace
 	}
diff --git a/pkg/sql/rowexec/ordinality.go b/pkg/sql/rowexec/ordinality.go
index c404de0bdb24..b86dc98f4771 100644
--- a/pkg/sql/rowexec/ordinality.go
+++ b/pkg/sql/rowexec/ordinality.go
@@ -18,7 +18,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
-	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 )
 
 // ordinalityProcessor is the processor of the WITH ORDINALITY operator, which
@@ -67,7 +66,7 @@ func newOrdinalityProcessor(
 		return nil, err
 	}
 
-	if sp := tracing.SpanFromContext(ctx); sp != nil && sp.IsVerbose() {
+	if execinfra.ShouldCollectStats(ctx, flowCtx) {
 		o.input = newInputStatCollector(o.input)
 		o.ExecStatsForTrace = o.execStatsForTrace
 	}
diff --git a/pkg/sql/rowexec/sorter.go b/pkg/sql/rowexec/sorter.go
index 13b3076bade4..7bdc6e98bef8 100644
--- a/pkg/sql/rowexec/sorter.go
+++ b/pkg/sql/rowexec/sorter.go
@@ -22,7 +22,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
 	"github.com/cockroachdb/cockroach/pkg/util/optional"
-	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 )
 
@@ -54,7 +53,7 @@ func (s *sorterBase) init(
 	opts execinfra.ProcStateOpts,
 ) error {
 	ctx := flowCtx.EvalCtx.Ctx()
-	if sp := tracing.SpanFromContext(ctx); sp != nil && sp.IsVerbose() {
+	if execinfra.ShouldCollectStats(ctx, flowCtx) {
 		input = newInputStatCollector(input)
 		s.ExecStatsForTrace = s.execStatsForTrace
 	}
diff --git a/pkg/sql/rowexec/tablereader.go b/pkg/sql/rowexec/tablereader.go
index df53b4fb9595..2e2456158723 100644
--- a/pkg/sql/rowexec/tablereader.go
+++ b/pkg/sql/rowexec/tablereader.go
@@ -25,7 +25,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/optional"
-	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 )
 
@@ -159,7 +158,7 @@ func newTableReader(
 		tr.spans[i] = s.Span
 	}
 
-	if sp := tracing.SpanFromContext(flowCtx.EvalCtx.Ctx()); sp != nil && sp.IsVerbose() {
+	if execinfra.ShouldCollectStats(flowCtx.EvalCtx.Ctx(), flowCtx) {
 		tr.fetcher = newRowFetcherStatCollector(&fetcher)
 		tr.ExecStatsForTrace = tr.execStatsForTrace
 	} else {
diff --git a/pkg/sql/rowexec/tablereader_test.go b/pkg/sql/rowexec/tablereader_test.go
index af472b0dfaa1..ee1fbd9527e6 100644
--- a/pkg/sql/rowexec/tablereader_test.go
+++ b/pkg/sql/rowexec/tablereader_test.go
@@ -351,6 +351,8 @@ func TestTableReaderDrain(t *testing.T) {
 // we properly set the limit on the underlying Fetcher/KVFetcher).
 func TestLimitScans(t *testing.T) {
 	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
 	ctx := context.Background()
 
 	s, sqlDB, kvDB := serverutils.StartServer(t, base.TestServerArgs{
@@ -394,6 +396,7 @@ func TestLimitScans(t *testing.T) {
 	sp.SetVerbose(true)
 	ctx = tracing.ContextWithSpan(ctx, sp)
 	flowCtx.EvalCtx.Context = ctx
+	flowCtx.CollectStats = true
 
 	tr, err := newTableReader(&flowCtx, 0 /* processorID */, &spec, &post, nil /* output */)
 	if err != nil {
diff --git a/pkg/sql/rowexec/windower.go b/pkg/sql/rowexec/windower.go
index 755061d8d716..60000e4b95fb 100644
--- a/pkg/sql/rowexec/windower.go
+++ b/pkg/sql/rowexec/windower.go
@@ -29,7 +29,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
 	"github.com/cockroachdb/cockroach/pkg/util/optional"
-	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 )
 
@@ -201,7 +200,7 @@ func newWindower(
 	// them to reuse the same shared memory account with the windower.
 	evalCtx.SingleDatumAggMemAccount = &w.acc
 
-	if sp := tracing.SpanFromContext(ctx); sp != nil && sp.IsVerbose() {
+	if execinfra.ShouldCollectStats(ctx, flowCtx) {
 		w.input = newInputStatCollector(w.input)
 		w.ExecStatsForTrace = w.execStatsForTrace
 	}

From 32f93ec0191600b4aab6d12d555f48b3044e2f12 Mon Sep 17 00:00:00 2001
From: Alfonso Subiotto Marques <alfonso@cockroachlabs.com>
Date: Thu, 28 Jan 2021 12:39:50 +0100
Subject: [PATCH 2/4] sql: add cluster setting to sample execution stats

This commit adds sql.statement_stats.sample_rate, which defines a probability
that a statement will be sampled.

This commit also adds MaxMemoryUsage to the StatementStatistics DB Console
protobuf.

Release note: None
---
 pkg/roachpb/app_stats.go        |  14 +-
 pkg/roachpb/app_stats.pb.go     | 294 ++++++++++++++++++++------------
 pkg/roachpb/app_stats.proto     |  17 +-
 pkg/sql/BUILD.bazel             |   1 +
 pkg/sql/conn_executor.go        |   6 +
 pkg/sql/conn_executor_exec.go   |   2 +-
 pkg/sql/instrumentation.go      |  61 +++++--
 pkg/sql/instrumentation_test.go |  64 +++++++
 8 files changed, 337 insertions(+), 122 deletions(-)
 create mode 100644 pkg/sql/instrumentation_test.go

diff --git a/pkg/roachpb/app_stats.go b/pkg/roachpb/app_stats.go
index 71292693c162..cedf73531d49 100644
--- a/pkg/roachpb/app_stats.go
+++ b/pkg/roachpb/app_stats.go
@@ -129,7 +129,16 @@ func (s *StatementStatistics) Add(other *StatementStatistics) {
 	s.OverheadLat.Add(other.OverheadLat, s.Count, other.Count)
 	s.BytesRead.Add(other.BytesRead, s.Count, other.Count)
 	s.RowsRead.Add(other.RowsRead, s.Count, other.Count)
-	s.BytesSentOverNetwork.Add(other.BytesSentOverNetwork, s.Count, other.Count)
+
+	// Execution stats collected using a sampling approach.
+	statCollectionCount := s.ExecStatCollectionCount
+	if statCollectionCount == 0 && other.ExecStatCollectionCount == 0 {
+		// If both are zero, artificially set the receiver's count to one to avoid
+		// division by zero in Add.
+		statCollectionCount = 1
+	}
+	s.BytesSentOverNetwork.Add(other.BytesSentOverNetwork, s.ExecStatCollectionCount, statCollectionCount)
+	s.MaxMemUsage.Add(other.MaxMemUsage, s.ExecStatCollectionCount, statCollectionCount)
 
 	if other.SensitiveInfo.LastErr != "" {
 		s.SensitiveInfo.LastErr = other.SensitiveInfo.LastErr
@@ -157,5 +166,6 @@ func (s *StatementStatistics) AlmostEqual(other *StatementStatistics, eps float6
 		s.SensitiveInfo.Equal(other.SensitiveInfo) &&
 		s.BytesRead.AlmostEqual(other.BytesRead, eps) &&
 		s.RowsRead.AlmostEqual(other.RowsRead, eps) &&
-		s.BytesSentOverNetwork.AlmostEqual(other.BytesSentOverNetwork, eps)
+		s.BytesSentOverNetwork.AlmostEqual(other.BytesSentOverNetwork, eps) &&
+		s.MaxMemUsage.AlmostEqual(other.MaxMemUsage, eps)
 }
diff --git a/pkg/roachpb/app_stats.pb.go b/pkg/roachpb/app_stats.pb.go
index 476b06944f26..9cfc5ff61ab0 100644
--- a/pkg/roachpb/app_stats.pb.go
+++ b/pkg/roachpb/app_stats.pb.go
@@ -26,6 +26,9 @@ var _ = time.Kitchen
 // proto package needs to be updated.
 const _ = proto.GoGoProtoPackageIsVersion2 // please upgrade the proto package
 
+// StatementStatistics represent the statement statistics sent to the DB
+// Console for a given statement fingerprint. Note that these stats are cleared
+// ever diagnostics.sql_stat_reset_interval.
 // N.B. When fields are added to this struct, make sure to update
 // (*StatementStatistics).Add and (*StatementStatistics).AlmostEqual
 // in app_stats.go.
@@ -81,13 +84,22 @@ type StatementStatistics struct {
 	RowsRead NumericStat `protobuf:"bytes,16,opt,name=rows_read,json=rowsRead" json:"rows_read"`
 	// BytesSentOverNetwork collects the number of bytes sent over the network.
 	BytesSentOverNetwork NumericStat `protobuf:"bytes,17,opt,name=bytes_sent_over_network,json=bytesSentOverNetwork" json:"bytes_sent_over_network"`
+	// MaxMemUsage collects the maximum memory usage that occurred on a node.
+	MaxMemUsage NumericStat `protobuf:"bytes,18,opt,name=max_mem_usage,json=maxMemUsage" json:"max_mem_usage"`
+	// ExecStatCollectionCount keeps track of how many times execution stats were
+	// recorded for this statement. Since this collection follows a sampling
+	// approach, this number is not necessarily equal to Count. Used to calculate
+	// the mean of the following NumericStat values:
+	//  bytes_sent_over_network
+	//  max_mem_usage
+	ExecStatCollectionCount int64 `protobuf:"varint,19,opt,name=exec_stat_collection_count,json=execStatCollectionCount" json:"exec_stat_collection_count"`
 }
 
 func (m *StatementStatistics) Reset()         { *m = StatementStatistics{} }
 func (m *StatementStatistics) String() string { return proto.CompactTextString(m) }
 func (*StatementStatistics) ProtoMessage()    {}
 func (*StatementStatistics) Descriptor() ([]byte, []int) {
-	return fileDescriptor_app_stats_1e7bd1c14de5d1e0, []int{0}
+	return fileDescriptor_app_stats_ad5cbe238e742d0d, []int{0}
 }
 func (m *StatementStatistics) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -136,7 +148,7 @@ func (m *TransactionStatistics) Reset()         { *m = TransactionStatistics{} }
 func (m *TransactionStatistics) String() string { return proto.CompactTextString(m) }
 func (*TransactionStatistics) ProtoMessage()    {}
 func (*TransactionStatistics) Descriptor() ([]byte, []int) {
-	return fileDescriptor_app_stats_1e7bd1c14de5d1e0, []int{1}
+	return fileDescriptor_app_stats_ad5cbe238e742d0d, []int{1}
 }
 func (m *TransactionStatistics) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -176,7 +188,7 @@ func (m *SensitiveInfo) Reset()         { *m = SensitiveInfo{} }
 func (m *SensitiveInfo) String() string { return proto.CompactTextString(m) }
 func (*SensitiveInfo) ProtoMessage()    {}
 func (*SensitiveInfo) Descriptor() ([]byte, []int) {
-	return fileDescriptor_app_stats_1e7bd1c14de5d1e0, []int{2}
+	return fileDescriptor_app_stats_ad5cbe238e742d0d, []int{2}
 }
 func (m *SensitiveInfo) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -218,7 +230,7 @@ func (m *NumericStat) Reset()         { *m = NumericStat{} }
 func (m *NumericStat) String() string { return proto.CompactTextString(m) }
 func (*NumericStat) ProtoMessage()    {}
 func (*NumericStat) Descriptor() ([]byte, []int) {
-	return fileDescriptor_app_stats_1e7bd1c14de5d1e0, []int{3}
+	return fileDescriptor_app_stats_ad5cbe238e742d0d, []int{3}
 }
 func (m *NumericStat) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -257,7 +269,7 @@ func (m *StatementStatisticsKey) Reset()         { *m = StatementStatisticsKey{}
 func (m *StatementStatisticsKey) String() string { return proto.CompactTextString(m) }
 func (*StatementStatisticsKey) ProtoMessage()    {}
 func (*StatementStatisticsKey) Descriptor() ([]byte, []int) {
-	return fileDescriptor_app_stats_1e7bd1c14de5d1e0, []int{4}
+	return fileDescriptor_app_stats_ad5cbe238e742d0d, []int{4}
 }
 func (m *StatementStatisticsKey) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -297,7 +309,7 @@ func (m *CollectedStatementStatistics) Reset()         { *m = CollectedStatement
 func (m *CollectedStatementStatistics) String() string { return proto.CompactTextString(m) }
 func (*CollectedStatementStatistics) ProtoMessage()    {}
 func (*CollectedStatementStatistics) Descriptor() ([]byte, []int) {
-	return fileDescriptor_app_stats_1e7bd1c14de5d1e0, []int{5}
+	return fileDescriptor_app_stats_ad5cbe238e742d0d, []int{5}
 }
 func (m *CollectedStatementStatistics) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -337,7 +349,7 @@ func (m *CollectedTransactionStatistics) Reset()         { *m = CollectedTransac
 func (m *CollectedTransactionStatistics) String() string { return proto.CompactTextString(m) }
 func (*CollectedTransactionStatistics) ProtoMessage()    {}
 func (*CollectedTransactionStatistics) Descriptor() ([]byte, []int) {
-	return fileDescriptor_app_stats_1e7bd1c14de5d1e0, []int{6}
+	return fileDescriptor_app_stats_ad5cbe238e742d0d, []int{6}
 }
 func (m *CollectedTransactionStatistics) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -377,7 +389,7 @@ func (m *ExplainTreePlanNode) Reset()         { *m = ExplainTreePlanNode{} }
 func (m *ExplainTreePlanNode) String() string { return proto.CompactTextString(m) }
 func (*ExplainTreePlanNode) ProtoMessage()    {}
 func (*ExplainTreePlanNode) Descriptor() ([]byte, []int) {
-	return fileDescriptor_app_stats_1e7bd1c14de5d1e0, []int{7}
+	return fileDescriptor_app_stats_ad5cbe238e742d0d, []int{7}
 }
 func (m *ExplainTreePlanNode) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -411,7 +423,7 @@ func (m *ExplainTreePlanNode_Attr) Reset()         { *m = ExplainTreePlanNode_At
 func (m *ExplainTreePlanNode_Attr) String() string { return proto.CompactTextString(m) }
 func (*ExplainTreePlanNode_Attr) ProtoMessage()    {}
 func (*ExplainTreePlanNode_Attr) Descriptor() ([]byte, []int) {
-	return fileDescriptor_app_stats_1e7bd1c14de5d1e0, []int{7, 0}
+	return fileDescriptor_app_stats_ad5cbe238e742d0d, []int{7, 0}
 }
 func (m *ExplainTreePlanNode_Attr) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -450,7 +462,7 @@ func (m *TxnStats) Reset()         { *m = TxnStats{} }
 func (m *TxnStats) String() string { return proto.CompactTextString(m) }
 func (*TxnStats) ProtoMessage()    {}
 func (*TxnStats) Descriptor() ([]byte, []int) {
-	return fileDescriptor_app_stats_1e7bd1c14de5d1e0, []int{8}
+	return fileDescriptor_app_stats_ad5cbe238e742d0d, []int{8}
 }
 func (m *TxnStats) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -700,6 +712,21 @@ func (m *StatementStatistics) MarshalTo(dAtA []byte) (int, error) {
 		return 0, err
 	}
 	i += n10
+	dAtA[i] = 0x92
+	i++
+	dAtA[i] = 0x1
+	i++
+	i = encodeVarintAppStats(dAtA, i, uint64(m.MaxMemUsage.Size()))
+	n11, err := m.MaxMemUsage.MarshalTo(dAtA[i:])
+	if err != nil {
+		return 0, err
+	}
+	i += n11
+	dAtA[i] = 0x98
+	i++
+	dAtA[i] = 0x1
+	i++
+	i = encodeVarintAppStats(dAtA, i, uint64(m.ExecStatCollectionCount))
 	return i, nil
 }
 
@@ -727,35 +754,35 @@ func (m *TransactionStatistics) MarshalTo(dAtA []byte) (int, error) {
 	dAtA[i] = 0x1a
 	i++
 	i = encodeVarintAppStats(dAtA, i, uint64(m.NumRows.Size()))
-	n11, err := m.NumRows.MarshalTo(dAtA[i:])
+	n12, err := m.NumRows.MarshalTo(dAtA[i:])
 	if err != nil {
 		return 0, err
 	}
-	i += n11
+	i += n12
 	dAtA[i] = 0x22
 	i++
 	i = encodeVarintAppStats(dAtA, i, uint64(m.ServiceLat.Size()))
-	n12, err := m.ServiceLat.MarshalTo(dAtA[i:])
+	n13, err := m.ServiceLat.MarshalTo(dAtA[i:])
 	if err != nil {
 		return 0, err
 	}
-	i += n12
+	i += n13
 	dAtA[i] = 0x2a
 	i++
 	i = encodeVarintAppStats(dAtA, i, uint64(m.RetryLat.Size()))
-	n13, err := m.RetryLat.MarshalTo(dAtA[i:])
+	n14, err := m.RetryLat.MarshalTo(dAtA[i:])
 	if err != nil {
 		return 0, err
 	}
-	i += n13
+	i += n14
 	dAtA[i] = 0x32
 	i++
 	i = encodeVarintAppStats(dAtA, i, uint64(m.CommitLat.Size()))
-	n14, err := m.CommitLat.MarshalTo(dAtA[i:])
+	n15, err := m.CommitLat.MarshalTo(dAtA[i:])
 	if err != nil {
 		return 0, err
 	}
-	i += n14
+	i += n15
 	return i, nil
 }
 
@@ -781,19 +808,19 @@ func (m *SensitiveInfo) MarshalTo(dAtA []byte) (int, error) {
 	dAtA[i] = 0x12
 	i++
 	i = encodeVarintAppStats(dAtA, i, uint64(m.MostRecentPlanDescription.Size()))
-	n15, err := m.MostRecentPlanDescription.MarshalTo(dAtA[i:])
+	n16, err := m.MostRecentPlanDescription.MarshalTo(dAtA[i:])
 	if err != nil {
 		return 0, err
 	}
-	i += n15
+	i += n16
 	dAtA[i] = 0x1a
 	i++
 	i = encodeVarintAppStats(dAtA, i, uint64(github_com_gogo_protobuf_types.SizeOfStdTime(m.MostRecentPlanTimestamp)))
-	n16, err := github_com_gogo_protobuf_types.StdTimeMarshalTo(m.MostRecentPlanTimestamp, dAtA[i:])
+	n17, err := github_com_gogo_protobuf_types.StdTimeMarshalTo(m.MostRecentPlanTimestamp, dAtA[i:])
 	if err != nil {
 		return 0, err
 	}
-	i += n16
+	i += n17
 	return i, nil
 }
 
@@ -907,19 +934,19 @@ func (m *CollectedStatementStatistics) MarshalTo(dAtA []byte) (int, error) {
 	dAtA[i] = 0xa
 	i++
 	i = encodeVarintAppStats(dAtA, i, uint64(m.Key.Size()))
-	n17, err := m.Key.MarshalTo(dAtA[i:])
+	n18, err := m.Key.MarshalTo(dAtA[i:])
 	if err != nil {
 		return 0, err
 	}
-	i += n17
+	i += n18
 	dAtA[i] = 0x12
 	i++
 	i = encodeVarintAppStats(dAtA, i, uint64(m.Stats.Size()))
-	n18, err := m.Stats.MarshalTo(dAtA[i:])
+	n19, err := m.Stats.MarshalTo(dAtA[i:])
 	if err != nil {
 		return 0, err
 	}
-	i += n18
+	i += n19
 	dAtA[i] = 0x18
 	i++
 	i = encodeVarintAppStats(dAtA, i, uint64(m.ID))
@@ -955,11 +982,11 @@ func (m *CollectedTransactionStatistics) MarshalTo(dAtA []byte) (int, error) {
 	dAtA[i] = 0x1a
 	i++
 	i = encodeVarintAppStats(dAtA, i, uint64(m.Stats.Size()))
-	n19, err := m.Stats.MarshalTo(dAtA[i:])
+	n20, err := m.Stats.MarshalTo(dAtA[i:])
 	if err != nil {
 		return 0, err
 	}
-	i += n19
+	i += n20
 	return i, nil
 }
 
@@ -1056,11 +1083,11 @@ func (m *TxnStats) MarshalTo(dAtA []byte) (int, error) {
 	dAtA[i] = 0x12
 	i++
 	i = encodeVarintAppStats(dAtA, i, uint64(m.TxnTimeSec.Size()))
-	n20, err := m.TxnTimeSec.MarshalTo(dAtA[i:])
+	n21, err := m.TxnTimeSec.MarshalTo(dAtA[i:])
 	if err != nil {
 		return 0, err
 	}
-	i += n20
+	i += n21
 	dAtA[i] = 0x18
 	i++
 	i = encodeVarintAppStats(dAtA, i, uint64(m.CommittedCount))
@@ -1112,6 +1139,9 @@ func (m *StatementStatistics) Size() (n int) {
 	n += 2 + l + sovAppStats(uint64(l))
 	l = m.BytesSentOverNetwork.Size()
 	n += 2 + l + sovAppStats(uint64(l))
+	l = m.MaxMemUsage.Size()
+	n += 2 + l + sovAppStats(uint64(l))
+	n += 2 + sovAppStats(uint64(m.ExecStatCollectionCount))
 	return n
 }
 
@@ -1717,6 +1747,55 @@ func (m *StatementStatistics) Unmarshal(dAtA []byte) error {
 				return err
 			}
 			iNdEx = postIndex
+		case 18:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field MaxMemUsage", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowAppStats
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= (int(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthAppStats
+			}
+			postIndex := iNdEx + msglen
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if err := m.MaxMemUsage.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		case 19:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ExecStatCollectionCount", wireType)
+			}
+			m.ExecStatCollectionCount = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowAppStats
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.ExecStatCollectionCount |= (int64(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
 		default:
 			iNdEx = preIndex
 			skippy, err := skipAppStats(dAtA[iNdEx:])
@@ -3167,80 +3246,83 @@ var (
 	ErrIntOverflowAppStats   = fmt.Errorf("proto: integer overflow")
 )
 
-func init() { proto.RegisterFile("roachpb/app_stats.proto", fileDescriptor_app_stats_1e7bd1c14de5d1e0) }
-
-var fileDescriptor_app_stats_1e7bd1c14de5d1e0 = []byte{
-	// 1145 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x56, 0xcd, 0x6e, 0x23, 0x45,
-	0x10, 0xce, 0xf8, 0x27, 0xb6, 0xcb, 0x76, 0xb2, 0x3b, 0xfb, 0x37, 0x58, 0x91, 0x1d, 0xac, 0x5d,
-	0x6d, 0x56, 0x80, 0x23, 0x45, 0x5c, 0x00, 0x65, 0x7f, 0x9c, 0xec, 0xc1, 0x4b, 0x14, 0xc0, 0x8e,
-	0x84, 0xc4, 0x65, 0xd4, 0x99, 0x29, 0x27, 0xad, 0xcc, 0x5f, 0xba, 0xdb, 0x8e, 0x7d, 0xe7, 0x01,
-	0xf6, 0x11, 0xb8, 0xf1, 0x10, 0x1c, 0xb9, 0xe4, 0x82, 0xb4, 0xc7, 0x15, 0x87, 0x00, 0xce, 0x85,
-	0x27, 0xe0, 0xc0, 0x09, 0x75, 0xf7, 0x8c, 0xb1, 0xbd, 0x8e, 0x98, 0xe5, 0x66, 0x57, 0xd5, 0xf7,
-	0x75, 0x57, 0xd5, 0x57, 0xd5, 0x03, 0x0f, 0x58, 0x48, 0x9c, 0xd3, 0xe8, 0x78, 0x9b, 0x44, 0x91,
-	0xcd, 0x05, 0x11, 0xbc, 0x15, 0xb1, 0x50, 0x84, 0x66, 0xd5, 0x09, 0x9d, 0x33, 0xe5, 0x6c, 0xf1,
-	0x73, 0xaf, 0x76, 0xf7, 0x24, 0x3c, 0x09, 0x95, 0x67, 0x5b, 0xfe, 0xd2, 0x41, 0xb5, 0xc6, 0x49,
-	0x18, 0x9e, 0x78, 0xb8, 0xad, 0xfe, 0x1d, 0x0f, 0xfa, 0xdb, 0x82, 0xfa, 0xc8, 0x05, 0xf1, 0x23,
-	0x1d, 0xd0, 0xfc, 0xb1, 0x00, 0x77, 0x7a, 0x82, 0x08, 0xf4, 0x31, 0x10, 0xf2, 0x07, 0xe5, 0x82,
-	0x3a, 0xdc, 0xac, 0x41, 0xde, 0x09, 0x07, 0x81, 0xb0, 0x8c, 0x4d, 0x63, 0x2b, 0xdb, 0xce, 0x5d,
-	0x5e, 0x35, 0x56, 0xba, 0xda, 0x64, 0x7e, 0x0a, 0x77, 0xfa, 0x94, 0x71, 0x61, 0x13, 0x21, 0xd0,
-	0x8f, 0x84, 0xad, 0x23, 0x33, 0x33, 0x91, 0xb7, 0x55, 0xc0, 0x0b, 0xed, 0xdf, 0x53, 0xa8, 0x47,
-	0x50, 0xf6, 0xc9, 0xc8, 0x66, 0x28, 0x18, 0x45, 0x6e, 0x65, 0x67, 0xa2, 0xc1, 0x27, 0xa3, 0xae,
-	0xb6, 0x9b, 0x1f, 0xc3, 0xba, 0x87, 0x27, 0xc4, 0x19, 0xdb, 0x1e, 0xe1, 0xc2, 0x46, 0xc6, 0xac,
-	0xdc, 0xa6, 0xb1, 0x55, 0x8a, 0x43, 0xab, 0xda, 0x79, 0x40, 0xb8, 0x78, 0xc9, 0x98, 0xf9, 0x05,
-	0x14, 0x83, 0x81, 0x6f, 0xb3, 0xf0, 0x82, 0x5b, 0xf9, 0x4d, 0x63, 0xab, 0xbc, 0x53, 0x6b, 0xcd,
-	0xd5, 0xa5, 0x75, 0x38, 0xf0, 0x91, 0x51, 0x47, 0xa6, 0x16, 0x53, 0x14, 0x82, 0x81, 0xdf, 0x0d,
-	0x2f, 0xb8, 0xb9, 0x0b, 0xa5, 0x88, 0x30, 0x8e, 0xb6, 0x47, 0x84, 0xb5, 0x9a, 0x12, 0x5d, 0x54,
-	0x90, 0x03, 0x22, 0xe4, 0xd9, 0x91, 0x47, 0x02, 0x85, 0x2e, 0xa4, 0x3d, 0x5b, 0x22, 0x24, 0xf8,
-	0x33, 0x28, 0xb0, 0x81, 0xc6, 0x16, 0x53, 0x62, 0x57, 0xd9, 0x40, 0x41, 0x5f, 0x40, 0x99, 0x23,
-	0x1b, 0x52, 0x47, 0x5f, 0xbc, 0x94, 0x12, 0x0e, 0x31, 0x48, 0x52, 0xec, 0x41, 0x25, 0x1c, 0x22,
-	0x3b, 0x45, 0xe2, 0x2a, 0x0e, 0x48, 0xc9, 0x51, 0x4e, 0x50, 0x92, 0x64, 0x17, 0xac, 0x85, 0x4e,
-	0xd9, 0x0c, 0x5d, 0xe2, 0x08, 0x74, 0xad, 0xf2, 0x4c, 0xcb, 0xee, 0xcd, 0xb5, 0xac, 0x1b, 0x87,
-	0x98, 0x1d, 0x58, 0xe3, 0x18, 0x70, 0x2a, 0xe8, 0x10, 0x6d, 0x1a, 0xf4, 0x43, 0xab, 0xa2, 0x6e,
-	0xb1, 0xb1, 0x70, 0x8b, 0x5e, 0x12, 0xd4, 0x09, 0xfa, 0x61, 0xa2, 0x02, 0x3e, 0x6b, 0x34, 0x9f,
-	0x01, 0x1c, 0x8f, 0x05, 0x72, 0x9b, 0x21, 0x71, 0xad, 0xf5, 0x94, 0xc9, 0x94, 0x14, 0xa6, 0x8b,
-	0xc4, 0x95, 0x4a, 0x90, 0x12, 0xd2, 0xf8, 0x5b, 0x69, 0x95, 0x20, 0x21, 0x0a, 0xfe, 0x2d, 0x3c,
-	0xd0, 0xe7, 0x73, 0x0c, 0x84, 0x2d, 0x6b, 0x64, 0x07, 0x28, 0x2e, 0x42, 0x76, 0x66, 0xdd, 0x4e,
-	0x49, 0x76, 0x57, 0x11, 0xf4, 0x30, 0x10, 0x5f, 0x0d, 0x91, 0x1d, 0x6a, 0xf4, 0xab, 0x5c, 0xb1,
-	0x7a, 0x6b, 0xed, 0x55, 0xae, 0xb8, 0x76, 0x6b, 0xbd, 0xf9, 0x6b, 0x06, 0xee, 0x1d, 0x31, 0x12,
-	0x70, 0xe2, 0x08, 0x1a, 0x06, 0x29, 0x67, 0x75, 0x61, 0xea, 0x32, 0x37, 0x4c, 0xdd, 0xec, 0x1c,
-	0x65, 0xdf, 0x77, 0x8e, 0x16, 0x04, 0x99, 0xfb, 0x1f, 0x82, 0x94, 0x0d, 0x40, 0xc1, 0xc6, 0x8a,
-	0x20, 0x9f, 0xba, 0x01, 0x12, 0x22, 0xe1, 0xcf, 0x00, 0x9c, 0xd0, 0xf7, 0xa9, 0x78, 0xaf, 0x51,
-	0x2e, 0x69, 0xcc, 0x01, 0x11, 0xcd, 0xef, 0x33, 0x50, 0x9d, 0x13, 0x9a, 0xd9, 0x80, 0xe2, 0x74,
-	0x01, 0x19, 0x33, 0x6a, 0x2e, 0x78, 0xf1, 0xea, 0xa1, 0xb0, 0xe1, 0x87, 0x5c, 0xd8, 0x0c, 0x1d,
-	0xd9, 0x75, 0xb5, 0x0a, 0x5c, 0xe4, 0x0e, 0xa3, 0x91, 0xec, 0x8f, 0x2a, 0x75, 0x79, 0xa7, 0xb9,
-	0x70, 0x8b, 0x97, 0xa3, 0xc8, 0x23, 0x34, 0x38, 0x62, 0x88, 0x5f, 0x7b, 0x24, 0x38, 0x0c, 0x5d,
-	0x8c, 0x89, 0x3f, 0x90, 0x6c, 0x5d, 0x45, 0x26, 0x3d, 0xfb, 0xff, 0x52, 0x99, 0x04, 0x6a, 0xef,
-	0x1c, 0x35, 0x5d, 0xe4, 0xd3, 0x7e, 0xe9, 0x55, 0xdf, 0x4a, 0x56, 0x7d, 0xeb, 0x28, 0x89, 0x68,
-	0x17, 0xe5, 0x01, 0xaf, 0x7f, 0x6b, 0x18, 0xdd, 0x07, 0xf3, 0x87, 0x4c, 0x43, 0x3e, 0xcf, 0xfd,
-	0xf9, 0x43, 0xc3, 0x68, 0x76, 0xa1, 0x3c, 0x53, 0x26, 0xd3, 0x82, 0x9c, 0x8f, 0x24, 0x50, 0xf9,
-	0x1b, 0xf1, 0x35, 0x95, 0xc5, 0x7c, 0x02, 0x55, 0x7e, 0x3e, 0x20, 0x0c, 0x5d, 0xdb, 0xa5, 0xfd,
-	0xbe, 0x16, 0x56, 0x12, 0x52, 0x89, 0x5d, 0xfb, 0xd2, 0xd3, 0xbc, 0x36, 0xe0, 0xfe, 0x92, 0x17,
-	0xe6, 0x4b, 0x1c, 0x4b, 0xe1, 0x9e, 0x0f, 0x90, 0x8d, 0xe7, 0x0a, 0xac, 0x4d, 0xe6, 0x7d, 0xc8,
-	0x92, 0x28, 0x52, 0xbc, 0x89, 0x47, 0x1a, 0xcc, 0x3a, 0x14, 0x5c, 0xca, 0x45, 0xef, 0x9b, 0x03,
-	0x95, 0x78, 0x31, 0x69, 0x4b, 0x6c, 0x34, 0x37, 0x60, 0xb5, 0x4f, 0xa8, 0x87, 0xae, 0xd2, 0x61,
-	0xe2, 0x8e, 0x6d, 0x92, 0x35, 0x8c, 0xb4, 0xc2, 0x12, 0x97, 0x34, 0x98, 0x8f, 0xa1, 0x42, 0xfd,
-	0xc8, 0xa3, 0x0e, 0x15, 0xb6, 0x18, 0x05, 0x4a, 0x42, 0x49, 0x40, 0x39, 0xf1, 0x1c, 0x8d, 0x02,
-	0x49, 0x30, 0x44, 0x47, 0xed, 0xfb, 0x29, 0xc1, 0x10, 0x9d, 0xe6, 0xcf, 0x06, 0x6c, 0xec, 0x85,
-	0x9e, 0x87, 0x72, 0xb7, 0x2d, 0x7b, 0x50, 0x77, 0x21, 0x7b, 0x86, 0x3a, 0xd3, 0xf2, 0xce, 0xa3,
-	0xc5, 0x1d, 0xb7, 0xb4, 0x3e, 0x09, 0xff, 0x19, 0x8e, 0xcd, 0xa7, 0x90, 0x57, 0x8f, 0xff, 0x0d,
-	0xb2, 0x5a, 0x42, 0x90, 0x94, 0x53, 0xc1, 0xcc, 0x87, 0x90, 0xa1, 0xae, 0xaa, 0x58, 0xae, 0x7d,
-	0x57, 0x3a, 0x26, 0x57, 0x8d, 0x4c, 0x67, 0xff, 0xef, 0xab, 0xc6, 0x6a, 0x4f, 0xf8, 0xa2, 0xb3,
-	0xdf, 0xcd, 0x50, 0xb7, 0xf9, 0x93, 0x01, 0xf5, 0x69, 0x16, 0xcb, 0x97, 0xcd, 0x33, 0xa8, 0xf2,
-	0xe4, 0x30, 0x9b, 0xba, 0xdc, 0x32, 0x36, 0xb3, 0x5b, 0xb9, 0x76, 0x6d, 0x72, 0xd5, 0xa8, 0x4c,
-	0x6f, 0xd1, 0xd9, 0xe7, 0x33, 0xcc, 0x95, 0x29, 0xa0, 0xe3, 0xf2, 0x1b, 0x1b, 0xfb, 0x3c, 0xc9,
-	0x50, 0xeb, 0xf9, 0xe1, 0x42, 0x86, 0x4b, 0x6f, 0x33, 0x97, 0x63, 0xf3, 0x2f, 0x03, 0xee, 0x2c,
-	0x99, 0x2f, 0x29, 0xe3, 0x80, 0xf8, 0x38, 0xa7, 0x32, 0x65, 0x31, 0x77, 0x21, 0x4f, 0x84, 0x60,
-	0xb2, 0xaa, 0xd9, 0xad, 0xf2, 0xce, 0xe3, 0xff, 0x1e, 0xd6, 0xd6, 0x0b, 0x21, 0x58, 0x57, 0xa3,
-	0xcc, 0xa7, 0x50, 0x74, 0x4e, 0xa9, 0xe7, 0x32, 0x0c, 0xac, 0xac, 0x62, 0x48, 0x31, 0xee, 0xdd,
-	0x29, 0xa6, 0xf6, 0x1c, 0x72, 0x92, 0x4e, 0x96, 0x24, 0xd1, 0x46, 0x69, 0xb6, 0xe9, 0x35, 0xc8,
-	0x0f, 0x89, 0x37, 0xc0, 0xb9, 0x62, 0x69, 0x93, 0x1e, 0xd8, 0x78, 0x6c, 0x7f, 0x31, 0xa0, 0x78,
-	0x34, 0x52, 0x75, 0xe1, 0xe6, 0x87, 0x50, 0x12, 0xa3, 0xc0, 0x7e, 0xf7, 0x45, 0x28, 0x8a, 0x51,
-	0xa0, 0x3f, 0xc5, 0xda, 0x50, 0x91, 0x21, 0x72, 0x85, 0xd8, 0x1c, 0x9d, 0x58, 0x53, 0x29, 0x36,
-	0xb6, 0x18, 0xa9, 0x95, 0xd1, 0x43, 0xc7, 0xfc, 0x04, 0xd6, 0xf5, 0xfa, 0x14, 0xe8, 0xc6, 0x87,
-	0xcd, 0x7e, 0xd2, 0xad, 0x4d, 0x9d, 0xfa, 0xc8, 0x8f, 0x60, 0x6d, 0x3a, 0x60, 0x3a, 0x3a, 0x37,
-	0x13, 0x5d, 0x4d, 0x7c, 0x2a, 0xb8, 0xfd, 0xe4, 0xf2, 0x8f, 0xfa, 0xca, 0xe5, 0xa4, 0x6e, 0xbc,
-	0x99, 0xd4, 0x8d, 0xb7, 0x93, 0xba, 0xf1, 0xfb, 0xa4, 0x6e, 0xbc, 0xbe, 0xae, 0xaf, 0xbc, 0xb9,
-	0xae, 0xaf, 0xbc, 0xbd, 0xae, 0xaf, 0x7c, 0x57, 0x88, 0xbf, 0x8a, 0xff, 0x09, 0x00, 0x00, 0xff,
-	0xff, 0x8b, 0x9b, 0xd9, 0x8f, 0x1f, 0x0b, 0x00, 0x00,
+func init() { proto.RegisterFile("roachpb/app_stats.proto", fileDescriptor_app_stats_ad5cbe238e742d0d) }
+
+var fileDescriptor_app_stats_ad5cbe238e742d0d = []byte{
+	// 1199 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x56, 0x4f, 0x6f, 0x1b, 0xc5,
+	0x1b, 0xce, 0xda, 0x4e, 0x62, 0xbf, 0xb6, 0x93, 0x76, 0xd2, 0x36, 0xfb, 0xb3, 0x22, 0x3b, 0x3f,
+	0xab, 0x55, 0x53, 0x01, 0x8e, 0x14, 0x71, 0x01, 0x94, 0xb6, 0x49, 0xd3, 0x43, 0x4a, 0x28, 0xe0,
+	0x04, 0x21, 0x71, 0x59, 0x4d, 0x77, 0x5f, 0xa7, 0xa3, 0xec, 0xbf, 0xce, 0x8c, 0x5d, 0xe7, 0xce,
+	0x07, 0xe8, 0x47, 0xe0, 0x23, 0x70, 0xe7, 0xc8, 0xa5, 0x17, 0xa4, 0x1e, 0x2b, 0x0e, 0x01, 0x9c,
+	0x0b, 0x9f, 0x80, 0x03, 0x27, 0x34, 0x33, 0xbb, 0xcb, 0xda, 0x38, 0x62, 0xcb, 0xcd, 0x7e, 0xdf,
+	0xe7, 0x79, 0x66, 0xe6, 0x9d, 0xe7, 0x7d, 0x67, 0x61, 0x9d, 0x47, 0xd4, 0x7d, 0x1e, 0x3f, 0xdb,
+	0xa6, 0x71, 0xec, 0x08, 0x49, 0xa5, 0xe8, 0xc5, 0x3c, 0x92, 0x11, 0x69, 0xba, 0x91, 0x7b, 0xa6,
+	0x93, 0x3d, 0xf1, 0xc2, 0x6f, 0xdd, 0x38, 0x8d, 0x4e, 0x23, 0x9d, 0xd9, 0x56, 0xbf, 0x0c, 0xa8,
+	0xd5, 0x39, 0x8d, 0xa2, 0x53, 0x1f, 0xb7, 0xf5, 0xbf, 0x67, 0xc3, 0xc1, 0xb6, 0x64, 0x01, 0x0a,
+	0x49, 0x83, 0xd8, 0x00, 0xba, 0xdf, 0x57, 0x61, 0xed, 0x58, 0x52, 0x89, 0x01, 0x86, 0x52, 0xfd,
+	0x60, 0x42, 0x32, 0x57, 0x90, 0x16, 0x2c, 0xba, 0xd1, 0x30, 0x94, 0xb6, 0xb5, 0x69, 0x6d, 0x95,
+	0xf7, 0x2b, 0xaf, 0x2f, 0x3a, 0x0b, 0x7d, 0x13, 0x22, 0x1f, 0xc2, 0xda, 0x80, 0x71, 0x21, 0x1d,
+	0x2a, 0x25, 0x06, 0xb1, 0x74, 0x0c, 0xb2, 0x94, 0x43, 0x5e, 0xd7, 0x80, 0x3d, 0x93, 0x7f, 0xa4,
+	0x59, 0x77, 0xa0, 0x1e, 0xd0, 0xb1, 0xc3, 0x51, 0x72, 0x86, 0xc2, 0x2e, 0xe7, 0xd0, 0x10, 0xd0,
+	0x71, 0xdf, 0xc4, 0xc9, 0xfb, 0xb0, 0xea, 0xe3, 0x29, 0x75, 0xcf, 0x1d, 0x9f, 0x0a, 0xe9, 0x20,
+	0xe7, 0x76, 0x65, 0xd3, 0xda, 0xaa, 0x25, 0xd0, 0xa6, 0x49, 0x1e, 0x51, 0x21, 0x1f, 0x73, 0x4e,
+	0x3e, 0x81, 0x6a, 0x38, 0x0c, 0x1c, 0x1e, 0xbd, 0x14, 0xf6, 0xe2, 0xa6, 0xb5, 0x55, 0xdf, 0x69,
+	0xf5, 0xa6, 0xea, 0xd2, 0x7b, 0x3a, 0x0c, 0x90, 0x33, 0x57, 0x1d, 0x2d, 0x91, 0x58, 0x0e, 0x87,
+	0x41, 0x3f, 0x7a, 0x29, 0xc8, 0x2e, 0xd4, 0x62, 0xca, 0x05, 0x3a, 0x3e, 0x95, 0xf6, 0x52, 0x41,
+	0x76, 0x55, 0x53, 0x8e, 0xa8, 0x54, 0x6b, 0xc7, 0x3e, 0x0d, 0x35, 0x7b, 0xb9, 0xe8, 0xda, 0x8a,
+	0xa1, 0xc8, 0x1f, 0xc1, 0x32, 0x1f, 0x1a, 0x6e, 0xb5, 0x20, 0x77, 0x89, 0x0f, 0x35, 0x75, 0x0f,
+	0xea, 0x02, 0xf9, 0x88, 0xb9, 0x66, 0xe3, 0xb5, 0x82, 0x74, 0x48, 0x48, 0x4a, 0xe2, 0x11, 0x34,
+	0xa2, 0x11, 0xf2, 0xe7, 0x48, 0x3d, 0xad, 0x01, 0x05, 0x35, 0xea, 0x29, 0x4b, 0x89, 0xec, 0x82,
+	0x3d, 0x73, 0x53, 0x0e, 0x47, 0x8f, 0xba, 0x12, 0x3d, 0xbb, 0x9e, 0xbb, 0xb2, 0x9b, 0x53, 0x57,
+	0xd6, 0x4f, 0x20, 0xe4, 0x10, 0x56, 0x04, 0x86, 0x82, 0x49, 0x36, 0x42, 0x87, 0x85, 0x83, 0xc8,
+	0x6e, 0xe8, 0x5d, 0x6c, 0xcc, 0xec, 0xe2, 0x38, 0x05, 0x1d, 0x86, 0x83, 0x28, 0x75, 0x81, 0xc8,
+	0x07, 0xc9, 0x03, 0x80, 0x67, 0xe7, 0x12, 0x85, 0xc3, 0x91, 0x7a, 0xf6, 0x6a, 0xc1, 0xc3, 0xd4,
+	0x34, 0xa7, 0x8f, 0xd4, 0x53, 0x4e, 0x50, 0x16, 0x32, 0xfc, 0x6b, 0x45, 0x9d, 0xa0, 0x28, 0x9a,
+	0xfe, 0x35, 0xac, 0x9b, 0xf5, 0x05, 0x86, 0xd2, 0x51, 0x35, 0x72, 0x42, 0x94, 0x2f, 0x23, 0x7e,
+	0x66, 0x5f, 0x2f, 0x28, 0x76, 0x43, 0x0b, 0x1c, 0x63, 0x28, 0x3f, 0x1f, 0x21, 0x7f, 0x6a, 0xd8,
+	0xe4, 0x00, 0x9a, 0xaa, 0x67, 0x02, 0x0c, 0x9c, 0xa1, 0xa0, 0xa7, 0x68, 0x93, 0xa2, 0x17, 0x15,
+	0xd0, 0xf1, 0x67, 0x18, 0x7c, 0xa5, 0x48, 0x64, 0x0f, 0x5a, 0x38, 0x46, 0x57, 0x4f, 0x0f, 0xc7,
+	0x8d, 0x7c, 0x1f, 0x5d, 0xc9, 0xa2, 0x30, 0x69, 0xdb, 0xb5, 0x5c, 0x23, 0xae, 0x2b, 0x9c, 0x92,
+	0x79, 0x94, 0xa1, 0x74, 0xf3, 0x3e, 0xa9, 0x54, 0x9b, 0xd7, 0x56, 0x9e, 0x54, 0xaa, 0x2b, 0xd7,
+	0x56, 0xbb, 0x3f, 0x97, 0xe0, 0xe6, 0x09, 0xa7, 0xa1, 0xa0, 0x1a, 0x50, 0x70, 0x68, 0xcc, 0xb4,
+	0x7f, 0xe9, 0x8a, 0xf6, 0xcf, 0x37, 0x74, 0xf9, 0x5d, 0x1b, 0x7a, 0xa6, 0x33, 0x2a, 0xff, 0xa1,
+	0x33, 0x94, 0x13, 0x50, 0xf2, 0x73, 0x2d, 0xb0, 0x58, 0xd8, 0x09, 0x8a, 0xa2, 0xe8, 0x0f, 0x00,
+	0xdc, 0x28, 0x08, 0x98, 0x7c, 0xa7, 0x99, 0x52, 0x33, 0x9c, 0x23, 0x2a, 0xbb, 0xdf, 0x96, 0xa0,
+	0x39, 0xe5, 0x78, 0xd2, 0x81, 0x6a, 0x36, 0x09, 0xad, 0x5c, 0x5b, 0x2d, 0xfb, 0xc9, 0x0c, 0x64,
+	0xb0, 0x11, 0x44, 0x42, 0x3a, 0x1c, 0x5d, 0x65, 0x3f, 0x3d, 0x93, 0x3c, 0x14, 0x2e, 0x67, 0xb1,
+	0xba, 0x1f, 0x5d, 0xea, 0xfa, 0x4e, 0x77, 0x66, 0x17, 0x8f, 0xc7, 0xb1, 0x4f, 0x59, 0x78, 0xc2,
+	0x11, 0xbf, 0xf0, 0x69, 0xf8, 0x34, 0xf2, 0x30, 0x11, 0xfe, 0x9f, 0x52, 0xeb, 0x6b, 0x31, 0x95,
+	0x39, 0xf8, 0x5b, 0x8a, 0x50, 0x68, 0xfd, 0x63, 0xa9, 0xec, 0x45, 0xc9, 0xee, 0xcb, 0xbc, 0x39,
+	0xbd, 0xf4, 0xcd, 0xe9, 0x9d, 0xa4, 0x88, 0xfd, 0xaa, 0x5a, 0xe0, 0xd5, 0x2f, 0x1d, 0xab, 0xbf,
+	0x3e, 0xbd, 0x48, 0x06, 0xf9, 0xb8, 0xf2, 0xfb, 0x77, 0x1d, 0xab, 0xdb, 0x87, 0x7a, 0xae, 0x4c,
+	0xc4, 0x86, 0x4a, 0x80, 0x34, 0xd4, 0xe7, 0xb7, 0x92, 0x6d, 0xea, 0x08, 0xb9, 0x07, 0x4d, 0xf1,
+	0x62, 0x48, 0x39, 0x7a, 0x8e, 0xc7, 0x06, 0x03, 0x63, 0xac, 0x14, 0xd2, 0x48, 0x52, 0x07, 0x2a,
+	0xd3, 0xbd, 0xb4, 0xe0, 0xd6, 0x9c, 0xa7, 0xee, 0x53, 0x3c, 0x57, 0xc6, 0x7d, 0x31, 0x44, 0x7e,
+	0x3e, 0x55, 0x60, 0x13, 0x22, 0xb7, 0xa0, 0x4c, 0xe3, 0x58, 0xeb, 0xa6, 0x19, 0x15, 0x20, 0x6d,
+	0x58, 0xf6, 0x98, 0x90, 0xc7, 0x5f, 0x1e, 0xe9, 0x83, 0x57, 0xd3, 0x6b, 0x49, 0x82, 0x64, 0x03,
+	0x96, 0x06, 0x94, 0xf9, 0xe8, 0x69, 0x1f, 0xa6, 0xe9, 0x24, 0xa6, 0x54, 0xa3, 0xd8, 0x38, 0x2c,
+	0x4d, 0xa9, 0x00, 0xb9, 0x0b, 0x0d, 0x16, 0xc4, 0x3e, 0x73, 0x99, 0x74, 0xe4, 0x38, 0xd4, 0x16,
+	0x4a, 0x01, 0xf5, 0x34, 0x73, 0x32, 0x0e, 0x95, 0xc0, 0x08, 0x5d, 0xfd, 0xf0, 0x64, 0x02, 0x23,
+	0x74, 0xbb, 0x3f, 0x5a, 0xb0, 0x91, 0x74, 0x2f, 0x7a, 0xf3, 0x5e, 0xf6, 0x5d, 0x28, 0x9f, 0xa1,
+	0x39, 0x69, 0x7d, 0xe7, 0xce, 0xec, 0xb0, 0x9d, 0x5b, 0x9f, 0x54, 0xff, 0x0c, 0xcf, 0xc9, 0x7d,
+	0x58, 0xd4, 0x5f, 0x21, 0x57, 0xd8, 0x6a, 0x8e, 0x40, 0x5a, 0x4e, 0x4d, 0x23, 0xb7, 0xa1, 0xc4,
+	0x3c, 0x5d, 0xb1, 0xca, 0xfe, 0x0d, 0x95, 0x98, 0x5c, 0x74, 0x4a, 0x87, 0x07, 0x7f, 0x5e, 0x74,
+	0x96, 0x8e, 0x65, 0x20, 0x0f, 0x0f, 0xfa, 0x25, 0xe6, 0x75, 0x7f, 0xb0, 0xa0, 0x9d, 0x9d, 0x62,
+	0xfe, 0xb0, 0x79, 0x00, 0x4d, 0x91, 0x2e, 0xe6, 0x30, 0x4f, 0xd8, 0xd6, 0x66, 0x79, 0xab, 0xb2,
+	0xdf, 0x9a, 0x5c, 0x74, 0x1a, 0xd9, 0x2e, 0x0e, 0x0f, 0x44, 0x4e, 0xb9, 0x91, 0x11, 0x0e, 0x3d,
+	0x71, 0xe5, 0xc5, 0x3e, 0x4c, 0x4f, 0x68, 0xfc, 0x7c, 0x7b, 0xe6, 0x84, 0x73, 0x77, 0x33, 0x75,
+	0xc6, 0xee, 0x1f, 0x16, 0xac, 0xcd, 0xe9, 0x2f, 0x65, 0xe3, 0x90, 0x06, 0x38, 0xe5, 0x32, 0x1d,
+	0x21, 0xbb, 0xb0, 0x48, 0xa5, 0xe4, 0xaa, 0xaa, 0xe5, 0xad, 0xfa, 0xce, 0xdd, 0x7f, 0x6f, 0xd6,
+	0xde, 0x9e, 0x94, 0xbc, 0x6f, 0x58, 0xe4, 0x3e, 0x54, 0xdd, 0xe7, 0xcc, 0xf7, 0x38, 0x86, 0x76,
+	0x59, 0x2b, 0x14, 0x68, 0xf7, 0x7e, 0xc6, 0x69, 0x3d, 0x84, 0x8a, 0x92, 0x53, 0x25, 0x49, 0xbd,
+	0x51, 0xcb, 0x5f, 0x7a, 0x0b, 0x16, 0x47, 0xd4, 0x1f, 0xe2, 0x54, 0xb1, 0x4c, 0xc8, 0x34, 0x6c,
+	0xd2, 0xb6, 0x3f, 0x59, 0x50, 0x3d, 0x19, 0xeb, 0xba, 0x08, 0xf2, 0x7f, 0xa8, 0xc9, 0x71, 0xfa,
+	0xca, 0xe4, 0x5f, 0x84, 0xaa, 0x1c, 0x9b, 0x67, 0x85, 0xec, 0x43, 0x43, 0x41, 0xd4, 0x08, 0x71,
+	0x04, 0xba, 0x89, 0xa7, 0x0a, 0x4c, 0x6c, 0x39, 0xd6, 0x23, 0xe3, 0x18, 0x5d, 0xf2, 0x01, 0xac,
+	0x9a, 0xf1, 0x29, 0xd1, 0x4b, 0x16, 0xcb, 0x7f, 0x5b, 0xae, 0x64, 0x49, 0xb3, 0xe4, 0x7b, 0xb0,
+	0x92, 0x35, 0x98, 0x41, 0x57, 0x72, 0xe8, 0x66, 0x9a, 0xd3, 0xe0, 0xfd, 0x7b, 0xaf, 0x7f, 0x6b,
+	0x2f, 0xbc, 0x9e, 0xb4, 0xad, 0x37, 0x93, 0xb6, 0xf5, 0x76, 0xd2, 0xb6, 0x7e, 0x9d, 0xb4, 0xad,
+	0x57, 0x97, 0xed, 0x85, 0x37, 0x97, 0xed, 0x85, 0xb7, 0x97, 0xed, 0x85, 0x6f, 0x96, 0x93, 0xcf,
+	0xf3, 0xbf, 0x02, 0x00, 0x00, 0xff, 0xff, 0xa2, 0xb5, 0xaa, 0x8f, 0xa8, 0x0b, 0x00, 0x00,
 }
diff --git a/pkg/roachpb/app_stats.proto b/pkg/roachpb/app_stats.proto
index 690f6016a662..bd8a568b3500 100644
--- a/pkg/roachpb/app_stats.proto
+++ b/pkg/roachpb/app_stats.proto
@@ -15,6 +15,9 @@ option go_package = "roachpb";
 import "gogoproto/gogo.proto";
 import "google/protobuf/timestamp.proto";
 
+// StatementStatistics represent the statement statistics sent to the DB
+// Console for a given statement fingerprint. Note that these stats are cleared
+// ever diagnostics.sql_stat_reset_interval.
 // N.B. When fields are added to this struct, make sure to update
 // (*StatementStatistics).Add and (*StatementStatistics).AlmostEqual
 // in app_stats.go.
@@ -89,7 +92,19 @@ message StatementStatistics {
   // BytesSentOverNetwork collects the number of bytes sent over the network.
   optional NumericStat bytes_sent_over_network = 17 [(gogoproto.nullable) = false];
 
-  // Note: be sure to update `sql/app_stats.go` when adding/removing fields here!
+  // MaxMemUsage collects the maximum memory usage that occurred on a node.
+  optional NumericStat max_mem_usage = 18 [(gogoproto.nullable) = false];
+
+  // ExecStatCollectionCount keeps track of how many times execution stats were
+  // recorded for this statement. Since this collection follows a sampling
+  // approach, this number is not necessarily equal to Count. Used to calculate
+  // the mean of the following NumericStat values:
+  //  bytes_sent_over_network
+  //  max_mem_usage
+  optional int64 exec_stat_collection_count = 19 [(gogoproto.nullable) = false];
+
+  // Note: be sure to update `sql/app_stats.go` and the comment above
+  // exec_stat_collection_count when adding/removing fields here!
 }
 
 message TransactionStatistics {
diff --git a/pkg/sql/BUILD.bazel b/pkg/sql/BUILD.bazel
index 94d85b53a113..2b6645a9b902 100644
--- a/pkg/sql/BUILD.bazel
+++ b/pkg/sql/BUILD.bazel
@@ -417,6 +417,7 @@ go_test(
         "explain_test.go",
         "explain_tree_test.go",
         "indexbackfiller_test.go",
+        "instrumentation_test.go",
         "internal_test.go",
         "main_test.go",
         "materialized_view_test.go",
diff --git a/pkg/sql/conn_executor.go b/pkg/sql/conn_executor.go
index 66446b9ff8e1..7d105584d3b0 100644
--- a/pkg/sql/conn_executor.go
+++ b/pkg/sql/conn_executor.go
@@ -15,6 +15,7 @@ import (
 	"fmt"
 	"io"
 	"math"
+	"math/rand"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -595,6 +596,7 @@ func (s *Server) newConnExecutor(
 		// ctxHolder will be reset at the start of run(). We only define
 		// it here so that an early call to close() doesn't panic.
 		ctxHolder:                 ctxHolder{connCtx: ctx},
+		rng:                       rand.New(rand.NewSource(timeutil.Now().UnixNano())),
 		executorType:              executorTypeExec,
 		hasCreatedTemporarySchema: false,
 		stmtDiagnosticsRecorder:   s.cfg.StmtDiagnosticsRecorder,
@@ -1077,6 +1079,10 @@ type connExecutor struct {
 	// transactions.
 	statsCollector *sqlStatsCollector
 
+	// rng is used to generate random numbers. An example usage is to determine
+	// whether to sample execution stats.
+	rng *rand.Rand
+
 	// mu contains of all elements of the struct that can be changed
 	// after initialization, and may be accessed from another thread.
 	mu struct {
diff --git a/pkg/sql/conn_executor_exec.go b/pkg/sql/conn_executor_exec.go
index ddb548114391..4b71efdac75d 100644
--- a/pkg/sql/conn_executor_exec.go
+++ b/pkg/sql/conn_executor_exec.go
@@ -383,7 +383,7 @@ func (ex *connExecutor) execStmtInOpenState(
 	var needFinish bool
 	ctx, needFinish = ih.Setup(
 		ctx, ex.server.cfg, ex.appStats, p, ex.stmtDiagnosticsRecorder,
-		stmt.AnonymizedStr, os.ImplicitTxn.Get(),
+		stmt.AnonymizedStr, os.ImplicitTxn.Get(), ex.rng,
 	)
 	if needFinish {
 		sql := stmt.SQL
diff --git a/pkg/sql/instrumentation.go b/pkg/sql/instrumentation.go
index 7240b2ab5019..005125c4c827 100644
--- a/pkg/sql/instrumentation.go
+++ b/pkg/sql/instrumentation.go
@@ -14,10 +14,12 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"math/rand"
 
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/server/telemetry"
+	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/execstats"
@@ -30,6 +32,19 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing/tracingpb"
+	"github.com/cockroachdb/errors"
+)
+
+var collectStmtStatsSampleRate = settings.RegisterFloatSetting(
+	"sql.statement_stats.sample_rate",
+	"the probability that a given statement will collect execution statistics (displayed in the DB Console)",
+	0,
+	func(f float64) error {
+		if f < 0 || f > 1 {
+			return errors.New("value must be between 0 and 1 inclusive")
+		}
+		return nil
+	},
 )
 
 // instrumentationHelper encapsulates the logic around extracting information
@@ -122,6 +137,7 @@ func (ih *instrumentationHelper) Setup(
 	stmtDiagnosticsRecorder *stmtdiagnostics.Registry,
 	fingerprint string,
 	implicitTxn bool,
+	rng *rand.Rand,
 ) (newCtx context.Context, needFinish bool) {
 	ih.fingerprint = fingerprint
 	ih.implicitTxn = implicitTxn
@@ -148,8 +164,26 @@ func (ih *instrumentationHelper) Setup(
 
 	ih.savePlanForStats = appStats.shouldSaveLogicalPlanDescription(fingerprint, implicitTxn)
 
+	if sp := tracing.SpanFromContext(ctx); sp != nil && sp.IsVerbose() {
+		// If verbose tracing was enabled at a higher level, stats collection is
+		// enabled so that stats are shown in the traces, but no extra work is
+		// needed by the instrumentationHelper.
+		ih.collectExecStats = true
+		return ctx, false
+	}
+
+	if statsSampleRate := collectStmtStatsSampleRate.Get(&cfg.Settings.SV); statsSampleRate > 0 {
+		ih.collectExecStats = statsSampleRate > rng.Float64()
+	}
+
 	if !ih.collectBundle && ih.withStatementTrace == nil && ih.outputMode == unmodifiedOutput {
-		// TODO(asubiotto): Create a span for stat collection in future commit.
+		if ih.collectExecStats {
+			// If we need to collect stats, create a non-verbose child span. Stats
+			// will be added as structured metadata and processed in Finish.
+			ih.origCtx = ctx
+			newCtx, ih.sp = tracing.EnsureChildSpan(ctx, cfg.AmbientCtx.Tracer, "traced statement")
+			return newCtx, true
+		}
 		return ctx, false
 	}
 
@@ -197,18 +231,21 @@ func (ih *instrumentationHelper) Finish(
 		for _, flowInfo := range p.curPlan.distSQLFlowInfos {
 			flowMetadata = append(flowMetadata, flowInfo.flowMetadata)
 		}
-		queryLevelStats, error := execstats.GetQueryLevelStats(trace, cfg.TestingKnobs.DeterministicExplainAnalyze, flowMetadata)
-		if error != nil {
-			log.VInfof(ctx, 1, "error getting query level stats for statement %s: %+v", ast, error)
+		queryLevelStats, err := execstats.GetQueryLevelStats(trace, cfg.TestingKnobs.DeterministicExplainAnalyze, flowMetadata)
+		if err != nil {
+			log.VInfof(ctx, 1, "error getting query level stats for statement %s: %+v", ast, err)
+		} else {
+			stmtStats.mu.Lock()
+			stmtStats.mu.data.ExecStatCollectionCount++
+			// Record trace-related statistics.
+			stmtStats.mu.data.BytesSentOverNetwork.Record(
+				stmtStats.mu.data.ExecStatCollectionCount, float64(queryLevelStats.NetworkBytesSent),
+			)
+			stmtStats.mu.data.MaxMemUsage.Record(
+				stmtStats.mu.data.ExecStatCollectionCount, float64(queryLevelStats.MaxMemUsage),
+			)
+			stmtStats.mu.Unlock()
 		}
-
-		stmtStats.mu.Lock()
-		// Record trace-related statistics. A count of 1 is passed given that this
-		// statistic is only recorded when statement diagnostics are enabled.
-		// TODO(asubiotto): NumericStat properties will be properly calculated
-		//  once this statistic is always collected.
-		stmtStats.mu.data.BytesSentOverNetwork.Record(1 /* count */, float64(queryLevelStats.NetworkBytesSent))
-		stmtStats.mu.Unlock()
 	}
 
 	var bundle diagnosticsBundle
diff --git a/pkg/sql/instrumentation_test.go b/pkg/sql/instrumentation_test.go
new file mode 100644
index 000000000000..561f3fab48ac
--- /dev/null
+++ b/pkg/sql/instrumentation_test.go
@@ -0,0 +1,64 @@
+// Copyright 2021 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package sql
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSampledStatsCollection(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	st := cluster.MakeTestingClusterSettings()
+	s, db, _ := serverutils.StartServer(t, base.TestServerArgs{Settings: st})
+	defer s.Stopper().Stop(ctx)
+
+	sqlutils.CreateTable(
+		t, db, "test", "x INT", 10, sqlutils.ToRowFn(sqlutils.RowIdxFn),
+	)
+
+	const query = "SELECT * FROM test.test ORDER BY x"
+
+	collectStmtStatsSampleRate.Override(&st.SV, 0)
+
+	r, err := db.Query(query)
+	require.NoError(t, err)
+	require.NoError(t, r.Close())
+
+	collectStmtStatsSampleRate.Override(&st.SV, 1)
+
+	r, err = db.Query(query)
+	require.NoError(t, err)
+	require.NoError(t, r.Close())
+
+	sqlServer := s.SQLServer().(*Server)
+	appStats := sqlServer.sqlStats.getStatsForApplication("")
+	require.NotNil(t, appStats, "could not find app stats for default app")
+	stmtStats, _ := appStats.getStatsForStmt(query, true /* implicitTxn */, nil /* err */, false /* createIfNonexistent */)
+	require.NotNil(t, stmtStats, "could not find stmt stats for %s", query)
+
+	stmtStats.mu.Lock()
+	defer stmtStats.mu.Unlock()
+	require.Equal(t, int64(2), stmtStats.mu.data.Count, "expected to have collected two sets of general stats")
+	require.Equal(t, int64(1), stmtStats.mu.data.ExecStatCollectionCount, "expected to have collected exactly one set of execution stats")
+	require.Greater(t, stmtStats.mu.data.BytesRead.Mean, float64(0), "expected statement to have read at least one byte")
+}

From 2615ca99a94956f841cfb793f0e1396b1fac1792 Mon Sep 17 00:00:00 2001
From: Alfonso Subiotto Marques <alfonso@cockroachlabs.com>
Date: Thu, 28 Jan 2021 12:39:50 +0100
Subject: [PATCH 3/4] ui: update UI to handle newly-added protobuf fields

Release note: None
---
 pkg/ui/src/util/appStats.spec.ts              | 10 ++++++
 pkg/ui/src/util/appStats.ts                   | 33 +++++++++++++++----
 .../src/views/statements/statements.spec.tsx  |  3 ++
 3 files changed, 39 insertions(+), 7 deletions(-)

diff --git a/pkg/ui/src/util/appStats.spec.ts b/pkg/ui/src/util/appStats.spec.ts
index ad56b4e29709..cbc028eb9d93 100644
--- a/pkg/ui/src/util/appStats.spec.ts
+++ b/pkg/ui/src/util/appStats.spec.ts
@@ -160,6 +160,7 @@ function randomStats(sensitiveInfo?: ISensitiveInfo): StatementStatistics {
   const count = randomInt(1000);
   const first_attempt_count = randomInt(count);
   const max_retries = randomInt(count - first_attempt_count);
+  const exec_stat_collection_count = randomInt(count);
 
   return {
     count: Long.fromNumber(count),
@@ -174,6 +175,7 @@ function randomStats(sensitiveInfo?: ISensitiveInfo): StatementStatistics {
     bytes_read: randomStat(),
     rows_read: randomStat(),
     sensitive_info: sensitiveInfo || makeSensitiveInfo(null, null),
+    exec_stat_collection_count: Long.fromNumber(exec_stat_collection_count),
   };
 }
 
@@ -225,6 +227,14 @@ describe("combineStatementStats", () => {
 
     assert.equal(ab_c.count.toString(), ac_b.count.toString());
     assert.equal(ab_c.count.toString(), bc_a.count.toString());
+    assert.equal(
+      ab_c.exec_stat_collection_count.toString(),
+      ac_b.exec_stat_collection_count.toString(),
+    );
+    assert.equal(
+      ab_c.exec_stat_collection_count.toString(),
+      bc_a.exec_stat_collection_count.toString(),
+    );
 
     assert.equal(
       ab_c.first_attempt_count.toString(),
diff --git a/pkg/ui/src/util/appStats.ts b/pkg/ui/src/util/appStats.ts
index 523227e0e60c..a4b3a829ee33 100644
--- a/pkg/ui/src/util/appStats.ts
+++ b/pkg/ui/src/util/appStats.ts
@@ -61,6 +61,13 @@ export function addStatementStats(
 ): Required<StatementStatistics> {
   const countA = FixLong(a.count).toInt();
   const countB = FixLong(b.count).toInt();
+  let execStatCountA = FixLong(a.exec_stat_collection_count).toInt();
+  const execStatCountB = FixLong(b.exec_stat_collection_count).toInt();
+  if (execStatCountA === 0 && execStatCountB === 0) {
+    // If both counts are zero, artificially set the one count to one to avoid
+    // division by zero when calculating the mean in addNumericStats.
+    execStatCountA = 1;
+  }
   return {
     count: a.count.add(b.count),
     first_attempt_count: a.first_attempt_count.add(b.first_attempt_count),
@@ -78,20 +85,32 @@ export function addStatementStats(
       countA,
       countB,
     ),
+    bytes_read: addNumericStats(a.bytes_read, b.bytes_read, countA, countB),
+    rows_read: addNumericStats(a.rows_read, b.rows_read, countA, countB),
+    sensitive_info: coalesceSensitiveInfo(a.sensitive_info, b.sensitive_info),
+    legacy_last_err: "",
+    legacy_last_err_redacted: "",
     bytes_sent_over_network:
       a.bytes_sent_over_network && b.bytes_sent_over_network
         ? addNumericStats(
             a.bytes_sent_over_network,
             b.bytes_sent_over_network,
-            countA,
-            countB,
+            execStatCountA,
+            execStatCountB,
           )
         : null,
-    bytes_read: addNumericStats(a.bytes_read, b.bytes_read, countA, countB),
-    rows_read: addNumericStats(a.rows_read, b.rows_read, countA, countB),
-    sensitive_info: coalesceSensitiveInfo(a.sensitive_info, b.sensitive_info),
-    legacy_last_err: "",
-    legacy_last_err_redacted: "",
+    max_mem_usage:
+      a.max_mem_usage && b.max_mem_usage
+        ? addNumericStats(
+            a.max_mem_usage,
+            b.max_mem_usage,
+            execStatCountA,
+            execStatCountB,
+          )
+        : null,
+    exec_stat_collection_count: a.exec_stat_collection_count.add(
+      b.exec_stat_collection_count,
+    ),
   };
 }
 
diff --git a/pkg/ui/src/views/statements/statements.spec.tsx b/pkg/ui/src/views/statements/statements.spec.tsx
index 570f29234150..39b8b877c54f 100644
--- a/pkg/ui/src/views/statements/statements.spec.tsx
+++ b/pkg/ui/src/views/statements/statements.spec.tsx
@@ -490,6 +490,9 @@ function makeStats() {
     sensitive_info: makeEmptySensitiveInfo(),
     rows_read: makeStat(),
     bytes_read: makeStat(),
+    bytes_sent_over_network: makeStat(),
+    max_mem_usage: makeStat(),
+    exec_stat_collection_count: Long.fromNumber(10),
   };
 }
 

From 0d79e70beb9ee349d29d92128c59c26f1cd383eb Mon Sep 17 00:00:00 2001
From: irfan sharif <irfanmahmoudsharif@gmail.com>
Date: Thu, 28 Jan 2021 09:03:33 -0500
Subject: [PATCH 4/4] colexec,bazel: pin the `types` dependency in generated
 files

This is a workaround for bazel auto-generated code. goimports does not
automatically pick up the right packages when run within the bazel
sandbox, so we have to pin it by hand.

Release note: None
---
 pkg/sql/colexec/crossjoiner.eg.go   | 1 +
 pkg/sql/colexec/crossjoiner_tmpl.go | 1 +
 2 files changed, 2 insertions(+)

diff --git a/pkg/sql/colexec/crossjoiner.eg.go b/pkg/sql/colexec/crossjoiner.eg.go
index 6d7739936637..20807f870621 100644
--- a/pkg/sql/colexec/crossjoiner.eg.go
+++ b/pkg/sql/colexec/crossjoiner.eg.go
@@ -23,6 +23,7 @@ import (
 // pick up the right packages when run within the bazel sandbox.
 var (
 	_ = typeconv.DatumVecCanonicalTypeFamily
+	_ = types.BoolFamily
 )
 
 // buildFromLeftInput builds part of the output of a cross join that comes from
diff --git a/pkg/sql/colexec/crossjoiner_tmpl.go b/pkg/sql/colexec/crossjoiner_tmpl.go
index 31f44a75c79a..6f87523fcc07 100644
--- a/pkg/sql/colexec/crossjoiner_tmpl.go
+++ b/pkg/sql/colexec/crossjoiner_tmpl.go
@@ -34,6 +34,7 @@ import (
 // pick up the right packages when run within the bazel sandbox.
 var (
 	_ = typeconv.DatumVecCanonicalTypeFamily
+	_ = types.BoolFamily
 )
 
 // buildFromLeftInput builds part of the output of a cross join that comes from