chore(storage): allow append when change tracking was enabled (#15427)

* allow append when change tracking was enabled

* chore: more code comments & minor refactoring

---------

Co-authored-by: dantengsky <dantengsky@gmail.com>

Author: zhyass

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

@@ -13,13 +13,13 @@
 // limitations under the License.
 
 mod multi_table_insert_commit;
-pub use multi_table_insert_commit::CommitMultiTableInsert;
 mod sink_commit;
 mod transform_merge_commit_meta;
 mod transform_mutation_aggregator;
 mod transform_serialize_block;
 mod transform_serialize_segment;
 
+pub use multi_table_insert_commit::CommitMultiTableInsert;
 pub use sink_commit::CommitSink;
 pub use transform_merge_commit_meta::TransformMergeCommitMeta;
 pub use transform_mutation_aggregator::TableMutationAggregator;

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

@@ -32,11 +32,12 @@ use databend_common_pipeline_sinks::AsyncSink;
 use databend_storages_common_table_meta::meta::TableSnapshot;
 use databend_storages_common_table_meta::meta::Versioned;
 
-use super::TransformMergeCommitMeta;
-use crate::operations::common::CommitMeta;
 use crate::operations::AppendGenerator;
+use crate::operations::CommitMeta;
 use crate::operations::SnapshotGenerator;
+use crate::operations::TransformMergeCommitMeta;
 use crate::FuseTable;
+
 pub struct CommitMultiTableInsert {
     commit_metas: HashMap<u64, CommitMeta>,
     tables: HashMap<u64, Arc<dyn Table>>,

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

@@ -48,10 +48,11 @@ use log::warn;
 use opendal::Operator;
 
 use crate::io::TableMetaLocationGenerator;
-use crate::operations::common::AbortOperation;
-use crate::operations::common::CommitMeta;
-use crate::operations::common::SnapshotGenerator;
 use crate::operations::set_backoff;
+use crate::operations::AbortOperation;
+use crate::operations::AppendGenerator;
+use crate::operations::CommitMeta;
+use crate::operations::SnapshotGenerator;
 use crate::operations::TruncateGenerator;
 use crate::operations::TruncateMode;
 use crate::FuseTable;
@@ -71,7 +72,7 @@ enum State {
         snapshot: TableSnapshot,
         table_info: TableInfo,
     },
-    AbortOperation,
+    AbortOperation(ErrorCode),
     Finish,
 }
 
@@ -205,6 +206,13 @@ where F: SnapshotGenerator + Send + 'static
             .downcast_ref::<TruncateGenerator>()
             .is_some_and(|gen| !matches!(gen.mode(), TruncateMode::Delete))
     }
+
+    fn is_append_only_txn(&self) -> bool {
+        self.snapshot_gen
+            .as_any()
+            .downcast_ref::<AppendGenerator>()
+            .is_some()
+    }
 }
 
 #[async_trait::async_trait]
@@ -229,7 +237,7 @@ where F: SnapshotGenerator + Send + 'static
             State::FillDefault
                 | State::TryCommit { .. }
                 | State::RefreshTable
-                | State::AbortOperation
+                | State::AbortOperation(_)
         ) {
             return Ok(Event::Async);
         }
