kvstorage: factor out SubsumeReplica storage call

Epic: none
Release note: none

Author: pav-kv

pkg/kv/kvserver/kvstorage/destroy.go

@@ -174,3 +174,28 @@ func DestroyReplica(
 		NextReplicaID: next, // NB: NextReplicaID > 0
 	})
 }
+
+// SubsumeReplica is like DestroyReplica, but it does not delete the user keys
+// (and the corresponding system and lock table keys). The latter are inherited
+// by the subsuming range.
+//
+// Returns SelectOpts which can be used to reflect on the key spans that this
+// function clears.
+// TODO(pav-kv): get rid of SelectOpts.
+func SubsumeReplica(
+	ctx context.Context, reader storage.Reader, writer storage.Writer, info DestroyReplicaInfo,
+) (rditer.SelectOpts, error) {
+	// NB: set MustUseClearRange to true because this call is used for generating
+	// SSTables when ingesting a snapshot, which requires Clears and Puts to be
+	// written in key order. DestroyReplica sets RangeTombstoneKey after clearing
+	// the unreplicated span which may contain higher keys.
+	opts := ClearRangeDataOptions{
+		ClearReplicatedByRangeID:   true,
+		ClearUnreplicatedByRangeID: true,
+		MustUseClearRange:          true,
+	}
+	return rditer.SelectOpts{
+		ReplicatedByRangeID:   opts.ClearReplicatedByRangeID,
+		UnreplicatedByRangeID: opts.ClearUnreplicatedByRangeID,
+	}, DestroyReplica(ctx, reader, writer, info, MergedTombstoneReplicaID, opts)
+}

pkg/kv/kvserver/snapshot_apply_prepare.go

@@ -150,20 +150,9 @@ func (s *snapWriteBuilder) clearSubsumedReplicaDiskData(ctx context.Context) err
 	for _, sub := range s.subsume {
 		// We have to create an SST for the subsumed replica's range-id local keys.
 		if err := s.writeSST(ctx, func(ctx context.Context, w storage.Writer) error {
-			// NOTE: We set mustClearRange to true because we are setting
-			// RangeTombstoneKey. Since Clears and Puts need to be done in increasing
-			// order of keys, it is not safe to use ClearRangeIter.
-			opts := kvstorage.ClearRangeDataOptions{
-				ClearReplicatedByRangeID:   true,
-				ClearUnreplicatedByRangeID: true,
-				MustUseClearRange:          true,
-			}
-			s.cleared = append(s.cleared, rditer.Select(sub.RangeID, rditer.SelectOpts{
-				ReplicatedByRangeID:   opts.ClearReplicatedByRangeID,
-				UnreplicatedByRangeID: opts.ClearUnreplicatedByRangeID,
-			})...)
-			// NB: Actually clear RangeID local key spans.
-			return kvstorage.DestroyReplica(ctx, reader, w, sub, kvstorage.MergedTombstoneReplicaID, opts)
+			opts, err := kvstorage.SubsumeReplica(ctx, reader, w, sub)
+			s.cleared = append(s.cleared, rditer.Select(sub.RangeID, opts)...)
+			return err
 		}); err != nil {
 			return err
 		}