sql: add index_bytes_written metric

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.

Author: andy-kimball

docs/generated/metrics/metrics.yaml

@@ -8763,6 +8763,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

pkg/roachprod/agents/opentelemetry/cockroachdb_metrics.go

@@ -2169,6 +2169,12 @@ var cockroachdbMetrics = map[string]string{
     "sql_statements_active_internal": "sql.statements.active.internal",
     "sql_statements_auto_retry_count": "sql.statements.auto_retry.count",
     "sql_statements_auto_retry_count_internal": "sql.statements.auto_retry.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_bytes_written_count": "sql.statements.index_bytes_written.count",
+    "sql_statements_index_bytes_written_count_internal": "sql.statements.index_bytes_written.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_rows_read_count": "sql.statements.rows_read.count",
     "sql_statements_rows_read_count_internal": "sql.statements.rows_read.count.internal",
     "sql_stats_activity_update_latency": "sql.stats.activity.update.latency",

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),

pkg/sql/conn_executor_exec.go

@@ -3275,10 +3275,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
@@ -3289,6 +3289,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
@@ -3304,6 +3307,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

pkg/sql/delete.go

@@ -255,6 +255,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) returnsRowsAffected() bool {
 	return !d.run.rowsNeeded
 }

pkg/sql/delete_range.go

@@ -65,6 +65,11 @@ func (d *deleteRangeNode) indexRowsWritten() int64 {
 	return d.rowsAffected()
 }
 
+func (d *deleteRangeNode) indexBytesWritten() int64 {
+	// No bytes counted as written for a deletion.
+	return 0
+}
+
 func (d *deleteRangeNode) returnsRowsAffected() bool {
 	// DeleteRange always returns the number of rows deleted.
 	return true

pkg/sql/delete_swap.go

@@ -134,6 +134,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) returnsRowsAffected() bool {
 	return !d.run.rowsNeeded
 }

pkg/sql/distsql_running.go

@@ -1624,6 +1624,7 @@ func forwardInnerQueryStats(f metadataForwarder, stats topLevelQueryStats) {
 	meta.Metrics.RowsRead = stats.rowsRead
 	meta.Metrics.RowsWritten = stats.rowsWritten
 	meta.Metrics.IndexRowsWritten = stats.indexRowsWritten
+	meta.Metrics.IndexBytesWritten = stats.indexBytesWritten
 	// stats.networkEgressEstimate and stats.clientTime are ignored since they
 	// only matter at the "true" top-level query (and actually should be zero
 	// here anyway).
@@ -1673,6 +1674,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)

pkg/sql/distsql_running_test.go

