sql: add index_rows_written metric

andy-kimball · andy-kimball · commit 30e5c2a57c53 · 2025-10-30T07:47:00.000-07:00
Add new `sql.statements.index_rows_written.count` metric that counts the number
of primary and secondary index rows modified by SQL statements.

Epic: AOP-30

Release note (sql change): Added sql.statements.index_rows_written.count metric
that counts the number of primary and secondary index rows 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_rows_written.count
+      exported_name: sql_statements_index_rows_written_count
+      description: Number of primary and secondary index rows modified by SQL statements
+      y_axis_label: SQL Statements
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: sql.statements.index_rows_written.count.internal
+      exported_name: sql_statements_index_rows_written_count_internal
+      description: Number of primary and secondary index rows modified by SQL statements (internal queries)
+      y_axis_label: SQL Internal Statements
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: sql.statements.rows_read.count
       exported_name: sql_statements_rows_read_count
       description: Number of rows read by SQL statements
diff --git a/pkg/roachprod/opentelemetry/cockroachdb_metrics.go b/pkg/roachprod/opentelemetry/cockroachdb_metrics.go
@@ -1876,6 +1876,8 @@ var cockroachdbMetrics = map[string]string{
 	"sql_statements_rows_read_count_internal":                     "sql.statements.rows_read.count.internal",
 	"sql_statements_bytes_read_count":                             "sql.statements.bytes_read.count",
 	"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_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
@@ -627,6 +627,7 @@ func makeMetrics(internal bool, sv *settings.Values) Metrics {
 			StatementRetryCount:               metric.NewCounter(getMetricMeta(MetaStatementRetry, internal)),
 			StatementRowsRead:                 metric.NewCounter(getMetricMeta(MetaStatementRowsRead, internal)),
 			StatementBytesRead:                metric.NewCounter(getMetricMeta(MetaStatementBytesRead, internal)),
+			StatementIndexRowsWritten:         metric.NewCounter(getMetricMeta(MetaStatementIndexRowsWritten, internal)),
 		},
 		StartedStatementCounters:  makeStartedStatementCounters(internal),
 		ExecutedStatementCounters: makeExecutedStatementCounters(internal),
diff --git a/pkg/sql/conn_executor_exec.go b/pkg/sql/conn_executor_exec.go
@@ -3279,12 +3279,20 @@ 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 disk.
+	// bytesRead is the number of bytes read from primary and secondary indexes.
 	bytesRead int64
-	// rowsRead is the number of rows read from disk.
+	// rowsRead is the number of rows read from primary and secondary indexes.
 	rowsRead int64
-	// rowsWritten is the number of rows written.
+	// 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
+	// includes rows read from secondary indexes, while rowsWritten does not
+	// include rows written to secondary indexes. This matches the behavior of
+	// EXPLAIN ANALYZE and SQL "rows affected".
 	rowsWritten int64
+	// indexRowsWritten is the number of rows written to primary and secondary
+	// indexes. It is always >= rowsWritten.
+	indexRowsWritten 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
@@ -3298,6 +3306,7 @@ func (s *topLevelQueryStats) add(other *topLevelQueryStats) {
 	s.bytesRead += other.bytesRead
 	s.rowsRead += other.rowsRead
 	s.rowsWritten += other.rowsWritten
+	s.indexRowsWritten += other.indexRowsWritten
 	s.networkEgressEstimate += other.networkEgressEstimate
 	s.clientTime += other.clientTime
 }
diff --git a/pkg/sql/delete.go b/pkg/sql/delete.go
@@ -259,6 +259,10 @@ func (d *deleteNode) rowsWritten() int64 {
 	return d.run.td.rowsWritten
 }
 
+func (d *deleteNode) indexRowsWritten() int64 {
+	return d.run.td.indexRowsWritten
+}
+
 func (d *deleteNode) enableAutoCommit() {
 	d.run.td.enableAutoCommit()
 }
diff --git a/pkg/sql/delete_range.go b/pkg/sql/delete_range.go
@@ -83,6 +83,12 @@ func (d *deleteRangeNode) rowsWritten() int64 {
 	return int64(d.rowCount)
 }
 
+func (d *deleteRangeNode) indexRowsWritten() int64 {
+	// Same as rowsWritten, because deleteRangeNode only applies to primary index
+	// rows (it is not used if there's a secondary index on the table).
+	return int64(d.rowCount)
+}
+
 // 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
@@ -143,6 +143,10 @@ func (d *deleteSwapNode) rowsWritten() int64 {
 	return d.run.td.rowsWritten
 }
 
+func (d *deleteSwapNode) indexRowsWritten() int64 {
+	return d.run.td.indexRowsWritten
+}
+
 func (d *deleteSwapNode) enableAutoCommit() {
 	d.run.td.enableAutoCommit()
 }
diff --git a/pkg/sql/distsql_running.go b/pkg/sql/distsql_running.go
@@ -1606,6 +1606,7 @@ func (r *DistSQLReceiver) pushMeta(meta *execinfrapb.ProducerMetadata) execinfra
 		r.stats.bytesRead += meta.Metrics.BytesRead
 		r.stats.rowsRead += meta.Metrics.RowsRead
 		r.stats.rowsWritten += meta.Metrics.RowsWritten
+		r.stats.indexRowsWritten += meta.Metrics.IndexRowsWritten
 
 		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
@@ -1526,6 +1526,12 @@ var (
 		Measurement: "SQL Statements",
 		Unit:        metric.Unit_BYTES,
 	}
+	MetaStatementIndexRowsWritten = metric.Metadata{
+		Name:        "sql.statements.index_rows_written.count",
+		Help:        "Number of primary and secondary index rows modified by SQL statements",
+		Measurement: "SQL Statements",
+		Unit:        metric.Unit_COUNT,
+	}
 )
 
 func getMetricMeta(meta metric.Metadata, internal bool) metric.Metadata {
diff --git a/pkg/sql/execinfrapb/data.proto b/pkg/sql/execinfrapb/data.proto
@@ -333,19 +333,26 @@ message RemoteProducerMetadata {
   }
   // Metrics are unconditionally emitted by table readers.
   message Metrics {
-    // Total number of bytes read while executing a statement.
+    // BytesRead is the number of primary and secondary index bytes read while
+    // executing a statement.
     optional int64 bytes_read = 1 [(gogoproto.nullable) = false];
-    // Total number of rows read while executing a statement.
+    // RowsRead is the number of primary and secondary index rows read while
+    // executing a statement.
     optional int64 rows_read = 2 [(gogoproto.nullable) = false];
-    // Total number of rows modified while executing a statement.
+    // RowsWritten is the number of primary index rows modified while executing
+    // a statement. It does not include secondary index rows.
     optional int64 rows_written = 3 [(gogoproto.nullable) = false];
+    // 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];
     // 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
   }
   oneof value {
     RangeInfos range_info = 1;
@@ -360,7 +367,7 @@ message RemoteProducerMetadata {
     TracingAggregatorEvents tracing_aggregator_events= 12;
   }
   reserved 6, 10;
-  // NEXT ID: 13
+  // NEXT ID: 14
 }
 
 // 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
@@ -92,6 +92,10 @@ type EngineMetrics struct {
 	// StatementBytesRead counts the number of bytes scanned by SQL statements
 	// from primary and secondary indexes.
 	StatementBytesRead *metric.Counter
+
+	// StatementIndexRowsWritten counts the number of primary and secondary index
+	// rows modified by SQL statements.
+	StatementIndexRowsWritten *metric.Counter
 }
 
 // EngineMetrics implements the metric.Struct interface.
@@ -209,6 +213,7 @@ func (ex *connExecutor) recordStatementSummary(
 	// Update SQL statement metrics.
 	ex.metrics.EngineMetrics.StatementRowsRead.Inc(stats.rowsRead)
 	ex.metrics.EngineMetrics.StatementBytesRead.Inc(stats.bytesRead)
+	ex.metrics.EngineMetrics.StatementIndexRowsWritten.Inc(stats.indexRowsWritten)
 
 	if ex.statsCollector.EnabledForTransaction() {
 		recordedStmtStats := &sqlstats.RecordedStmtStats{
diff --git a/pkg/sql/insert.go b/pkg/sql/insert.go
@@ -388,3 +388,7 @@ func (n *insertNode) enableAutoCommit() {
 func (n *insertNode) rowsWritten() int64 {
 	return n.run.ti.rowsWritten
 }
+
+func (n *insertNode) indexRowsWritten() int64 {
+	return n.run.ti.indexRowsWritten
+}
diff --git a/pkg/sql/insert_fast_path.go b/pkg/sql/insert_fast_path.go
@@ -548,6 +548,10 @@ func (n *insertFastPathNode) rowsWritten() int64 {
 	return n.run.ti.rowsWritten
 }
 
+func (n *insertFastPathNode) indexRowsWritten() int64 {
+	return n.run.ti.indexRowsWritten
+}
+
 // 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
@@ -218,21 +218,14 @@ func TestStatementMetrics(t *testing.T) {
 	runner := sqlutils.MakeSQLRunner(sqlDB)
 	s := srv.ApplicationLayer()
 
-	runner.Exec(t, `CREATE DATABASE db`)
-	runner.Exec(t, `CREATE TABLE db.t (x INT PRIMARY KEY, y INT, a STRING[], v VECTOR(2))`)
-	runner.Exec(t, `CREATE INDEX ON db.t (y)`)
-	runner.Exec(t, `CREATE INVERTED INDEX ON db.t (a)`)
-	runner.Exec(t, `CREATE VECTOR INDEX ON db.t (v)`)
-	runner.Exec(t, `INSERT INTO db.t VALUES (1, 10, ARRAY['apple', 'orange'], '[1, 2]')`)
-	runner.Exec(t, `INSERT INTO db.t VALUES (2, 20, ARRAY['banana'], '[3, 4]')`)
-	runner.Exec(t, `INSERT INTO db.t VALUES (3, 30, ARRAY['apple', 'pear', 'mango'], '[5, 6]')`)
-
 	testCases := []struct {
-		query    string
-		rowsRead int64
+		query            string
+		rowsRead         int64
+		indexRowsWritten int64
 		// If true, then skip metamorphic testing for this query.
 		skipMetamorphic bool
 	}{
+		// Read-only statements.
 		{query: `SELECT * FROM db.t WHERE x = 1`, rowsRead: 1},
 		{query: `SELECT * FROM db.t WHERE x <= 2`, rowsRead: 2},
 		{query: `SELECT * FROM db.t WHERE a @> ARRAY['apple']`, rowsRead: 4},
@@ -244,20 +237,58 @@ func TestStatementMetrics(t *testing.T) {
 			skipMetamorphic: true,
 		},
 		{query: `SELECT * FROM db.t ORDER BY v <-> '[2, 2]' LIMIT 1`, rowsRead: 3},
-	}
 
-	for _, tc := range testCases {
-		t.Run(tc.query, func(t *testing.T) {
-			for _, vectorized := range []string{"off", "on"} {
-				runner.Exec(t, "SET vectorize = $1", vectorized)
-				t.Run(fmt.Sprintf("vectorize = %s", vectorized), func(t *testing.T) {
+		// Mutation statements.
+		{query: `INSERT INTO db.t VALUES (4, 40, ARRAY['orange', 'cherry'], '[7, 8]')`, indexRowsWritten: 5},
+		{
+			query:            `INSERT INTO db.t VALUES (5, 50, ARRAY['orange', 'cherry']), (6, 60, NULL)`,
+			indexRowsWritten: 6,
+		},
+		// Expect to read 2 rows - one in secondary index, one in primary index.
+		// Expect to write 3 rows - update primary index, delete and insert in
+		// "y" secondary index.
+		{query: `UPDATE db.t SET y = 100 WHERE y = 40`, rowsRead: 2, indexRowsWritten: 3},
+		// Expect to read 4 rows - two in secondary index, two in primary index.
+		// Expect to write 5 rows - update primary index, delete and insert two
+		// rows in "y" secondary index.
+		{query: `UPDATE db.t SET y = 100 WHERE y >= 10 AND y <= 20`, rowsRead: 4, indexRowsWritten: 6},
+		{
+			query:            `UPSERT INTO db.t VALUES (6, 60, ARRAY['date'], '[100, 110]')`,
+			rowsRead:         1,
+			indexRowsWritten: 3,
+		},
+		{
+			query:            `INSERT INTO db.t VALUES (6, 600) ON CONFLICT (x) DO UPDATE SET a = ARRAY['strawberry']`,
+			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},
+	}
+
+	for _, vectorized := range []string{"off", "on"} {
+		t.Run(fmt.Sprintf("vectorize = %s", vectorized), func(t *testing.T) {
+			runner.Exec(t, "SET vectorize = $1", vectorized)
+			runner.Exec(t, `DROP DATABASE IF EXISTS db`)
+			runner.Exec(t, `CREATE DATABASE db`)
+			runner.Exec(t, `CREATE TABLE db.t (x INT PRIMARY KEY, y INT, a STRING[], v VECTOR(2))`)
+			runner.Exec(t, `CREATE INDEX ON db.t (y)`)
+			runner.Exec(t, `CREATE INVERTED INDEX ON db.t (a)`)
+			runner.Exec(t, `CREATE VECTOR INDEX ON db.t (v)`)
+			runner.Exec(t, `INSERT INTO db.t VALUES (1, 10, ARRAY['apple', 'orange'], '[1, 2]')`)
+			runner.Exec(t, `INSERT INTO db.t VALUES (2, 20, ARRAY['banana'], '[3, 4]')`)
+			runner.Exec(t, `INSERT INTO db.t VALUES (3, 30, ARRAY['apple', 'pear', 'mango'], '[5, 6]')`)
+
+			for _, tc := range testCases {
+				t.Run(tc.query, func(t *testing.T) {
 					if tc.skipMetamorphic && metamorphic.IsMetamorphicBuild() {
 						return
 					}
 
 					// Get initial values of the counters.
 					lastRowsRead := s.MustGetSQLCounter(sql.MetaStatementRowsRead.Name)
 					lastBytesRead := s.MustGetSQLCounter(sql.MetaStatementBytesRead.Name)
+					lastRowsWritten := s.MustGetSQLCounter(sql.MetaStatementIndexRowsWritten.Name)
 
 					runner.Exec(t, tc.query)
 
@@ -270,8 +301,16 @@ func TestStatementMetrics(t *testing.T) {
 					// If there were rows read, then expect bytes read to have
 					// increased. Don't check for a specific value, since there are
 					// too many factors that can change this.
-					currBytesRead := s.MustGetSQLCounter(sql.MetaStatementBytesRead.Name)
-					require.Greater(t, currBytesRead, lastBytesRead)
+					if tc.rowsRead > 0 {
+						currBytesRead := s.MustGetSQLCounter(sql.MetaStatementBytesRead.Name)
+						require.Greater(t, currBytesRead, lastBytesRead)
+					}
+
+					// Test the rows written metric.
+					actual, err = checkCounterDelta(
+						s, sql.MetaStatementIndexRowsWritten, lastRowsWritten, tc.indexRowsWritten)
+					require.NoError(t, err)
+					lastRowsWritten = actual
 				})
 			}
 		})
diff --git a/pkg/sql/plan.go b/pkg/sql/plan.go
@@ -147,9 +147,15 @@ func (n *singleInputPlanNode) SetInput(i int, p planNode) error {
 type mutationPlanNode interface {
 	planNode
 
-	// rowsWritten returns the number of rows modified by this planNode. It
-	// should only be called once Next returns false.
+	// rowsWritten returns the number of table rows modified by this planNode.
+	// It does not include rows written to secondary indexes. It should only be
+	// called once Next returns false.
 	rowsWritten() int64
+
+	// indexRowsWritten returns the number of primary and secondary index rows
+	// modified by this planNode. It is always >= rowsWritten. It should only be
+	// called once Next returns false.
+	indexRowsWritten() 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
@@ -154,6 +154,14 @@ func (s *serializeNode) rowsWritten() int64 {
 	return m.rowsWritten()
 }
 
+func (s *serializeNode) indexRowsWritten() int64 {
+	m, ok := s.source.(mutationPlanNode)
+	if !ok {
+		return 0
+	}
+	return m.indexRowsWritten()
+}
+
 // requireSpool implements the planNodeRequireSpool interface.
 func (s *serializeNode) requireSpool() {}
 
@@ -223,3 +231,11 @@ func (r *rowCountNode) rowsWritten() int64 {
 	}
 	return m.rowsWritten()
 }
+
+func (r *rowCountNode) indexRowsWritten() int64 {
+	m, ok := r.source.(mutationPlanNode)
+	if !ok {
+		return 0
+	}
+	return m.indexRowsWritten()
+}
diff --git a/pkg/sql/plan_node_to_row_source.go b/pkg/sql/plan_node_to_row_source.go
@@ -266,6 +266,7 @@ func (p *planNodeToRowSource) trailingMetaCallback() []execinfrapb.ProducerMetad
 		if m, ok := p.node.(mutationPlanNode); ok {
 			metrics := execinfrapb.GetMetricsMeta()
 			metrics.RowsWritten = m.rowsWritten()
+			metrics.IndexRowsWritten = m.indexRowsWritten()
 			meta = []execinfrapb.ProducerMetadata{{Metrics: metrics}}
 		}
 	}
diff --git a/pkg/sql/spool.go b/pkg/sql/spool.go
@@ -110,3 +110,11 @@ func (s *spoolNode) rowsWritten() int64 {
 	}
 	return m.rowsWritten()
 }
+
+func (s *spoolNode) indexRowsWritten() int64 {
+	m, ok := s.input.(mutationPlanNode)
+	if !ok {
+		return 0
+	}
+	return m.indexRowsWritten()
+}
diff --git a/pkg/sql/tablewriter.go b/pkg/sql/tablewriter.go
diff --git a/pkg/sql/update.go b/pkg/sql/update.go
diff --git a/pkg/sql/update_swap.go b/pkg/sql/update_swap.go
diff --git a/pkg/sql/upsert.go b/pkg/sql/upsert.go

Original file line number	Diff line number	Diff line change
`@@ -259,6 +259,10 @@ func (d *deleteNode) rowsWritten() int64 {`
`259`	`259`	`return d.run.td.rowsWritten`
`260`	`260`	`}`
`261`	`261`
	`262`	`+func (d *deleteNode) indexRowsWritten() int64 {`
	`263`	`+ return d.run.td.indexRowsWritten`
	`264`	`+}`
	`265`	`+`
`262`	`266`	`func (d *deleteNode) enableAutoCommit() {`
`263`	`267`	`d.run.td.enableAutoCommit()`
`264`	`268`	`}`
Original file line number	Diff line number	Diff line change
`@@ -143,6 +143,10 @@ func (d *deleteSwapNode) rowsWritten() int64 {`
`143`	`143`	`return d.run.td.rowsWritten`
`144`	`144`	`}`
`145`	`145`
	`146`	`+func (d *deleteSwapNode) indexRowsWritten() int64 {`
	`147`	`+ return d.run.td.indexRowsWritten`
	`148`	`+}`
	`149`	`+`
`146`	`150`	`func (d *deleteSwapNode) enableAutoCommit() {`
`147`	`151`	`d.run.td.enableAutoCommit()`
`148`	`152`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1526,6 +1526,12 @@ var (`
`1526`	`1526`	`Measurement: "SQL Statements",`
`1527`	`1527`	`Unit: metric.Unit_BYTES,`
`1528`	`1528`	`}`
	`1529`	`+ MetaStatementIndexRowsWritten = metric.Metadata{`
	`1530`	`+ Name: "sql.statements.index_rows_written.count",`
	`1531`	`+ Help: "Number of primary and secondary index rows modified by SQL statements",`
	`1532`	`+ Measurement: "SQL Statements",`
	`1533`	`+ Unit: metric.Unit_COUNT,`
	`1534`	`+ }`
`1529`	`1535`	`)`
`1530`	`1536`
`1531`	`1537`	`func getMetricMeta(meta metric.Metadata, internal bool) metric.Metadata {`