feat: snapshots generated in multi txn has same prev_snapshot_id

src/query/service/tests/it/storages/fuse/conflict.rs

@@ -24,6 +24,7 @@ use databend_common_storages_fuse::operations::SnapshotChanges;
 use databend_common_storages_fuse::operations::SnapshotGenerator;
 use databend_storages_common_table_meta::meta::Statistics;
 use databend_storages_common_table_meta::meta::TableSnapshot;
+use databend_storages_common_txn::TxnManager;
 
 #[test]
 /// base snapshot contains segments 1, 2, 3,
@@ -62,6 +63,8 @@ fn test_unresolvable_delete_conflict() {
         None,
         Some(Arc::new(latest_snapshot)),
         None,
+        TxnManager::init(),
+        0,
     );
     assert!(result.is_err());
 }
@@ -151,6 +154,8 @@ fn test_resolvable_delete_conflict() {
         None,
         Some(Arc::new(latest_snapshot)),
         None,
+        TxnManager::init(),
+        0,
     );
     let snapshot = result.unwrap();
     let expected = vec![("8".to_string(), 1), ("4".to_string(), 1)];
@@ -255,6 +260,8 @@ fn test_resolvable_replace_conflict() {
         None,
         Some(Arc::new(latest_snapshot)),
         None,
+        TxnManager::init(),
+        0,
     );
     let snapshot = result.unwrap();
     let expected = vec![

src/query/storages/fuse/Cargo.toml

@@ -47,6 +47,7 @@ databend-storages-common-index = { workspace = true }
 databend-storages-common-io = { workspace = true }
 databend-storages-common-pruner = { workspace = true }
 databend-storages-common-table-meta = { workspace = true }
+databend-storages-common-txn = { workspace = true }
 enum-as-inner = "0.5"
 futures = { workspace = true }
 futures-util = { workspace = true }

src/query/storages/fuse/src/operations/common/generators/append_generator.rs

@@ -113,7 +113,7 @@ impl SnapshotGenerator for AppendGenerator {
         Ok(())
     }
 
-    fn generate_new_snapshot(
+    fn do_generate_new_snapshot(
         &self,
         schema: TableSchema,
         cluster_key_meta: Option<ClusterKey>,

src/query/storages/fuse/src/operations/common/generators/mutation_generator.rs

@@ -56,7 +56,7 @@ impl SnapshotGenerator for MutationGenerator {
         self.conflict_resolve_ctx = ctx;
     }
 
-    fn generate_new_snapshot(
+    fn do_generate_new_snapshot(
         &self,
         schema: TableSchema,
         cluster_key_meta: Option<ClusterKey>,

src/query/storages/fuse/src/operations/common/generators/snapshot_generator.rs

@@ -19,6 +19,7 @@ use databend_common_exception::Result;
 use databend_common_expression::TableSchema;
 use databend_storages_common_table_meta::meta::ClusterKey;
 use databend_storages_common_table_meta::meta::TableSnapshot;
+use databend_storages_common_txn::TxnManagerRef;
 
 use crate::operations::common::ConflictResolveContext;
 
@@ -43,5 +44,25 @@ pub trait SnapshotGenerator {
         cluster_key_meta: Option<ClusterKey>,
         previous: Option<Arc<TableSnapshot>>,
         prev_table_seq: Option<u64>,
+        txn_mgr: TxnManagerRef,
+        table_id: u64,
+    ) -> Result<TableSnapshot> {
+        let previous_of_previous = previous.as_ref().and_then(|prev| prev.prev_snapshot_id);
+        let mut snapshot =
+            self.do_generate_new_snapshot(schema, cluster_key_meta, previous, prev_table_seq)?;
+        let guard = txn_mgr.lock();
+        // If a table is updated multi times in a transaction, the previous snapshot is always the snapshot before the transaction.
+        if guard.is_active() && guard.get_table_from_buffer_by_id(table_id).is_some() {
+            snapshot.prev_snapshot_id = previous_of_previous;
+        }
+        Ok(snapshot)
+    }
+
+    fn do_generate_new_snapshot(
+        &self,
+        schema: TableSchema,
+        cluster_key_meta: Option<ClusterKey>,
+        previous: Option<Arc<TableSnapshot>>,
+        prev_table_seq: Option<u64>,
     ) -> Result<TableSnapshot>;
 }

src/query/storages/fuse/src/operations/common/generators/truncate_generator.rs

@@ -54,7 +54,7 @@ impl SnapshotGenerator for TruncateGenerator {
         self
     }
 
-    fn generate_new_snapshot(
+    fn do_generate_new_snapshot(
         &self,
         schema: TableSchema,
         cluster_key_meta: Option<ClusterKey>,

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

@@ -34,6 +34,7 @@ use databend_common_meta_types::MatchSeq;
 use databend_common_pipeline_sinks::AsyncSink;
 use databend_storages_common_table_meta::meta::TableSnapshot;
 use databend_storages_common_table_meta::meta::Versioned;
+use databend_storages_common_txn::TxnManagerRef;
 use log::debug;
 use log::error;
 use log::info;
@@ -90,7 +91,12 @@ impl AsyncSink for CommitMultiTableInsert {
             snapshot_generator.set_conflict_resolve_context(commit_meta.conflict_resolve_context);
             let table = self.tables.get(&table_id).unwrap();
             update_table_metas.push((
-                build_update_table_meta_req(table.as_ref(), &snapshot_generator).await?,
+                build_update_table_meta_req(
+                    table.as_ref(),
+                    &snapshot_generator,
+                    self.ctx.txn_mgr(),
+                )
+                .await?,
                 table.get_table_info().clone(),
             ));
             snapshot_generators.insert(table_id, snapshot_generator);
@@ -173,6 +179,7 @@ impl AsyncSink for CommitMultiTableInsert {
                                 *req = build_update_table_meta_req(
                                     table.as_ref(),
                                     snapshot_generators.get(&tid).unwrap(),
+                                    self.ctx.txn_mgr(),
                                 )
                                 .await?;
                                 break;
@@ -227,6 +234,7 @@ impl AsyncSink for CommitMultiTableInsert {
 async fn build_update_table_meta_req(
     table: &dyn Table,
     snapshot_generator: &AppendGenerator,
+    txn_mgr: TxnManagerRef,
 ) -> Result<UpdateTableMetaReq> {
     let fuse_table = FuseTable::try_from_table(table)?;
     let previous = fuse_table.read_table_snapshot().await?;
@@ -235,6 +243,8 @@ async fn build_update_table_meta_req(
         fuse_table.cluster_key_meta.clone(),
         previous,
         Some(fuse_table.table_info.ident.seq),
+        txn_mgr,
+        table.get_id(),
     )?;
 
     // write snapshot

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

@@ -286,6 +286,8 @@ where F: SnapshotGenerator + Send + 'static
                     cluster_key_meta,
                     previous,
                     Some(table_info.ident.seq),
+                    self.ctx.txn_mgr(),
+                    table_info.ident.table_id,
                 ) {
                     Ok(snapshot) => {
                         self.state = State::TryCommit {

tests/sqllogictests/suites/base/09_fuse_engine/09_0100_insert_multi_table.sql

@@ -325,6 +325,11 @@ begin transaction;
 statement ok
 insert into s values(3,4),(1,2),(5,6);
 
+query I
+select count(*) from fuse_snapshot('default', 's');
+----
+1
+
 query II
 INSERT FIRST
     WHEN c3 = 5 THEN
@@ -335,6 +340,16 @@ SELECT * from s;
 ----
 1 2
 
+query I
+select count(*) from fuse_snapshot('default', 't1');
+----
+1
+
+query I
+select count(*) from fuse_snapshot('default', 't2');
+----
+1
+
 query II
 select * from t1 order by c1;
 ----
@@ -349,6 +364,22 @@ select * from t2;
 statement ok
 rollback;
 
+query I
+select count(*) from fuse_snapshot('default', 's');
+----
+0
+
+query I
+select count(*) from fuse_snapshot('default', 't1');
+----
+0
+
+query I
+select count(*) from fuse_snapshot('default', 't2');
+----
+0
+
+
 query II
 select * from t1 order by c1;
 ----

tests/sqllogictests/suites/base/14_transaction/14_0003_merge_into.test

@@ -30,11 +30,21 @@ INSERT INTO employees VALUES
     (3, 'Charlie', 'Finance'),
     (4, 'David', 'HR');
 
+query I
+select count(*) from fuse_snapshot('test_txn_merge_into','employees');
+----
+1
+
 statement ok
 INSERT INTO salaries VALUES
     (1, 50000.00),
     (2, 60000.00);
 
+query I
+select count(*) from fuse_snapshot('test_txn_merge_into','salaries');
+----
+1
+
 statement ok
 MERGE INTO salaries
     USING (SELECT * FROM employees) AS employees
@@ -49,6 +59,11 @@ MERGE INTO salaries
         INSERT (employee_id, salary)
             VALUES (employees.employee_id, 55000.00);
 
+query I
+select count(*) from fuse_snapshot('test_txn_merge_into','salaries');
+----
+1
+
 query IF
 SELECT employee_id, salary FROM salaries order by employee_id;
 ----
@@ -60,6 +75,16 @@ SELECT employee_id, salary FROM salaries order by employee_id;
 statement ok
 COMMIT;
 
+query I
+select count(*) from fuse_snapshot('test_txn_merge_into','salaries');
+----
+1
+
+query I
+select count(*) from fuse_snapshot('test_txn_merge_into','employees');
+----
+1
+
 query IF
 SELECT employee_id, salary FROM salaries order by employee_id;
 ----
@@ -145,4 +170,4 @@ SELECT employee_id, salary FROM salaries order by employee_id;
 
 
 statement ok
-drop database test_txn_merge_into;
+drop database test_txn_merge_into;

tests/sqllogictests/suites/ee/06_ee_stream/06_0002_stream_txn_consume.test

@@ -54,30 +54,90 @@ select a from s_append_only;
 2
 3
 
+query I
+select count(*) from fuse_snapshot('test_txn_stream','t_append_only');
+----
+1
+
 statement ok
 BEGIN;
 
 statement ok
 INSERT INTO t_append_only VALUES(4), (5);
 
+query I
+select count(*) from fuse_snapshot('test_txn_stream','t_append_only');
+----
+2
+
 statement ok
 INSERT INTO t_consume_append_only_1 SELECT a FROM s_append_only;
 
+query I
+select count(*) from fuse_snapshot('test_txn_stream','t_consume_append_only_1');
+----
+1
+
 statement ok
 INSERT INTO t_consume_append_only_2 SELECT a FROM s_append_only;
 
+query I
+select count(*) from fuse_snapshot('test_txn_stream','t_consume_append_only_2');
+----
+1
+
 statement ok
 INSERT INTO t_append_only VALUES(6), (7);
 
+query I
+select count(*) from fuse_snapshot('test_txn_stream','t_append_only');
+----
+2
+
 statement ok
 INSERT INTO t_consume_append_only_3 SELECT a FROM s_append_only;
 
+query I
+select count(*) from fuse_snapshot('test_txn_stream','t_consume_append_only_3');
+----
+1
+
 statement ok
 INSERT INTO t_consume_append_only_4 SELECT a FROM s_append_only_1;
 
+query I
+select count(*) from fuse_snapshot('test_txn_stream','t_consume_append_only_4');
+----
+1
+
 statement ok
 COMMIT;
 
+query I
+select count(*) from fuse_snapshot('test_txn_stream','t_append_only');
+----
+2
+
+query I
+select count(*) from fuse_snapshot('test_txn_stream','t_consume_append_only_1');
+----
+1
+
+query I
+select count(*) from fuse_snapshot('test_txn_stream','t_consume_append_only_2');
+----
+1
+
+query I
+select count(*) from fuse_snapshot('test_txn_stream','t_consume_append_only_3');
+----
+1
+
+query I
+select count(*) from fuse_snapshot('test_txn_stream','t_consume_append_only_4');
+----
+1
+
 statement ok
 INSERT INTO t_consume_append_only_5 SELECT a FROM s_append_only;
 
@@ -473,4 +533,3 @@ select str from target_2 order by str;
 
 statement ok
 drop database test_txn_stream;
-