@@ -1226,7 +1226,7 @@ func TestTopLevelQueryStats(t *testing.T) {
 	var testQuery atomic.Value
 	// The callback will send number of rows read and rows written (for each
 	// ProducerMetadata.Metrics object) on these channels, respectively.
-	rowsReadCh, rowsWrittenCh := make(chan int64), make(chan int64)
+	rowsReadCh, rowsWrittenCh, indexRowsWrittenCh := make(chan int64), make(chan int64), make(chan int64)
 	srv, sqlDB, _ := serverutils.StartServer(t, base.TestServerArgs{
 		Knobs: base.TestingKnobs{
 			SQLExecutor: &ExecutorTestingKnobs{
@@ -1238,6 +1238,7 @@ func TestTopLevelQueryStats(t *testing.T) {
 						if meta != nil && meta.Metrics != nil {
 							rowsReadCh <- meta.Metrics.RowsRead
 							rowsWrittenCh <- meta.Metrics.RowsWritten
+							indexRowsWrittenCh <- meta.Metrics.IndexRowsWritten
 						}
 						return row, batch, meta
 					}
@@ -1254,23 +1255,25 @@ CREATE TABLE t (k INT PRIMARY KEY, i INT, v INT, INDEX(i));
 INSERT INTO t SELECT i, 1, 1 FROM generate_series(1, 10) AS g(i);
 CREATE FUNCTION no_reads() RETURNS INT AS 'SELECT 1' LANGUAGE SQL;
 CREATE FUNCTION reads() RETURNS INT AS 'SELECT count(*) FROM t' LANGUAGE SQL;
-CREATE FUNCTION write(x INT) RETURNS INT AS 'INSERT INTO t VALUES (x); SELECT x' LANGUAGE SQL;
+CREATE FUNCTION write(x INT) RETURNS INT AS 'INSERT INTO t VALUES (x, x); SELECT x' LANGUAGE SQL;
 `); err != nil {
 		t.Fatal(err)
 	}
 
 	for _, tc := range []struct {
-		name           string
-		query          string
-		setup, cleanup string // optional
-		expRowsRead    int64
-		expRowsWritten int64
+		name                string
+		query               string
+		setup, cleanup      string // optional
+		expRowsRead         int64
+		expRowsWritten      int64
+		expIndexRowsWritten int64
 	}{
 		{
-			name:           "simple read",
-			query:          "SELECT k FROM t",
-			expRowsRead:    10,
-			expRowsWritten: 0,
+			name:                "simple read",
+			query:               "SELECT k FROM t",
+			expRowsRead:         10,
+			expRowsWritten:      0,
+			expIndexRowsWritten: 0,
 		},
 		{
 			name:    "routine and index join (used to be powered by streamer)",
@@ -1279,47 +1282,54 @@ CREATE FUNCTION write(x INT) RETURNS INT AS 'INSERT INTO t VALUES (x); SELECT x'
 			cleanup: "RESET distsql",
 			// 10 rows for secondary index, 10 for index join into primary, and
 			// then for each row do ten-row-scan in the routine.
-			expRowsRead:    120,
-			expRowsWritten: 0,
+			expRowsRead:         120,
+			expRowsWritten:      0,
+			expIndexRowsWritten: 0,
 		},
 		{
-			name:           "simple write",
-			query:          "INSERT INTO t SELECT generate_series(11, 42)",
-			expRowsRead:    0,
-			expRowsWritten: 32,
+			name:                "simple write",
+			query:               "INSERT INTO t SELECT generate_series(11, 42)",
+			expRowsRead:         0,
+			expRowsWritten:      32,
+			expIndexRowsWritten: 64,
 		},
 		{
 			name: "read with apply join",
 			query: `SELECT (
     WITH foo AS MATERIALIZED (SELECT k FROM t AS x WHERE x.k = y.k)
     SELECT * FROM foo
   ) FROM t AS y`,
-			expRowsRead:    84, // scanning the table twice
-			expRowsWritten: 0,
+			expRowsRead:         84, // scanning the table twice
+			expRowsWritten:      0,
+			expIndexRowsWritten: 0,
 		},
 		{
-			name:           "routine, no reads",
-			query:          "SELECT no_reads()",
-			expRowsRead:    0,
-			expRowsWritten: 0,
+			name:                "routine, no reads",
+			query:               "SELECT no_reads()",
+			expRowsRead:         0,
+			expRowsWritten:      0,
+			expIndexRowsWritten: 0,
 		},
 		{
-			name:           "routine, reads",
-			query:          "SELECT reads()",
-			expRowsRead:    42,
-			expRowsWritten: 0,
+			name:                "routine, reads",
+			query:               "SELECT reads()",
+			expRowsRead:         42,
+			expRowsWritten:      0,
+			expIndexRowsWritten: 0,
 		},
 		{
-			name:           "routine, write",
-			query:          "SELECT write(43)",
-			expRowsRead:    0,
-			expRowsWritten: 1,
+			name:                "routine, write",
+			query:               "SELECT write(43)",
+			expRowsRead:         0,
+			expRowsWritten:      1,
+			expIndexRowsWritten: 2,
 		},
 		{
-			name:           "routine, multiple reads and writes",
-			query:          "SELECT reads(), write(44), reads(), write(45), write(46), reads()",
-			expRowsRead:    133, // first read is 43 rows, second is 44, third is 46
-			expRowsWritten: 3,
+			name:                "routine, multiple reads and writes",
+			query:               "SELECT reads(), write(44), reads(), write(45), write(46), reads()",
+			expRowsRead:         133, // first read is 43 rows, second is 44, third is 46
+			expRowsWritten:      3,
+			expIndexRowsWritten: 6,
 		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
@@ -1344,21 +1354,24 @@ CREATE FUNCTION write(x INT) RETURNS INT AS 'INSERT INTO t VALUES (x); SELECT x'
 			}()
 			// In the main goroutine, loop until the query is completed while
 			// accumulating the top-level query stats.
-			var rowsRead, rowsWritten int64
+			var rowsRead, rowsWritten, indexRowsWritten int64
 		LOOP:
 			for {
 				select {
 				case read := <-rowsReadCh:
 					rowsRead += read
 				case written := <-rowsWrittenCh:
 					rowsWritten += written
+				case written := <-indexRowsWrittenCh:
+					indexRowsWritten += written
 				case err := <-errCh:
 					require.NoError(t, err)
 					break LOOP
 				}
 			}
 			require.Equal(t, tc.expRowsRead, rowsRead)
 			require.Equal(t, tc.expRowsWritten, rowsWritten)
+			require.Equal(t, tc.expIndexRowsWritten, indexRowsWritten)
 		})
 	}
 }

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 {