Updates

Author: alamb

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

@@ -104,10 +104,6 @@ mod tests {
             unimplemented!()
         }
 
-        async fn get(&self, location: &Path) -> object_store::Result<GetResult> {
-            self.get_opts(location, GetOptions::default()).await
-        }
-
         async fn get_opts(
             &self,
             location: &Path,
@@ -147,14 +143,6 @@ mod tests {
             unimplemented!()
         }
 
-        async fn head(&self, _location: &Path) -> object_store::Result<ObjectMeta> {
-            unimplemented!()
-        }
-
-        async fn delete(&self, _location: &Path) -> object_store::Result<()> {
-            unimplemented!()
-        }
-
         fn list(
             &self,
             _prefix: Option<&Path>,
@@ -169,15 +157,10 @@ mod tests {
             unimplemented!()
         }
 
-        async fn copy(&self, _from: &Path, _to: &Path) -> object_store::Result<()> {
-            unimplemented!()
-        }
-
-        async fn copy_if_not_exists(
+        fn delete_stream(
             &self,
-            _from: &Path,
-            _to: &Path,
-        ) -> object_store::Result<()> {
+            locations: BoxStream<'static, object_store::Result<Path>>,
+        ) -> BoxStream<'static, object_store::Result<Path>> {
             unimplemented!()
         }
     }

datafusion/datasource-arrow/src/file_format.rs

@@ -63,7 +63,8 @@ use datafusion_session::Session;
 use futures::StreamExt;
 use futures::stream::BoxStream;
 use object_store::{
-    GetOptions, GetRange, GetResultPayload, ObjectMeta, ObjectStore, path::Path,
+    GetOptions, GetRange, GetResultPayload, ObjectMeta, ObjectStore, ObjectStoreExt,
+    path::Path,
 };
 use tokio::io::AsyncWriteExt;
 

datafusion/datasource-arrow/src/source.rs

@@ -56,7 +56,7 @@ use datafusion_datasource::file_stream::FileOpenFuture;
 use datafusion_datasource::file_stream::FileOpener;
 use futures::{StreamExt, TryStreamExt};
 use itertools::Itertools;
-use object_store::{GetOptions, GetRange, GetResultPayload, ObjectStore};
+use object_store::{GetOptions, GetRange, GetResultPayload, ObjectStore, ObjectStoreExt};
 
 /// Enum indicating which Arrow IPC format to use
 #[derive(Clone, Copy, Debug)]

datafusion/datasource-avro/src/file_format.rs

@@ -41,7 +41,7 @@ use datafusion_physical_plan::ExecutionPlan;
 use datafusion_session::Session;
 
 use async_trait::async_trait;
-use object_store::{GetResultPayload, ObjectMeta, ObjectStore};
+use object_store::{GetResultPayload, ObjectMeta, ObjectStore, ObjectStoreExt};
 
 #[derive(Default)]
 /// Factory struct used to create [`AvroFormat`]

datafusion/datasource-avro/src/source.rs

@@ -33,7 +33,7 @@ use datafusion_physical_expr_common::sort_expr::LexOrdering;
 use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
 use datafusion_physical_plan::projection::ProjectionExprs;
 
-use object_store::ObjectStore;
+use object_store::{ObjectStore, ObjectStoreExt};
 
 /// AvroSource holds the extra configuration that is necessary for opening avro files
 #[derive(Clone)]

datafusion/datasource-csv/src/file_format.rs

@@ -60,7 +60,9 @@ use bytes::{Buf, Bytes};
 use datafusion_datasource::source::DataSourceExec;
 use futures::stream::BoxStream;
 use futures::{Stream, StreamExt, TryStreamExt, pin_mut};
-use object_store::{ObjectMeta, ObjectStore, delimited::newline_delimited_stream};
+use object_store::{
+    ObjectMeta, ObjectStore, ObjectStoreExt, delimited::newline_delimited_stream,
+};
 use regex::Regex;
 
 #[derive(Default)]

datafusion/datasource-json/src/file_format.rs

@@ -60,7 +60,7 @@ use datafusion_session::Session;
 use async_trait::async_trait;
 use bytes::{Buf, Bytes};
 use datafusion_datasource::source::DataSourceExec;
-use object_store::{GetResultPayload, ObjectMeta, ObjectStore};
+use object_store::{GetResultPayload, ObjectMeta, ObjectStore, ObjectStoreExt};
 
 #[derive(Default)]
 /// Factory struct used to create [JsonFormat]

datafusion/datasource-parquet/src/file_format.rs

@@ -70,7 +70,7 @@ use futures::future::BoxFuture;
 use futures::{FutureExt, StreamExt, TryStreamExt};
 use object_store::buffered::BufWriter;
 use object_store::path::Path;
-use object_store::{ObjectMeta, ObjectStore};
+use object_store::{ObjectMeta, ObjectStore, ObjectStoreExt};
 use parquet::arrow::arrow_writer::{
     ArrowColumnChunk, ArrowColumnWriter, ArrowLeafColumn, ArrowRowGroupWriterFactory,
     ArrowWriterOptions, compute_leaves,

datafusion/datasource-parquet/src/opener.rs

@@ -984,7 +984,7 @@ mod test {
     };
     use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
     use futures::{Stream, StreamExt};
-    use object_store::{ObjectStore, memory::InMemory, path::Path};
+    use object_store::{ObjectStore, ObjectStoreExt, memory::InMemory, path::Path};
     use parquet::arrow::ArrowWriter;
     use parquet::file::properties::WriterProperties;
 

datafusion/datasource-parquet/src/row_group_filter.rs

@@ -1537,7 +1537,7 @@ mod tests {
         pruning_predicate: &PruningPredicate,
     ) -> Result<RowGroupAccessPlanFilter> {
         use datafusion_datasource::PartitionedFile;
-        use object_store::{ObjectMeta, ObjectStore};
+        use object_store::{ObjectMeta, ObjectStoreExt};
 
         let object_meta = ObjectMeta {
             location: object_store::path::Path::parse(file_name).expect("creating path"),