sql: add index_bytes_written metric

andy-kimball · andy-kimball · commit 89f5c678317c · 2025-10-30T07:48:11.000-07:00
Add new `sql.statements.index_bytes_written.count` metric that counts the number
of primary and secondary index bytes modified by SQL statements.

Epic: AOP-30

Release note (sql change): Added sql.statements.index_bytes_written.count metric
that counts the number of primary and secondary index bytes modified by SQL
statements.
diff --git a/docs/generated/metrics/metrics.yaml b/docs/generated/metrics/metrics.yaml
@@ -8969,6 +8969,22 @@ layers:
       unit: BYTES
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: sql.statements.index_bytes_written.count
+      exported_name: sql_statements_index_bytes_written_count
+      description: Number of primary and secondary index bytes modified by SQL statements
+      y_axis_label: SQL Statements
+      type: COUNTER
+      unit: BYTES
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: sql.statements.index_bytes_written.count.internal
+      exported_name: sql_statements_index_bytes_written_count_internal
+      description: Number of primary and secondary index bytes modified by SQL statements (internal queries)
+      y_axis_label: SQL Internal Statements
+      type: COUNTER
+      unit: BYTES
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: sql.statements.index_rows_written.count
       exported_name: sql_statements_index_rows_written_count
       description: Number of primary and secondary index rows modified by SQL statements
