Merge #155549 #156334

155549: kvstorage: make replica destruction always sorted r=arulajmani a=pav-kv

This PR refactors replica destruction helpers in `kvstorage` (`DestroyReplica` and `SubsumeReplica`) so that they always generate storage writes ordered by engine key. This is to support the most restrictive user of `SubsumeReplica`: `prepareSnapApply` feeds these writes into SSTables (which assert on the keys order), part of snapshot ingestion path.

Other byproducts of this PR:

- `ClearRangeData` and its options are removed due to being redundant and not flexible.
- A decision is made to split the unreplicated RangeID-local keyspace into "state machine" and "raft", separated by the `RangeTombstone` key. This leaves us with only one exception: `RaftReplicaID` key in the middle of the "raft" keyspace is said to be a "state machine" key. We can migrate it to the correct half later, together with other storage migrations.
- The loop unwinding in `destroyReplicaImpl` brings us very close to logical separation of engines in this function. We now only need to handle the `RaftReplicaID` key correctly.
- The heuristic from `ClearRangeWithHeuristic` is now always used with the same threshold. There are no forced `ClearRawRange`s in the snapshot path anymore, unless opted in by tests.

Related to #152845

156334: sql: add read/write metrics r=yuzefovich a=andy-kimball

### 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.

### sql: add index_rows_written metric

Add new `sql.statements.index_rows_written.count` metric that counts the number
of primary and secondary index rows modified by SQL statements.

### sql: add bytes_read metric

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

Co-authored-by: Pavel Kalinnikov <pavel@cockroachlabs.com>
Co-authored-by: Andrew Kimball <andyk@cockroachlabs.com>

Author: 3 people

docs/generated/metrics/metrics.yaml

@@ -8747,17 +8747,65 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: sql.statements.bytes_read.count
+      exported_name: sql_statements_bytes_read_count
+      description: Number of bytes read by SQL statements
+      y_axis_label: SQL Statements
+      type: COUNTER
+      unit: BYTES
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: sql.statements.bytes_read.count.internal
+      exported_name: sql_statements_bytes_read_count_internal
+      description: Number of bytes read 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_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
+      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 from primary and secondary indexes
+      description: Number of rows read by SQL statements
       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)
+      description: Number of rows read by SQL statements (internal queries)
       y_axis_label: SQL Internal Statements
       type: COUNTER
       unit: COUNT

pkg/keys/doc.go

