migration: introduce primitives for below-raft migrations

docs/generated/settings/settings.html

@@ -100,6 +100,6 @@
 <tr><td><code>trace.debug.enable</code></td><td>boolean</td><td><code>false</code></td><td>if set, traces for recent requests can be seen at https://<ui>/debug/requests</td></tr>
 <tr><td><code>trace.lightstep.token</code></td><td>string</td><td><code></code></td><td>if set, traces go to Lightstep using this token</td></tr>
 <tr><td><code>trace.zipkin.collector</code></td><td>string</td><td><code></code></td><td>if set, traces go to the given Zipkin instance (example: '127.0.0.1:9411'); ignored if trace.lightstep.token is set</td></tr>
-<tr><td><code>version</code></td><td>version</td><td><code>20.2-10</code></td><td>set the active cluster version in the format '<major>.<minor>'</td></tr>
+<tr><td><code>version</code></td><td>version</td><td><code>20.2-16</code></td><td>set the active cluster version in the format '<major>.<minor>'</td></tr>
 </tbody>
 </table>

pkg/clusterversion/cockroach_versions.go

@@ -206,6 +206,22 @@ const (
 	VirtualComputedColumns
 	// CPutInline is conditional put support for inline values.
 	CPutInline
+	// ReplicaVersions enables the versioning of Replica state.
+	ReplicaVersions
+	// TruncatedAndRangeAppliedStateMigration is part of the migration to stop
+	// using the legacy truncated state within KV. After the migration, we'll be
+	// using the unreplicated truncated state and the RangeAppliedState on all
+	// ranges. Callers that wish to assert on there no longer being any legacy
+	// will be able to do so after PostTruncatedAndRangeAppliedStateMigration is
+	// active. This lets remove any holdover code handling the possibility of
+	// replicated truncated state in 21.2.
+	//
+	// TODO(irfansharif): Do the above in 21.2.
+	TruncatedAndRangeAppliedStateMigration
+	// PostTruncatedAndRangeAppliedStateMigration is used to purge all replicas
+	// using the replicated legacy TruncatedState. It's also used in asserting
+	// that no replicated truncated state representation is found.
+	PostTruncatedAndRangeAppliedStateMigration
 
 	// Step (1): Add new versions here.
 )
@@ -340,6 +356,18 @@ var versionsSingleton = keyedVersions([]keyedVersion{
 		Key:     CPutInline,
 		Version: roachpb.Version{Major: 20, Minor: 2, Internal: 10},
 	},
+	{
+		Key:     ReplicaVersions,
+		Version: roachpb.Version{Major: 20, Minor: 2, Internal: 12},
+	},
+	{
+		Key:     TruncatedAndRangeAppliedStateMigration,
+		Version: roachpb.Version{Major: 20, Minor: 2, Internal: 14},
+	},
+	{
+		Key:     PostTruncatedAndRangeAppliedStateMigration,
+		Version: roachpb.Version{Major: 20, Minor: 2, Internal: 16},
+	},
 
 	// Step (2): Add new versions here.
 })

pkg/keys/constants.go

