Update to object_store 0.4

datafusion/common/Cargo.toml

@@ -41,7 +41,7 @@ pyarrow = ["pyo3"]
 apache-avro = { version = "0.14", features = ["snappy"], optional = true }
 arrow = { version = "19.0.0", features = ["prettyprint"] }
 cranelift-module = { version = "0.86.1", optional = true }
-object_store = { version = "0.3", optional = true }
+object_store = { version = "0.4", optional = true }
 ordered-float = "3.0"
 parquet = { version = "19.0.0", features = ["arrow"], optional = true }
 pyo3 = { version = "0.16", optional = true }

datafusion/core/Cargo.toml

@@ -75,7 +75,7 @@ lazy_static = { version = "^1.4.0" }
 log = "^0.4"
 num-traits = { version = "0.2", optional = true }
 num_cpus = "1.13.0"
-object_store = "0.3.0"
+object_store = "0.4.0"
 ordered-float = "3.0"
 parking_lot = "0.12"
 parquet = { version = "19.0.0", features = ["arrow", "async"] }

datafusion/core/src/datasource/file_format/parquet.rs

@@ -575,7 +575,8 @@ mod tests {
     use futures::StreamExt;
     use object_store::local::LocalFileSystem;
     use object_store::path::Path;
-    use object_store::{GetResult, ListResult};
+    use object_store::{GetResult, ListResult, MultipartId};
+    use tokio::io::AsyncWrite;
 
     #[tokio::test]
     async fn read_merged_batches() -> Result<()> {
@@ -649,6 +650,22 @@ mod tests {
             Err(object_store::Error::NotImplemented)
         }
 
+        async fn put_multipart(
+            &self,
+            _location: &Path,
+        ) -> object_store::Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)>
+        {
+            Err(object_store::Error::NotImplemented)
+        }
+
+        async fn abort_multipart(
+            &self,
+            _location: &Path,
+            _multipart_id: &MultipartId,
+        ) -> object_store::Result<()> {
+            Err(object_store::Error::NotImplemented)
+        }
+
         async fn get(&self, _location: &Path) -> object_store::Result<GetResult> {
             Err(object_store::Error::NotImplemented)
         }

datafusion/core/src/physical_plan/file_format/chunked_store.rs

@@ -20,11 +20,12 @@ use bytes::Bytes;
 use futures::stream::BoxStream;
 use futures::StreamExt;
 use object_store::path::Path;
-use object_store::Result;
 use object_store::{GetResult, ListResult, ObjectMeta, ObjectStore};
+use object_store::{MultipartId, Result};
 use std::fmt::{Debug, Display, Formatter};
 use std::ops::Range;
 use std::sync::Arc;
+use tokio::io::AsyncWrite;
 
 /// Wraps a [`ObjectStore`] and makes its get response return chunks
 ///
@@ -53,6 +54,21 @@ impl ObjectStore for ChunkedStore {
         self.inner.put(location, bytes).await
     }
 
+    async fn put_multipart(
+        &self,
+        location: &Path,
+    ) -> Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
+        self.inner.put_multipart(location).await
+    }
+
+    async fn abort_multipart(
+        &self,
+        location: &Path,
+        multipart_id: &MultipartId,
+    ) -> Result<()> {
+        self.inner.abort_multipart(location, multipart_id).await
+    }
+
     async fn get(&self, location: &Path) -> Result<GetResult> {
         let bytes = self.inner.get(location).await?.bytes().await?;
         let mut offset = 0;

datafusion/core/src/physical_plan/file_format/parquet.rs

@@ -369,6 +369,30 @@ impl AsyncFileReader for ParquetFileReader {
             .boxed()
     }
 
+    fn get_byte_ranges(
+        &mut self,
+        ranges: Vec<Range<usize>>,
+    ) -> BoxFuture<'_, parquet::errors::Result<Vec<Bytes>>>
+    where
+        Self: Send,
+    {
+        let total = ranges.iter().map(|r| r.end - r.start).sum();
+        self.metrics.bytes_scanned.add(total);
+
+        async move {
+            self.store
+                .get_ranges(&self.meta.location, &ranges)
+                .await
+                .map_err(|e| {
+                    ParquetError::General(format!(
+                        "AsyncChunkReader::get_byte_ranges error: {}",
+                        e
+                    ))
+                })
+        }
+        .boxed()
+    }
+
     fn get_metadata(
         &mut self,
     ) -> BoxFuture<'_, parquet::errors::Result<Arc<ParquetMetaData>>> {

datafusion/core/tests/path_partition.rs

@@ -40,7 +40,10 @@ use datafusion::{
 use datafusion_common::ScalarValue;
 use futures::stream::BoxStream;
 use futures::{stream, StreamExt};
-use object_store::{path::Path, GetResult, ListResult, ObjectMeta, ObjectStore};
+use object_store::{
+    path::Path, GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore,
+};
+use tokio::io::AsyncWrite;
 
 #[tokio::test]
 async fn parquet_distinct_partition_col() -> Result<()> {
@@ -516,6 +519,21 @@ impl ObjectStore for MirroringObjectStore {
         unimplemented!()
     }
 
+    async fn put_multipart(
+        &self,
+        _location: &Path,
+    ) -> object_store::Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
+        unimplemented!()
+    }
+
+    async fn abort_multipart(
+        &self,
+        _location: &Path,
+        _multipart_id: &MultipartId,
+    ) -> object_store::Result<()> {
+        unimplemented!()
+    }
+
     async fn get(&self, location: &Path) -> object_store::Result<GetResult> {
         self.files.iter().find(|x| *x == location.as_ref()).unwrap();
         let path = std::path::PathBuf::from(&self.mirrored_file);