chore: remove synchronous read methods and update opendal dependency

Author: dqhl76

Cargo.lock

@@ -385,7 +385,7 @@ num-derive = "0.4.2"
 num-traits = "0.2.19"
 num_cpus = "1.17"
 object = "0.36.5"
-object_store_opendal = { version = "0.52.0" }
+object_store_opendal = { version = "0.54.1" }
 once_cell = "1.15.0"
 opendal = { version = "0.54.1", features = [
     "layers-fastrace",

Cargo.toml

@@ -535,12 +535,11 @@ async fn fs_list_until_prefix(
     gc_root_meta_ts: Option<DateTime<Utc>>,
 ) -> Result<Vec<Entry>> {
     // Fetch ALL entries from the path and sort them by path in lexicographical order.
-    let lister = dal.blocking().lister(path)?;
+    let mut lister = dal.lister(path).await?;
     let mut entries = Vec::new();
-    for item in lister {
-        let entry = item?;
-        if entry.metadata().is_file() {
-            entries.push(entry);
+    while let Some(item) = lister.try_next().await? {
+        if item.metadata().is_file() {
+            entries.push(item);
         }
     }
     entries.sort_by(|l, r| l.path().cmp(r.path()));

src/query/ee/src/storages/fuse/operations/vacuum_table_v2.rs

@@ -113,62 +113,4 @@ impl MergeIOReader {
 
         Ok(read_res)
     }
-
-    pub fn sync_merge_io_read(
-        read_settings: &ReadSettings,
-        op: Operator,
-        location: &str,
-        raw_ranges: &[(ColumnId, Range<u64>)],
-    ) -> Result<MergeIOReadResult> {
-        let path = location.to_string();
-
-        // Build merged read ranges.
-        let ranges = raw_ranges
-            .iter()
-            .map(|(_, r)| r.clone())
-            .collect::<Vec<_>>();
-        let range_merger = RangeMerger::from_iter(
-            ranges,
-            read_settings.max_gap_size,
-            read_settings.max_range_size,
-            Some(read_settings.parquet_fast_read_bytes),
-        );
-        let merged_ranges = range_merger.ranges();
-
-        // Read merged range data.
-        let mut io_res = Vec::with_capacity(merged_ranges.len());
-        for (idx, range) in merged_ranges.iter().enumerate() {
-            let buf = op
-                .blocking()
-                .read_with(location)
-                .range(range.clone())
-                .call()?;
-            io_res.push((idx, buf));
-        }
-
-        let owner_memory = OwnerMemory::create(io_res);
-
-        let mut columns_chunk_offsets = HashMap::with_capacity(raw_ranges.len());
-        for (raw_idx, raw_range) in raw_ranges {
-            let column_id = *raw_idx as ColumnId;
-            let column_range = raw_range.start..raw_range.end;
-
-            // Find the range index and Range from merged ranges.
-            let (merged_range_idx, merged_range) = range_merger.get(column_range.clone()).ok_or_else(|| ErrorCode::Internal(format!(
-                "It's a terrible bug, not found raw range:[{:?}], path:{} from merged ranges\n: {:?}",
-                column_range, path, merged_ranges
-            )))?;
-
-            // Fetch the raw data for the raw range.
-            let start = (column_range.start - merged_range.start) as usize;
-            let end = (column_range.end - merged_range.start) as usize;
-            let range = start..end;
-            columns_chunk_offsets.insert(column_id, (merged_range_idx, range));
-        }
-
-        let read_res =
-            MergeIOReadResult::create(owner_memory, columns_chunk_offsets, location.to_string());
-
-        Ok(read_res)
-    }
 }