@@ -259,33 +267,50 @@ where F: SnapshotGenerator + Send + 'static
                 cluster_key_meta,
                 table_info,
             } => {
-                if !self.change_tracking && self.table.change_tracking_enabled() {
-                    // If change tracing is disabled when the txn start, but is enabled when commit,
+                let change_tracking_enabled_during_commit = {
+                    let no_change_tracking_at_beginning = !self.change_tracking;
+                    // note that `self.table` might be refreshed if commit retried
+                    let change_tracking_enabled_now = self.table.change_tracking_enabled();
+
+                    no_change_tracking_at_beginning && change_tracking_enabled_now
+                };
+
+                if !self.is_append_only_txn() && change_tracking_enabled_during_commit {
+                    // If change tracking is not enabled when the txn start, but is enabled when committing,
                     // then the txn should be aborted.
-                    error!("commit mutation failed cause change tracking is enabled when commit");
-                    self.state = State::AbortOperation;
-                } else {
-                    let schema = self.table.schema().as_ref().clone();
-                    match self.snapshot_gen.generate_new_snapshot(
-                        schema,
-                        cluster_key_meta,
-                        previous,
-                        Some(table_info.ident.seq),
-                    ) {
-                        Ok(snapshot) => {
-                            self.state = State::TryCommit {
-                                data: snapshot.to_bytes()?,
-                                snapshot,
-                                table_info,
-                            };
-                        }
-                        Err(e) => {
-                            error!(
-                                "commit mutation failed after {} retries, error: {:?}",
-                                self.retries, e,
-                            );
-                            self.state = State::AbortOperation;
-                        }
+                    // For mutations other than append-only, stream column values (like _origin_block_id)
+                    // must be properly generated. If not, CDC will not function as expected.
+                    self.state = State::AbortOperation(ErrorCode::StorageOther(
+                        "commit failed because change tracking was enabled during the commit process",
+                    ));
+                    return Ok(());
+                }
+
+                // now:
+                // - either current txn IS append-only
+                //    even if this is a conflict txn T (in the meaning of table version) has been
+                // commited, which has changed the change-tracking state from disabled to enabled,
+                // merging with transaction T is still safe, since the CDC mechanism allows it.
+                // - or change-tracking state is NOT changed.
+                //    in this case, we only need standard conflict resolution.
+                // therefore, we can safely proceed.
+
+                let schema = self.table.schema().as_ref().clone();
+                match self.snapshot_gen.generate_new_snapshot(
+                    schema,
+                    cluster_key_meta,
+                    previous,
+                    Some(table_info.ident.seq),
+                ) {
+                    Ok(snapshot) => {
+                        self.state = State::TryCommit {
+                            data: snapshot.to_bytes()?,
+                            snapshot,
+                            table_info,
+                        };
+                    }
+                    Err(e) => {
+                        self.state = State::AbortOperation(e);
                     }
                 }
             }
@@ -312,9 +337,10 @@ where F: SnapshotGenerator + Send + 'static
                         .map_or(true, |previous| previous.snapshot_id != prev_snapshot_id)
                 });
                 if snapshot_has_changed {
-                    error!("commit mutation failed cause snapshot has changed when commit");
                     // if snapshot has changed abort operation
-                    self.state = State::AbortOperation;
+                    self.state = State::AbortOperation(ErrorCode::StorageOther(
+                        "commit failed because the snapshot had changed during the commit process",
+                    ));
                 } else {
                     self.snapshot_gen
                         .fill_default_values(schema, &previous)
@@ -333,11 +359,7 @@ where F: SnapshotGenerator + Send + 'static
                     self.state = State::FillDefault;
                 }
                 Err(e) => {
-                    error!(
-                        "commit mutation failed cause get lock failed, error: {:?}",
-                        e
-                    );
-                    self.state = State::AbortOperation;
+                    self.state = State::AbortOperation(e);
                 }
             },
             State::TryCommit {
@@ -448,7 +470,7 @@ where F: SnapshotGenerator + Send + 'static
                                 if FuseTable::no_side_effects_in_meta_store(&e) {
                                     // if we are sure that table state inside metastore has not been
                                     // modified by this operation, abort this operation.
-                                    self.state = State::AbortOperation;
+                                    self.state = State::AbortOperation(e);
                                 } else {
                                     return Err(ErrorCode::OCCRetryFailure(format!(
                                         "can not fulfill the tx after retries({} times, {} ms), aborted. table name {}, identity {}",
@@ -481,18 +503,20 @@ where F: SnapshotGenerator + Send + 'static
                     table_info: fuse_table.table_info.clone(),
                 };
             }
-            State::AbortOperation => {
+            State::AbortOperation(e) => {
                 let duration = self.start_time.elapsed();
                 metrics_inc_commit_aborts();
                 // todo: use histogram when it ready
                 metrics_inc_commit_milliseconds(duration.as_millis());
                 let op = self.abort_operation.clone();
                 op.abort(self.ctx.clone(), self.dal.clone()).await?;
-                return Err(ErrorCode::StorageOther(format!(
-                    "transaction aborted after {} retries, which took {} ms",
+                error!(
+                    "transaction aborted after {} retries, which took {} ms, cause: {:?}",
                     self.retries,
-                    duration.as_millis()
-                )));
+                    duration.as_millis(),
+                    e
+                );
+                return Err(e);
             }
             _ => return Err(ErrorCode::Internal("It's a bug.")),
         }