cargo fmt

Author: H-Plus-Time

parquet/src/arrow/async_reader/metadata.rs

@@ -52,7 +52,6 @@ impl<F: MetadataFetch> MetadataLoader<F> {
     ///
     /// See [`fetch_parquet_metadata`] for the meaning of the individual parameters
     pub async fn load(mut fetch: F, prefetch: Option<usize>) -> Result<Self> {
-
         let suffix = fetch.fetch(GetRange::Suffix(prefetch.unwrap_or(8))).await?;
         let suffix_len = suffix.len();
 
@@ -85,7 +84,11 @@ impl<F: MetadataFetch> MetadataLoader<F> {
         })
     }
 
-    pub async fn load_absolute(mut fetch: F, file_size: usize, prefetch: Option<usize>) -> Result<Self> {
+    pub async fn load_absolute(
+        mut fetch: F,
+        file_size: usize,
+        prefetch: Option<usize>,
+    ) -> Result<Self> {
         todo!()
     }
 
@@ -222,7 +225,7 @@ where
     let fetch = MetadataFetchFn(fetch);
     let loader = match file_size {
         Some(file_size) => MetadataLoader::load_absolute(fetch, file_size, prefetch).await?,
-        None => MetadataLoader::load(fetch, prefetch).await?
+        None => MetadataLoader::load(fetch, prefetch).await?,
     };
     Ok(loader.finish())
 }
@@ -241,7 +244,9 @@ mod tests {
         let range = match range {
             GetRange::Bounded(range) => range,
             GetRange::Offset(offset) => offset..file_size,
-            GetRange::Suffix(end_offset) => (file_size.saturating_sub(end_offset.try_into().unwrap())..file_size)
+            GetRange::Suffix(end_offset) => {
+                (file_size.saturating_sub(end_offset.try_into().unwrap())..file_size)
+            }
         };
         file.seek(SeekFrom::Start(range.start as _))?;
         let len = range.end - range.start;
@@ -264,7 +269,9 @@ mod tests {
             futures::future::ready(read_range(&mut file, range))
         };
 
-        let actual = fetch_parquet_metadata(&mut fetch, None, None).await.unwrap();
+        let actual = fetch_parquet_metadata(&mut fetch, None, None)
+            .await
+            .unwrap();
         assert_eq!(actual.file_metadata().schema(), expected);
         assert_eq!(fetch_count.load(Ordering::SeqCst), 2);
 
@@ -352,7 +359,7 @@ mod tests {
         assert_eq!(fetch_count.load(Ordering::SeqCst), 1);
         let metadata = loader.finish();
         assert!(metadata.offset_index().is_some() && metadata.column_index().is_some());
-        
+
         // Prefetch more than enough
         fetch_count.store(0, Ordering::SeqCst);
         let f = MetadataFetchFn(&mut fetch);

parquet/src/arrow/async_reader/mod.rs

@@ -200,11 +200,11 @@ impl<T: AsyncRead + AsyncSeek + Unpin + Send> AsyncFileReader for T {
                 GetRange::Suffix(end_offset) => {
                     self.seek(SeekFrom::End(-(end_offset as i64))).await?;
                     Some(end_offset)
-                },
+                }
                 GetRange::Offset(offset) => {
                     self.seek(SeekFrom::Start(offset as u64)).await?;
                     None
-                },
+                }
                 GetRange::Bounded(range) => {
                     self.seek(SeekFrom::Start(range.start as u64)).await?;
                     Some(range.end - range.start)
@@ -406,7 +406,10 @@ impl<T: AsyncFileReader + Send + 'static> ParquetRecordBatchStreamBuilder<T> {
         };
 
         let buffer = match column_metadata.bloom_filter_length() {
-            Some(length) => self.input.0.get_bytes((offset..offset + length as usize).into()),
+            Some(length) => self
+                .input
+                .0
+                .get_bytes((offset..offset + length as usize).into()),
             None => self
                 .input
                 .0
@@ -441,7 +444,9 @@ impl<T: AsyncFileReader + Send + 'static> ParquetRecordBatchStreamBuilder<T> {
                 })?;
                 self.input
                     .0
-                    .get_bytes((bitset_offset as usize..bitset_offset as usize + bitset_length).into())
+                    .get_bytes(
+                        (bitset_offset as usize..bitset_offset as usize + bitset_length).into(),
+                    )
                     .await?
             }
         };
@@ -989,7 +994,9 @@ mod tests {
             let range = match range {
                 GetRange::Bounded(range) => range,
                 GetRange::Offset(offset) => offset..self.data.len(),
-                GetRange::Suffix(end_offset) => self.data.len().saturating_sub(end_offset)..self.data.len()
+                GetRange::Suffix(end_offset) => {
+                    self.data.len().saturating_sub(end_offset)..self.data.len()
+                }
             };
             futures::future::ready(Ok(self.data.slice(range))).boxed()
         }

parquet/src/arrow/async_reader/store.rs

@@ -23,7 +23,7 @@ use futures::future::BoxFuture;
 use futures::{FutureExt, TryFutureExt};
 
 use object_store::GetOptions;
-use object_store::{ObjectStore, path::Path};
+use object_store::{path::Path, ObjectStore};
 
 use crate::arrow::async_reader::{AsyncFileReader, MetadataLoader};
 use crate::errors::Result;
@@ -118,10 +118,16 @@ impl AsyncFileReader for ParquetObjectReader {
     fn get_bytes(&mut self, range: GetRange) -> BoxFuture<'_, Result<Bytes>> {
         async move {
             self.store
-                .get_opts(&self.location, GetOptions {
-                    range: Some(range.into()),
-                    ..Default::default()
-                }).await?.bytes().await
+                .get_opts(
+                    &self.location,
+                    GetOptions {
+                        range: Some(range.into()),
+                        ..Default::default()
+                    },
+                )
+                .await?
+                .bytes()
+                .await
         }
         .map_err(|e| e.into())
         .boxed()