diff --git a/pkg/kv/kvclient/kvcoord/txn_coord_sender_test.go b/pkg/kv/kvclient/kvcoord/txn_coord_sender_test.go
index 24cd33850b6e..27687979fece 100644
--- a/pkg/kv/kvclient/kvcoord/txn_coord_sender_test.go
+++ b/pkg/kv/kvclient/kvcoord/txn_coord_sender_test.go
@@ -378,12 +378,12 @@ func verifyCleanup(
 				return fmt.Errorf("expected no heartbeat")
 			}
 		}
-		_, intent, err := storage.MVCCGet(ctx, eng, key, hlc.MaxTimestamp, storage.MVCCGetOptions{
+		intentRes, err := storage.MVCCGet(ctx, eng, key, hlc.MaxTimestamp, storage.MVCCGetOptions{
 			Inconsistent: true,
 		})
 		require.NoError(t, err)
-		if intent != nil {
-			return fmt.Errorf("found unexpected write intent: %s", intent)
+		if intentRes.Intent != nil {
+			return fmt.Errorf("found unexpected write intent: %s", intentRes.Intent)
 		}
 		return nil
 	})
diff --git a/pkg/kv/kvserver/batcheval/cmd_end_transaction_test.go b/pkg/kv/kvserver/batcheval/cmd_end_transaction_test.go
index e67a7a5d4f47..36f282bc6e0e 100644
--- a/pkg/kv/kvserver/batcheval/cmd_end_transaction_test.go
+++ b/pkg/kv/kvserver/batcheval/cmd_end_transaction_test.go
@@ -1062,17 +1062,17 @@ func TestPartialRollbackOnEndTransaction(t *testing.T) {
 
 		// The second write has been rolled back; verify that the remaining
 		// value is from the first write.
-		res, i, err := storage.MVCCGet(ctx, batch, k, ts2, storage.MVCCGetOptions{})
+		res, err := storage.MVCCGet(ctx, batch, k, ts2, storage.MVCCGetOptions{})
 		if err != nil {
 			t.Fatal(err)
 		}
-		if i != nil {
-			t.Errorf("found intent, expected none: %+v", i)
+		if res.Intent != nil {
+			t.Errorf("found intent, expected none: %+v", res.Intent)
 		}
-		if res == nil {
+		if res.Value == nil {
 			t.Errorf("no value found, expected one")
 		} else {
-			s, err := res.GetBytes()
+			s, err := res.Value.GetBytes()
 			if err != nil {
 				t.Fatal(err)
 			}
diff --git a/pkg/kv/kvserver/batcheval/cmd_get.go b/pkg/kv/kvserver/batcheval/cmd_get.go
index 8a6506ce11aa..04c76685eb8e 100644
--- a/pkg/kv/kvserver/batcheval/cmd_get.go
+++ b/pkg/kv/kvserver/batcheval/cmd_get.go
@@ -32,27 +32,7 @@ func Get(
 	h := cArgs.Header
 	reply := resp.(*roachpb.GetResponse)
 
-	if h.MaxSpanRequestKeys < 0 || h.TargetBytes < 0 {
-		// Receipt of a GetRequest with negative MaxSpanRequestKeys or TargetBytes
-		// indicates that the request was part of a batch that has already exhausted
-		// its limit, which means that we should *not* serve the request and return
-		// a ResumeSpan for this GetRequest.
-		//
-		// This mirrors the logic in MVCCScan, though the logic in MVCCScan is
-		// slightly lower in the stack.
-		reply.ResumeSpan = &roachpb.Span{Key: args.Key}
-		if h.MaxSpanRequestKeys < 0 {
-			reply.ResumeReason = roachpb.RESUME_KEY_LIMIT
-		} else if h.TargetBytes < 0 {
-			reply.ResumeReason = roachpb.RESUME_BYTE_LIMIT
-		}
-		return result.Result{}, nil
-	}
-
-	var val *roachpb.Value
-	var intent *roachpb.Intent
-	var err error
-	val, intent, err = storage.MVCCGet(ctx, reader, args.Key, h.Timestamp, storage.MVCCGetOptions{
+	getRes, err := storage.MVCCGet(ctx, reader, args.Key, h.Timestamp, storage.MVCCGetOptions{
 		Inconsistent:          h.ReadConsistency != roachpb.CONSISTENT,
 		SkipLocked:            h.WaitPolicy == lock.WaitPolicy_SkipLocked,
 		Txn:                   h.Txn,
@@ -61,29 +41,27 @@ func Get(
 		MemoryAccount:         cArgs.EvalCtx.GetResponseMemoryAccount(),
 		LockTable:             cArgs.Concurrency,
 		DontInterleaveIntents: cArgs.DontInterleaveIntents,
+		MaxKeys:               cArgs.Header.MaxSpanRequestKeys,
+		TargetBytes:           cArgs.Header.TargetBytes,
+		AllowEmpty:            cArgs.Header.AllowEmpty,
 	})
 	if err != nil {
 		return result.Result{}, err
 	}
-	if val != nil {
-		// NB: This calculation is different from Scan, since Scan responses include
-		// the key/value pair while Get only includes the value.
-		numBytes := int64(len(val.RawBytes))
-		if h.TargetBytes > 0 && h.AllowEmpty && numBytes > h.TargetBytes {
-			reply.ResumeSpan = &roachpb.Span{Key: args.Key}
-			reply.ResumeReason = roachpb.RESUME_BYTE_LIMIT
-			reply.ResumeNextBytes = numBytes
-			return result.Result{}, nil
-		}
-		reply.NumKeys = 1
-		reply.NumBytes = numBytes
+	reply.ResumeSpan = getRes.ResumeSpan
+	reply.ResumeReason = getRes.ResumeReason
+	reply.ResumeNextBytes = getRes.ResumeNextBytes
+	reply.NumKeys = getRes.NumKeys
+	reply.NumBytes = getRes.NumBytes
+	if reply.ResumeSpan != nil {
+		return result.Result{}, nil
 	}
 	var intents []roachpb.Intent
-	if intent != nil {
-		intents = append(intents, *intent)
+	if getRes.Intent != nil {
+		intents = append(intents, *getRes.Intent)
 	}
 
-	reply.Value = val
+	reply.Value = getRes.Value
 	if h.ReadConsistency == roachpb.READ_UNCOMMITTED {
 		var intentVals []roachpb.KeyValue
 		// NOTE: MVCCGet uses a Prefix iterator, so we want to use one in
@@ -103,7 +81,7 @@ func Get(
 	}
 
 	var res result.Result
-	if args.KeyLocking != lock.None && h.Txn != nil && val != nil {
+	if args.KeyLocking != lock.None && h.Txn != nil && getRes.Value != nil {
 		acq := roachpb.MakeLockAcquisition(h.Txn, args.Key, lock.Unreplicated)
 		res.Local.AcquiredLocks = []roachpb.LockAcquisition{acq}
 	}
diff --git a/pkg/kv/kvserver/batcheval/cmd_refresh.go b/pkg/kv/kvserver/batcheval/cmd_refresh.go
index 4a3c525f621a..548215a22ad2 100644
--- a/pkg/kv/kvserver/batcheval/cmd_refresh.go
+++ b/pkg/kv/kvserver/batcheval/cmd_refresh.go
@@ -54,15 +54,15 @@ func Refresh(
 	// specifying consistent=false. Note that we include tombstones,
 	// which must be considered as updates on refresh.
 	log.VEventf(ctx, 2, "refresh %s @[%s-%s]", args.Span(), refreshFrom, refreshTo)
-	val, intent, err := storage.MVCCGet(ctx, reader, args.Key, refreshTo, storage.MVCCGetOptions{
+	res, err := storage.MVCCGet(ctx, reader, args.Key, refreshTo, storage.MVCCGetOptions{
 		Inconsistent: true,
 		Tombstones:   true,
 	})
 
 	if err != nil {
 		return result.Result{}, err
-	} else if val != nil {
-		if ts := val.Timestamp; refreshFrom.Less(ts) {
+	} else if res.Value != nil {
+		if ts := res.Value.Timestamp; refreshFrom.Less(ts) {
 			return result.Result{},
 				roachpb.NewRefreshFailedError(roachpb.RefreshFailedError_REASON_COMMITTED_VALUE, args.Key, ts)
 		}
@@ -70,9 +70,9 @@ func Refresh(
 
 	// Check if an intent which is not owned by this transaction was written
 	// at or beneath the refresh timestamp.
-	if intent != nil && intent.Txn.ID != h.Txn.ID {
+	if res.Intent != nil && res.Intent.Txn.ID != h.Txn.ID {
 		return result.Result{}, roachpb.NewRefreshFailedError(roachpb.RefreshFailedError_REASON_INTENT,
-			intent.Key, intent.Txn.WriteTimestamp)
+			res.Intent.Key, res.Intent.Txn.WriteTimestamp)
 	}
 
 	return result.Result{}, nil
diff --git a/pkg/kv/kvserver/batcheval/cmd_refresh_range_test.go b/pkg/kv/kvserver/batcheval/cmd_refresh_range_test.go
index ac4bba3eebbb..1f46da432606 100644
--- a/pkg/kv/kvserver/batcheval/cmd_refresh_range_test.go
+++ b/pkg/kv/kvserver/batcheval/cmd_refresh_range_test.go
@@ -188,12 +188,12 @@ func TestRefreshRangeTimeBoundIterator(t *testing.T) {
 	// have previously performed a consistent read at the lower time-bound to
 	// prove that there are no intents present that would be missed by the time-
 	// bound iterator.
-	if val, intent, err := storage.MVCCGet(ctx, db, k, ts1, storage.MVCCGetOptions{}); err != nil {
+	if res, err := storage.MVCCGet(ctx, db, k, ts1, storage.MVCCGetOptions{}); err != nil {
 		t.Fatal(err)
-	} else if intent != nil {
+	} else if res.Intent != nil {
 		t.Fatalf("got unexpected intent: %v", intent)
-	} else if !val.EqualTagAndData(v) {
-		t.Fatalf("expected %v, got %v", v, val)
+	} else if !res.Value.EqualTagAndData(v) {
+		t.Fatalf("expected %v, got %v", v, res.Value)
 	}
 
 	// Now the real test: a transaction at ts2 has been pushed to ts3
diff --git a/pkg/kv/kvserver/batcheval/cmd_resolve_intent_test.go b/pkg/kv/kvserver/batcheval/cmd_resolve_intent_test.go
index 94d4e5ff6a98..13f9a70faabd 100644
--- a/pkg/kv/kvserver/batcheval/cmd_resolve_intent_test.go
+++ b/pkg/kv/kvserver/batcheval/cmd_resolve_intent_test.go
@@ -253,17 +253,17 @@ func TestResolveIntentAfterPartialRollback(t *testing.T) {
 
 		// The second write has been rolled back; verify that the remaining
 		// value is from the first write.
-		res, i, err := storage.MVCCGet(ctx, batch, k, ts2, storage.MVCCGetOptions{})
+		res, err := storage.MVCCGet(ctx, batch, k, ts2, storage.MVCCGetOptions{})
 		if err != nil {
 			t.Fatal(err)
 		}
-		if i != nil {
-			t.Errorf("%s: found intent, expected none: %+v", k, i)
+		if res.Intent != nil {
+			t.Errorf("%s: found intent, expected none: %+v", k, res.Intent)
 		}
-		if res == nil {
+		if res.Value == nil {
 			t.Errorf("%s: no value found, expected one", k)
 		} else {
-			s, err := res.GetBytes()
+			s, err := res.Value.GetBytes()
 			if err != nil {
 				t.Fatal(err)
 			}
diff --git a/pkg/kv/kvserver/batcheval/cmd_subsume.go b/pkg/kv/kvserver/batcheval/cmd_subsume.go
index db5baae593e0..093249b10706 100644
--- a/pkg/kv/kvserver/batcheval/cmd_subsume.go
+++ b/pkg/kv/kvserver/batcheval/cmd_subsume.go
@@ -100,17 +100,17 @@ func Subsume(
 	// the maximum timestamp to ensure that we see an intent if one exists,
 	// regardless of what timestamp it is written at.
 	descKey := keys.RangeDescriptorKey(desc.StartKey)
-	_, intent, err := storage.MVCCGet(ctx, readWriter, descKey, hlc.MaxTimestamp,
+	intentRes, err := storage.MVCCGet(ctx, readWriter, descKey, hlc.MaxTimestamp,
 		storage.MVCCGetOptions{Inconsistent: true})
 	if err != nil {
 		return result.Result{}, errors.Wrap(err, "fetching local range descriptor")
-	} else if intent == nil {
+	} else if intentRes.Intent == nil {
 		return result.Result{}, errors.Errorf("range missing intent on its local descriptor")
 	}
-	val, _, err := storage.MVCCGetAsTxn(ctx, readWriter, descKey, intent.Txn.WriteTimestamp, intent.Txn)
+	valRes, err := storage.MVCCGetAsTxn(ctx, readWriter, descKey, intentRes.Intent.Txn.WriteTimestamp, intentRes.Intent.Txn)
 	if err != nil {
 		return result.Result{}, errors.Wrap(err, "fetching local range descriptor as txn")
-	} else if val != nil {
+	} else if valRes.Value != nil {
 		return result.Result{}, errors.Errorf("non-deletion intent on local range descriptor")
 	}
 
diff --git a/pkg/kv/kvserver/batcheval/intent.go b/pkg/kv/kvserver/batcheval/intent.go
index 8e943f116029..19f170fb7dd5 100644
--- a/pkg/kv/kvserver/batcheval/intent.go
+++ b/pkg/kv/kvserver/batcheval/intent.go
@@ -75,17 +75,17 @@ func readProvisionalVal(
 	ctx context.Context, reader storage.Reader, usePrefixIter bool, intent *roachpb.Intent,
 ) (roachpb.KeyValue, error) {
 	if usePrefixIter {
-		val, _, err := storage.MVCCGetAsTxn(
+		valRes, err := storage.MVCCGetAsTxn(
 			ctx, reader, intent.Key, intent.Txn.WriteTimestamp, intent.Txn,
 		)
 		if err != nil {
 			return roachpb.KeyValue{}, err
 		}
-		if val == nil {
+		if valRes.Value == nil {
 			// Intent is a deletion.
 			return roachpb.KeyValue{}, nil
 		}
-		return roachpb.KeyValue{Key: intent.Key, Value: *val}, nil
+		return roachpb.KeyValue{Key: intent.Key, Value: *valRes.Value}, nil
 	}
 	res, err := storage.MVCCScanAsTxn(
 		ctx, reader, intent.Key, intent.Key.Next(), intent.Txn.WriteTimestamp, intent.Txn,
diff --git a/pkg/kv/kvserver/client_merge_test.go b/pkg/kv/kvserver/client_merge_test.go
index 311f452399ca..cb610ef12b8b 100644
--- a/pkg/kv/kvserver/client_merge_test.go
+++ b/pkg/kv/kvserver/client_merge_test.go
@@ -353,7 +353,7 @@ func mergeWithData(t *testing.T, retries int64) {
 
 	// Verify no intents remains on range descriptor keys.
 	for _, key := range []roachpb.Key{keys.RangeDescriptorKey(lhsDesc.StartKey), keys.RangeDescriptorKey(rhsDesc.StartKey)} {
-		if _, _, err := storage.MVCCGet(
+		if _, err := storage.MVCCGet(
 			ctx, store.Engine(), key, store.Clock().Now(), storage.MVCCGetOptions{},
 		); err != nil {
 			t.Fatal(err)
diff --git a/pkg/kv/kvserver/client_replica_test.go b/pkg/kv/kvserver/client_replica_test.go
index 3c633fab771d..d6000e805291 100644
--- a/pkg/kv/kvserver/client_replica_test.go
+++ b/pkg/kv/kvserver/client_replica_test.go
@@ -207,12 +207,12 @@ func TestLeaseholdersRejectClockUpdateWithJump(t *testing.T) {
 	if advance := ts3.GoTime().Sub(ts2.GoTime()); advance != 0 {
 		t.Fatalf("expected clock not to advance, but it advanced by %s", advance)
 	}
-	val, _, err := storage.MVCCGet(context.Background(), store.Engine(), key, ts3,
+	valRes, err := storage.MVCCGet(context.Background(), store.Engine(), key, ts3,
 		storage.MVCCGetOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if a, e := mustGetInt(val), incArgs.Increment*numCmds; a != e {
+	if a, e := mustGetInt(valRes.Value), incArgs.Increment*numCmds; a != e {
 		t.Errorf("expected %d, got %d", e, a)
 	}
 }
diff --git a/pkg/kv/kvserver/client_split_test.go b/pkg/kv/kvserver/client_split_test.go
index 6fb8ee84ba4b..6119ba97a40d 100644
--- a/pkg/kv/kvserver/client_split_test.go
+++ b/pkg/kv/kvserver/client_split_test.go
@@ -364,7 +364,7 @@ func TestStoreRangeSplitIntents(t *testing.T) {
 		t.Fatal(err)
 	}
 	for _, key := range []roachpb.Key{keys.RangeDescriptorKey(roachpb.RKeyMin), keys.RangeDescriptorKey(splitKeyAddr)} {
-		if _, _, err := storage.MVCCGet(
+		if _, err := storage.MVCCGet(
 			ctx, store.Engine(), key, store.Clock().Now(), storage.MVCCGetOptions{},
 		); err != nil {
 			t.Errorf("failed to read consistent range descriptor for key %s: %+v", key, err)
@@ -667,7 +667,7 @@ func TestStoreRangeSplitIdempotency(t *testing.T) {
 		t.Fatal(err)
 	}
 	for _, key := range []roachpb.Key{keys.RangeDescriptorKey(roachpb.RKeyMin), keys.RangeDescriptorKey(splitKeyAddr)} {
-		if _, _, err := storage.MVCCGet(
+		if _, err := storage.MVCCGet(
 			context.Background(), store.Engine(), key, store.Clock().Now(), storage.MVCCGetOptions{},
 		); err != nil {
 			t.Fatal(err)
diff --git a/pkg/kv/kvserver/loqrecovery/apply.go b/pkg/kv/kvserver/loqrecovery/apply.go
index 0dc1df5ed174..62c7ce546e7b 100644
--- a/pkg/kv/kvserver/loqrecovery/apply.go
+++ b/pkg/kv/kvserver/loqrecovery/apply.go
@@ -177,9 +177,9 @@ func applyReplicaUpdate(
 	// there will be keys not represented by any ranges or vice
 	// versa).
 	key := keys.RangeDescriptorKey(update.StartKey.AsRKey())
-	value, intent, err := storage.MVCCGet(
+	res, err := storage.MVCCGet(
 		ctx, readWriter, key, clock.Now(), storage.MVCCGetOptions{Inconsistent: true})
-	if value == nil {
+	if res.Value == nil {
 		return PrepareReplicaReport{}, errors.Errorf(
 			"failed to find a range descriptor for range %v", key)
 	}
@@ -187,7 +187,7 @@ func applyReplicaUpdate(
 		return PrepareReplicaReport{}, err
 	}
 	var localDesc roachpb.RangeDescriptor
-	if err := value.GetProto(&localDesc); err != nil {
+	if err := res.Value.GetProto(&localDesc); err != nil {
 		return PrepareReplicaReport{}, err
 	}
 	// Sanity check that this is indeed the right range.
@@ -213,7 +213,7 @@ func applyReplicaUpdate(
 	// we won't be able to do MVCCPut later during recovery for the new
 	// descriptor. It should have no effect on the recovery process itself as
 	// transaction would be rolled back anyways.
-	if intent != nil {
+	if res.Intent != nil {
 		// We rely on the property that transactions involving the range
 		// descriptor always start on the range-local descriptor's key. When there
 		// is an intent, this means that it is likely that the transaction did not
@@ -251,24 +251,24 @@ func applyReplicaUpdate(
 		// plan is trivially achievable, due to any of the above problems. But
 		// in the common case, we do expect one to exist.
 		report.AbortedTransaction = true
-		report.AbortedTransactionID = intent.Txn.ID
+		report.AbortedTransactionID = res.Intent.Txn.ID
 
 		// A crude form of the intent resolution process: abort the
 		// transaction by deleting its record.
-		txnKey := keys.TransactionKey(intent.Txn.Key, intent.Txn.ID)
+		txnKey := keys.TransactionKey(res.Intent.Txn.Key, res.Intent.Txn.ID)
 		if _, err := storage.MVCCDelete(ctx, readWriter, &ms, txnKey, hlc.Timestamp{}, hlc.ClockTimestamp{}, nil); err != nil {
 			return PrepareReplicaReport{}, err
 		}
 		update := roachpb.LockUpdate{
-			Span:   roachpb.Span{Key: intent.Key},
-			Txn:    intent.Txn,
+			Span:   roachpb.Span{Key: res.Intent.Key},
+			Txn:    res.Intent.Txn,
 			Status: roachpb.ABORTED,
 		}
 		if _, err := storage.MVCCResolveWriteIntent(ctx, readWriter, &ms, update); err != nil {
 			return PrepareReplicaReport{}, err
 		}
 		report.AbortedTransaction = true
-		report.AbortedTransactionID = intent.Txn.ID
+		report.AbortedTransactionID = res.Intent.Txn.ID
 	}
 	newDesc := localDesc
 	replicas := []roachpb.ReplicaDescriptor{
diff --git a/pkg/kv/kvserver/replica.go b/pkg/kv/kvserver/replica.go
index 1b8421a31468..f19f79353f70 100644
--- a/pkg/kv/kvserver/replica.go
+++ b/pkg/kv/kvserver/replica.go
@@ -1805,18 +1805,18 @@ func (r *Replica) maybeWatchForMergeLocked(ctx context.Context) (bool, error) {
 	// if one exists, regardless of what timestamp it is written at.
 	desc := r.descRLocked()
 	descKey := keys.RangeDescriptorKey(desc.StartKey)
-	_, intent, err := storage.MVCCGet(ctx, r.Engine(), descKey, hlc.MaxTimestamp,
+	intentRes, err := storage.MVCCGet(ctx, r.Engine(), descKey, hlc.MaxTimestamp,
 		storage.MVCCGetOptions{Inconsistent: true})
 	if err != nil {
 		return false, err
-	} else if intent == nil {
+	} else if intentRes.Intent == nil {
 		return false, nil
 	}
-	val, _, err := storage.MVCCGetAsTxn(
-		ctx, r.Engine(), descKey, intent.Txn.WriteTimestamp, intent.Txn)
+	valRes, err := storage.MVCCGetAsTxn(
+		ctx, r.Engine(), descKey, intentRes.Intent.Txn.WriteTimestamp, intentRes.Intent.Txn)
 	if err != nil {
 		return false, err
-	} else if val != nil {
+	} else if valRes.Value != nil {
 		return false, nil
 	}
 
@@ -1833,7 +1833,7 @@ func (r *Replica) maybeWatchForMergeLocked(ctx context.Context) (bool, error) {
 		return true, nil
 	}
 	r.mu.mergeComplete = mergeCompleteCh
-	r.mu.mergeTxnID = intent.Txn.ID
+	r.mu.mergeTxnID = intentRes.Intent.Txn.ID
 	// The RHS of a merge is not permitted to quiesce while a mergeComplete
 	// channel is installed. (If the RHS is quiescent when the merge commits, any
 	// orphaned followers would fail to queue themselves for GC.) Unquiesce the
@@ -1854,11 +1854,11 @@ func (r *Replica) maybeWatchForMergeLocked(ctx context.Context) (bool, error) {
 			b := &kv.Batch{}
 			b.Header.Timestamp = r.Clock().Now()
 			b.AddRawRequest(&roachpb.PushTxnRequest{
-				RequestHeader: roachpb.RequestHeader{Key: intent.Txn.Key},
+				RequestHeader: roachpb.RequestHeader{Key: intentRes.Intent.Txn.Key},
 				PusherTxn: roachpb.Transaction{
 					TxnMeta: enginepb.TxnMeta{Priority: enginepb.MinTxnPriority},
 				},
-				PusheeTxn: intent.Txn,
+				PusheeTxn: intentRes.Intent.Txn,
 				PushType:  roachpb.PUSH_ABORT,
 			})
 			if err := r.store.DB().Run(ctx, b); err != nil {
@@ -1882,7 +1882,7 @@ func (r *Replica) maybeWatchForMergeLocked(ctx context.Context) (bool, error) {
 		switch pushTxnRes.PusheeTxn.Status {
 		case roachpb.PENDING, roachpb.STAGING:
 			log.Fatalf(ctx, "PushTxn returned while merge transaction %s was still %s",
-				intent.Txn.ID.Short(), pushTxnRes.PusheeTxn.Status)
+				intentRes.Intent.Txn.ID.Short(), pushTxnRes.PusheeTxn.Status)
 		case roachpb.COMMITTED:
 			// If PushTxn claims that the transaction committed, then the transaction
 			// definitely committed.
diff --git a/pkg/kv/kvserver/replica_rangefeed.go b/pkg/kv/kvserver/replica_rangefeed.go
index 6612431fee31..4d01a0e263c1 100644
--- a/pkg/kv/kvserver/replica_rangefeed.go
+++ b/pkg/kv/kvserver/replica_rangefeed.go
@@ -535,14 +535,14 @@ func populatePrevValsInLogicalOpLog(
 
 		// Read the previous value from the prev Reader. Unlike the new value
 		// (see handleLogicalOpLogRaftMuLocked), this one may be missing.
-		prevVal, _, err := storage.MVCCGet(
+		prevValRes, err := storage.MVCCGet(
 			ctx, prevReader, key, ts, storage.MVCCGetOptions{Tombstones: true, Inconsistent: true},
 		)
 		if err != nil {
 			return errors.Wrapf(err, "consuming %T for key %v @ ts %v", op, key, ts)
 		}
-		if prevVal != nil {
-			*prevValPtr = prevVal.RawBytes
+		if prevValRes.Value != nil {
+			*prevValPtr = prevValRes.Value.RawBytes
 		} else {
 			*prevValPtr = nil
 		}
@@ -631,8 +631,8 @@ func (r *Replica) handleLogicalOpLogRaftMuLocked(
 		// Read the value directly from the Reader. This is performed in the
 		// same raftMu critical section that the logical op's corresponding
 		// WriteBatch is applied, so the value should exist.
-		val, _, vh, err := storage.MVCCGetWithValueHeader(ctx, reader, key, ts, storage.MVCCGetOptions{Tombstones: true})
-		if val == nil && err == nil {
+		valRes, vh, err := storage.MVCCGetWithValueHeader(ctx, reader, key, ts, storage.MVCCGetOptions{Tombstones: true})
+		if valRes.Value == nil && err == nil {
 			err = errors.New("value missing in reader")
 		}
 		if err != nil {
@@ -645,7 +645,7 @@ func (r *Replica) handleLogicalOpLogRaftMuLocked(
 		if vhf != nil {
 			vhf(key, nil, ts, vh)
 		}
-		*valPtr = val.RawBytes
+		*valPtr = valRes.Value.RawBytes
 	}
 
 	// Pass the ops to the rangefeed processor.
diff --git a/pkg/server/node_tombstone_storage.go b/pkg/server/node_tombstone_storage.go
index edc966ad2dee..312a97c44a40 100644
--- a/pkg/server/node_tombstone_storage.go
+++ b/pkg/server/node_tombstone_storage.go
@@ -59,16 +59,16 @@ func (s *nodeTombstoneStorage) IsDecommissioned(
 	// No cache hit.
 	k := s.key(nodeID)
 	for _, eng := range s.engs {
-		v, _, err := storage.MVCCGet(ctx, eng, k, hlc.Timestamp{}, storage.MVCCGetOptions{})
+		valRes, err := storage.MVCCGet(ctx, eng, k, hlc.Timestamp{}, storage.MVCCGetOptions{})
 		if err != nil {
 			return time.Time{}, err
 		}
-		if v == nil {
+		if valRes.Value == nil {
 			// Not found.
 			continue
 		}
 		var tsp hlc.Timestamp
-		if err := v.GetProto(&tsp); err != nil {
+		if err := valRes.Value.GetProto(&tsp); err != nil {
 			return time.Time{}, err
 		}
 		// Found, offer to cache and return.
diff --git a/pkg/storage/bench_test.go b/pkg/storage/bench_test.go
index 3c58af1bac28..5a10ff57debb 100644
--- a/pkg/storage/bench_test.go
+++ b/pkg/storage/bench_test.go
@@ -845,11 +845,11 @@ func runMVCCGet(ctx context.Context, b *testing.B, opts mvccBenchData, useBatch
 		key := roachpb.Key(encoding.EncodeUvarintAscending(keyBuf[:4], uint64(keyIdx)))
 		walltime := int64(5 * (rand.Int31n(int32(opts.numVersions)) + 1))
 		ts := hlc.Timestamp{WallTime: walltime}
-		if v, _, err := MVCCGet(ctx, r, key, ts, MVCCGetOptions{}); err != nil {
+		if valRes, err := MVCCGet(ctx, r, key, ts, MVCCGetOptions{}); err != nil {
 			b.Fatalf("failed get: %+v", err)
-		} else if v == nil {
+		} else if valRes.Value == nil {
 			b.Fatalf("failed get (key not found): %d@%d", keyIdx, walltime)
-		} else if valueBytes, err := v.GetBytes(); err != nil {
+		} else if valueBytes, err := valRes.Value.GetBytes(); err != nil {
 			b.Fatal(err)
 		} else if len(valueBytes) != opts.valueBytes {
 			b.Fatalf("unexpected value size: %d", len(valueBytes))
@@ -1144,7 +1144,7 @@ func runMVCCGetMergedValue(
 
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		_, _, err := MVCCGet(ctx, eng, keys[rand.Intn(numKeys)], timestamp, MVCCGetOptions{})
+		_, err := MVCCGet(ctx, eng, keys[rand.Intn(numKeys)], timestamp, MVCCGetOptions{})
 		if err != nil {
 			b.Fatal(err)
 		}
diff --git a/pkg/storage/engine_test.go b/pkg/storage/engine_test.go
index 3a5dc3d7f823..19451058c2eb 100644
--- a/pkg/storage/engine_test.go
+++ b/pkg/storage/engine_test.go
@@ -191,11 +191,11 @@ func TestEngineBatchStaleCachedIterator(t *testing.T) {
 		// invalidate the iterator's cache), and if it reports its cached
 		// result back, we'll see the (newly deleted) value (due to the
 		// failure mode above).
-		if v, _, err := MVCCGet(context.Background(), batch, key,
+		if valRes, err := MVCCGet(context.Background(), batch, key,
 			hlc.Timestamp{}, MVCCGetOptions{}); err != nil {
 			t.Fatal(err)
-		} else if v != nil {
-			t.Fatalf("expected no value, got %+v", v)
+		} else if valRes.Value != nil {
+			t.Fatalf("expected no value, got %+v", valRes.Value)
 		}
 	}
 }
diff --git a/pkg/storage/metamorphic/operations.go b/pkg/storage/metamorphic/operations.go
index 0105dd19ee2e..cf8986c9ae51 100644
--- a/pkg/storage/metamorphic/operations.go
+++ b/pkg/storage/metamorphic/operations.go
@@ -181,7 +181,7 @@ func (m mvccGetOp) run(ctx context.Context) string {
 	// TODO(itsbilal): Specify these bools as operands instead of having a
 	// separate operation for inconsistent cases. This increases visibility for
 	// anyone reading the output file.
-	val, intent, err := storage.MVCCGet(ctx, reader, m.key, m.ts, storage.MVCCGetOptions{
+	res, err := storage.MVCCGet(ctx, reader, m.key, m.ts, storage.MVCCGetOptions{
 		Inconsistent: m.inconsistent,
 		Tombstones:   true,
 		Txn:          txn,
@@ -189,7 +189,7 @@ func (m mvccGetOp) run(ctx context.Context) string {
 	if err != nil {
 		return fmt.Sprintf("error: %s", err)
 	}
-	return fmt.Sprintf("val = %v, intent = %v", val, intent)
+	return fmt.Sprintf("val = %v, intent = %v", res.Value, res.Intent)
 }
 
 type mvccPutOp struct {
diff --git a/pkg/storage/mvcc.go b/pkg/storage/mvcc.go
index 54a23edb9263..dd907f2f54f4 100644
--- a/pkg/storage/mvcc.go
+++ b/pkg/storage/mvcc.go
@@ -804,14 +804,14 @@ func MVCCGetProto(
 	opts MVCCGetOptions,
 ) (bool, error) {
 	// TODO(tschottdorf): Consider returning skipped intents to the caller.
-	value, _, mvccGetErr := MVCCGet(ctx, reader, key, timestamp, opts)
-	found := value != nil
+	valueRes, mvccGetErr := MVCCGet(ctx, reader, key, timestamp, opts)
+	found := valueRes.Value != nil
 	// If we found a result, parse it regardless of the error returned by MVCCGet.
 	if found && msg != nil {
 		// If the unmarshal failed, return its result. Otherwise, pass
 		// through the underlying error (which may be a WriteIntentError
 		// to be handled specially alongside the returned value).
-		if err := value.GetProto(msg); err != nil {
+		if err := valueRes.Value.GetProto(msg); err != nil {
 			return found, err
 		}
 	}
@@ -900,6 +900,39 @@ type MVCCGetOptions struct {
 	// or not. It is usually set by read-only requests that have resolved their
 	// conflicts before they begin their MVCC scan.
 	DontInterleaveIntents bool
+	// MaxKeys is the maximum number of kv pairs returned from this operation.
+	// The non-negative value represents an unbounded Get. The value -1 returns
+	// no keys in the result and a ResumeSpan equal to the request span is
+	// returned.
+	MaxKeys int64
+	// TargetBytes is a byte threshold to limit the amount of data pulled into
+	// memory during a Get operation. The zero value indicates no limit. The
+	// value -1 returns no keys in the result. A positive value represents an
+	// unbounded Get unless AllowEmpty is set. If an empty result is returned,
+	// then a ResumeSpan equal to the request span is returned.
+	TargetBytes int64
+	// AllowEmpty will return an empty result if the request key exceeds the
+	// TargetBytes limit.
+	AllowEmpty bool
+}
+
+// MVCCGetResult bundles return values for the MVCCGet family of functions.
+type MVCCGetResult struct {
+	// The most recent value for the specified key whose timestamp is less than
+	// or equal to the supplied timestamp. If no such value exists, nil is
+	// returned instead.
+	Value *roachpb.Value
+	// In inconsistent mode, the intent if an intent is encountered. In
+	// consistent mode, an intent will generate a WriteIntentError with the
+	// intent embedded within and the intent parameter will be nil.
+	Intent *roachpb.Intent
+	// See the documentation for roachpb.ResponseHeader for information on
+	// these parameters.
+	ResumeSpan      *roachpb.Span
+	ResumeReason    roachpb.ResumeReason
+	ResumeNextBytes int64
+	NumKeys         int64
+	NumBytes        int64
 }
 
 func (opts *MVCCGetOptions) validate() error {
@@ -952,9 +985,11 @@ func newMVCCIterator(
 	return reader.NewMVCCIterator(iterKind, opts)
 }
 
-// MVCCGet returns the most recent value for the specified key whose timestamp
-// is less than or equal to the supplied timestamp. If no such value exists, nil
-// is returned instead.
+// MVCCGet returns a MVCCGetResult.
+//
+// The first field of MVCCGetResult contains the most recent value for the
+// specified key whose timestamp is less than or equal to the supplied
+// timestamp. If no such value exists, nil is returned instead.
 //
 // In tombstones mode, if the most recent value is a deletion tombstone, the
 // result will be a non-nil roachpb.Value whose RawBytes field is nil.
@@ -967,9 +1002,9 @@ func newMVCCIterator(
 // above existing point keys.
 //
 // In inconsistent mode, if an intent is encountered, it will be placed in the
-// dedicated return parameter. By contrast, in consistent mode, an intent will
-// generate a WriteIntentError with the intent embedded within, and the intent
-// result parameter will be nil.
+// intent field. By contrast, in consistent mode, an intent will generate a
+// WriteIntentError with the intent embedded within, and the intent result
+// parameter will be nil.
 //
 // Note that transactional gets must be consistent. Put another way, only
 // non-transactional gets may be inconsistent.
@@ -991,16 +1026,30 @@ func newMVCCIterator(
 // the read timestamp.
 func MVCCGet(
 	ctx context.Context, reader Reader, key roachpb.Key, timestamp hlc.Timestamp, opts MVCCGetOptions,
-) (*roachpb.Value, *roachpb.Intent, error) {
-	value, intent, _, err := MVCCGetWithValueHeader(ctx, reader, key, timestamp, opts)
-	return value, intent, err
+) (MVCCGetResult, error) {
+	res, _, err := MVCCGetWithValueHeader(ctx, reader, key, timestamp, opts)
+	return res, err
 }
 
 // MVCCGetWithValueHeader is like MVCCGet, but in addition returns the
 // MVCCValueHeader for the value.
 func MVCCGetWithValueHeader(
 	ctx context.Context, reader Reader, key roachpb.Key, timestamp hlc.Timestamp, opts MVCCGetOptions,
-) (*roachpb.Value, *roachpb.Intent, enginepb.MVCCValueHeader, error) {
+) (MVCCGetResult, enginepb.MVCCValueHeader, error) {
+	var result MVCCGetResult
+	if opts.MaxKeys < 0 || opts.TargetBytes < 0 {
+		// Receipt of a GetRequest with negative MaxKeys or TargetBytes indicates
+		// that the request was part of a batch that has already exhausted its
+		// limit, which means that we should *not* serve the request and return a
+		// ResumeSpan for this GetRequest.
+		result.ResumeSpan = &roachpb.Span{Key: key}
+		if opts.MaxKeys < 0 {
+			result.ResumeReason = roachpb.RESUME_KEY_LIMIT
+		} else if opts.TargetBytes < 0 {
+			result.ResumeReason = roachpb.RESUME_BYTE_LIMIT
+		}
+		return result, enginepb.MVCCValueHeader{}, nil
+	}
 	iter := newMVCCIterator(
 		reader, timestamp, false /* rangeKeyMasking */, opts.DontInterleaveIntents, IterOptions{
 			KeyTypes: IterKeyTypePointsAndRanges,
@@ -1009,7 +1058,23 @@ func MVCCGetWithValueHeader(
 	)
 	defer iter.Close()
 	value, intent, vh, err := mvccGetWithValueHeader(ctx, iter, key, timestamp, opts)
-	return value.ToPointer(), intent, vh, err
+	val := value.ToPointer()
+	if err == nil && val != nil {
+		// NB: This calculation is different from Scan, since Scan responses include
+		// the key/value pair while Get only includes the value.
+		numBytes := int64(len(val.RawBytes))
+		if opts.TargetBytes > 0 && opts.AllowEmpty && numBytes > opts.TargetBytes {
+			result.ResumeSpan = &roachpb.Span{Key: key}
+			result.ResumeReason = roachpb.RESUME_BYTE_LIMIT
+			result.ResumeNextBytes = numBytes
+			return result, enginepb.MVCCValueHeader{}, nil
+		}
+		result.NumKeys = 1
+		result.NumBytes = numBytes
+	}
+	result.Value = val
+	result.Intent = intent
+	return result, vh, err
 }
 
 // gcassert:inline
@@ -1119,7 +1184,7 @@ func MVCCGetAsTxn(
 	key roachpb.Key,
 	timestamp hlc.Timestamp,
 	txnMeta enginepb.TxnMeta,
-) (*roachpb.Value, *roachpb.Intent, error) {
+) (MVCCGetResult, error) {
 	return MVCCGet(ctx, reader, key, timestamp, MVCCGetOptions{
 		Txn: &roachpb.Transaction{
 			TxnMeta:                txnMeta,
diff --git a/pkg/storage/mvcc_history_test.go b/pkg/storage/mvcc_history_test.go
index dc931c979183..7e175c1ff12b 100644
--- a/pkg/storage/mvcc_history_test.go
+++ b/pkg/storage/mvcc_history_test.go
@@ -95,7 +95,7 @@ var (
 // merge          [t=<name>] [ts=<int>[,<int>]] [resolve [status=<txnstatus>]] k=<key> v=<string> [raw]
 // put            [t=<name>] [ts=<int>[,<int>]] [localTs=<int>[,<int>]] [resolve [status=<txnstatus>]] k=<key> v=<string> [raw]
 // put_rangekey   ts=<int>[,<int>] [localTs=<int>[,<int>]] k=<key> end=<key>
-// get            [t=<name>] [ts=<int>[,<int>]]                         [resolve [status=<txnstatus>]] k=<key> [inconsistent] [skipLocked] [tombstones] [failOnMoreRecent] [localUncertaintyLimit=<int>[,<int>]] [globalUncertaintyLimit=<int>[,<int>]]
+// get            [t=<name>] [ts=<int>[,<int>]]                         [resolve [status=<txnstatus>]] k=<key> [inconsistent] [skipLocked] [tombstones] [failOnMoreRecent] [localUncertaintyLimit=<int>[,<int>]] [globalUncertaintyLimit=<int>[,<int>]] [maxKeys=<int>] [targetBytes=<int>] [allowEmpty]
 // scan           [t=<name>] [ts=<int>[,<int>]]                         [resolve [status=<txnstatus>]] k=<key> [end=<key>] [inconsistent] [skipLocked] [tombstones] [reverse] [failOnMoreRecent] [localUncertaintyLimit=<int>[,<int>]] [globalUncertaintyLimit=<int>[,<int>]] [max=<max>] [targetbytes=<target>] [wholeRows[=<int>]] [allowEmpty]
 // export         [k=<key>] [end=<key>] [ts=<int>[,<int>]] [kTs=<int>[,<int>]] [startTs=<int>[,<int>]] [maxIntents=<int>] [allRevisions] [targetSize=<int>] [maxSize=<int>] [stopMidKey] [fingerprint]
 //
@@ -1230,17 +1230,26 @@ func cmdGet(e *evalCtx) error {
 		}
 		opts.Uncertainty.GlobalLimit = txn.GlobalUncertaintyLimit
 	}
+	if e.hasArg("maxKeys") {
+		e.scanArg("maxKeys", &opts.MaxKeys)
+	}
+	if e.hasArg("targetBytes") {
+		e.scanArg("targetBytes", &opts.TargetBytes)
+	}
+	if e.hasArg("allowEmpty") {
+		opts.AllowEmpty = true
+	}
 
 	return e.withReader(func(r storage.Reader) error {
-		val, intent, err := storage.MVCCGet(e.ctx, r, key, ts, opts)
+		res, err := storage.MVCCGet(e.ctx, r, key, ts, opts)
 		// NB: the error is returned below. This ensures the test can
 		// ascertain no result is populated in the intent when an error
 		// occurs.
-		if intent != nil {
-			e.results.buf.Printf("get: %v -> intent {%s}\n", key, intent.Txn)
+		if res.Intent != nil {
+			e.results.buf.Printf("get: %v -> intent {%s}\n", key, res.Intent.Txn)
 		}
-		if val != nil {
-			e.results.buf.Printf("get: %v -> %v @%v\n", key, val.PrettyPrint(), val.Timestamp)
+		if res.Value != nil {
+			e.results.buf.Printf("get: %v -> %v @%v\n", key, res.Value.PrettyPrint(), res.Value.Timestamp)
 		} else {
 			e.results.buf.Printf("get: %v -> <no data>\n", key)
 		}
diff --git a/pkg/storage/mvcc_test.go b/pkg/storage/mvcc_test.go
index c1bf794c0716..9603a1a80aba 100644
--- a/pkg/storage/mvcc_test.go
+++ b/pkg/storage/mvcc_test.go
@@ -239,12 +239,12 @@ func TestMVCCGetNotExist(t *testing.T) {
 	engine := NewDefaultInMemForTesting()
 	defer engine.Close()
 
-	value, _, err := MVCCGet(context.Background(), engine, testKey1, hlc.Timestamp{Logical: 1},
+	valueRes, err := MVCCGet(context.Background(), engine, testKey1, hlc.Timestamp{Logical: 1},
 		MVCCGetOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if value != nil {
+	if valueRes.Value != nil {
 		t.Fatal("the value should be empty")
 	}
 }
@@ -273,11 +273,11 @@ func TestMVCCGetNoMoreOldVersion(t *testing.T) {
 	if err := MVCCPut(ctx, engine, nil, testKey2, hlc.Timestamp{WallTime: 1}, hlc.ClockTimestamp{}, value2, nil); err != nil {
 		t.Fatal(err)
 	}
-	value, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 2}, MVCCGetOptions{})
+	valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 2}, MVCCGetOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if value != nil {
+	if valueRes.Value != nil {
 		t.Fatal("the value should be empty")
 	}
 }
@@ -293,11 +293,11 @@ func TestMVCCGetWithValueHeader(t *testing.T) {
 	if err := MVCCPut(ctx, engine, nil, testKey1, hlc.Timestamp{WallTime: 1, Logical: 1}, hlc.ClockTimestamp{WallTime: 1}, value1, nil); err != nil {
 		t.Fatal(err)
 	}
-	value, _, vh, err := MVCCGetWithValueHeader(ctx, engine, testKey1, hlc.Timestamp{WallTime: 2}, MVCCGetOptions{})
+	valueRes, vh, err := MVCCGetWithValueHeader(ctx, engine, testKey1, hlc.Timestamp{WallTime: 2}, MVCCGetOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if value == nil {
+	if valueRes.Value == nil {
 		t.Fatal("the value should not be empty")
 	}
 	require.Equal(t, hlc.ClockTimestamp{WallTime: 1}, vh.LocalTimestamp)
@@ -308,33 +308,33 @@ func TestMVCCGetWithValueHeader(t *testing.T) {
 	}
 
 	// Read the latest version which should be deleted.
-	value, _, vh, err = MVCCGetWithValueHeader(ctx, engine, testKey1, hlc.Timestamp{WallTime: 4}, MVCCGetOptions{})
+	valueRes, vh, err = MVCCGetWithValueHeader(ctx, engine, testKey1, hlc.Timestamp{WallTime: 4}, MVCCGetOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if value != nil {
+	if valueRes.Value != nil {
 		t.Fatal("the value should be empty")
 	}
 	require.Zero(t, vh.LocalTimestamp)
 	// Read the latest version with tombstone.
-	value, _, vh, err = MVCCGetWithValueHeader(ctx, engine, testKey1, hlc.Timestamp{WallTime: 4},
+	valueRes, vh, err = MVCCGetWithValueHeader(ctx, engine, testKey1, hlc.Timestamp{WallTime: 4},
 		MVCCGetOptions{Tombstones: true})
 	if err != nil {
 		t.Fatal(err)
-	} else if value == nil || len(value.RawBytes) != 0 {
-		t.Fatalf("the value should be non-nil with empty RawBytes; got %+v", value)
+	} else if valueRes.Value == nil || len(valueRes.Value.RawBytes) != 0 {
+		t.Fatalf("the value should be non-nil with empty RawBytes; got %+v", valueRes.Value)
 	}
 
 	require.Equal(t, hlc.ClockTimestamp{WallTime: 2, Logical: 1}, vh.LocalTimestamp)
 
 	// Read the old version which should still exist.
 	for _, logical := range []int32{0, math.MaxInt32} {
-		value, _, vh, err := MVCCGetWithValueHeader(ctx, engine, testKey1, hlc.Timestamp{WallTime: 2, Logical: logical},
+		valueRes, vh, err := MVCCGetWithValueHeader(ctx, engine, testKey1, hlc.Timestamp{WallTime: 2, Logical: logical},
 			MVCCGetOptions{})
 		if err != nil {
 			t.Fatal(err)
 		}
-		if value == nil {
+		if valueRes.Value == nil {
 			t.Fatal("the value should not be empty")
 		}
 		require.Equal(t, hlc.ClockTimestamp{WallTime: 1}, vh.LocalTimestamp)
@@ -361,31 +361,31 @@ func TestMVCCWriteWithOlderTimestampAfterDeletionOfNonexistentKey(t *testing.T)
 		t.Fatal(err)
 	}
 
-	value, _, err := MVCCGet(context.Background(), engine, testKey1, hlc.Timestamp{WallTime: 2},
+	valueRes, err := MVCCGet(context.Background(), engine, testKey1, hlc.Timestamp{WallTime: 2},
 		MVCCGetOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
 	// The attempted write at ts(1,0) was performed at ts(3,1), so we should
 	// not see it at ts(2,0).
-	if value != nil {
-		t.Fatalf("value present at TS = %s", value.Timestamp)
+	if valueRes.Value != nil {
+		t.Fatalf("value present at TS = %s", valueRes.Value.Timestamp)
 	}
 
 	// Read the latest version which will be the value written with the timestamp pushed.
-	value, _, err = MVCCGet(context.Background(), engine, testKey1, hlc.Timestamp{WallTime: 4},
+	valueRes, err = MVCCGet(context.Background(), engine, testKey1, hlc.Timestamp{WallTime: 4},
 		MVCCGetOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if value == nil {
+	if valueRes.Value == nil {
 		t.Fatal("value doesn't exist")
 	}
-	if !bytes.Equal(value.RawBytes, value1.RawBytes) {
-		t.Errorf("expected %q; got %q", value1.RawBytes, value.RawBytes)
+	if !bytes.Equal(valueRes.Value.RawBytes, value1.RawBytes) {
+		t.Errorf("expected %q; got %q", value1.RawBytes, valueRes.Value.RawBytes)
 	}
-	if expTS := (hlc.Timestamp{WallTime: 3, Logical: 1}); value.Timestamp != expTS {
-		t.Fatalf("timestamp was not pushed: %s, expected %s", value.Timestamp, expTS)
+	if expTS := (hlc.Timestamp{WallTime: 3, Logical: 1}); valueRes.Value.Timestamp != expTS {
+		t.Fatalf("timestamp was not pushed: %s, expected %s", valueRes.Value.Timestamp, expTS)
 	}
 }
 
@@ -403,17 +403,17 @@ func TestMVCCInlineWithTxn(t *testing.T) {
 	}
 
 	// Now verify inline get.
-	value, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{}, MVCCGetOptions{})
+	valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{}, MVCCGetOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if !reflect.DeepEqual(value1, *value) {
-		t.Errorf("the inline value should be %v; got %v", value1, *value)
+	if !reflect.DeepEqual(value1, *valueRes.Value) {
+		t.Errorf("the inline value should be %v; got %v", value1, *valueRes.Value)
 	}
 
 	// Verify inline get with txn does still work (this will happen on a
 	// scan if the distributed sender is forced to wrap it in a txn).
-	if _, _, err = MVCCGet(ctx, engine, testKey1, hlc.Timestamp{}, MVCCGetOptions{
+	if _, err = MVCCGet(ctx, engine, testKey1, hlc.Timestamp{}, MVCCGetOptions{
 		Txn: txn1,
 	}); err != nil {
 		t.Error(err)
@@ -457,11 +457,11 @@ func TestMVCCGetAndDeleteInTxn(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if value, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 2}, MVCCGetOptions{
+	if valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 2}, MVCCGetOptions{
 		Txn: txn,
 	}); err != nil {
 		t.Fatal(err)
-	} else if value == nil {
+	} else if valueRes.Value == nil {
 		t.Fatal("the value should not be empty")
 	}
 
@@ -472,29 +472,29 @@ func TestMVCCGetAndDeleteInTxn(t *testing.T) {
 	}
 
 	// Read the latest version which should be deleted.
-	if value, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 4}, MVCCGetOptions{
+	if valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 4}, MVCCGetOptions{
 		Txn: txn,
 	}); err != nil {
 		t.Fatal(err)
-	} else if value != nil {
+	} else if valueRes.Value != nil {
 		t.Fatal("the value should be empty")
 	}
 	// Read the latest version with tombstone.
-	if value, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 4}, MVCCGetOptions{
+	if valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 4}, MVCCGetOptions{
 		Tombstones: true,
 		Txn:        txn,
 	}); err != nil {
 		t.Fatal(err)
-	} else if value == nil || len(value.RawBytes) != 0 {
-		t.Fatalf("the value should be non-nil with empty RawBytes; got %+v", value)
+	} else if valueRes.Value == nil || len(valueRes.Value.RawBytes) != 0 {
+		t.Fatalf("the value should be non-nil with empty RawBytes; got %+v", valueRes.Value)
 	}
 
 	// Read the old version which shouldn't exist, as within a
 	// transaction, we delete previous values.
-	if value, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 2}, MVCCGetOptions{}); err != nil {
+	if valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 2}, MVCCGetOptions{}); err != nil {
 		t.Fatal(err)
-	} else if value != nil {
-		t.Fatalf("expected value nil, got: %s", value)
+	} else if valueRes.Value != nil {
+		t.Fatalf("expected value nil, got: %s", valueRes.Value)
 	}
 }
 
@@ -510,11 +510,11 @@ func TestMVCCGetWriteIntentError(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if _, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{}); err == nil {
+	if _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{}); err == nil {
 		t.Fatal("cannot read the value of a write intent without TxnID")
 	}
 
-	if _, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{
+	if _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{
 		Txn: txn2,
 	}); err == nil {
 		t.Fatal("cannot read the value of a write intent from a different TxnID")
@@ -671,7 +671,7 @@ func TestMVCCGetInconsistent(t *testing.T) {
 	}
 
 	// A get with consistent=false should fail in a txn.
-	if _, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{
+	if _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{
 		Inconsistent: true,
 		Txn:          txn1,
 	}); err == nil {
@@ -680,18 +680,18 @@ func TestMVCCGetInconsistent(t *testing.T) {
 
 	// Inconsistent get will fetch value1 for any timestamp.
 	for _, ts := range []hlc.Timestamp{{WallTime: 1}, {WallTime: 2}} {
-		val, intent, err := MVCCGet(ctx, engine, testKey1, ts, MVCCGetOptions{Inconsistent: true})
+		res, err := MVCCGet(ctx, engine, testKey1, ts, MVCCGetOptions{Inconsistent: true})
 		if ts.Less(hlc.Timestamp{WallTime: 2}) {
 			if err != nil {
 				t.Fatal(err)
 			}
 		} else {
-			if intent == nil || !intent.Key.Equal(testKey1) {
-				t.Fatalf("expected %v, but got %v", testKey1, intent)
+			if res.Intent == nil || !res.Intent.Key.Equal(testKey1) {
+				t.Fatalf("expected %v, but got %v", testKey1, res.Intent)
 			}
 		}
-		if !bytes.Equal(val.RawBytes, value1.RawBytes) {
-			t.Errorf("@%s expected %q; got %q", ts, value1.RawBytes, val.RawBytes)
+		if !bytes.Equal(res.Value.RawBytes, value1.RawBytes) {
+			t.Errorf("@%s expected %q; got %q", ts, value1.RawBytes, res.Value.RawBytes)
 		}
 	}
 
@@ -699,13 +699,13 @@ func TestMVCCGetInconsistent(t *testing.T) {
 	if err := MVCCPut(ctx, engine, nil, testKey2, txn2.ReadTimestamp, hlc.ClockTimestamp{}, value1, txn2); err != nil {
 		t.Fatal(err)
 	}
-	val, intent, err := MVCCGet(ctx, engine, testKey2, hlc.Timestamp{WallTime: 2},
+	res, err := MVCCGet(ctx, engine, testKey2, hlc.Timestamp{WallTime: 2},
 		MVCCGetOptions{Inconsistent: true})
-	if intent == nil || !intent.Key.Equal(testKey2) {
+	if res.Intent == nil || !res.Intent.Key.Equal(testKey2) {
 		t.Fatal(err)
 	}
-	if val != nil {
-		t.Errorf("expected empty val; got %+v", val)
+	if res.Value != nil {
+		t.Errorf("expected empty val; got %+v", res.Value)
 	}
 }
 
@@ -863,7 +863,7 @@ func TestMVCCInvalidateIterator(t *testing.T) {
 			var err error
 			switch which {
 			case "get":
-				_, _, err = MVCCGet(ctx, batch, key, ts2, MVCCGetOptions{})
+				_, err = MVCCGet(ctx, batch, key, ts2, MVCCGetOptions{})
 			case "scan":
 				_, err = MVCCScan(ctx, batch, key, roachpb.KeyMax, ts2, MVCCScanOptions{})
 			case "findSplitKey":
@@ -967,7 +967,7 @@ func mvccScanTest(ctx context.Context, t *testing.T, engine Engine) {
 		t.Fatalf("resumeSpan = %+v", resumeSpan)
 	}
 
-	if _, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{
+	if _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{
 		Txn: txn2,
 	}); err != nil {
 		t.Fatal(err)
@@ -2258,25 +2258,25 @@ func TestMVCCInitPut(t *testing.T) {
 		// will be present.
 		{ts: hlc.Timestamp{WallTime: 1}, expTS: hlc.Timestamp{Logical: 5}},
 	} {
-		value, _, err := MVCCGet(ctx, engine, testKey1, check.ts, MVCCGetOptions{})
+		valueRes, err := MVCCGet(ctx, engine, testKey1, check.ts, MVCCGetOptions{})
 		if err != nil {
 			t.Fatal(err)
 		}
-		if !bytes.Equal(value1.RawBytes, value.RawBytes) {
+		if !bytes.Equal(value1.RawBytes, valueRes.Value.RawBytes) {
 			t.Fatalf("the value %s in get result does not match the value %s in request",
-				value1.RawBytes, value.RawBytes)
+				value1.RawBytes, valueRes.Value.RawBytes)
 		}
-		if value.Timestamp != check.expTS {
-			t.Errorf("value at timestamp %s seen, expected %s", value.Timestamp, check.expTS)
+		if valueRes.Value.Timestamp != check.expTS {
+			t.Errorf("value at timestamp %s seen, expected %s", valueRes.Value.Timestamp, check.expTS)
 		}
 	}
 
-	value, _, pErr := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 0}, MVCCGetOptions{})
+	valueRes, pErr := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 0}, MVCCGetOptions{})
 	if pErr != nil {
 		t.Fatal(pErr)
 	}
-	if value != nil {
-		t.Fatalf("%v present at old timestamp", value)
+	if valueRes.Value != nil {
+		t.Fatalf("%v present at old timestamp", valueRes.Value)
 	}
 }
 
@@ -2604,15 +2604,15 @@ func TestMVCCResolveTxn(t *testing.T) {
 	}
 
 	{
-		value, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 1}, MVCCGetOptions{
+		valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 1}, MVCCGetOptions{
 			Txn: txn1,
 		})
 		if err != nil {
 			t.Fatal(err)
 		}
-		if !bytes.Equal(value1.RawBytes, value.RawBytes) {
+		if !bytes.Equal(value1.RawBytes, valueRes.Value.RawBytes) {
 			t.Fatalf("the value %s in get result does not match the value %s in request",
-				value1.RawBytes, value.RawBytes)
+				value1.RawBytes, valueRes.Value.RawBytes)
 		}
 	}
 
@@ -2623,13 +2623,13 @@ func TestMVCCResolveTxn(t *testing.T) {
 	}
 
 	{
-		value, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 1}, MVCCGetOptions{})
+		valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 1}, MVCCGetOptions{})
 		if err != nil {
 			t.Fatal(err)
 		}
-		if !bytes.Equal(value1.RawBytes, value.RawBytes) {
+		if !bytes.Equal(value1.RawBytes, valueRes.Value.RawBytes) {
 			t.Fatalf("the value %s in get result does not match the value %s in request",
-				value1.RawBytes, value.RawBytes)
+				value1.RawBytes, valueRes.Value.RawBytes)
 		}
 	}
 }
@@ -2661,12 +2661,12 @@ func TestMVCCResolveNewerIntent(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	value, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 2}, MVCCGetOptions{})
+	valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 2}, MVCCGetOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if !bytes.Equal(value1.RawBytes, value.RawBytes) {
-		t.Fatalf("expected value1 bytes; got %q", value.RawBytes)
+	if !bytes.Equal(value1.RawBytes, valueRes.Value.RawBytes) {
+		t.Fatalf("expected value1 bytes; got %q", valueRes.Value.RawBytes)
 	}
 }
 
@@ -2710,7 +2710,7 @@ func TestMVCCResolveIntentTxnTimestampMismatch(t *testing.T) {
 		{intent.Txn.WriteTimestamp, true},
 		{hlc.MaxTimestamp, true},
 	} {
-		_, _, err := MVCCGet(ctx, engine, testKey1, test.Timestamp, MVCCGetOptions{})
+		_, err := MVCCGet(ctx, engine, testKey1, test.Timestamp, MVCCGetOptions{})
 		if errors.HasType(err, (*roachpb.WriteIntentError)(nil)) != test.found {
 			t.Fatalf("%d: expected write intent error: %t, got %v", i, test.found, err)
 		}
@@ -2759,10 +2759,10 @@ func TestMVCCConditionalPutOldTimestamp(t *testing.T) {
 	}
 	// Verify new value was actually written at (3, 1).
 	ts := hlc.Timestamp{WallTime: 3, Logical: 1}
-	value, _, err := MVCCGet(ctx, engine, testKey1, ts, MVCCGetOptions{})
-	if err != nil || value.Timestamp != ts || !bytes.Equal(value3.RawBytes, value.RawBytes) {
+	valueRes, err := MVCCGet(ctx, engine, testKey1, ts, MVCCGetOptions{})
+	if err != nil || valueRes.Value.Timestamp != ts || !bytes.Equal(value3.RawBytes, valueRes.Value.RawBytes) {
 		t.Fatalf("expected err=nil (got %s), timestamp=%s (got %s), value=%q (got %q)",
-			err, value.Timestamp, ts, value3.RawBytes, value.RawBytes)
+			err, valueRes.Value.Timestamp, ts, value3.RawBytes, valueRes.Value.RawBytes)
 	}
 }
 
@@ -2793,14 +2793,14 @@ func TestMVCCMultiplePutOldTimestamp(t *testing.T) {
 		t.Errorf("expected WriteTooOldError on Put; got %v", err)
 	}
 	// Verify new value was actually written at (3, 1).
-	value, _, err := MVCCGet(ctx, engine, testKey1, hlc.MaxTimestamp, MVCCGetOptions{Txn: txn})
+	valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.MaxTimestamp, MVCCGetOptions{Txn: txn})
 	if err != nil {
 		t.Fatal(err)
 	}
 	expTS := hlc.Timestamp{WallTime: 3, Logical: 1}
-	if value.Timestamp != expTS || !bytes.Equal(value2.RawBytes, value.RawBytes) {
+	if valueRes.Value.Timestamp != expTS || !bytes.Equal(value2.RawBytes, valueRes.Value.RawBytes) {
 		t.Fatalf("expected timestamp=%s (got %s), value=%q (got %q)",
-			value.Timestamp, expTS, value2.RawBytes, value.RawBytes)
+			valueRes.Value.Timestamp, expTS, value2.RawBytes, valueRes.Value.RawBytes)
 	}
 
 	// Put again and verify no WriteTooOldError, but timestamp should continue
@@ -2811,13 +2811,13 @@ func TestMVCCMultiplePutOldTimestamp(t *testing.T) {
 		t.Error(err)
 	}
 	// Verify new value was actually written at (3, 1).
-	value, _, err = MVCCGet(ctx, engine, testKey1, hlc.MaxTimestamp, MVCCGetOptions{Txn: txn})
+	valueRes, err = MVCCGet(ctx, engine, testKey1, hlc.MaxTimestamp, MVCCGetOptions{Txn: txn})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if value.Timestamp != expTS || !bytes.Equal(value3.RawBytes, value.RawBytes) {
+	if valueRes.Value.Timestamp != expTS || !bytes.Equal(value3.RawBytes, valueRes.Value.RawBytes) {
 		t.Fatalf("expected timestamp=%s (got %s), value=%q (got %q)",
-			value.Timestamp, expTS, value3.RawBytes, value.RawBytes)
+			valueRes.Value.Timestamp, expTS, value3.RawBytes, valueRes.Value.RawBytes)
 	}
 }
 
@@ -2871,13 +2871,13 @@ func TestMVCCPutOldOrigTimestampNewCommitTimestamp(t *testing.T) {
 
 	// Verify new value was actually written at the transaction's provisional
 	// commit timestamp.
-	value, _, err := MVCCGet(ctx, engine, testKey1, hlc.MaxTimestamp, MVCCGetOptions{Txn: txn})
+	valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.MaxTimestamp, MVCCGetOptions{Txn: txn})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if value.Timestamp != expTS || !bytes.Equal(value2.RawBytes, value.RawBytes) {
+	if valueRes.Value.Timestamp != expTS || !bytes.Equal(value2.RawBytes, valueRes.Value.RawBytes) {
 		t.Fatalf("expected timestamp=%s (got %s), value=%q (got %q)",
-			value.Timestamp, expTS, value2.RawBytes, value.RawBytes)
+			valueRes.Value.Timestamp, expTS, value2.RawBytes, valueRes.Value.RawBytes)
 	}
 }
 
@@ -2902,12 +2902,12 @@ func TestMVCCAbortTxn(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if value, _, err := MVCCGet(
+	if valueRes, err := MVCCGet(
 		ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{},
 	); err != nil {
 		t.Fatal(err)
-	} else if value != nil {
-		t.Fatalf("expected the value to be empty: %s", value)
+	} else if valueRes.Value != nil {
+		t.Fatalf("expected the value to be empty: %s", valueRes.Value)
 	}
 	require.Empty(t, mvccGetRaw(t, engine, mvccKey(testKey1)))
 }
@@ -2940,23 +2940,23 @@ func TestMVCCAbortTxnWithPreviousVersion(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if _, intent, err := MVCCGet(ctx, engine, testKey1, hlc.MaxTimestamp, MVCCGetOptions{
+	if intentRes, err := MVCCGet(ctx, engine, testKey1, hlc.MaxTimestamp, MVCCGetOptions{
 		Inconsistent: true,
 	}); err != nil {
 		t.Fatal(err)
-	} else if intent != nil {
-		t.Fatalf("expected no intent, got: %s", intent)
+	} else if intentRes.Intent != nil {
+		t.Fatalf("expected no intent, got: %s", intentRes.Intent)
 	}
 
-	if value, _, err := MVCCGet(
+	if valueRes, err := MVCCGet(
 		ctx, engine, testKey1, hlc.Timestamp{WallTime: 3}, MVCCGetOptions{},
 	); err != nil {
 		t.Fatal(err)
-	} else if expTS := (hlc.Timestamp{WallTime: 1}); value.Timestamp != expTS {
-		t.Fatalf("expected timestamp %+v == %+v", value.Timestamp, expTS)
-	} else if !bytes.Equal(value2.RawBytes, value.RawBytes) {
+	} else if expTS := (hlc.Timestamp{WallTime: 1}); valueRes.Value.Timestamp != expTS {
+		t.Fatalf("expected timestamp %+v == %+v", valueRes.Value.Timestamp, expTS)
+	} else if !bytes.Equal(value2.RawBytes, valueRes.Value.RawBytes) {
 		t.Fatalf("the value %q in get result does not match the value %q in request",
-			value.RawBytes, value2.RawBytes)
+			valueRes.Value.RawBytes, value2.RawBytes)
 	}
 }
 
@@ -3016,10 +3016,10 @@ func TestMVCCWriteWithDiffTimestampsAndEpochs(t *testing.T) {
 		t.Fatalf("expected write too old error with actual ts %s; got %s", expTS, wtoErr.ActualTimestamp)
 	}
 	// Verify value was actually written at (1, 1).
-	value, _, err := MVCCGet(ctx, engine, testKey1, expTS, MVCCGetOptions{})
-	if err != nil || value.Timestamp != expTS || !bytes.Equal(value4.RawBytes, value.RawBytes) {
+	valueRes, err := MVCCGet(ctx, engine, testKey1, expTS, MVCCGetOptions{})
+	if err != nil || valueRes.Value.Timestamp != expTS || !bytes.Equal(value4.RawBytes, valueRes.Value.RawBytes) {
 		t.Fatalf("expected err=nil (got %s), timestamp=%s (got %s), value=%q (got %q)",
-			err, value.Timestamp, expTS, value4.RawBytes, value.RawBytes)
+			err, valueRes.Value.Timestamp, expTS, value4.RawBytes, valueRes.Value.RawBytes)
 	}
 	// Now write an intent with exactly the same timestamp--ties also get WriteTooOldError.
 	err = MVCCPut(ctx, engine, nil, testKey1, txn2.ReadTimestamp, hlc.ClockTimestamp{}, value5, txn2)
@@ -3030,27 +3030,27 @@ func TestMVCCWriteWithDiffTimestampsAndEpochs(t *testing.T) {
 		t.Fatalf("expected write too old error with actual ts %s; got %s", intentTS, wtoErr.ActualTimestamp)
 	}
 	// Verify intent value was actually written at (1, 2).
-	value, _, err = MVCCGet(ctx, engine, testKey1, intentTS, MVCCGetOptions{Txn: txn2})
-	if err != nil || value.Timestamp != intentTS || !bytes.Equal(value5.RawBytes, value.RawBytes) {
+	valueRes, err = MVCCGet(ctx, engine, testKey1, intentTS, MVCCGetOptions{Txn: txn2})
+	if err != nil || valueRes.Value.Timestamp != intentTS || !bytes.Equal(value5.RawBytes, valueRes.Value.RawBytes) {
 		t.Fatalf("expected err=nil (got %s), timestamp=%s (got %s), value=%q (got %q)",
-			err, value.Timestamp, intentTS, value5.RawBytes, value.RawBytes)
+			err, valueRes.Value.Timestamp, intentTS, value5.RawBytes, valueRes.Value.RawBytes)
 	}
 	// Attempt to read older timestamp; should fail.
-	value, _, err = MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 0}, MVCCGetOptions{})
-	if value != nil || err != nil {
-		t.Fatalf("expected value nil, err nil; got %+v, %v", value, err)
+	valueRes, err = MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 0}, MVCCGetOptions{})
+	if valueRes.Value != nil || err != nil {
+		t.Fatalf("expected value nil, err nil; got %+v, %v", valueRes.Value, err)
 	}
 	// Read at correct timestamp.
-	value, _, err = MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{})
+	valueRes, err = MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if expTS := (hlc.Timestamp{WallTime: 1}); value.Timestamp != expTS {
-		t.Fatalf("expected timestamp %+v == %+v", value.Timestamp, expTS)
+	if expTS := (hlc.Timestamp{WallTime: 1}); valueRes.Value.Timestamp != expTS {
+		t.Fatalf("expected timestamp %+v == %+v", valueRes.Value.Timestamp, expTS)
 	}
-	if !bytes.Equal(value3.RawBytes, value.RawBytes) {
+	if !bytes.Equal(value3.RawBytes, valueRes.Value.RawBytes) {
 		t.Fatalf("the value %s in get result does not match the value %s in request",
-			value3.RawBytes, value.RawBytes)
+			value3.RawBytes, valueRes.Value.RawBytes)
 	}
 }
 
@@ -3091,7 +3091,7 @@ func TestMVCCGetWithDiffEpochs(t *testing.T) {
 	}
 	for i, test := range testCases {
 		t.Run(strconv.Itoa(i), func(t *testing.T) {
-			value, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 2}, MVCCGetOptions{
+			valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 2}, MVCCGetOptions{
 				Txn: test.txn,
 			})
 			if test.expErr {
@@ -3100,8 +3100,8 @@ func TestMVCCGetWithDiffEpochs(t *testing.T) {
 				} else if !errors.HasType(err, (*roachpb.WriteIntentError)(nil)) {
 					t.Errorf("test %d: expected write intent error; got %v", i, err)
 				}
-			} else if err != nil || value == nil || !bytes.Equal(test.expValue.RawBytes, value.RawBytes) {
-				t.Errorf("test %d: expected value %q, err nil; got %+v, %v", i, test.expValue.RawBytes, value, err)
+			} else if err != nil || valueRes.Value == nil || !bytes.Equal(test.expValue.RawBytes, valueRes.Value.RawBytes) {
+				t.Errorf("test %d: expected value %q, err nil; got %+v, %v", i, test.expValue.RawBytes, valueRes.Value, err)
 			}
 		})
 	}
@@ -3170,9 +3170,9 @@ func TestMVCCGetWithDiffEpochsAndTimestamps(t *testing.T) {
 	}
 	for i, test := range testCases {
 		t.Run(strconv.Itoa(i), func(t *testing.T) {
-			value, _, err := MVCCGet(ctx, engine, testKey1, test.readTS, MVCCGetOptions{Txn: test.txn})
-			if err != nil || value == nil || !bytes.Equal(test.expValue.RawBytes, value.RawBytes) {
-				t.Errorf("test %d: expected value %q, err nil; got %+v, %v", i, test.expValue.RawBytes, value, err)
+			valueRes, err := MVCCGet(ctx, engine, testKey1, test.readTS, MVCCGetOptions{Txn: test.txn})
+			if err != nil || valueRes.Value == nil || !bytes.Equal(test.expValue.RawBytes, valueRes.Value.RawBytes) {
+				t.Errorf("test %d: expected value %q, err nil; got %+v, %v", i, test.expValue.RawBytes, valueRes.Value, err)
 			}
 		})
 	}
@@ -3191,7 +3191,7 @@ func TestMVCCGetWithOldEpoch(t *testing.T) {
 	if err := MVCCPut(ctx, engine, nil, testKey1, txn1e2.ReadTimestamp, hlc.ClockTimestamp{}, value2, txn1e2); err != nil {
 		t.Fatal(err)
 	}
-	_, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 2}, MVCCGetOptions{
+	_, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 2}, MVCCGetOptions{
 		Txn: txn1,
 	})
 	if err == nil {
@@ -3294,11 +3294,11 @@ func TestMVCCGetWithPushedTimestamp(t *testing.T) {
 		t.Fatal(err)
 	}
 	// Attempt to read using naive txn's previous timestamp.
-	value, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 1}, MVCCGetOptions{
+	valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 1}, MVCCGetOptions{
 		Txn: txn1,
 	})
-	if err != nil || value == nil || !bytes.Equal(value.RawBytes, value1.RawBytes) {
-		t.Errorf("expected value %q, err nil; got %+v, %v", value1.RawBytes, value, err)
+	if err != nil || valueRes.Value == nil || !bytes.Equal(valueRes.Value.RawBytes, value1.RawBytes) {
+		t.Errorf("expected value %q, err nil; got %+v, %v", value1.RawBytes, valueRes.Value, err)
 	}
 }
 
@@ -3328,19 +3328,19 @@ func TestMVCCResolveWithDiffEpochs(t *testing.T) {
 
 	// Verify key1 is empty, as resolution with epoch 2 would have
 	// aborted the epoch 1 intent.
-	value, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 1}, MVCCGetOptions{})
-	if value != nil || err != nil {
-		t.Errorf("expected value nil, err nil; got %+v, %v", value, err)
+	valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 1}, MVCCGetOptions{})
+	if valueRes.Value != nil || err != nil {
+		t.Errorf("expected value nil, err nil; got %+v, %v", valueRes.Value, err)
 	}
 
 	// Key2 should be committed.
-	value, _, err = MVCCGet(ctx, engine, testKey2, hlc.Timestamp{Logical: 1}, MVCCGetOptions{})
+	valueRes, err = MVCCGet(ctx, engine, testKey2, hlc.Timestamp{Logical: 1}, MVCCGetOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if !bytes.Equal(value2.RawBytes, value.RawBytes) {
+	if !bytes.Equal(value2.RawBytes, valueRes.Value.RawBytes) {
 		t.Fatalf("the value %s in get result does not match the value %s in request",
-			value2.RawBytes, value.RawBytes)
+			value2.RawBytes, valueRes.Value.RawBytes)
 	}
 }
 
@@ -3356,15 +3356,15 @@ func TestMVCCResolveWithUpdatedTimestamp(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	value, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{
+	valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{
 		Txn: txn1,
 	})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if !bytes.Equal(value1.RawBytes, value.RawBytes) {
+	if !bytes.Equal(value1.RawBytes, valueRes.Value.RawBytes) {
 		t.Fatalf("the value %s in get result does not match the value %s in request",
-			value1.RawBytes, value.RawBytes)
+			value1.RawBytes, valueRes.Value.RawBytes)
 	}
 
 	// Resolve with a higher commit timestamp -- this should rewrite the
@@ -3375,21 +3375,21 @@ func TestMVCCResolveWithUpdatedTimestamp(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	value, _, err = MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 1}, MVCCGetOptions{})
-	if value != nil || err != nil {
-		t.Fatalf("expected both value and err to be nil: %+v, %v", value, err)
+	valueRes, err = MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 1}, MVCCGetOptions{})
+	if valueRes.Value != nil || err != nil {
+		t.Fatalf("expected both value and err to be nil: %+v, %v", valueRes.Value, err)
 	}
 
-	value, _, err = MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{})
+	valueRes, err = MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{})
 	if err != nil {
 		t.Error(err)
 	}
-	if expTS := (hlc.Timestamp{WallTime: 1}); value.Timestamp != expTS {
-		t.Fatalf("expected timestamp %+v == %+v", value.Timestamp, expTS)
+	if expTS := (hlc.Timestamp{WallTime: 1}); valueRes.Value.Timestamp != expTS {
+		t.Fatalf("expected timestamp %+v == %+v", valueRes.Value.Timestamp, expTS)
 	}
-	if !bytes.Equal(value1.RawBytes, value.RawBytes) {
+	if !bytes.Equal(value1.RawBytes, valueRes.Value.RawBytes) {
 		t.Fatalf("the value %s in get result does not match the value %s in request",
-			value1.RawBytes, value.RawBytes)
+			value1.RawBytes, valueRes.Value.RawBytes)
 	}
 }
 
@@ -3404,15 +3404,15 @@ func TestMVCCResolveWithPushedTimestamp(t *testing.T) {
 	if err := MVCCPut(ctx, engine, nil, testKey1, txn1.ReadTimestamp, hlc.ClockTimestamp{}, value1, txn1); err != nil {
 		t.Fatal(err)
 	}
-	value, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{
+	valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{
 		Txn: txn1,
 	})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if !bytes.Equal(value1.RawBytes, value.RawBytes) {
+	if !bytes.Equal(value1.RawBytes, valueRes.Value.RawBytes) {
 		t.Fatalf("the value %s in get result does not match the value %s in request",
-			value1.RawBytes, value.RawBytes)
+			value1.RawBytes, valueRes.Value.RawBytes)
 	}
 
 	// Resolve with a higher commit timestamp, but with still-pending transaction.
@@ -3423,24 +3423,24 @@ func TestMVCCResolveWithPushedTimestamp(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	value, _, err = MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{})
-	if value != nil || err == nil {
-		t.Fatalf("expected both value nil and err to be a writeIntentError: %+v", value)
+	valueRes, err = MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{})
+	if valueRes.Value != nil || err == nil {
+		t.Fatalf("expected both value nil and err to be a writeIntentError: %+v", valueRes.Value)
 	}
 
 	// Can still fetch the value using txn1.
-	value, _, err = MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{
+	valueRes, err = MVCCGet(ctx, engine, testKey1, hlc.Timestamp{WallTime: 1}, MVCCGetOptions{
 		Txn: txn1,
 	})
 	if err != nil {
 		t.Error(err)
 	}
-	if expTS := (hlc.Timestamp{WallTime: 1}); value.Timestamp != expTS {
-		t.Fatalf("expected timestamp %+v == %+v", value.Timestamp, expTS)
+	if expTS := (hlc.Timestamp{WallTime: 1}); valueRes.Value.Timestamp != expTS {
+		t.Fatalf("expected timestamp %+v == %+v", valueRes.Value.Timestamp, expTS)
 	}
-	if !bytes.Equal(value1.RawBytes, value.RawBytes) {
+	if !bytes.Equal(value1.RawBytes, valueRes.Value.RawBytes) {
 		t.Fatalf("the value %s in get result does not match the value %s in request",
-			value1.RawBytes, value.RawBytes)
+			value1.RawBytes, valueRes.Value.RawBytes)
 	}
 }
 
@@ -3513,45 +3513,45 @@ func TestMVCCResolveTxnRange(t *testing.T) {
 	}
 
 	{
-		value, _, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 1}, MVCCGetOptions{})
+		valueRes, err := MVCCGet(ctx, engine, testKey1, hlc.Timestamp{Logical: 1}, MVCCGetOptions{})
 		if err != nil {
 			t.Fatal(err)
 		}
-		if !bytes.Equal(value1.RawBytes, value.RawBytes) {
+		if !bytes.Equal(value1.RawBytes, valueRes.Value.RawBytes) {
 			t.Fatalf("the value %s in get result does not match the value %s in request",
-				value1.RawBytes, value.RawBytes)
+				value1.RawBytes, valueRes.Value.RawBytes)
 		}
 	}
 	{
-		value, _, err := MVCCGet(ctx, engine, testKey2, hlc.Timestamp{Logical: 1}, MVCCGetOptions{})
+		valueRes, err := MVCCGet(ctx, engine, testKey2, hlc.Timestamp{Logical: 1}, MVCCGetOptions{})
 		if err != nil {
 			t.Fatal(err)
 		}
-		if !bytes.Equal(value2.RawBytes, value.RawBytes) {
+		if !bytes.Equal(value2.RawBytes, valueRes.Value.RawBytes) {
 			t.Fatalf("the value %s in get result does not match the value %s in request",
-				value2.RawBytes, value.RawBytes)
+				value2.RawBytes, valueRes.Value.RawBytes)
 		}
 	}
 	{
-		value, _, err := MVCCGet(ctx, engine, testKey3, hlc.Timestamp{Logical: 1}, MVCCGetOptions{
+		valueRes, err := MVCCGet(ctx, engine, testKey3, hlc.Timestamp{Logical: 1}, MVCCGetOptions{
 			Txn: txn2,
 		})
 		if err != nil {
 			t.Fatal(err)
 		}
-		if !bytes.Equal(value3.RawBytes, value.RawBytes) {
+		if !bytes.Equal(value3.RawBytes, valueRes.Value.RawBytes) {
 			t.Fatalf("the value %s in get result does not match the value %s in request",
-				value3.RawBytes, value.RawBytes)
+				value3.RawBytes, valueRes.Value.RawBytes)
 		}
 	}
 	{
-		value, _, err := MVCCGet(ctx, engine, testKey4, hlc.Timestamp{Logical: 1}, MVCCGetOptions{})
+		valueRes, err := MVCCGet(ctx, engine, testKey4, hlc.Timestamp{Logical: 1}, MVCCGetOptions{})
 		if err != nil {
 			t.Fatal(err)
 		}
-		if !bytes.Equal(value4.RawBytes, value.RawBytes) {
+		if !bytes.Equal(value4.RawBytes, valueRes.Value.RawBytes) {
 			t.Fatalf("the value %s in get result does not match the value %s in request",
-				value1.RawBytes, value.RawBytes)
+				value1.RawBytes, valueRes.Value.RawBytes)
 		}
 	}
 }
@@ -3607,11 +3607,11 @@ func TestMVCCResolveTxnRangeResume(t *testing.T) {
 	// Check that the intents are actually gone by trying to read above them
 	// using txn2.
 	for i := 0; i < 18; i += 3 {
-		val, intent, err := MVCCGet(ctx, engine, roachpb.Key(fmt.Sprintf("%02d%d", i, i)),
+		res, err := MVCCGet(ctx, engine, roachpb.Key(fmt.Sprintf("%02d%d", i, i)),
 			txn2.ReadTimestamp, MVCCGetOptions{Txn: txn2})
-		require.NotNil(t, val)
+		require.NotNil(t, res.Value)
 		require.NoError(t, err)
-		require.Nil(t, intent)
+		require.Nil(t, res.Intent)
 	}
 }
 
@@ -5870,13 +5870,13 @@ func TestResolveIntentWithLowerEpoch(t *testing.T) {
 	}
 
 	// Check that the intent was not cleared.
-	_, intent, err := MVCCGet(ctx, engine, testKey1, hlc.MaxTimestamp, MVCCGetOptions{
+	intentRes, err := MVCCGet(ctx, engine, testKey1, hlc.MaxTimestamp, MVCCGetOptions{
 		Inconsistent: true,
 	})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if intent == nil {
+	if intentRes.Intent == nil {
 		t.Fatal("intent should not be cleared by resolve intent request with lower epoch")
 	}
 }
@@ -5959,10 +5959,10 @@ func TestMVCCTimeSeriesPartialMerge(t *testing.T) {
 			}
 		}
 
-		if v, _, err := MVCCGet(ctx, engine, k, hlc.Timestamp{}, MVCCGetOptions{}); err != nil {
+		if valueRes, err := MVCCGet(ctx, engine, k, hlc.Timestamp{}, MVCCGetOptions{}); err != nil {
 			t.Fatal(err)
 		} else {
-			vals[i] = v
+			vals[i] = valueRes.Value
 		}
 	}
 
diff --git a/pkg/storage/testdata/mvcc_histories/get_pagination b/pkg/storage/testdata/mvcc_histories/get_pagination
new file mode 100644
index 000000000000..98fe2539b71c
--- /dev/null
+++ b/pkg/storage/testdata/mvcc_histories/get_pagination
@@ -0,0 +1,52 @@
+# Test MaxKeys and TargetBytes for get.
+
+# Put some test data.
+run ok
+put k=a v=a ts=1
+put k=b v=bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb ts=1
+----
+>> at end:
+data: "a"/1.000000000,0 -> /BYTES/a
+data: "b"/1.000000000,0 -> /BYTES/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb
+
+# Return none since maxKeys < 0.
+run ok
+get k=a ts=2 maxKeys=-1
+----
+get: "a" -> <no data>
+
+# Return value since maxKeys >= 0.
+run ok
+get k=a ts=2 maxKeys=1
+----
+get: "a" -> /BYTES/a @1.000000000,0
+
+# Return none since targetBytes < 0.
+run ok
+get k=a ts=2 targetBytes=-1
+----
+get: "a" -> <no data>
+
+# Return none since targetBytes is insufficient and allowEmpty is true.
+run ok
+get k=b ts=2 targetBytes=1 allowEmpty
+----
+get: "b" -> <no data>
+
+# Return value since targetBytes is sufficient and allowEmpty is true.
+run ok
+get k=a ts=2 targetBytes=100 allowEmpty
+----
+get: "a" -> /BYTES/a @1.000000000,0
+
+# Return value since targetBytes is insufficient and allowEmpty is false.
+run ok
+get k=b ts=2 targetBytes=1
+----
+get: "b" -> /BYTES/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb @1.000000000,0
+
+# Return value since targetBytes is sufficient and allowEmpty is false.
+run ok
+get k=a ts=2 targetBytes=100
+----
+get: "a" -> /BYTES/a @1.000000000,0
diff --git a/pkg/testutils/testcluster/testcluster.go b/pkg/testutils/testcluster/testcluster.go
index 165204abd621..578771e85e18 100644
--- a/pkg/testutils/testcluster/testcluster.go
+++ b/pkg/testutils/testcluster/testcluster.go
@@ -1473,16 +1473,16 @@ func (tc *TestCluster) ReadIntFromStores(key roachpb.Key) []int64 {
 	results := make([]int64, len(tc.Servers))
 	for i, server := range tc.Servers {
 		err := server.Stores().VisitStores(func(s *kvserver.Store) error {
-			val, _, err := storage.MVCCGet(context.Background(), s.Engine(), key,
+			valRes, err := storage.MVCCGet(context.Background(), s.Engine(), key,
 				server.Clock().Now(), storage.MVCCGetOptions{})
 			if err != nil {
 				log.VEventf(context.Background(), 1, "store %d: error reading from key %s: %s", s.StoreID(), key, err)
-			} else if val == nil {
+			} else if valRes.Value == nil {
 				log.VEventf(context.Background(), 1, "store %d: missing key %s", s.StoreID(), key)
 			} else {
-				results[i], err = val.GetInt()
+				results[i], err = valRes.Value.GetInt()
 				if err != nil {
-					log.Errorf(context.Background(), "store %d: error decoding %s from key %s: %+v", s.StoreID(), val, key, err)
+					log.Errorf(context.Background(), "store %d: error decoding %s from key %s: %+v", s.StoreID(), valRes.Value, key, err)
 				}
 			}
 			return nil