Split executeRange function

Signed-off-by: Marek Siarkowicz <[email protected]>

Author: serathius

server/etcdserver/txn/txn.go

@@ -152,32 +152,45 @@ func Range(ctx context.Context, lg *zap.Logger, kv mvcc.KV, r *pb.RangeRequest)
 func executeRange(ctx context.Context, lg *zap.Logger, txnRead mvcc.TxnRead, r *pb.RangeRequest) (*pb.RangeResponse, error) {
 	trace := traceutil.Get(ctx)
 
-	resp := &pb.RangeResponse{}
-	resp.Header = &pb.ResponseHeader{}
+	limit := rangeLimit(r)
+	ro := mvcc.RangeOptions{
+		Limit: limit,
+		Rev:   r.Revision,
+		Count: r.CountOnly,
+	}
+
+	rr, err := txnRead.Range(ctx, r.Key, mkGteRange(r.RangeEnd), ro)
+	if err != nil {
+		return nil, err
+	}
+
+	filterRangeResults(rr, r)
+	sortRangeResults(rr, r, lg)
+	trace.Step("filter and sort the key-value pairs")
+
+	resp := asembleRangeResponse(rr, r)
+	trace.Step("assemble the response")
+
+	return resp, nil
+}
 
+func rangeLimit(r *pb.RangeRequest) int64 {
 	limit := r.Limit
 	if r.SortOrder != pb.RangeRequest_NONE ||
 		r.MinModRevision != 0 || r.MaxModRevision != 0 ||
 		r.MinCreateRevision != 0 || r.MaxCreateRevision != 0 {
 		// fetch everything; sort and truncate afterwards
 		limit = 0
 	}
+
 	if limit > 0 {
 		// fetch one extra for 'more' flag
 		limit = limit + 1
 	}
+	return limit
+}
 
-	ro := mvcc.RangeOptions{
-		Limit: limit,
-		Rev:   r.Revision,
-		Count: r.CountOnly,
-	}
-
-	rr, err := txnRead.Range(ctx, r.Key, mkGteRange(r.RangeEnd), ro)
-	if err != nil {
-		return nil, err
-	}
-
+func filterRangeResults(rr *mvcc.RangeResult, r *pb.RangeRequest) {
 	if r.MaxModRevision != 0 {
 		f := func(kv *mvccpb.KeyValue) bool { return kv.ModRevision > r.MaxModRevision }
 		pruneKVs(rr, f)
@@ -194,7 +207,9 @@ func executeRange(ctx context.Context, lg *zap.Logger, txnRead mvcc.TxnRead, r *
 		f := func(kv *mvccpb.KeyValue) bool { return kv.CreateRevision < r.MinCreateRevision }
 		pruneKVs(rr, f)
 	}
+}
 
+func sortRangeResults(rr *mvcc.RangeResult, r *pb.RangeRequest, lg *zap.Logger) {
 	sortOrder := r.SortOrder
 	if r.SortTarget != pb.RangeRequest_KEY && sortOrder == pb.RangeRequest_NONE {
 		// Since current mvcc.Range implementation returns results
@@ -207,46 +222,55 @@ func executeRange(ctx context.Context, lg *zap.Logger, txnRead mvcc.TxnRead, r *
 		// don't re-sort when target is 'KEY' and order is ASCEND
 		sortOrder = pb.RangeRequest_NONE
 	}
-	if sortOrder != pb.RangeRequest_NONE {
-		var sorter sort.Interface
-		switch {
-		case r.SortTarget == pb.RangeRequest_KEY:
-			sorter = &kvSortByKey{&kvSort{rr.KVs}}
-		case r.SortTarget == pb.RangeRequest_VERSION:
-			sorter = &kvSortByVersion{&kvSort{rr.KVs}}
-		case r.SortTarget == pb.RangeRequest_CREATE:
-			sorter = &kvSortByCreate{&kvSort{rr.KVs}}
-		case r.SortTarget == pb.RangeRequest_MOD:
-			sorter = &kvSortByMod{&kvSort{rr.KVs}}
-		case r.SortTarget == pb.RangeRequest_VALUE:
-			sorter = &kvSortByValue{&kvSort{rr.KVs}}
-		default:
-			lg.Panic("unexpected sort target", zap.Int32("sort-target", int32(r.SortTarget)))
-		}
-		switch {
-		case sortOrder == pb.RangeRequest_ASCEND:
-			sort.Sort(sorter)
-		case sortOrder == pb.RangeRequest_DESCEND:
-			sort.Sort(sort.Reverse(sorter))
-		}
+
+	if sortOrder == pb.RangeRequest_NONE {
+		return
 	}
 
+	var sorter sort.Interface
+	switch r.SortTarget {
+	case pb.RangeRequest_KEY:
+		sorter = &kvSortByKey{&kvSort{rr.KVs}}
+	case pb.RangeRequest_VERSION:
+		sorter = &kvSortByVersion{&kvSort{rr.KVs}}
+	case pb.RangeRequest_CREATE:
+		sorter = &kvSortByCreate{&kvSort{rr.KVs}}
+	case pb.RangeRequest_MOD:
+		sorter = &kvSortByMod{&kvSort{rr.KVs}}
+	case pb.RangeRequest_VALUE:
+		sorter = &kvSortByValue{&kvSort{rr.KVs}}
+	default:
+		// This should ideally not happen if request validation is done prior.
+		lg.Panic("unexpected sort target", zap.Int32("sort-target", int32(r.SortTarget)))
+		return // Defensive return after panic, though panic will stop execution.
+	}
+
+	switch sortOrder {
+	case pb.RangeRequest_ASCEND:
+		sort.Sort(sorter)
+	case pb.RangeRequest_DESCEND:
+		sort.Sort(sort.Reverse(sorter))
+	}
+}
+
+func asembleRangeResponse(rr *mvcc.RangeResult, r *pb.RangeRequest) *pb.RangeResponse {
+	resp := &pb.RangeResponse{Header: &pb.ResponseHeader{}}
 	if r.Limit > 0 && len(rr.KVs) > int(r.Limit) {
 		rr.KVs = rr.KVs[:r.Limit]
 		resp.More = true
 	}
-	trace.Step("filter and sort the key-value pairs")
+
 	resp.Header.Revision = rr.Rev
 	resp.Count = int64(rr.Count)
+
 	resp.Kvs = make([]*mvccpb.KeyValue, len(rr.KVs))
 	for i := range rr.KVs {
 		if r.KeysOnly {
 			rr.KVs[i].Value = nil
 		}
 		resp.Kvs[i] = &rr.KVs[i]
 	}
-	trace.Step("assemble the response")
-	return resp, nil
+	return resp
 }
 
 func Txn(ctx context.Context, lg *zap.Logger, rt *pb.TxnRequest, txnModeWriteWithSharedBuffer bool, kv mvcc.KV, lessor lease.Lessor) (*pb.TxnResponse, *traceutil.Trace, error) {