diff --git a/pkg/roachprod/opentelemetry/cockroachdb_metrics.go b/pkg/roachprod/opentelemetry/cockroachdb_metrics.go
@@ -1878,6 +1878,8 @@ var cockroachdbMetrics = map[string]string{
 	"sql_statements_bytes_read_count_internal":                    "sql.statements.bytes_read.count.internal",
 	"sql_statements_index_rows_written_count":                     "sql.statements.index_rows_written.count",
 	"sql_statements_index_rows_written_count_internal":            "sql.statements.index_rows_written.count.internal",
+	"sql_statements_index_bytes_written_count":                    "sql.statements.index_bytes_written.count",
+	"sql_statements_index_bytes_written_count_internal":           "sql.statements.index_bytes_written.count.internal",
 	"sql_stats_activity_update_latency":                           "sql.stats.activity.update.latency",
 	"sql_stats_activity_update_latency_bucket":                    "sql.stats.activity.update.latency.bucket",
 	"sql_stats_activity_update_latency_count":                     "sql.stats.activity.update.latency.count",
diff --git a/pkg/sql/conn_executor.go b/pkg/sql/conn_executor.go
@@ -628,6 +628,7 @@ func makeMetrics(internal bool, sv *settings.Values) Metrics {
 			StatementRowsRead:                 metric.NewCounter(getMetricMeta(MetaStatementRowsRead, internal)),
 			StatementBytesRead:                metric.NewCounter(getMetricMeta(MetaStatementBytesRead, internal)),
 			StatementIndexRowsWritten:         metric.NewCounter(getMetricMeta(MetaStatementIndexRowsWritten, internal)),
+			StatementIndexBytesWritten:        metric.NewCounter(getMetricMeta(MetaStatementIndexBytesWritten, internal)),
 		},
 		StartedStatementCounters:  makeStartedStatementCounters(internal),
 		ExecutedStatementCounters: makeExecutedStatementCounters(internal),
diff --git a/pkg/sql/conn_executor_exec.go b/pkg/sql/conn_executor_exec.go
@@ -3279,10 +3279,10 @@ func (ex *connExecutor) makeExecPlan(
 
 // topLevelQueryStats returns some basic statistics about the run of the query.
 type topLevelQueryStats struct {
-	// bytesRead is the number of bytes read from primary and secondary indexes.
-	bytesRead int64
 	// rowsRead is the number of rows read from primary and secondary indexes.
 	rowsRead int64
+	// bytesRead is the number of bytes read from primary and secondary indexes.
+	bytesRead int64
 	// rowsWritten is the number of rows written to the primary index. It does not
 	// include rows written to secondary indexes.
 	// NB: There is an asymmetry between rowsRead and rowsWritten - rowsRead
@@ -3293,6 +3293,9 @@ type topLevelQueryStats struct {
 	// indexRowsWritten is the number of rows written to primary and secondary
 	// indexes. It is always >= rowsWritten.
 	indexRowsWritten int64
+	// indexBytesWritten is the number of bytes written to primary and secondary
+	// indexes.
+	indexBytesWritten int64
 	// networkEgressEstimate is an estimate for the number of bytes sent to the
 	// client. It is used for estimating the number of RUs consumed by a query.
 	networkEgressEstimate int64
@@ -3306,6 +3309,7 @@ func (s *topLevelQueryStats) add(other *topLevelQueryStats) {
 	s.bytesRead += other.bytesRead
 	s.rowsRead += other.rowsRead
 	s.rowsWritten += other.rowsWritten
+	s.indexBytesWritten += other.indexBytesWritten
 	s.indexRowsWritten += other.indexRowsWritten
 	s.networkEgressEstimate += other.networkEgressEstimate
 	s.clientTime += other.clientTime
diff --git a/pkg/sql/delete.go b/pkg/sql/delete.go
@@ -263,6 +263,11 @@ func (d *deleteNode) indexRowsWritten() int64 {
 	return d.run.td.indexRowsWritten
 }
 
+func (d *deleteNode) indexBytesWritten() int64 {
+	// No bytes counted as written for a deletion.
+	return 0
+}
+
 func (d *deleteNode) enableAutoCommit() {
 	d.run.td.enableAutoCommit()
 }
diff --git a/pkg/sql/delete_range.go b/pkg/sql/delete_range.go
@@ -89,6 +89,11 @@ func (d *deleteRangeNode) indexRowsWritten() int64 {
 	return int64(d.rowCount)
 }
 
+func (d *deleteRangeNode) indexBytesWritten() int64 {
+	// No bytes counted as written for a deletion.
+	return 0
+}
+
 // startExec implements the planNode interface.
 func (d *deleteRangeNode) startExec(params runParams) error {
 	if err := params.p.cancelChecker.Check(); err != nil {
diff --git a/pkg/sql/delete_swap.go b/pkg/sql/delete_swap.go
@@ -147,6 +147,11 @@ func (d *deleteSwapNode) indexRowsWritten() int64 {
 	return d.run.td.indexRowsWritten
 }
 
+func (d *deleteSwapNode) indexBytesWritten() int64 {
+	// No bytes counted as written for a deletion.
+	return 0
+}
+
 func (d *deleteSwapNode) enableAutoCommit() {
 	d.run.td.enableAutoCommit()
 }
diff --git a/pkg/sql/distsql_running.go b/pkg/sql/distsql_running.go
@@ -1607,6 +1607,7 @@ func (r *DistSQLReceiver) pushMeta(meta *execinfrapb.ProducerMetadata) execinfra
 		r.stats.rowsRead += meta.Metrics.RowsRead
 		r.stats.rowsWritten += meta.Metrics.RowsWritten
 		r.stats.indexRowsWritten += meta.Metrics.IndexRowsWritten
+		r.stats.indexBytesWritten += meta.Metrics.IndexBytesWritten
 
 		if sm, ok := r.scanStageEstimateMap[meta.Metrics.StageID]; ok {
 			sm.rowsRead += uint64(meta.Metrics.RowsRead)
diff --git a/pkg/sql/exec_util.go b/pkg/sql/exec_util.go
@@ -1532,6 +1532,12 @@ var (
 		Measurement: "SQL Statements",
 		Unit:        metric.Unit_COUNT,
 	}
+	MetaStatementIndexBytesWritten = metric.Metadata{
+		Name:        "sql.statements.index_bytes_written.count",
+		Help:        "Number of primary and secondary index bytes modified by SQL statements",
+		Measurement: "SQL Statements",
+		Unit:        metric.Unit_BYTES,
+	}
 )
 
 func getMetricMeta(meta metric.Metadata, internal bool) metric.Metadata {
diff --git a/pkg/sql/execinfrapb/data.proto b/pkg/sql/execinfrapb/data.proto
@@ -345,14 +345,17 @@ message RemoteProducerMetadata {
     // IndexRowsWritten is the number of primary and secondary index rows
     // modified while executing a statement. It is always >= RowsWritten.
     optional int64 index_rows_written = 5 [(gogoproto.nullable) = false];
+    // IndexBytesWritten is the number of primary and secondary index bytes
+    // modified while executing a statement.
+    optional int64 index_bytes_written = 6 [(gogoproto.nullable) = false];
     // Stage identifier that produced these metrics. This is used to aggregate
     // row counts across distributed table reader processors for misestimate
     // logging. Note that other disk-reading processors (index and lookup joins,
     // inverted joins, zigzag joins) do not set this field. This is safe because
     // we enumerate physical plan stages starting at 1, so the default value of
     // 0 isn't conflated with any actual plan stage.
     optional int32 stage_id = 4 [(gogoproto.nullable) = false, (gogoproto.customname) = "StageID"];
-    // NEXT ID: 6
+    // NEXT ID: 7
   }
   oneof value {
     RangeInfos range_info = 1;
@@ -367,7 +370,7 @@ message RemoteProducerMetadata {
     TracingAggregatorEvents tracing_aggregator_events= 12;
   }
   reserved 6, 10;
-  // NEXT ID: 14
+  // NEXT ID: 13
 }
 
 // DistSQLDrainingInfo represents the DistSQL draining state that gets gossiped
diff --git a/pkg/sql/executor_statement_metrics.go b/pkg/sql/executor_statement_metrics.go
@@ -96,6 +96,10 @@ type EngineMetrics struct {
 	// StatementIndexRowsWritten counts the number of primary and secondary index
 	// rows modified by SQL statements.
 	StatementIndexRowsWritten *metric.Counter
+
+	// StatementIndexBytesWritten counts the number of primary and secondary index
+	// bytes modified by SQL statements.
+	StatementIndexBytesWritten *metric.Counter
 }
 
 // EngineMetrics implements the metric.Struct interface.
@@ -214,6 +218,7 @@ func (ex *connExecutor) recordStatementSummary(
 	ex.metrics.EngineMetrics.StatementRowsRead.Inc(stats.rowsRead)
 	ex.metrics.EngineMetrics.StatementBytesRead.Inc(stats.bytesRead)
 	ex.metrics.EngineMetrics.StatementIndexRowsWritten.Inc(stats.indexRowsWritten)
+	ex.metrics.EngineMetrics.StatementIndexBytesWritten.Inc(stats.indexBytesWritten)
 
 	if ex.statsCollector.EnabledForTransaction() {
 		recordedStmtStats := &sqlstats.RecordedStmtStats{
diff --git a/pkg/sql/insert.go b/pkg/sql/insert.go
@@ -392,3 +392,7 @@ func (n *insertNode) rowsWritten() int64 {
 func (n *insertNode) indexRowsWritten() int64 {
 	return n.run.ti.indexRowsWritten
 }
+
+func (n *insertNode) indexBytesWritten() int64 {
+	return n.run.ti.indexBytesWritten
+}
diff --git a/pkg/sql/insert_fast_path.go b/pkg/sql/insert_fast_path.go
@@ -552,6 +552,10 @@ func (n *insertFastPathNode) indexRowsWritten() int64 {
 	return n.run.ti.indexRowsWritten
 }
 
+func (n *insertFastPathNode) indexBytesWritten() int64 {
+	return n.run.ti.indexBytesWritten
+}
+
 // See planner.autoCommit.
 func (n *insertFastPathNode) enableAutoCommit() {
 	n.run.ti.enableAutoCommit()
diff --git a/pkg/sql/metric_test.go b/pkg/sql/metric_test.go
@@ -224,6 +224,9 @@ func TestStatementMetrics(t *testing.T) {
 		indexRowsWritten int64
 		// If true, then skip metamorphic testing for this query.
 		skipMetamorphic bool
+		// If true, then this is a deletion case, in which case indexBytesWritten
+		// is expected to be zero.
+		isDelete bool
 	}{
 		// Read-only statements.
 		{query: `SELECT * FROM db.t WHERE x = 1`, rowsRead: 1},
@@ -262,8 +265,15 @@ func TestStatementMetrics(t *testing.T) {
 			rowsRead:         1,
 			indexRowsWritten: 3,
 		},
-		{query: `DELETE FROM db.t WHERE x = 2`, rowsRead: 1, indexRowsWritten: 4},
-		{query: `DELETE FROM db.t WHERE a @> ARRAY['apple']`, rowsRead: 4, indexRowsWritten: 11},
+		{query: `DELETE FROM db.t WHERE x = 2`, rowsRead: 1, indexRowsWritten: 4, isDelete: true},
+		{
+			query:            `DELETE FROM db.t WHERE a @> ARRAY['apple']`,
+			rowsRead:         4,
+			indexRowsWritten: 11,
+			isDelete:         true,
+		},
+		// CREATE TABLE does not count rows or bytes written.
+		{query: `CREATE TABLE db.t2 (x, y) AS SELECT 1, 10 UNION SELECT 2, 20`},
 	}
 
 	for _, vectorized := range []string{"off", "on"} {
@@ -289,14 +299,14 @@ func TestStatementMetrics(t *testing.T) {
 					lastRowsRead := s.MustGetSQLCounter(sql.MetaStatementRowsRead.Name)
 					lastBytesRead := s.MustGetSQLCounter(sql.MetaStatementBytesRead.Name)
 					lastRowsWritten := s.MustGetSQLCounter(sql.MetaStatementIndexRowsWritten.Name)
+					lastBytesWritten := s.MustGetSQLCounter(sql.MetaStatementIndexBytesWritten.Name)
 
 					runner.Exec(t, tc.query)
 
 					// Test the rows read metric.
-					actual, err := checkCounterDelta(
+					_, err := checkCounterDelta(
 						s, sql.MetaStatementRowsRead, lastRowsRead, tc.rowsRead)
 					require.NoError(t, err)
-					lastRowsRead = actual
 
 					// If there were rows read, then expect bytes read to have
 					// increased. Don't check for a specific value, since there are
@@ -307,10 +317,22 @@ func TestStatementMetrics(t *testing.T) {
 					}
 
 					// Test the rows written metric.
-					actual, err = checkCounterDelta(
+					_, err = checkCounterDelta(
 						s, sql.MetaStatementIndexRowsWritten, lastRowsWritten, tc.indexRowsWritten)
 					require.NoError(t, err)
-					lastRowsWritten = actual
+
+					// If there were index rows written, then expect bytes written
+					// to have increased (or not in the delete case). Don't check
+					// for a specific value, since there are too many factors that
+					// can change this.
+					if tc.indexRowsWritten > 0 {
+						currBytesWritten := s.MustGetSQLCounter(sql.MetaStatementIndexBytesWritten.Name)
+						if !tc.isDelete {
+							require.Greater(t, currBytesWritten, lastBytesWritten)
+						} else {
+							require.Equal(t, currBytesWritten, lastBytesWritten)
+						}
+					}
 				})
 			}
 		})
diff --git a/pkg/sql/plan.go b/pkg/sql/plan.go
@@ -156,6 +156,11 @@ type mutationPlanNode interface {
 	// modified by this planNode. It is always >= rowsWritten. It should only be
 	// called once Next returns false.
 	indexRowsWritten() int64
+
+	// indexBytesWritten returns the number of primary and secondary index bytes
+	// modified by this planNode. It should only be called once Next returns
+	// false.
+	indexBytesWritten() int64
 }
 
 // planNodeFastPath is implemented by nodes that can perform all their
diff --git a/pkg/sql/plan_batch.go b/pkg/sql/plan_batch.go
@@ -162,6 +162,14 @@ func (s *serializeNode) indexRowsWritten() int64 {
 	return m.indexRowsWritten()
 }
 
+func (s *serializeNode) indexBytesWritten() int64 {
+	m, ok := s.source.(mutationPlanNode)
+	if !ok {
+		return 0
+	}
+	return m.indexBytesWritten()
+}
+
 // requireSpool implements the planNodeRequireSpool interface.
 func (s *serializeNode) requireSpool() {}
 
@@ -239,3 +247,11 @@ func (r *rowCountNode) indexRowsWritten() int64 {
 	}
 	return m.indexRowsWritten()
 }
+
+func (r *rowCountNode) indexBytesWritten() int64 {
+	m, ok := r.source.(mutationPlanNode)
+	if !ok {
+		return 0
+	}
+	return m.indexBytesWritten()
+}
diff --git a/pkg/sql/plan_node_to_row_source.go b/pkg/sql/plan_node_to_row_source.go
@@ -267,6 +267,7 @@ func (p *planNodeToRowSource) trailingMetaCallback() []execinfrapb.ProducerMetad
 			metrics := execinfrapb.GetMetricsMeta()
 			metrics.RowsWritten = m.rowsWritten()
 			metrics.IndexRowsWritten = m.indexRowsWritten()
+			metrics.IndexBytesWritten = m.indexBytesWritten()
 			meta = []execinfrapb.ProducerMetadata{{Metrics: metrics}}
 		}
 	}
diff --git a/pkg/sql/spool.go b/pkg/sql/spool.go
@@ -118,3 +118,11 @@ func (s *spoolNode) indexRowsWritten() int64 {
 	}
 	return m.indexRowsWritten()
 }
+
+func (s *spoolNode) indexBytesWritten() int64 {
+	m, ok := s.input.(mutationPlanNode)
+	if !ok {
+		return 0
+	}
+	return m.indexBytesWritten()
+}
diff --git a/pkg/sql/tablewriter.go b/pkg/sql/tablewriter.go
@@ -71,6 +71,9 @@ type tableWriterBase struct {
 	// indexRowsWritten tracks the number of primary and secondary index rows
 	// written by this tableWriterBase so far. It is always >= rowsWritten.
 	indexRowsWritten int64
+	// indexBytesWritten tracks the number of primary and secondary index bytes
+	// written by this tableWriterBase so far.
+	indexBytesWritten int64
 	// rowsWrittenLimit if positive indicates that
 	// `transaction_rows_written_err` is enabled. The limit will be checked in
 	// finalize() before deciding whether it is safe to auto commit (if auto
@@ -151,11 +154,12 @@ func (tb *tableWriterBase) flushAndStartNewBatch(ctx context.Context) error {
 	if err := tb.tryDoResponseAdmission(ctx); err != nil {
 		return err
 	}
+	tb.rowsWritten += int64(tb.currentBatchSize)
+	tb.lastBatchSize = tb.currentBatchSize
 	// The mutation operators add one request to the KV batch for each index
 	// entry that's written.
 	tb.indexRowsWritten += int64(len(tb.b.Requests()))
-	tb.rowsWritten += int64(tb.currentBatchSize)
-	tb.lastBatchSize = tb.currentBatchSize
+	tb.indexBytesWritten += int64(tb.b.ApproximateMutationBytes())
 	tb.currentBatchSize = 0
 	tb.initNewBatch()
 	return nil
@@ -165,8 +169,9 @@ func (tb *tableWriterBase) flushAndStartNewBatch(ctx context.Context) error {
 func (tb *tableWriterBase) finalize(ctx context.Context) (err error) {
 	// NB: unlike flushAndStartNewBatch, we don't bother with admission control
 	// for response processing when finalizing.
-	tb.indexRowsWritten += int64(len(tb.b.Requests()))
 	tb.rowsWritten += int64(tb.currentBatchSize)
+	tb.indexRowsWritten += int64(len(tb.b.Requests()))
+	tb.indexBytesWritten += int64(tb.b.ApproximateMutationBytes())
 	if tb.autoCommit == autoCommitEnabled &&
 		// We can only auto commit if the rows written guardrail is disabled or
 		// we haven't exceeded the specified limit (the optimizer is responsible
diff --git a/pkg/sql/update.go b/pkg/sql/update.go
@@ -309,6 +309,10 @@ func (u *updateNode) indexRowsWritten() int64 {
 	return u.run.tu.indexRowsWritten
 }
 
+func (u *updateNode) indexBytesWritten() int64 {
+	return u.run.tu.indexBytesWritten
+}
+
 func (u *updateNode) enableAutoCommit() {
 	u.run.tu.enableAutoCommit()
 }
diff --git a/pkg/sql/update_swap.go b/pkg/sql/update_swap.go
@@ -147,6 +147,10 @@ func (u *updateSwapNode) indexRowsWritten() int64 {
 	return u.run.tu.indexRowsWritten
 }
 
+func (u *updateSwapNode) indexBytesWritten() int64 {
+	return u.run.tu.indexBytesWritten
+}
+
 func (u *updateSwapNode) enableAutoCommit() {
 	u.run.tu.enableAutoCommit()
 }
diff --git a/pkg/sql/upsert.go b/pkg/sql/upsert.go
@@ -237,6 +237,10 @@ func (n *upsertNode) indexRowsWritten() int64 {
 	return n.run.tw.indexRowsWritten
 }
 
+func (n *upsertNode) indexBytesWritten() int64 {
+	return n.run.tw.indexBytesWritten
+}
+
 func (n *upsertNode) enableAutoCommit() {
 	n.run.tw.enableAutoCommit()
 }

Original file line number	Diff line number	Diff line change
`@@ -263,6 +263,11 @@ func (d *deleteNode) indexRowsWritten() int64 {`
`263`	`263`	`return d.run.td.indexRowsWritten`
`264`	`264`	`}`
`265`	`265`
	`266`	`+func (d *deleteNode) indexBytesWritten() int64 {`
	`267`	`+ // No bytes counted as written for a deletion.`
	`268`	`+ return 0`
	`269`	`+}`
	`270`	`+`
`266`	`271`	`func (d *deleteNode) enableAutoCommit() {`
`267`	`272`	`d.run.td.enableAutoCommit()`
`268`	`273`	`}`
Original file line number	Diff line number	Diff line change
`@@ -147,6 +147,11 @@ func (d *deleteSwapNode) indexRowsWritten() int64 {`
`147`	`147`	`return d.run.td.indexRowsWritten`
`148`	`148`	`}`
`149`	`149`
	`150`	`+func (d *deleteSwapNode) indexBytesWritten() int64 {`
	`151`	`+ // No bytes counted as written for a deletion.`
	`152`	`+ return 0`
	`153`	`+}`
	`154`	`+`
`150`	`155`	`func (d *deleteSwapNode) enableAutoCommit() {`
`151`	`156`	`d.run.td.enableAutoCommit()`
`152`	`157`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1532,6 +1532,12 @@ var (`
`1532`	`1532`	`Measurement: "SQL Statements",`
`1533`	`1533`	`Unit: metric.Unit_COUNT,`
`1534`	`1534`	`}`
	`1535`	`+ MetaStatementIndexBytesWritten = metric.Metadata{`
	`1536`	`+ Name: "sql.statements.index_bytes_written.count",`
	`1537`	`+ Help: "Number of primary and secondary index bytes modified by SQL statements",`
	`1538`	`+ Measurement: "SQL Statements",`
	`1539`	`+ Unit: metric.Unit_BYTES,`
	`1540`	`+ }`
`1535`	`1541`	`)`
`1536`	`1542`
`1537`	`1543`	`func getMetricMeta(meta metric.Metadata, internal bool) metric.Metadata {`