Refactor: remove unnecessary async for ArrowReader::read (apache#1608)

## Which issue does this PR close?

- Closes #.

## What changes are included in this PR?


## Are these changes tested?

Co-authored-by: Renjie Liu <[email protected]>

Author: 2 people

crates/iceberg/src/arrow/reader.rs

@@ -138,7 +138,7 @@ pub struct ArrowReader {
 impl ArrowReader {
     /// Take a stream of FileScanTasks and reads all the files.
     /// Returns a stream of Arrow RecordBatches containing the data from the files
-    pub async fn read(self, tasks: FileScanTaskStream) -> Result<ArrowRecordBatchStream> {
+    pub fn read(self, tasks: FileScanTaskStream) -> Result<ArrowRecordBatchStream> {
         let file_io = self.file_io.clone();
         let batch_size = self.batch_size;
         let concurrency_limit_data_files = self.concurrency_limit_data_files;
@@ -1751,7 +1751,6 @@ message schema {
 
         let result = reader
             .read(tasks)
-            .await
             .unwrap()
             .try_collect::<Vec<RecordBatch>>()
             .await

crates/iceberg/src/scan/mod.rs

@@ -437,10 +437,7 @@ impl TableScan {
             arrow_reader_builder = arrow_reader_builder.with_batch_size(batch_size);
         }
 
-        arrow_reader_builder
-            .build()
-            .read(self.plan_files().await?)
-            .await
+        arrow_reader_builder.build().read(self.plan_files().await?)
     }
 
     /// Returns a reference to the column names of the table scan.
@@ -1332,14 +1329,12 @@ pub mod tests {
         let batch_stream = reader
             .clone()
             .read(Box::pin(stream::iter(vec![Ok(plan_task.remove(0))])))
-            .await
             .unwrap();
         let batch_1: Vec<_> = batch_stream.try_collect().await.unwrap();
 
         let reader = ArrowReaderBuilder::new(fixture.table.file_io().clone()).build();
         let batch_stream = reader
             .read(Box::pin(stream::iter(vec![Ok(plan_task.remove(0))])))
-            .await
             .unwrap();
         let batch_2: Vec<_> = batch_stream.try_collect().await.unwrap();