Storage: Fast path when list with metadata.name=XYZ (#111630)

pkg/storage/unified/resource/server.go

@@ -943,6 +943,11 @@ func (s *server) List(ctx context.Context, req *resourcepb.ListRequest) (*resour
 		}
 	}
 
+	// Fast path for getting single value in a list
+	if rsp := s.tryFastPathList(ctx, req); rsp != nil {
+		return rsp, nil
+	}
+
 	if req.Limit < 1 {
 		req.Limit = 500 // default max 500 items in a page
 	}
@@ -1036,6 +1041,40 @@ func (s *server) List(ctx context.Context, req *resourcepb.ListRequest) (*resour
 	return rsp, err
 }
 
+// Some list queries can be calculated with simple reads
+func (s *server) tryFastPathList(ctx context.Context, req *resourcepb.ListRequest) *resourcepb.ListResponse {
+	if req.Source != resourcepb.ListRequest_STORE || req.Options.Key.Namespace == "" {
+		return nil
+	}
+
+	for _, v := range req.Options.Fields {
+		if v.Key == "metadata.name" && v.Operator == `=` {
+			if len(v.Values) == 1 {
+				read := &resourcepb.ReadRequest{
+					Key:             req.Options.Key,
+					ResourceVersion: req.ResourceVersion,
+				}
+				read.Key.Name = v.Values[0]
+				found, err := s.Read(ctx, read)
+				if err != nil {
+					return &resourcepb.ListResponse{Error: AsErrorResult(err)}
+				}
+
+				// Return a value when it exists
+				rsp := &resourcepb.ListResponse{}
+				if len(found.Value) > 0 {
+					rsp.Items = []*resourcepb.ResourceWrapper{{
+						Value:           found.Value,
+						ResourceVersion: found.ResourceVersion,
+					}}
+				}
+				return rsp
+			}
+		}
+	}
+	return nil
+}
+
 // isTrashItemAuthorized checks if the user has access to the trash item.
 func (s *server) isTrashItemAuthorized(ctx context.Context, iter ListIterator, trashChecker claims.ItemChecker) bool {
 	user, ok := claims.AuthInfoFrom(ctx)