Merge pull request #18579 from nvanbenschoten/nvanbenschoten/lookupKeys

storage: fix declareKeysRangeLookup to include all possible keys

pkg/storage/batcheval/cmd_range_lookup.go

@@ -35,8 +35,60 @@ func init() {
 func declareKeysDeprecatedRangeLookup(
 	desc roachpb.RangeDescriptor, header roachpb.Header, req roachpb.Request, spans *spanset.SpanSet,
 ) {
-	DefaultDeclareKeys(desc, header, req, spans)
+	lookupReq := req.(*roachpb.DeprecatedRangeLookupRequest)
+
+	if keys.IsLocal(lookupReq.Key) {
+		// Error will be thrown during evaluation.
+		return
+	}
+	key := roachpb.RKey(lookupReq.Key)
+
+	// RangeLookupRequests depend on the range descriptor because they need
+	// to determine which range descriptors are within the local range.
 	spans.Add(spanset.SpanReadOnly, roachpb.Span{Key: keys.RangeDescriptorKey(desc.StartKey)})
+
+	// Both forward and reverse RangeLookupRequests scan forward initially. We are
+	// unable to bound this initial scan any further than from the lookup key to
+	// the end of the current descriptor.
+	scanBounds, err := rangeLookupScanBounds(&desc, key, false /* reverse */)
+	if err != nil {
+		// Errors will be caught during evaluation.
+		return
+	}
+	spans.Add(spanset.SpanReadOnly, roachpb.Span{
+		Key:    scanBounds.Key.AsRawKey(),
+		EndKey: scanBounds.EndKey.AsRawKey(),
+	})
+
+	if lookupReq.Reverse {
+		// A reverse RangeLookupRequest also scans backwards.
+		revScanBounds, err := rangeLookupScanBounds(&desc, key, true /* reverse */)
+		if err != nil {
+			// Errors will be caught during evaluation.
+			return
+		}
+		spans.Add(spanset.SpanReadOnly, roachpb.Span{
+			Key:    revScanBounds.Key.AsRawKey(),
+			EndKey: revScanBounds.EndKey.AsRawKey(),
+		})
+	}
+}
+
+// rangeLookupScanBounds returns the range [start,end) within the bounds of the
+// provided RangeDescriptor which the desired meta record can be found by means
+// of an engine scan.
+func rangeLookupScanBounds(
+	desc *roachpb.RangeDescriptor, key roachpb.RKey, reverse bool,
+) (roachpb.RSpan, error) {
+	boundsFn := keys.MetaScanBounds
+	if reverse {
+		boundsFn = keys.MetaReverseScanBounds
+	}
+	span, err := boundsFn(key)
+	if err != nil {
+		return roachpb.RSpan{}, err
+	}
+	return span.Intersect(desc)
 }
 
 // DeprecatedRangeLookup is used to look up RangeDescriptors - a RangeDescriptor
@@ -80,13 +132,11 @@ func DeprecatedRangeLookup(
 	h := cArgs.Header
 	reply := resp.(*roachpb.DeprecatedRangeLookupResponse)
 
-	key, err := keys.Addr(args.Key)
-	if err != nil {
-		return result.Result{}, err
-	}
-	if !key.Equal(args.Key) {
+	if keys.IsLocal(args.Key) {
 		return result.Result{}, errors.Errorf("illegal lookup of range-local key %q", args.Key)
 	}
+	key := roachpb.RKey(args.Key)
+
 	ts, txn, consistent, rangeCount := h.Timestamp, h.Txn, h.ReadConsistency == roachpb.CONSISTENT, int64(args.MaxRanges)
 	if rangeCount < 1 {
 		return result.Result{}, errors.Errorf("range lookup specified invalid maximum range count %d: must be > 0", rangeCount)
@@ -112,11 +162,7 @@ func DeprecatedRangeLookup(
 		// We want to search for the metadata key greater than
 		// args.Key. Scan for both the requested key and the keys immediately
 		// afterwards, up to MaxRanges.
-		span, err := keys.MetaScanBounds(key)
-		if err != nil {
-			return result.Result{}, err
-		}
-		span, err = span.Intersect(desc)
+		span, err := rangeLookupScanBounds(desc, key, false /* reverse */)
 		if err != nil {
 			return result.Result{}, err
 		}
@@ -167,11 +213,7 @@ func DeprecatedRangeLookup(
 		}
 
 		if key.Less(roachpb.RKey(keys.Meta2KeyMax)) {
-			span, err := keys.MetaScanBounds(key)
-			if err != nil {
-				return result.Result{}, err
-			}
-			span, err = span.Intersect(desc)
+			span, err := rangeLookupScanBounds(desc, key, false /* reverse */)
 			if err != nil {
 				return result.Result{}, err
 			}
@@ -186,11 +228,7 @@ func DeprecatedRangeLookup(
 		// We want to search for the metadata key just less or equal to
 		// args.Key. Scan in reverse order for both the requested key and the
 		// keys immediately backwards, up to MaxRanges.
-		span, err := keys.MetaReverseScanBounds(key)
-		if err != nil {
-			return result.Result{}, err
-		}
-		span, err = span.Intersect(desc)
+		span, err := rangeLookupScanBounds(desc, key, true /* reverse */)
 		if err != nil {
 			return result.Result{}, err
 		}