feat: enhancing concurrent conflict resolution capability for multi-statement transactions

[SkyFan2002]

I hereby agree to the terms of the CLA available at: https://docs.databend.com/dev/policies/cla/

Summary

This PR improves the commit process for multi-statement transactions by detecting resolvable conflicts whenever possible when they arise with other concurrent transactions and retrying the commit.

Generally, append-only modifications to tables (e.g., INSERT/COPY INTO) do not conflict with other transactions. For other modifications (e.g., MERGE INTO, COMPACT), conflicts may be resolved in some scenarios, while in others, a rollback is required.

Example

1. Prepare target table:

root@localhost:8000/default> create or replace table t(c int) row_per_block = "10" block_per_segment = "1";

create
or replace table t(c int) row_per_block = "10" block_per_segment = "1"

0 row written in 0.168 sec. Processed 0 row, 0 B (0 row/s, 0 B/s)

root@localhost:8000/default> insert into t select number from numbers(10);

insert into
  t
select
  number
from
  numbers(10)

╭─────────────────────────╮
│ number of rows inserted │
│          UInt64         │
├─────────────────────────┤
│                      10 │
╰─────────────────────────╯
10 rows written in 0.224 sec. Processed 10 rows, 42 B (44.64 rows/s, 187 B/s)

root@localhost:8000/default> insert into t select number from numbers(20) where number > 9;

insert into
  t
select
  number
from
  numbers(20)
where
  number > 9

╭─────────────────────────╮
│ number of rows inserted │
│          UInt64         │
├─────────────────────────┤
│                      10 │
╰─────────────────────────╯
10 rows written in 0.103 sec. Processed 10 rows, 42 B (97.09 rows/s, 407 B/s)

root@localhost:8000/default> select * from fuse_snapshot('default','t');

select
  *
from
  fuse_snapshot('default', 't')

╭────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│ snapshot_i │ snapshot_ │ format_ve │ previous_ │ segment_c │ block_cou │ row_count │ bytes_unc │ bytes_com │ index_siz │ timestamp │
│      d     │  location │   rsion   │ snapshot_ │    ount   │     nt    │   UInt64  │ ompressed │  pressed  │     e     │ Nullable( │
│   String   │   String  │   UInt64  │     id    │   UInt64  │   UInt64  │           │   UInt64  │   UInt64  │   UInt64  │ Timestamp │
│            │           │           │ Nullable( │           │           │           │           │           │           │     )     │
│            │           │           │  String)  │           │           │           │           │           │           │           │
├────────────┼───────────┼───────────┼───────────┼───────────┼───────────┼───────────┼───────────┼───────────┼───────────┼───────────┤
│ 0196face15 │ 1/2234/_s │         4 │ 0196facdf │         2 │         2 │        20 │        84 │       796 │       876 │ 2025-05-2 │
│ 24771aa245 │ s/h0196fa │           │ ff6704586 │           │           │           │           │           │           │ 3 01:42:1 │
│ 5cb74523c8 │ ce1524771 │           │ 83e4efb29 │           │           │           │           │           │           │ 2.004000  │
│ 17         │ aa2455cb7 │           │ 4267d     │           │           │           │           │           │           │           │
│            │ 4523c817_ │           │           │           │           │           │           │           │           │           │
│            │ v4.mpk    │           │           │           │           │           │           │           │           │           │
│ 0196facdff │ 1/2234/_s │         4 │ NULL      │         1 │         1 │        10 │        42 │       398 │       438 │ 2025-05-2 │
│ f670458683 │ s/h0196fa │           │           │           │           │           │           │           │           │ 3 01:42:0 │
│ e4efb29426 │ cdfff6704 │           │           │           │           │           │           │           │           │ 6.582000  │
│ 7d         │ 58683e4ef │           │           │           │           │           │           │           │           │           │
│            │ b294267d_ │           │           │           │           │           │           │           │           │           │
│            │ v4.mpk    │           │           │           │           │           │           │           │           │           │
╰────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
2 rows read in 0.057 sec. Processed 2 rows, 400 B (35.09 rows/s, 6.85 KiB/s)

2. Execute a merge into statement on the target table in a multi statement transaction(T1), but do not commit:

root@localhost:8000/default> begin;

begin

0 row read in 0.006 sec. Processed 0 row, 0 B (0 row/s, 0 B/s)

root@localhost:8000/default> merge into t using (select number from numbers(10)) s on t.c = s.number when matched then update set c = 1;

merge into t using (
  select
    number
  from
    numbers(10)
) s on t.c = s.number
when matched then
update
set
  c = 1

╭────────────────────────╮
│ number of rows updated │
│         UInt64         │
├────────────────────────┤
│                     10 │
╰────────────────────────╯
1 row read in 0.297 sec. Processed 30 row, 164 B (101.01 rows/s, 552 B/s)

3. Execute another transaction(T2) on the same table in another session, and commit:

root@localhost:8000/default> begin;

begin

0 row read in 0.004 sec. Processed 0 row, 0 B (0 row/s, 0 B/s)

root@localhost:8000/default> merge into t using (select number from numbers(20) where number > 9) s on t.c = s.number when matched then update set c = 2;

merge into t using (
  select
    number
  from
    numbers(20)
  where
    number > 9
) s on t.c = s.number
when matched then
update
set
  c = 2

╭────────────────────────╮
│ number of rows updated │
│         UInt64         │
├────────────────────────┤
│                     10 │
╰────────────────────────╯
1 row read in 0.320 sec. Processed 30 row, 202 B (93.75 rows/s, 631 B/s)

root@localhost:8000/default> commit;

commit

0 row read in 0.025 sec. Processed 0 row, 0 B (0 row/s, 0 B/s)

4. Commit T1 and validating data in tables

root@localhost:8000/default> commit;

commit

0 row read in 0.042 sec. Processed 0 row, 0 B (0 row/s, 0 B/s)

root@localhost:8000/default> select * from t;

select
  *
from
  t

╭─────────────────╮
│        c        │
│ Nullable(Int32) │
├─────────────────┤
│               1 │
│               1 │
│               1 │
│               1 │
│               1 │
│               1 │
│               1 │
│               1 │
│               1 │
│               1 │
│               2 │
│               2 │
│               2 │
│               2 │
│               2 │
│               2 │
│               2 │
│               2 │
│               2 │
│               2 │
╰─────────────────╯
20 rows read in 0.071 sec. Processed 20 rows, 84 B (281.69 rows/s, 1.16 KiB/s)

root@localhost:8000/default> select count(*) from t where c = 1;

select
  count(*)
from
  t
where
  c = 1

╭──────────╮
│ COUNT(*) │
│  UInt64  │
├──────────┤
│       10 │
╰──────────╯
1 row read in 0.088 sec. Processed 10 row, 42 B (113.64 rows/s, 477 B/s)

root@localhost:8000/default> select count(*) from t where c = 2;

select
  count(*)
from
  t
where
  c = 2

╭──────────╮
│ COUNT(*) │
│  UInt64  │
├──────────┤
│       10 │
╰──────────╯
1 row read in 0.069 sec. Processed 10 row, 42 B (144.93 rows/s, 608 B/s)

Tests

• Unit Test
• Logic Test
• Benchmark Test
• No Test - Explain why

Type of change

• Bug Fix (non-breaking change which fixes an issue)
• New Feature (non-breaking change which adds functionality)
• Breaking Change (fix or feature that could cause existing functionality not to work as expected)
• Documentation Update
• Refactoring
• Performance Improvement
• Other (please describe):

---

This change is 

[SkyFan2002]

long run test suites have beed updated: datafuse-extras/test-scripts#7
And tested:

*******************************
Running test : explicit-txn...
*******************************
warning: `/home/sky/.cargo/config` is deprecated in favor of `config.toml`
note: if you need to support cargo 1.38 or earlier, you can symlink `config` to `config.toml`
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.08s
     Running `target/debug/test-scripts explicit-txn`
[2025-05-26T03:55:54Z INFO  test_scripts] using DSN databend://root:@localhost:8000/default?sslmode=disable&enable_experimental_merge_into=1
All tests passed!
*******************************
Test explicit-txn succeeded.
*******************************

[github-actions]

Docker Image for PR

• tag: pr-17984-06fbce0-1748346586

note: this image tag is only available for internal use.