Merge #155820 #155981 #156103 #156106

155820: sql: add index_rows_read metric r=yuzefovich a=andy-kimball

Add new `sql.statements.index_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.index_rows_read.count metric that counts the number of index rows read by SQL statements.

155981: roachtest: recover panics in post test assertions r=golgeek a=DarrylWong

We previously were recovering panics in the main test goroutine, but not post tests assertions, which could cause an entire CI run to crash. This patch adds a similar recover to post tests assertions.

Fixes: none
Epic: none
Release note: none

156103: kv: add debugging obs for TestFlowControlSendQueueRangeFeed r=wenyihu6 a=arulajmani

I haven't been able to repro this. The test has failed twice in the last couple of days, so hopefully we'll be able to see why the rangefeed isn't ending up on n3 the next time this fails.

Closes #156064

Epic: none

Release note: None

156106: kv: deflake TestCheckConsistencyInconsistent r=pav-kv a=arulajmani

This patch skips invariant checks when iterating through range descriptors from a checkpoint to prevent spurious assertion failures. These assertion failures did not hold because checkpoints are not complete stores, so anything beyond the checkpoints span may not be considered a consistent snapshot.

Closes #155680

Epic: none

Release note: None

Co-authored-by: Andrew Kimball <[email protected]>
Co-authored-by: DarrylWong <[email protected]>
Co-authored-by: Arul Ajmani <[email protected]>

Author: 4 people

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/cmd/roachtest/test_runner.go

@@ -1703,6 +1703,15 @@ func (r *testRunner) postTestAssertions(
 	_ = r.stopper.RunAsyncTask(ctx, "test-post-assertions", func(ctx context.Context) {
 		defer close(postAssertCh)
 
+		defer func() {
+			// Unlike the main test goroutine, we _do_ want to log t.Fatal* calls here
+			// to make it clear that the post-test assertions failed. Otherwise, the fatal
+			// will be recorded as a normal test failure.
+			if r := recover(); r != nil {
+				postAssertionErr(fmt.Errorf("post-test assertion panicked: %v", r))
+			}
+		}()
+
 		// We collect all the admin health endpoints in parallel,
 		// and select the first one that succeeds to run the validation queries
 		statuses, err := c.HealthStatus(ctx, t.L(), c.CRDBNodes())

pkg/kv/kvclient/kvcoord/dist_sender_rangefeed.go

@@ -623,6 +623,7 @@ func newTransportForRange(
 		return nil, err
 	}
 	replicas.OptimizeReplicaOrder(ctx, ds.st, ds.nodeIDGetter(), ds.healthFunc, ds.latencyFunc, ds.locality)
+	log.VEventf(ctx, 2, "replica order for rangefeed transport is: %s", replicas)
 	opts := SendOptions{class: defRangefeedConnClass}
 	return ds.transportFactory(opts, replicas), nil
 }

pkg/kv/kvserver/consistency_queue_test.go

@@ -418,13 +418,14 @@ func TestCheckConsistencyInconsistent(t *testing.T) {
 
 		// Find the problematic range in the storage.
 		var desc *roachpb.RangeDescriptor
-		require.NoError(t, kvstorage.IterateRangeDescriptorsFromDisk(context.Background(), cpEng,
+		require.NoError(t, kvstorage.IterateRangeDescriptorsFromCheckpoint(context.Background(), cpEng,
 			func(rd roachpb.RangeDescriptor) error {
 				if rd.RangeID == resp.Result[0].RangeID {
 					desc = &rd
 				}
 				return nil
-			}))
+			},
+		))
 		require.NotNil(t, desc)
 
 		// Compute a checksum over the content of the problematic range.

pkg/kv/kvserver/flow_control_integration_test.go

@@ -3220,6 +3220,7 @@ func TestFlowControlSendQueueRangeSplitMerge(t *testing.T) {
 func TestFlowControlSendQueueRangeFeed(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
+	testutils.SetVModule(t, "dist_sender_mux_rangefeed=2,dist_sender_rangefeed=2")
 
 	// rangeFeed will create a rangefeed suitable for testing. It will start a
 	// rangefeed and return a function that can be used to stop it.

pkg/kv/kvserver/kvstorage/init.go

@@ -245,6 +245,28 @@ func ReadStoreIdent(ctx context.Context, eng storage.Engine) (roachpb.StoreIdent
 // that error.
 func IterateRangeDescriptorsFromDisk(
 	ctx context.Context, reader storage.Reader, fn func(desc roachpb.RangeDescriptor) error,
+) error {
+	return iterateRangeDescriptorsFromDiskHelper(ctx, reader, fn, buildutil.CrdbTestBuild)
+}
+
+// IterateRangeDescriptorsFromCheckpoint is like IterateRangeDescriptorsFromDisk
+// except it is intended to be used when reading from checkpoints. Checkpoints
+// do not correspond to full stores, and as such, certain invariant checks do
+// not hold for them.
+func IterateRangeDescriptorsFromCheckpoint(
+	ctx context.Context, reader storage.Reader, fn func(desc roachpb.RangeDescriptor) error,
+) error {
+	return iterateRangeDescriptorsFromDiskHelper(ctx, reader, fn, false /* performInvariantChecks */)
+}
+
+// iterateRangeDescriptorsFromDisk performs the actual heavy lifting of reading
+// range descriptors from the supplied storage.Reader. The caller may specify
+// whether or not invariant checks should be performed.
+func iterateRangeDescriptorsFromDiskHelper(
+	ctx context.Context,
+	reader storage.Reader,
+	fn func(desc roachpb.RangeDescriptor) error,
+	performInvariantChecks bool,
 ) error {
 	log.KvExec.Info(ctx, "beginning range descriptor iteration")
 
@@ -296,8 +318,11 @@ func IterateRangeDescriptorsFromDisk(
 				iter.SeekGE(storage.MVCCKey{Key: keys.RangeDescriptorKey(keys.MustAddr(startKey))})
 			} else {
 				// This case shouldn't happen in practice: we have a key that isn't
-				// associated with any range descriptor.
-				if buildutil.CrdbTestBuild {
+				// associated with any range descriptor. However, when reading from
+				// checkpoints (incomplete stores), this can happen because checkpoints
+				// only guarantee to include specific key spans and may have arbitrary
+				// keys outside those spans.
+				if performInvariantChecks {
 					return errors.AssertionFailedf("range local key %s outside of a known range", key.Key)
 				}
 				iter.NextKey()

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 {