fix review comment

Author: zhyass

src/query/storages/fuse/src/operations/commit.rs

@@ -46,12 +46,12 @@ use tracing::warn;
 use uuid::Uuid;
 
 use crate::io::write_meta;
+use crate::io::SegmentsIO;
 use crate::io::TableMetaLocationGenerator;
 use crate::operations::AppendOperationLogEntry;
 use crate::operations::TableOperationLog;
 use crate::statistics;
 use crate::statistics::merge_statistics;
-use crate::FuseSegmentIO;
 use crate::FuseTable;
 use crate::OPT_KEY_LEGACY_SNAPSHOT_LOC;
 use crate::OPT_KEY_SNAPSHOT_LOCATION;
@@ -425,7 +425,7 @@ impl FuseTable {
             }
 
             // Read all segments information in parallel.
-            let fuse_segment_io = FuseSegmentIO::create(ctx.clone(), self.operator.clone());
+            let fuse_segment_io = SegmentsIO::create(ctx.clone(), self.operator.clone());
             let results = fuse_segment_io.read_segments(&new_segments).await?;
             for result in results.iter() {
                 let segment = result.clone()?;

src/query/storages/fuse/src/operations/delete.rs

@@ -153,7 +153,7 @@ impl FuseTable {
             )
             .await
         {
-            abort_operation.abort(self.operator.clone()).await;
+            abort_operation.abort(ctx, self.operator.clone()).await?;
             return Err(e);
         }
         Ok(())

src/query/storages/fuse/src/operations/mutation/abort_operation.rs

@@ -12,9 +12,15 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
+use std::sync::Arc;
+
+use common_catalog::table_context::TableContext;
+use common_exception::Result;
 use common_fuse_meta::meta::BlockMeta;
 use opendal::Operator;
 
+use crate::io::Files;
+
 #[derive(Default, Clone, Debug)]
 pub struct AbortOperation {
     pub segments: Vec<String>,
@@ -37,15 +43,9 @@ impl AbortOperation {
         self
     }
 
-    pub async fn abort(self, operator: Operator) {
-        for block in self.blocks {
-            let _ = operator.object(&block).delete().await;
-        }
-        for index in self.bloom_filter_indexes {
-            let _ = operator.object(&index).delete().await;
-        }
-        for segment in self.segments {
-            let _ = operator.object(&segment).delete().await;
-        }
+    pub async fn abort(self, ctx: Arc<dyn TableContext>, operator: Operator) -> Result<()> {
+        let fuse_file = Files::create(ctx, operator);
+        let locations = vec![self.blocks, self.bloom_filter_indexes, self.segments].concat();
+        fuse_file.remove_file_in_batch(&locations).await
     }
 }

src/query/storages/fuse/src/operations/mutation/compact_mutator.rs

@@ -197,7 +197,9 @@ impl TableMutator for CompactMutator {
             .commit_mutation(ctx.clone(), self.base_snapshot.clone(), segments, summary)
             .await
         {
-            abort_operation.abort(self.data_accessor.clone()).await;
+            abort_operation
+                .abort(ctx, self.data_accessor.clone())
+                .await?;
             return Err(e);
         }
         Ok(())

src/query/storages/fuse/src/operations/mutation/recluster_mutator.rs

@@ -242,8 +242,8 @@ impl TableMutator for ReclusterMutator {
             .await
         {
             abort_operation
-                .abort(self.base_mutator.data_accessor.clone())
-                .await;
+                .abort(ctx, self.base_mutator.data_accessor.clone())
+                .await?;
             return Err(e);
         }
         Ok(())