roeap commented on code in PR #2374:
URL: https://github.com/apache/arrow-rs/pull/2374#discussion_r942933533
##########
object_store/src/azure.rs:
##########
@@ -277,79 +277,95 @@ impl ObjectStore for MicrosoftAzure {
}
async fn get(&self, location: &Path) -> Result<GetResult> {
- let blob = self
+ let loc = location.clone();
+ let mut stream = self
.container_client
- .as_blob_client(location.as_ref())
+ .blob_client(location.as_ref())
.get()
- .execute()
- .await
- .map_err(|err| {
- if check_err_not_found(&err) {
- return Error::NotFound {
- source: err,
- path: location.to_string(),
- };
- };
- Error::UnableToGetData {
- source: err,
- container: self.container_name.clone(),
- path: location.to_string(),
- }
- })?;
+ .into_stream()
+ .and_then(|chunk| chunk.data.collect())
+ .map_err(move |err| match err.kind() {
+ AzureErrorKind::HttpResponse {
+ status: StatusCode::NotFound,
+ ..
+ } => crate::Error::NotFound {
+ source: Box::new(err),
+ path: loc.to_string(),
+ },
+ _ => crate::Error::Generic {
+ source: Box::new(err),
+ store: "MicrosoftAzure",
+ },
+ })
+ .boxed();
- Ok(GetResult::Stream(
- futures::stream::once(async move { Ok(blob.data) }).boxed(),
- ))
+ let first = stream.next().await.transpose()?.unwrap_or_default();
+ Ok(GetResult::Stream(Box::pin(
+ futures::stream::once(async { Ok(first) }).chain(stream),
+ )))
Review Comment:
Somehow this felt a bit strange to collect the first element just to add it
to a stream right after, but I saw no other obvious way to have the method
raise an error. Before this we are never actually touching any results...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
