Formatting!

src/arrow1/ffi.rs

@@ -1,5 +1,3 @@
-
-
 use arrow::array::{Array, StructArray};
 use arrow::datatypes::{Field, Schema};
 use arrow::ffi::{self, from_ffi, to_ffi};
@@ -114,7 +112,7 @@ impl From<FFIArrowRecordBatch> for RecordBatch {
     fn from(value: FFIArrowRecordBatch) -> Self {
         let array_data = from_ffi(*value.array, &value.field).unwrap();
         let intermediate = StructArray::from(array_data);
-        
+
         RecordBatch::from(intermediate)
     }
 }

src/arrow1/reader_async.rs

@@ -71,7 +71,6 @@ pub async fn read_record_batch_stream(
         Some(_content_length) => _content_length,
         None => get_content_length(url.clone()).await?,
     };
-    let content_length = usize::try_from(content_length).unwrap();
     let reader = crate::common::fetch::create_reader(url, content_length, None);
 
     let builder = ParquetRecordBatchStreamBuilder::new(reader.compat()).await?;

src/arrow1/wasm.rs

@@ -1,5 +1,5 @@
 use crate::arrow1::error::WasmResult;
-use crate::arrow1::ffi::{FFIArrowRecordBatch, FFIArrowTable};
+use crate::arrow1::ffi::FFIArrowTable;
 use crate::utils::assert_parquet_file_not_empty;
 use wasm_bindgen::prelude::*;
 
@@ -81,6 +81,7 @@ pub async fn read_ffi_stream(
     url: String,
     content_length: Option<usize>,
 ) -> WasmResult<wasm_streams::readable::sys::ReadableStream> {
+    use crate::arrow1::ffi::FFIArrowRecordBatch;
     use futures::StreamExt;
     let parquet_stream =
         crate::arrow1::reader_async::read_record_batch_stream(url, content_length).await?;

src/arrow2/reader_async.rs

@@ -116,24 +116,26 @@ pub async fn read_row_group(
     Ok(output_file)
 }
 
-pub async fn read_record_batch_stream(url: String) -> Result<impl futures::Stream<Item = super::ffi::FFIArrowRecordBatch>> {
+pub async fn read_record_batch_stream(
+    url: String,
+) -> Result<impl futures::Stream<Item = super::ffi::FFIArrowRecordBatch>> {
     use async_stream::stream;
     let inner_stream = stream! {
         let metadata = read_metadata_async(url.clone(), None).await.unwrap();
         let compat_meta = crate::arrow2::metadata::FileMetaData::from(metadata.clone());
-    
+
         let arrow_schema = compat_meta.arrow_schema().unwrap_or_else(|_| {
             let bar: Vec<arrow2::datatypes::Field> = vec![];
             arrow2::datatypes::Schema::from(bar).into()
         });
         for row_group_meta in metadata.row_groups {
-            let foo = arrow_schema.clone().into();
-            let deserializer = _read_row_group(url.clone(), &row_group_meta, &foo).await.unwrap();
+            let schema = arrow_schema.clone().into();
+            let deserializer = _read_row_group(url.clone(), &row_group_meta, &schema).await.unwrap();
             for maybe_chunk in deserializer {
                 let chunk = maybe_chunk.unwrap();
                 yield super::ffi::FFIArrowRecordBatch::from_chunk(chunk, arrow_schema.clone().into());
             }
         }
     };
     Ok(inner_stream)
-}
+}

src/arrow2/wasm.rs

@@ -355,7 +355,8 @@ pub async fn read_ffi_stream(
     url: String,
 ) -> WasmResult<wasm_streams::readable::sys::ReadableStream> {
     use futures::StreamExt;
-    let stream = super::reader_async::read_record_batch_stream(url).await?.map(|batch| Ok(batch.into()));
+    let stream = super::reader_async::read_record_batch_stream(url)
+        .await?
+        .map(|batch| Ok(batch.into()));
     Ok(wasm_streams::ReadableStream::from_stream(stream).into_raw())
-
-}
+}