chore(storage): remove abort operation (#15557)

* chore: remove abort operation

* fix

Author: zhyass

src/query/service/src/pipelines/builders/builder_merge_into.rs

@@ -272,7 +272,6 @@ impl PipelineBuilder {
 
             // 6. serialize segment
             let serialize_segment_transform = TransformSerializeSegment::new(
-                self.ctx.clone(),
                 InputPort::create(),
                 OutputPort::create(),
                 table,
@@ -400,7 +399,6 @@ impl PipelineBuilder {
             table.get_cluster_stats_gen(self.ctx.clone(), 0, block_thresholds, None)?;
 
         let serialize_segment_transform = TransformSerializeSegment::new(
-            self.ctx.clone(),
             InputPort::create(),
             OutputPort::create(),
             table,

src/query/service/src/pipelines/builders/builder_replace_into.rs

@@ -134,7 +134,6 @@ impl PipelineBuilder {
         block_builder.source_schema = table.schema_with_stream();
 
         let serialize_segment_transform = TransformSerializeSegment::new(
-            self.ctx.clone(),
             InputPort::create(),
             OutputPort::create(),
             table,

src/query/service/src/pipelines/builders/transform_builder.rs

@@ -126,16 +126,9 @@ impl PipelineBuilder {
         table: Arc<dyn Table>,
         block_thresholds: BlockThresholds,
     ) -> Result<impl Fn(Arc<InputPort>, Arc<OutputPort>) -> Result<ProcessorPtr>> {
-        let ctx = self.ctx.clone();
         Ok(move |input, output| {
             let fuse_table = FuseTable::try_from_table(table.as_ref())?;
-            let proc = TransformSerializeSegment::new(
-                ctx.clone(),
-                input,
-                output,
-                fuse_table,
-                block_thresholds,
-            );
+            let proc = TransformSerializeSegment::new(input, output, fuse_table, block_thresholds);
             proc.into_processor()
         })
     }

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

@@ -56,7 +56,6 @@ use opendal::Operator;
 use crate::io::MetaWriter;
 use crate::io::SegmentsIO;
 use crate::io::TableMetaLocationGenerator;
-use crate::operations::common::AbortOperation;
 use crate::operations::common::AppendGenerator;
 use crate::operations::common::CommitSink;
 use crate::operations::common::ConflictResolveContext;
@@ -83,8 +82,7 @@ impl FuseTable {
         pipeline.try_resize(1)?;
 
         pipeline.add_transform(|input, output| {
-            let proc =
-                TransformSerializeSegment::new(ctx.clone(), input, output, self, block_thresholds);
+            let proc = TransformSerializeSegment::new(input, output, self, block_thresholds);
             proc.into_processor()
         })?;
 
@@ -281,7 +279,6 @@ impl FuseTable {
         base_snapshot: Arc<TableSnapshot>,
         base_segments: &[Location],
         base_summary: Statistics,
-        abort_operation: AbortOperation,
         max_retry_elapsed: Option<Duration>,
     ) -> Result<()> {
         let mut retries = 0;
@@ -369,9 +366,6 @@ impl FuseTable {
                                 concurrently_appended_segment_locations =
                                     &latest_snapshot.segments[range_of_newly_append];
                             } else {
-                                abort_operation
-                                    .abort(ctx.clone(), self.operator.clone())
-                                    .await?;
                                 metrics_inc_commit_mutation_unresolvable_conflict();
                                 break Err(ErrorCode::UnresolvableConflict(
                                     "segment compact conflict with other operations",
@@ -383,13 +377,10 @@ impl FuseTable {
                             continue;
                         }
                         None => {
-                            // Commit not fulfilled. try to abort the operations.
+                            // Commit not fulfilled, abort.
                             //
                             // Note that, here the last error we have seen is TableVersionMismatched,
                             // otherwise we should have been returned, thus it is safe to abort the operation here.
-                            abort_operation
-                                .abort(ctx.clone(), self.operator.clone())
-                                .await?;
                             break Err(ErrorCode::StorageOther(format!(
                                 "commit mutation failed after {} retries",
                                 retries

src/query/storages/fuse/src/operations/common/meta/abort_operation.rs

@@ -16,15 +16,15 @@ use databend_common_exception::ErrorCode;
 use databend_common_expression::BlockMetaInfo;
 use databend_common_expression::BlockMetaInfoDowncast;
 use databend_common_expression::DataBlock;
+use databend_storages_common_table_meta::meta::Location;
 
-use crate::operations::common::AbortOperation;
 use crate::operations::common::ConflictResolveContext;
 use crate::operations::common::SnapshotChanges;
 
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq)]
 pub struct CommitMeta {
     pub conflict_resolve_context: ConflictResolveContext,
-    pub abort_operation: AbortOperation,
+    pub new_segment_locs: Vec<Location>,
     pub table_id: u64,
 }
 
@@ -34,19 +34,19 @@ impl CommitMeta {
             conflict_resolve_context: ConflictResolveContext::ModifiedSegmentExistsInLatest(
                 SnapshotChanges::default(),
             ),
-            abort_operation: AbortOperation::default(),
+            new_segment_locs: vec![],
             table_id,
         }
     }
 
     pub fn new(
         conflict_resolve_context: ConflictResolveContext,
-        abort_operation: AbortOperation,
+        new_segment_locs: Vec<Location>,
         table_id: u64,
     ) -> Self {
         CommitMeta {
             conflict_resolve_context,
-            abort_operation,
+            new_segment_locs,
             table_id,
         }
     }

src/query/storages/fuse/src/operations/common/meta/commit_meta.rs

@@ -12,10 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod abort_operation;
 mod commit_meta;
 mod mutation_log;
 
-pub use abort_operation::AbortOperation;
 pub use commit_meta::*;
 pub use mutation_log::*;

src/query/storages/fuse/src/operations/common/meta/mod.rs

@@ -22,7 +22,6 @@ use databend_storages_common_table_meta::meta::BlockMeta;
 use databend_storages_common_table_meta::meta::FormatVersion;
 use databend_storages_common_table_meta::meta::Statistics;
 
-use crate::operations::common::AbortOperation;
 use crate::operations::mutation::BlockIndex;
 use crate::operations::mutation::CompactExtraInfo;
 use crate::operations::mutation::DeletedSegmentInfo;
@@ -38,7 +37,6 @@ pub enum MutationLogEntry {
     AppendSegment {
         segment_location: String,
         format_version: FormatVersion,
-        abort_operation: AbortOperation,
         summary: Statistics,
     },
     DeletedBlock {