refactor: spead up csv reader. (#15043)

Author: youngsofun

src/query/storages/stage/src/read/row_based/processors/reader.rs

@@ -34,6 +34,7 @@ use crate::read::row_based::batch::BytesBatch;
 struct FileState {
     file: OneFilePartition,
     offset: usize,
+    reader: opendal::Reader,
 }
 pub struct BytesReader {
     table_ctx: Arc<dyn TableContext>,
@@ -62,14 +63,8 @@ impl BytesReader {
     pub async fn read_batch(&mut self) -> Result<DataBlock> {
         if let Some(state) = &mut self.file_state {
             let end = min(self.read_batch_size + state.offset, state.file.size);
-            let mut reader = self
-                .op
-                .reader_with(&state.file.path)
-                .range((state.offset as u64)..(end as u64))
-                .await?;
-
             let mut buffer = vec![0u8; end - state.offset];
-            let n = read_full(&mut reader, &mut buffer[0..]).await?;
+            let n = read_full(&mut state.reader, &mut buffer[..]).await?;
             if n == 0 {
                 return Err(ErrorCode::BadBytes(format!(
                     "Unexpected EOF {} expect {} bytes, read only {} bytes.",
@@ -123,7 +118,13 @@ impl PrefetchAsyncSource for BytesReader {
                 None => return Ok(None),
             };
             let file = OneFilePartition::from_part(&part)?.clone();
-            self.file_state = Some(FileState { file, offset: 0 })
+
+            let reader = self.op.reader_with(&file.path).await?;
+            self.file_state = Some(FileState {
+                file,
+                reader,
+                offset: 0,
+            })
         }
         match self.read_batch().await {
             Ok(block) => Ok(Some(block)),