fix merge

Author: SkyFan2002

src/query/sql/src/planner/optimizer/optimizer.rs

@@ -46,6 +46,7 @@ use crate::optimizer::RuleID;
 use crate::optimizer::SExpr;
 use crate::optimizer::DEFAULT_REWRITE_RULES;
 use crate::planner::query_executor::QueryExecutor;
+use crate::plans::Append;
 use crate::plans::CopyIntoLocationPlan;
 use crate::plans::Exchange;
 use crate::plans::Join;
@@ -294,11 +295,20 @@ pub async fn optimize(mut opt_ctx: OptimizerContext, plan: Plan) -> Result<Plan>
             overwrite,
             forbid_occ_retry,
         } => {
+            let support_distributed_insert = {
+                let append: Append = s_expr.plan().clone().try_into()?;
+                let metadata = metadata.read();
+                metadata
+                    .table(append.table_index)
+                    .table()
+                    .support_distributed_insert()
+            };
             let enable_distributed = opt_ctx.enable_distributed_optimization
                 && opt_ctx
                     .table_ctx
                     .get_settings()
-                    .get_enable_distributed_copy()?;
+                    .get_enable_distributed_copy()?
+                && support_distributed_insert;
             info!(
                 "after optimization enable_distributed_copy? : {}",
                 enable_distributed

src/query/storages/fuse/src/operations/common/processors/sink_commit.rs

@@ -137,23 +137,18 @@ where F: SnapshotGenerator + Send + 'static
     }
 
     fn is_error_recoverable(&self, e: &ErrorCode) -> bool {
-        let code = e.code();
-        // When prev_snapshot_id is some, means it is an alter table column modification or truncate.
-        // In this case if commit to meta fail and error is TABLE_VERSION_MISMATCHED operation will be aborted.
-        if self.prev_snapshot_id.is_some() && code == ErrorCode::TABLE_VERSION_MISMATCHED {
+        if self.forbid_retry {
             return false;
         }
 
-        code == ErrorCode::TABLE_VERSION_MISMATCHED
-            || (self.purge && code == ErrorCode::STORAGE_NOT_FOUND)
+        e.code() == ErrorCode::TABLE_VERSION_MISMATCHED
+            || (self.purge && e.code() == ErrorCode::STORAGE_NOT_FOUND)
     }
 
     fn no_side_effects_in_meta_store(e: &ErrorCode) -> bool {
         // currently, the only error that we know,  which indicates there are no side effects
         // is TABLE_VERSION_MISMATCHED
         e.code() == ErrorCode::TABLE_VERSION_MISMATCHED
-    fn can_retry(&self, e: &ErrorCode) -> bool {
-        !self.forbid_retry && FuseTable::is_error_recoverable(e, self.purge)
     }
 
     fn read_meta(&mut self) -> Result<Event> {
@@ -448,7 +443,7 @@ where F: SnapshotGenerator + Send + 'static
                         info!("commit mutation success, targets {:?}", target_descriptions);
                         self.state = State::Finish;
                     }
-                    Err(e) if self.can_retry(&e) => {
+                    Err(e) if self.is_error_recoverable(&e) => {
                         let table_info = self.table.get_table_info();
                         match self.backoff.next_backoff() {
                             Some(d) => {