refactor: remove reduntant .compat() (#2949)

* refactor: remove reduntant `.compat()`

Signed-off-by: Zhenchi <zhongzc_arch@outlook.com>

* remove dep

Signed-off-by: Zhenchi <zhongzc_arch@outlook.com>

---------

Signed-off-by: Zhenchi <zhongzc_arch@outlook.com>

Cargo.lock

@@ -4888,7 +4888,6 @@ dependencies = [
  "aquamarine",
  "arc-swap",
  "async-channel",
- "async-compat",
  "async-stream",
  "async-trait",
  "bytes",
@@ -5622,7 +5621,6 @@ name = "operator"
 version = "0.4.4"
 dependencies = [
  "api",
- "async-compat",
  "async-trait",
  "auth",
  "catalog",

src/mito2/Cargo.toml

@@ -14,7 +14,6 @@ api.workspace = true
 aquamarine.workspace = true
 arc-swap = "1.6"
 async-channel = "1.9"
-async-compat = "0.2"
 async-stream.workspace = true
 async-trait = "0.1"
 bytes.workspace = true

src/mito2/src/sst/parquet/reader.rs

@@ -18,7 +18,6 @@ use std::collections::{HashSet, VecDeque};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
-use async_compat::{Compat, CompatExt};
 use async_trait::async_trait;
 use common_telemetry::debug;
 use common_time::range::TimestampRange;
@@ -123,8 +122,7 @@ impl ParquetReaderBuilder {
             .object_store
             .reader(&file_path)
             .await
-            .context(OpenDalSnafu)?
-            .compat();
+            .context(OpenDalSnafu)?;
         let mut reader = BufReader::new(reader);
         // Loads parquet metadata of the file.
         let parquet_meta = self.read_parquet_metadata(&mut reader, &file_path).await?;
@@ -288,7 +286,7 @@ struct RowGroupReaderBuilder {
     /// Metadata of the parquet file.
     parquet_meta: Arc<ParquetMetaData>,
     /// Reader to get data.
-    file_reader: BufReader<Compat<Reader>>,
+    file_reader: BufReader<Reader>,
     /// Projection mask.
     projection: ProjectionMask,
     /// Field levels to read.

src/operator/Cargo.toml

@@ -9,7 +9,6 @@ testing = []
 
 [dependencies]
 api.workspace = true
-async-compat = "0.2"
 async-trait = "0.1"
 auth.workspace = true
 catalog.workspace = true

src/operator/src/statement/copy_table_from.rs

@@ -16,7 +16,6 @@ use std::collections::HashMap;
 use std::future::Future;
 use std::sync::Arc;
 
-use async_compat::CompatExt;
 use common_base::readable_size::ReadableSize;
 use common_datasource::file_format::csv::{CsvConfigBuilder, CsvOpener};
 use common_datasource::file_format::json::JsonOpener;
@@ -205,8 +204,7 @@ impl StatementExecutor {
                     .reader(path)
                     .await
                     .context(error::ReadObjectSnafu { path })?;
-
-                let buf_reader = BufReader::new(reader.compat());
+                let buf_reader = BufReader::new(reader);
 
                 let builder = ParquetRecordBatchStreamBuilder::new(buf_reader)
                     .await