sql: add rows_read metric

Add new `sql.statements.rows_read.count` metric that counts the number
of index rows read by SQL statements. This is the same value that's
collected by SQL stats for each statement and transaction, except in
aggregated metric form.

Epic: none

Release note (sql change): Added sql.statements.rows_read.count metric
that counts the number of index rows read by SQL statements.

Author: andy-kimball

docs/generated/metrics/metrics.yaml

@@ -8747,6 +8747,22 @@ layers:
       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 from primary and secondary indexes
+      y_axis_label: SQL Statements
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: sql.statements.rows_read.count.internal
+      exported_name: sql_statements_rows_read_count_internal
+      description: Number of rows read by SQL statements from primary and secondary indexes (internal queries)
+      y_axis_label: SQL Internal Statements
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: sql.stats.activity.update.latency
       exported_name: sql_stats_activity_update_latency
       description: The latency of updates made by the SQL activity updater job. Includes failed update attempts

pkg/roachprod/agents/opentelemetry/cockroachdb_metrics.go

@@ -2169,6 +2169,8 @@ 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_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",
     "sql_stats_activity_update_latency_bucket": "sql.stats.activity.update.latency.bucket",
     "sql_stats_activity_update_latency_count": "sql.stats.activity.update.latency.count",

pkg/sql/BUILD.bazel

@@ -972,6 +972,7 @@ go_test(
         "//pkg/util/log/logconfig",
         "//pkg/util/log/logpb",
         "//pkg/util/log/logtestutils",
+        "//pkg/util/metamorphic",
         "//pkg/util/metric",
         "//pkg/util/mon",
         "//pkg/util/protoutil",

pkg/sql/conn_executor.go

@@ -625,6 +625,7 @@ func makeMetrics(internal bool, sv *settings.Values) Metrics {
 			FullTableOrIndexScanRejectedCount: metric.NewCounter(getMetricMeta(MetaFullTableOrIndexScanRejected, internal)),
 			TxnRetryCount:                     metric.NewCounter(getMetricMeta(MetaTxnRetry, internal)),
 			StatementRetryCount:               metric.NewCounter(getMetricMeta(MetaStatementRetry, internal)),
+			StatementRowsRead:                 metric.NewCounter(getMetricMeta(MetaStatementRowsRead, internal)),
 		},
 		StartedStatementCounters:  makeStartedStatementCounters(internal),
 		ExecutedStatementCounters: makeExecutedStatementCounters(internal),

pkg/sql/exec_util.go

@@ -1514,6 +1514,12 @@ var (
 		Measurement: "SQL Statements",
 		Unit:        metric.Unit_COUNT,
 	}
+	MetaStatementRowsRead = metric.Metadata{
+		Name:        "sql.statements.rows_read.count",
+		Help:        "Number of rows read by SQL statements from primary and secondary indexes",
+		Measurement: "SQL Statements",
+		Unit:        metric.Unit_COUNT,
+	}
 )
 
 func getMetricMeta(meta metric.Metadata, internal bool) metric.Metadata {

pkg/sql/executor_statement_metrics.go

@@ -83,6 +83,11 @@ type EngineMetrics struct {
 	// StatementRetryCount counts the number of automatic statement retries that
 	// have occurred under READ COMMITTED isolation.
 	StatementRetryCount *metric.Counter
+
+	// StatementRowsRead counts the number of rows read by SQL statements from
+	// primary and secondary indexes. Note that some secondary indexes can have
+	// multiple index rows per primary index row (e.g. inverted and vector).
+	StatementRowsRead *metric.Counter
 }
 
 // EngineMetrics implements the metric.Struct interface.
@@ -196,6 +201,10 @@ func (ex *connExecutor) recordStatementSummary(
 	if automaticRetryStmtCount > 0 {
 		autoRetryReason = planner.autoRetryStmtReason
 	}
+
+	// Update SQL statement metrics.
+	ex.metrics.EngineMetrics.StatementRowsRead.Inc(stats.rowsRead)
+
 	if ex.statsCollector.EnabledForTransaction() {
 		recordedStmtStats := &sqlstats.RecordedStmtStats{
 			FingerprintID:        stmtFingerprintID,

pkg/sql/metric_test.go

@@ -8,6 +8,7 @@ package sql_test
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"regexp"
 	"testing"
 
@@ -18,8 +19,10 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/lease"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"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/cockroachdb/cockroach/pkg/util/metamorphic"
 	"github.com/cockroachdb/cockroach/pkg/util/metric"
 	"github.com/stretchr/testify/require"
 )
@@ -205,6 +208,66 @@ func TestQueryCounts(t *testing.T) {
 	}
 }
 
+func TestStatementMetrics(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	params, _ := createTestServerParamsAllowTenants()
+	srv, sqlDB, _ := serverutils.StartServer(t, params)
+	defer srv.Stopper().Stop(context.Background())
+	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
+		// If true, then skip metamorphic testing for this query.
+		skipMetamorphic bool
+	}{
+		{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},
+		{
+			query:    `SELECT * FROM db.t@t_a_idx t1, db.t t2 WHERE t1.a @> t2.a`,
+			rowsRead: 12,
+			// Skip this variation under metamorphic testing because the inverted
+			// joiner batch size can affect number of rows read.
+			skipMetamorphic: true,
+		},
+		{query: `SELECT * FROM db.t ORDER BY v <-> '[2, 2]' LIMIT 1`, rowsRead: 3},
+	}
+
+	lastRowsRead := s.MustGetSQLCounter(sql.MetaStatementRowsRead.Name)
+
+	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) {
+					if tc.skipMetamorphic && metamorphic.IsMetamorphicBuild() {
+						return
+					}
+
+					runner.Exec(t, tc.query)
+					actual, err := checkCounterDelta(
+						s, sql.MetaStatementRowsRead, lastRowsRead, tc.rowsRead)
+					require.NoError(t, err)
+					lastRowsRead = actual
+				})
+			}
+		})
+	}
+}
+
 func TestAbortCountConflictingWrites(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)