@@ -93,6 +93,8 @@ var (
 	// LocalLeaseAppliedIndexLegacySuffix is the suffix for the applied lease
 	// index.
 	LocalLeaseAppliedIndexLegacySuffix = []byte("rlla")
+	// LocalRangeVersionSuffix is the suffix for the range version.
+	LocalRangeVersionSuffix = []byte("rver")
 	// LocalRangeStatsLegacySuffix is the suffix for range statistics.
 	LocalRangeStatsLegacySuffix = []byte("stat")
 	// localTxnSpanGCThresholdSuffix is DEPRECATED and remains to prevent reuse.

pkg/keys/doc.go

@@ -189,6 +189,7 @@ var _ = [...]interface{}{
 	RaftTruncatedStateLegacyKey, // "rftt"
 	RangeLeaseKey,               // "rll-"
 	LeaseAppliedIndexLegacyKey,  // "rlla"
+	RangeVersionKey,             // "rver"
 	RangeStatsLegacyKey,         // "stat"
 
 	//   2. Unreplicated range-ID local keys: These contain metadata that

pkg/keys/keys.go

@@ -286,6 +286,11 @@ func RangeLastGCKey(rangeID roachpb.RangeID) roachpb.Key {
 	return MakeRangeIDPrefixBuf(rangeID).RangeLastGCKey()
 }
 
+// RangeVersionKey returns a system-local for the range version.
+func RangeVersionKey(rangeID roachpb.RangeID) roachpb.Key {
+	return MakeRangeIDPrefixBuf(rangeID).RangeVersionKey()
+}
+
 // MakeRangeIDUnreplicatedPrefix creates a range-local key prefix from
 // rangeID for all unreplicated data.
 func MakeRangeIDUnreplicatedPrefix(rangeID roachpb.RangeID) roachpb.Key {
@@ -964,6 +969,11 @@ func (b RangeIDPrefixBuf) RangeLastGCKey() roachpb.Key {
 	return append(b.replicatedPrefix(), LocalRangeLastGCSuffix...)
 }
 
+// RangeVersionKey returns a system-local key for the range version.
+func (b RangeIDPrefixBuf) RangeVersionKey() roachpb.Key {
+	return append(b.replicatedPrefix(), LocalRangeVersionSuffix...)
+}
+
 // RangeTombstoneKey returns a system-local key for a range tombstone.
 func (b RangeIDPrefixBuf) RangeTombstoneKey() roachpb.Key {
 	return append(b.unreplicatedPrefix(), LocalRangeTombstoneSuffix...)

pkg/keys/printer.go

@@ -171,6 +171,7 @@ var (
 		{name: "RangeLease", suffix: LocalRangeLeaseSuffix},
 		{name: "RangeStats", suffix: LocalRangeStatsLegacySuffix},
 		{name: "RangeLastGC", suffix: LocalRangeLastGCSuffix},
+		{name: "RangeVersion", suffix: LocalRangeVersionSuffix},
 	}
 
 	rangeSuffixDict = []struct {

pkg/keys/printer_test.go

@@ -76,6 +76,7 @@ func TestPrettyPrint(t *testing.T) {
 		{keys.RangeLeaseKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/r/RangeLease", revertSupportUnknown},
 		{keys.RangeStatsLegacyKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/r/RangeStats", revertSupportUnknown},
 		{keys.RangeLastGCKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/r/RangeLastGC", revertSupportUnknown},
+		{keys.RangeVersionKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/r/RangeVersion", revertSupportUnknown},
 
 		{keys.RaftHardStateKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/u/RaftHardState", revertSupportUnknown},
 		{keys.RangeTombstoneKey(roachpb.RangeID(1000001)), "/Local/RangeID/1000001/u/RangeTombstone", revertSupportUnknown},

pkg/kv/batch.go

@@ -228,18 +228,10 @@ func (b *Batch) fillResults(ctx context.Context) {
 			case *roachpb.DeleteRequest:
 				row := &result.Rows[k]
 				row.Key = []byte(args.(*roachpb.DeleteRequest).Key)
-
 			case *roachpb.DeleteRangeRequest:
 				if result.Err == nil {
 					result.Keys = reply.(*roachpb.DeleteRangeResponse).Keys
 				}
-
-			default:
-				if result.Err == nil {
-					result.Err = errors.Errorf("unsupported reply: %T for %T",
-						reply, args)
-				}
-
 			// Nothing to do for all methods below as they do not generate
 			// any rows.
 			case *roachpb.EndTxnRequest:
@@ -265,6 +257,12 @@ func (b *Batch) fillResults(ctx context.Context) {
 			case *roachpb.ImportRequest:
 			case *roachpb.AdminScatterRequest:
 			case *roachpb.AddSSTableRequest:
+			case *roachpb.MigrateRequest:
+			default:
+				if result.Err == nil {
+					result.Err = errors.Errorf("unsupported reply: %T for %T",
+						reply, args)
+				}
 			}
 			// Fill up the resume span.
 			if result.Err == nil && reply != nil && reply.Header().ResumeSpan != nil {
@@ -786,3 +784,26 @@ func (b *Batch) addSSTable(
 	b.appendReqs(req)
 	b.initResult(1, 0, notRaw, nil)
 }
+
+// migrate is only exported on DB.
+func (b *Batch) migrate(s, e interface{}, version roachpb.Version) {
+	begin, err := marshalKey(s)
+	if err != nil {
+		b.initResult(0, 0, notRaw, err)
+		return
+	}
+	end, err := marshalKey(e)
+	if err != nil {
+		b.initResult(0, 0, notRaw, err)
+		return
+	}
+	req := &roachpb.MigrateRequest{
+		RequestHeader: roachpb.RequestHeader{
+			Key:    begin,
+			EndKey: end,
+		},
+		Version: version,
+	}
+	b.appendReqs(req)
+	b.initResult(1, 0, notRaw, nil)
+}

pkg/kv/db.go

@@ -681,6 +681,16 @@ func (db *DB) AddSSTable(
 	return getOneErr(db.Run(ctx, b), b)
 }
 
+// Migrate is used instruct all ranges overlapping with the provided keyspace to
+// exercise any relevant (below-raft) migrations in order for its range state to
+// conform to what's needed by the specified version. It's a core primitive used
+// in our migrations infrastructure to phase out legacy code below raft.
+func (db *DB) Migrate(ctx context.Context, begin, end interface{}, version roachpb.Version) error {
+	b := &Batch{}
+	b.migrate(begin, end, version)
+	return getOneErr(db.Run(ctx, b), b)
+}
+
 // sendAndFill is a helper which sends the given batch and fills its results,
 // returning the appropriate error which is either from the first failing call,
 // or an "internal" error.

pkg/kv/kvserver/BUILD.bazel

@@ -204,6 +204,7 @@ go_test(
         "client_lease_test.go",
         "client_merge_test.go",
         "client_metrics_test.go",
+        "client_migration_test.go",
         "client_protectedts_test.go",
         "client_raft_helpers_test.go",
         "client_raft_log_queue_test.go",

pkg/kv/kvserver/batcheval/BUILD.bazel

@@ -20,6 +20,7 @@ go_library(
         "cmd_lease_request.go",
         "cmd_lease_transfer.go",
         "cmd_merge.go",
+        "cmd_migrate.go",
         "cmd_push_txn.go",
         "cmd_put.go",
         "cmd_query_intent.go",

pkg/kv/kvserver/batcheval/cmd_end_transaction.go

@@ -901,7 +901,7 @@ func splitTriggerHelper(
 	// initial state. Additionally, since bothDeltaMS is tracking writes to
 	// both sides, we need to update it as well.
 	{
-		// Various pieces of code rely on a replica's lease never being unitialized,
+		// Various pieces of code rely on a replica's lease never being uninitialized,
 		// but it's more than that - it ensures that we properly initialize the
 		// timestamp cache, which is only populated on the lease holder, from that
 		// of the original Range.  We found out about a regression here the hard way
@@ -918,8 +918,9 @@ func splitTriggerHelper(
 		// - node two can illegally propose a write to 'd' at a lower timestamp.
 		//
 		// TODO(tschottdorf): why would this use r.store.Engine() and not the
-		// batch?
-		leftLease, err := MakeStateLoader(rec).LoadLease(ctx, rec.Engine())
+		// batch? We do the same thing for other usages of the state loader.
+		sl := MakeStateLoader(rec)
+		leftLease, err := sl.LoadLease(ctx, rec.Engine())
 		if err != nil {
 			return enginepb.MVCCStats{}, result.Result{}, errors.Wrap(err, "unable to load lease")
 		}
@@ -936,8 +937,7 @@ func splitTriggerHelper(
 		}
 		rightLease := leftLease
 		rightLease.Replica = replica
-
-		gcThreshold, err := MakeStateLoader(rec).LoadGCThreshold(ctx, rec.Engine())
+		gcThreshold, err := sl.LoadGCThreshold(ctx, rec.Engine())
 		if err != nil {
 			return enginepb.MVCCStats{}, result.Result{}, errors.Wrap(err, "unable to load GCThreshold")
 		}
@@ -968,6 +968,11 @@ func splitTriggerHelper(
 			truncStateType = stateloader.TruncatedStateLegacyReplicated
 		}
 
+		replicaVersion, err := sl.LoadVersion(ctx, rec.Engine())
+		if err != nil {
+			return enginepb.MVCCStats{}, result.Result{}, errors.Wrap(err, "unable to load GCThreshold")
+		}
+
 		// Writing the initial state is subtle since this also seeds the Raft
 		// group. It becomes more subtle due to proposer-evaluated Raft.
 		//
@@ -1000,7 +1005,7 @@ func splitTriggerHelper(
 
 		*h.AbsPostSplitRight(), err = stateloader.WriteInitialReplicaState(
 			ctx, batch, *h.AbsPostSplitRight(), split.RightDesc, rightLease,
-			*gcThreshold, truncStateType,
+			*gcThreshold, truncStateType, replicaVersion,
 		)
 		if err != nil {
 			return enginepb.MVCCStats{}, result.Result{}, errors.Wrap(err, "unable to write initial Replica state")