@@ -192,6 +192,16 @@ var schema = [...]interface{}{
 	//   pertain to just one replica of a range. They are unreplicated and
 	//   unaddressable. The typical example is the Raft log. They all share
 	//   `LocalRangeIDPrefix` and `localRangeIDUnreplicatedInfix`.
+	//
+	// WARNING: when adding a new key in this section, decide whether it should be
+	// classified as "raft" or "state machine" key, correspondingly to which
+	// engine it resides in:
+	//
+	//	- keys <= RangeTombstoneKey in this prefix are "state machine" engine keys
+	//	- keys > RangeTombstoneKey in this prefix are "raft" engine keys
+	//	- historical exception: RaftReplicaIDKey belongs to the state machine
+	//
+	// Failure to classify may result in replica state corruption in storage.
 	localRangeIDUnreplicatedInfix,  // "u"
 	RangeTombstoneKey,              // "rftb"
 	RaftHardStateKey,               // "rfth"

pkg/kv/kvserver/client_merge_test.go

@@ -3815,6 +3815,10 @@ func TestStoreRangeMergeRaftSnapshot(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
+	// The test expects replica destruction paths to use ranged clears rather than
+	// point clears.
+	defer kvstorage.TestingForceClearRange()()
+
 	testutils.RunTrueAndFalse(t, "rebalanceRHSAway", func(t *testing.T, rebalanceRHSAway bool) {
 		// We will be testing the SSTs written on store2's engine.
 		var receivingEng, sendingEng storage.Engine
@@ -3852,11 +3856,7 @@ func TestStoreRangeMergeRaftSnapshot(t *testing.T) {
 			// NOTE: There are no range-local keys or lock table keys, in [d,
 			// /Max) in the store we're sending a snapshot to, so we aren't
 			// expecting SSTs to clear those keys.
-			expectedSSTCount := 9
-			if len(sstNames) != expectedSSTCount {
-				return errors.Errorf("expected to ingest %d SSTs, got %d SSTs",
-					expectedSSTCount, len(sstNames))
-			}
+			require.Len(t, sstNames, 9)
 
 			// Only try to predict SSTs for:
 			// - The user keys in the snapshot
@@ -3900,11 +3900,11 @@ func TestStoreRangeMergeRaftSnapshot(t *testing.T) {
 				file := &storage.MemObject{}
 				writer := storage.MakeIngestionSSTWriter(ctx, st, file)
 				if i < len(keySpans)-1 {
-					// The last span is the MVCC span, and is always cleared via
-					// Excise. See MultiSSTWriter.
-					if err := writer.ClearRawRange(span.Key, span.EndKey, true /* pointKeys */, true /* rangeKeys */); err != nil {
-						return err
-					}
+					// The last span is the MVCC span, and is always cleared via Excise.
+					// See MultiSSTWriter.
+					require.NoError(t, writer.ClearRawRange(
+						span.Key, span.EndKey, true /* pointKeys */, true, /* rangeKeys */
+					))
 				}
 				sstFileWriters[string(span.Key)] = sstFileWriter{
 					span:   span,
@@ -3913,7 +3913,7 @@ func TestStoreRangeMergeRaftSnapshot(t *testing.T) {
 				}
 			}
 
-			if err := rditer.IterateReplicaKeySpans(ctx, inSnap.Desc, snapReader, rditer.SelectOpts{
+			require.NoError(t, rditer.IterateReplicaKeySpans(ctx, inSnap.Desc, snapReader, rditer.SelectOpts{
 				Ranged: rditer.SelectRangedOptions{
 					SystemKeys: true,
 					LockTable:  true,
@@ -3923,43 +3923,27 @@ func TestStoreRangeMergeRaftSnapshot(t *testing.T) {
 				UnreplicatedByRangeID: false,
 			}, func(iter storage.EngineIterator, span roachpb.Span) error {
 				fw, ok := sstFileWriters[string(span.Key)]
-				if !ok || !fw.span.Equal(span) {
-					return errors.Errorf("unexpected span %s", span)
-				}
+				require.True(t, ok)
+				require.Truef(t, fw.span.Equal(span), "unexpected span %s", span)
 				var err error
 				for ok := true; ok && err == nil; ok, err = iter.NextEngineKey() {
-					var key storage.EngineKey
-					if key, err = iter.UnsafeEngineKey(); err != nil {
-						return err
-					}
+					key, err := iter.UnsafeEngineKey()
+					require.NoError(t, err)
 					v, err := iter.UnsafeValue()
-					if err != nil {
-						return err
-					}
-					if err := fw.writer.PutEngineKey(key, v); err != nil {
-						return err
-					}
-				}
-				if err != nil {
-					return err
+					require.NoError(t, err)
+					require.NoError(t, fw.writer.PutEngineKey(key, v))
 				}
+				require.NoError(t, err)
 				return nil
-			}); err != nil {
-				return err
-			}
+			}))
 
 			for _, span := range keySpans {
 				fw := sstFileWriters[string(span.Key)]
-				if err := fw.writer.Finish(); err != nil {
-					return err
-				}
+				require.NoError(t, fw.writer.Finish())
 				expectedSSTs = append(expectedSSTs, fw.file.Data())
 			}
 
-			if len(expectedSSTs) != 5 {
-				return errors.Errorf("len of expectedSSTs should expected to be %d, but got %d",
-					5, len(expectedSSTs))
-			}
+			require.Len(t, expectedSSTs, 5)
 			// Keep the last one which contains the user keys.
 			expectedSSTs = expectedSSTs[len(expectedSSTs)-1:]
 
@@ -3973,23 +3957,24 @@ func TestStoreRangeMergeRaftSnapshot(t *testing.T) {
 				sstFile := &storage.MemObject{}
 				sst := storage.MakeIngestionSSTWriter(ctx, st, sstFile)
 				defer sst.Close()
+				// The snapshot code uses ClearRangeWithHeuristic with a threshold of 1
+				// to clear the range, but it will truncate the range tombstone to the
+				// first key. In this case, the first key is RangeGCThresholdKey, which
+				// doesn't yet exist in the engine, so we set it manually.
+				//
+				// The deletion also extends to the RangeTombstoneKey.
+				require.NoError(t, sst.ClearRawRange(
+					keys.RangeGCThresholdKey(rangeID),
+					keys.RangeTombstoneKey(rangeID),
+					true, false,
+				))
+				require.NoError(t, kvstorage.MakeStateLoader(rangeID).SetRangeTombstone(
+					context.Background(), &sst,
+					kvserverpb.RangeTombstone{NextReplicaID: math.MaxInt32},
+				))
 				{
-					// The snapshot code will use ClearRangeWithHeuristic with a
-					// threshold of 1 to clear the range, but this will truncate
-					// the range tombstone to the first key. In this case, the
-					// first key is RangeGCThresholdKey, which doesn't yet exist
-					// in the engine, so we write the Pebble range tombstone
-					// manually.
-					sl := rditer.Select(rangeID, rditer.SelectOpts{
-						ReplicatedByRangeID: true,
-					})
-					require.Len(t, sl, 1)
-					s := sl[0]
-					require.NoError(t, sst.ClearRawRange(keys.RangeGCThresholdKey(rangeID), s.EndKey, true, false))
-				}
-				{
-					// Ditto for the unreplicated version, where the first key
-					// happens to be the HardState.
+					// Ditto for the unreplicated version, where the first key happens to
+					// be the HardState.
 					sl := rditer.Select(rangeID, rditer.SelectOpts{
 						UnreplicatedByRangeID: true,
 					})
@@ -3998,15 +3983,7 @@ func TestStoreRangeMergeRaftSnapshot(t *testing.T) {
 					require.NoError(t, sst.ClearRawRange(keys.RaftHardStateKey(rangeID), s.EndKey, true, false))
 				}
 
-				if err := kvstorage.MakeStateLoader(rangeID).SetRangeTombstone(
-					context.Background(), &sst,
-					kvserverpb.RangeTombstone{NextReplicaID: math.MaxInt32},
-				); err != nil {
-					return err
-				}
-				if err := sst.Finish(); err != nil {
-					return err
-				}
+				require.NoError(t, sst.Finish())
 				expectedSSTs = append(expectedSSTs, sstFile.Data())
 			}
 
@@ -4018,23 +3995,20 @@ func TestStoreRangeMergeRaftSnapshot(t *testing.T) {
 				StartKey: roachpb.RKey(keyD),
 				EndKey:   roachpb.RKey(keyEnd),
 			}
-			if err := storage.ClearRangeWithHeuristic(
-				ctx, receivingEng, &sst, desc.StartKey.AsRawKey(), desc.EndKey.AsRawKey(), 64,
-			); err != nil {
-				return err
-			} else if err := sst.Finish(); err != nil {
-				return err
-			}
+			require.NoError(t, storage.ClearRangeWithHeuristic(
+				ctx, receivingEng, &sst,
+				desc.StartKey.AsRawKey(), desc.EndKey.AsRawKey(),
+				kvstorage.ClearRangeThresholdPointKeys(),
+			))
+			require.NoError(t, sst.Finish())
 			expectedSSTs = append(expectedSSTs, sstFile.Data())
 
 			// Iterate over all the tested SSTs and check that they're
 			// byte-by-byte equal.
 			var dumpDir string
 			for i := range sstNamesSubset {
 				actualSST, err := fs.ReadFile(receivingEng.Env(), sstNamesSubset[i])
-				if err != nil {
-					return err
-				}
+				require.NoError(t, err)
 				if !bytes.Equal(expectedSSTs[i], actualSST) { // intentionally not printing
 					t.Logf("%d=%s", i, sstNamesSubset[i])
 					if dumpDir == "" {