int RGWGetObj::read_user_manifest_part(rgw_bucket& bucket,
                                       const rgw_bucket_dir_entry& ent,
                                       RGWAccessControlPolicy * const bucket_acl,
                                       const boost::optional<Policy>& bucket_policy,
                                       const off_t start_ofs,
                                       const off_t end_ofs)
{
  ldout(s->cct, 20) << "user manifest obj=" << ent.key.name << "[" << ent.key.instance << "]" << dendl;
  RGWGetObj_CB cb(this);
  RGWGetObj_Filter* filter = &cb;
  boost::optional<RGWGetObj_Decompress> decompress;

  int64_t cur_ofs = start_ofs;
  int64_t cur_end = end_ofs;

  rgw_obj part(bucket, ent.key);

  map<string, bufferlist> attrs;

  uint64_t obj_size;
  RGWObjectCtx obj_ctx(store);
  RGWAccessControlPolicy obj_policy(s->cct);

  ldout(s->cct, 20) << "reading obj=" << part << " ofs=" << cur_ofs << " end=" << cur_end << dendl;

  obj_ctx.obj.set_atomic(part);
  store->set_prefetch_data(&obj_ctx, part);

  RGWRados::Object op_target(store, s->bucket_info, obj_ctx, part);
  RGWRados::Object::Read read_op(&op_target);

  read_op.conds.if_match = ent.meta.etag.c_str();
  read_op.params.attrs = &attrs;
  read_op.params.obj_size = &obj_size;

  op_ret = read_op.prepare();
  if (op_ret < 0)
    return op_ret;
  op_ret = read_op.range_to_ofs(ent.meta.accounted_size, cur_ofs, cur_end);
  if (op_ret < 0)
    return op_ret;
  bool need_decompress;
  op_ret = rgw_compression_info_from_attrset(attrs, need_decompress, cs_info);
  if (op_ret < 0) {
	  lderr(s->cct) << "ERROR: failed to decode compression info, cannot decompress" << dendl;
      return -EIO;
  }

  if (need_decompress)
  {
    if (cs_info.orig_size != ent.meta.accounted_size) {
      // hmm.. something wrong, object not as expected, abort!
      ldout(s->cct, 0) << "ERROR: expected cs_info.orig_size=" << cs_info.orig_size <<
          ", actual read size=" << ent.meta.size << dendl;
      return -EIO;
    }
    decompress.emplace(s->cct, &cs_info, partial_content, filter);
    filter = &*decompress;
  }
  else
  {
    if (obj_size != ent.meta.size) {
      // hmm.. something wrong, object not as expected, abort!
      ldout(s->cct, 0) << "ERROR: expected obj_size=" << obj_size << ", actual read size=" << ent.meta.size << dendl;
      return -EIO;
	  }
  }

  op_ret = rgw_policy_from_attrset(s->cct, attrs, &obj_policy);
  if (op_ret < 0)
    return op_ret;

  /* We can use global user_acl because LOs cannot have segments
   * stored inside different accounts. */
  if (s->system_request) {
    ldout(s->cct, 2) << "overriding permissions due to system operation" << dendl;
  } else if (s->auth.identity->is_admin_of(s->user->user_id)) {
    ldout(s->cct, 2) << "overriding permissions due to admin operation" << dendl;
  } else if (!verify_object_permission(s, part, s->user_acl.get(), bucket_acl,
				       &obj_policy, bucket_policy, action)) {
    return -EPERM;
  }

  if (ent.meta.size == 0) {
    return 0;
  }

  perfcounter->inc(l_rgw_get_b, cur_end - cur_ofs);
  filter->fixup_range(cur_ofs, cur_end);
  op_ret = read_op.iterate(cur_ofs, cur_end, filter);
  if (op_ret >= 0)
	  op_ret = filter->flush();
  return op_ret;
}