sql: add index_rows_written metric

andy-kimball · andy-kimball · commit cb789190e93a · 2025-10-28T13:44:27.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
@@ -8763,6 +8763,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/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
@@ -3275,12 +3275,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
@@ -3296,6 +3304,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
@@ -251,6 +251,10 @@ func (d *deleteNode) rowsWritten() int64 {
 	return d.run.rowsAffected()
 }
 
+func (d *deleteNode) indexRowsWritten() int64 {
+	return d.run.td.indexRowsWritten
+}
+
 func (d *deleteNode) returnsRowsAffected() bool {
 	return !d.run.rowsNeeded
 }
diff --git a/pkg/sql/delete_range.go b/pkg/sql/delete_range.go
@@ -59,6 +59,12 @@ func (d *deleteRangeNode) rowsWritten() int64 {
 	return d.rowsAffected()
 }
 
+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 d.rowsAffected()
+}
+
 func (d *deleteRangeNode) returnsRowsAffected() bool {
 	// DeleteRange always returns the number of rows deleted.
 	return true
diff --git a/pkg/sql/delete_swap.go b/pkg/sql/delete_swap.go
@@ -130,6 +130,10 @@ func (d *deleteSwapNode) rowsWritten() int64 {
 	return d.run.rowsAffected()
 }
 
+func (d *deleteSwapNode) indexRowsWritten() int64 {
+	return d.run.td.indexRowsWritten
+}
+
 func (d *deleteSwapNode) returnsRowsAffected() bool {
 	return !d.run.rowsNeeded
 }
diff --git a/pkg/sql/distsql_running.go b/pkg/sql/distsql_running.go
@@ -1623,6 +1623,7 @@ func forwardInnerQueryStats(f metadataForwarder, stats topLevelQueryStats) {
 	meta.Metrics.BytesRead = stats.bytesRead
 	meta.Metrics.RowsRead = stats.rowsRead
 	meta.Metrics.RowsWritten = stats.rowsWritten
+	meta.Metrics.IndexRowsWritten = stats.indexRowsWritten
 	// stats.networkEgressEstimate and stats.clientTime are ignored since they
 	// only matter at the "true" top-level query (and actually should be zero
 	// here anyway).
@@ -1671,6 +1672,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
@@ -379,6 +379,10 @@ func (n *insertNode) rowsWritten() int64 {
 	return n.run.rowsAffected()
 }
 
+func (n *insertNode) indexRowsWritten() int64 {
+	return n.run.ti.indexRowsWritten
+}
+
 func (n *insertNode) returnsRowsAffected() bool {
 	return !n.run.rowsNeeded
 }
diff --git a/pkg/sql/insert_fast_path.go b/pkg/sql/insert_fast_path.go
@@ -540,6 +540,10 @@ func (n *insertFastPathNode) rowsWritten() int64 {
 	return n.run.rowsAffected()
 }
 
+func (n *insertFastPathNode) indexRowsWritten() int64 {
+	return n.run.ti.indexRowsWritten
+}
+
 func (n *insertFastPathNode) returnsRowsAffected() bool {
 	return !n.run.rowsNeeded
 }
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
@@ -146,10 +146,16 @@ 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
+
 	// returnsRowsAffected indicates that the planNode returns the number of
 	// rows affected by the mutation, rather than the rows themselves.
 	returnsRowsAffected() bool
diff --git a/pkg/sql/plan_node_to_row_source.go b/pkg/sql/plan_node_to_row_source.go
@@ -260,6 +260,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/tablewriter.go b/pkg/sql/tablewriter.go
@@ -56,13 +56,17 @@ type tableWriterBase struct {
 	// maxBatchByteSize determines the maximum number of key and value bytes in
 	// the KV batch for a mutation operation.
 	maxBatchByteSize int
-	// currentBatchSize is the size of the current batch. It is updated on
-	// every row() call and is reset once a new batch is started.
+	// currentBatchSize is the size of the current SQL-level batch (i.e. not the
+	// KV-level batch). It is updated on every row() call and is reset once a new
+	// batch is started.
 	currentBatchSize int
-	// rowsWritten tracks the number of rows written by this tableWriterBase so
-	// far. This counter includes unsuccessful writes (e.g. those performed by
-	// swap mutations in the event of a nonexistent row).
+	// rowsWritten tracks the number of primary index rows written by this
+	// tableWriterBase so far. This counter includes unsuccessful writes (e.g.
+	// those performed by swap mutations in the event of a nonexistent row).
 	rowsWritten int64
+	// indexRowsWritten tracks the number of primary and secondary index rows
+	// written by this tableWriterBase so far. It is always >= rowsWritten.
+	indexRowsWritten 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
@@ -140,16 +144,20 @@ func (tb *tableWriterBase) flushAndStartNewBatch(ctx context.Context) error {
 	if err := tb.tryDoResponseAdmission(ctx); err != nil {
 		return err
 	}
-	tb.initNewBatch()
+	// 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.currentBatchSize = 0
+	tb.initNewBatch()
 	return nil
 }
 
 // finalize shares the common finalize() code between tableWriters.
 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)
 	if tb.autoCommit == autoCommitEnabled &&
 		// We can only auto commit if the rows written guardrail is disabled or
diff --git a/pkg/sql/update.go b/pkg/sql/update.go
@@ -297,6 +297,10 @@ func (u *updateNode) rowsWritten() int64 {
 	return u.run.rowsAffected()
 }
 
+func (u *updateNode) indexRowsWritten() int64 {
+	return u.run.tu.indexRowsWritten
+}
+
 func (u *updateNode) returnsRowsAffected() bool {
 	return !u.run.rowsNeeded
 }
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
`@@ -251,6 +251,10 @@ func (d *deleteNode) rowsWritten() int64 {`
`251`	`251`	`return d.run.rowsAffected()`
`252`	`252`	`}`
`253`	`253`
	`254`	`+func (d *deleteNode) indexRowsWritten() int64 {`
	`255`	`+ return d.run.td.indexRowsWritten`
	`256`	`+}`
	`257`	`+`
`254`	`258`	`func (d *deleteNode) returnsRowsAffected() bool {`
`255`	`259`	`return !d.run.rowsNeeded`
`256`	`260`	`}`
Original file line number	Diff line number	Diff line change
`@@ -130,6 +130,10 @@ func (d *deleteSwapNode) rowsWritten() int64 {`
`130`	`130`	`return d.run.rowsAffected()`
`131`	`131`	`}`
`132`	`132`
	`133`	`+func (d *deleteSwapNode) indexRowsWritten() int64 {`
	`134`	`+ return d.run.td.indexRowsWritten`
	`135`	`+}`
	`136`	`+`
`133`	`137`	`func (d *deleteSwapNode) returnsRowsAffected() bool {`
`134`	`138`	`return !d.run.rowsNeeded`
`135`	`139`	`}`
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 {`