1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424

import duckdb

conn = duckdb.connect(":memory:")
conn.execute("""
    CREATE TABLE accounts (
        id INTEGER, 
        name VARCHAR, 
        balance DECIMAL(10,2)
    )
""")
conn.execute("""
    INSERT INTO accounts VALUES 
    (1, 'Alice', 1000.00), 
    (2, 'Bob', 500.00)
""")

# 转账事务
conn.execute("BEGIN TRANSACTION")
conn.execute("UPDATE accounts SET balance = balance - 200 WHERE id = 1")
conn.execute("UPDATE accounts SET balance = balance + 200 WHERE id = 2")
conn.execute("COMMIT")

# 查看结果
print(conn.execute("SELECT * FROM accounts").fetchall())
# [(1, 'Alice', Decimal('800.00')), (2, 'Bob', Decimal('700.00'))]
```text


如果中途出错，可以回滚：

```python
conn.execute("BEGIN TRANSACTION")
conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
# 假设这里出错了
conn.execute("ROLLBACK")

# 数据恢复原状
print(conn.execute("SELECT * FROM accounts").fetchall())
# [(1, 'Alice', Decimal('800.00')), (2, 'Bob', Decimal('700.00'))]
```text


### 一致性（Consistency）

一致性靠约束检查保证。比如主键约束、外键约束、CHECK 约束，事务提交时会验证这些约束，违反约束的事务会被拒绝。

```python
conn.execute("""
    CREATE TABLE products (
        id INTEGER PRIMARY KEY,
        price DECIMAL(10,2) CHECK (price >= 0)
    )
""")

# 这条会失败，违反 CHECK 约束
conn.execute("BEGIN TRANSACTION")
try:
    conn.execute("INSERT INTO products VALUES (1, -10.00)")
    conn.execute("COMMIT")
except Exception as e:
    print(f"错误: {e}")
    conn.execute("ROLLBACK")
```text


### 隔离性（Isolation）

DuckDB 使用**快照隔离（Snapshot Isolation）**。每个事务开始时获得一个一致性快照，事务内的修改对其他事务不可见，直到提交。

```python
# 连接1开始事务
conn1 = duckdb.connect("test.db")
conn1.execute("BEGIN TRANSACTION")
conn1.execute("UPDATE accounts SET balance = 999 WHERE id = 1")

# 连接2读取数据（只读模式）
conn2 = duckdb.connect("test.db", config={"access_mode": "READ_ONLY"})
print(conn2.execute("SELECT balance FROM accounts WHERE id = 1").fetchone())
# (Decimal('800.00'),)  # 看不到连接1的修改

# 连接1提交
conn1.execute("COMMIT")

# 连接2现在能看到新值
print(conn2.execute("SELECT balance FROM accounts WHERE id = 1").fetchone())
# (Decimal('999.00'),)
```text


快照隔离可以防止：
- **脏读**：读到未提交的数据
- **不可重复读**：同一事务内两次读取结果不同
- **幻读**：同一查询返回新增的行

但有个限制：快照隔离不等于 SERIALIZABLE，存在 write skew 问题（两个事务同时读取相同数据，各自修改不同部分，导致逻辑不一致）。DuckDB 不支持 `SET TRANSACTION ISOLATION LEVEL`，无法切换到其他隔离级别。

### 持久性（Durability）

持久性靠 WAL（Write-Ahead Logging）保证。所有修改先写入 WAL 文件，提交时确保 WAL 记录持久化。即使数据库崩溃，重启后也能从 WAL 恢复。

```python
# 写入数据
conn = duckdb.connect("test.db")
conn.execute("CREATE TABLE test (id INTEGER)")
conn.execute("INSERT INTO test VALUES (1), (2), (3)")
conn.close()

# 重新连接，数据还在
conn = duckdb.connect("test.db")
print(conn.execute("SELECT * FROM test").fetchall())
# [(1,), (2,), (3,)]
```text


WAL 文件可以通过 `CHECKPOINT` 命令合并到主数据库：

```python
conn.execute("CHECKPOINT")
# WAL 文件清空，数据合并到主文件
```text


## 并发控制：乐观 vs 悲观

这是 DuckDB 跟传统数据库最大的区别。

### 传统数据库（悲观锁）

PostgreSQL、MySQL 使用悲观锁：
- 读操作获取共享锁（多个事务可以同时读）
- 写操作获取排他锁（其他事务必须等待）
- 冲突时阻塞等待，直到锁释放

### DuckDB（乐观并发控制）

DuckDB 使用**乐观并发控制（Optimistic Concurrency Control）**，灵感来自 Thomas Neumann 的论文《Fast Serializable Multi-Version Concurrency Control for Main-Memory Database Systems》。

工作原理：
1. 事务不持有锁，可以随时读写任何行
2. 更新数据时原地修改，但旧版本保存到 undo buffer
3. 提交时检查是否有冲突（其他事务也修改了同一行）
4. 有冲突就中止事务，没有冲突就提交

优势：
- 读查询不会被阻塞，适合分析型工作负载
- 没有锁管理的开销

劣势：
- 高并发写入场景冲突率高

### 实际测试：并发冲突率

我用 4 个线程同时更新同一行，每个线程 100 次：

```python
import threading
import duckdb

db_path = "conflict_test.db"
conn = duckdb.connect(db_path)
conn.execute("CREATE TABLE counter (id INTEGER, value INTEGER)")
conn.execute("INSERT INTO counter VALUES (1, 0)")
conn.close()

results = []

def update_counter(thread_id, iterations=100):
    conn = duckdb.connect(db_path)
    success = 0
    conflicts = 0
    for i in range(iterations):
        try:
            conn.execute("BEGIN TRANSACTION")
            current = conn.execute(
                "SELECT value FROM counter WHERE id = 1"
            ).fetchone()[0]
            conn.execute(
                f"UPDATE counter SET value = {current + 1} WHERE id = 1"
            )
            conn.execute("COMMIT")
            success += 1
        except Exception as e:
            if "conflict" in str(e).lower():
                conflicts += 1
            conn.execute("ROLLBACK")
    conn.close()
    return thread_id, success, conflicts

# 启动4个线程
threads = []
for i in range(4):
    t = threading.Thread(
        target=lambda tid=i: results.append(update_counter(tid, 100))
    )
    threads.append(t)
    t.start()

for t in threads:
    t.join()

# 打印结果
for thread_id, success, conflicts in results:
    print(f"线程{thread_id}: 成功{success}次, 冲突{conflicts}次")

# 最终值
conn = duckdb.connect(db_path)
final = conn.execute("SELECT value FROM counter WHERE id = 1").fetchone()[0]
print(f"最终计数器值: {final} (期望: 400)")
```text


运行结果：

```text
线程3: 成功26次, 冲突74次
线程1: 成功21次, 冲突79次
线程2: 成功29次, 冲突71次
线程0: 成功27次, 冲突73次

最终计数器值: 103 (期望: 400)
```text


4 个线程各 100 次更新，期望 400 次成功，实际只有 103 次。**冲突率 74%**。

这个结果说明 DuckDB 的乐观并发控制在高并发写入场景下表现不佳。如果业务场景有大量并发写入，需要实现重试机制。

### 重试机制实现

```python
import time

def transaction_with_retry(conn, max_retries=3):
    for attempt in range(max_retries):
        try:
            conn.execute("BEGIN TRANSACTION")
            # 执行操作
            current = conn.execute(
                "SELECT value FROM counter WHERE id = 1"
            ).fetchone()[0]
            conn.execute(
                f"UPDATE counter SET value = {current + 1} WHERE id = 1"
            )
            conn.execute("COMMIT")
            return True  # 成功
        except Exception as e:
            if "conflict" in str(e).lower():
                conn.execute("ROLLBACK")
                if attempt < max_retries - 1:
                    time.sleep(0.1 * (attempt + 1))  # 指数退避
                    continue
            raise
    return False  # 重试次数用尽
```text


## 性能优化

### 批量事务 vs 单条事务

事务本身有开销，批量操作可以显著减少这个开销。

```python
import duckdb
import time

conn = duckdb.connect(":memory:")
conn.execute("CREATE TABLE test (id INTEGER, value VARCHAR)")

# 方式1：1000条单独事务
start = time.time()
for i in range(1000):
    conn.execute("BEGIN TRANSACTION")
    conn.execute(f"INSERT INTO test VALUES ({i}, 'value{i}')")
    conn.execute("COMMIT")
single_time = time.time() - start

# 方式2：1000条批量事务
conn.execute("DELETE FROM test")
start = time.time()
conn.execute("BEGIN TRANSACTION")
for i in range(1000):
    conn.execute(f"INSERT INTO test VALUES ({i}, 'value{i}')")
conn.execute("COMMIT")
batch_time = time.time() - start

# 方式3：Prepared Statement 批量插入
conn.execute("DELETE FROM test")
start = time.time()
conn.execute("BEGIN TRANSACTION")
stmt = conn.prepare("INSERT INTO test VALUES (?, ?)")
for i in range(1000):
    stmt.execute([i, f"value{i}"])
conn.execute("COMMIT")
prepared_time = time.time() - start

print(f"单条事务: {single_time:.3f}秒")
print(f"批量事务: {batch_time:.3f}秒 ({single_time/batch_time:.1f}x)")
print(f"Prepared批量: {prepared_time:.3f}秒 ({single_time/prepared_time:.1f}x)")
```text


测试结果：

```text
单条事务: 0.718秒
批量事务: 0.346秒 (2.1x)
Prepared批量: 0.165秒 (4.3x)
```sql


批量事务比单条事务快 2.1 倍，Prepared Statement 批量插入快 4.3 倍。

### 性能优化建议

1. **减少事务数量**：把多个操作合并到一个事务
2. **使用 Prepared Statement**：减少 SQL 解析开销
3. **批量插入**：用 `INSERT INTO ... VALUES (...), (...), ...` 或 `COPY` 命令
4. **定期 CHECKPOINT**：控制 WAL 文件大小

## 踩坑记录

### 坑1：高并发写入冲突

**问题**：4 个线程并发更新同一行，冲突率 74%

**原因**：DuckDB 使用乐观并发控制，冲突时中止事务

**解决**：
- 实现重试机制（指数退避）
- 减少并发写入同一行
- 考虑使用队列串行化写入

### 坑2：不支持自定义隔离级别

**问题**：无法使用 `SET TRANSACTION ISOLATION LEVEL`

**原因**：DuckDB 只支持快照隔离

**影响**：无法选择更弱或更强的隔离级别

**解决**：接受快照隔离的限制，或考虑其他数据库

### 坑3：多进程不支持同时写入

**问题**：多个进程无法同时写入同一个数据库文件

**原因**：DuckDB 使用文件锁，只允许一个进程写入

**解决**：
- 使用单进程多线程
- 使用 `READ_ONLY` 模式多进程读取
- 考虑使用 MotherDuck（云端版本）

### 坑4：事务错误处理

**问题**：事务中发生错误后，需要手动 ROLLBACK

**原因**：DuckDB 不会自动回滚（除非某些特定错误）

**最佳实践**：

```python
try:
    conn.execute("BEGIN TRANSACTION")
    # 执行操作
    conn.execute("COMMIT")
except Exception as e:
    conn.execute("ROLLBACK")
    raise
```text

## 与其他数据库对比

| 特性 | DuckDB | PostgreSQL | MySQL | SQLite |
|------|--------|------------|-------|--------|
| 并发控制 | 乐观（MVCC） | 悲观（MVCC） | 悲观（锁） | 悲观（锁） |
| 隔离级别 | 快照隔离 | 4种可选 | 4种可选 | 可配置 |
| 多进程写入 | ❌ 不支持 | ✅ 支持 | ✅ 支持 | ❌ 不支持 |
| 事务冲突处理 | 中止事务 | 阻塞等待 | 阻塞等待 | 阻塞等待 |
| 适用场景 | 分析型（读多写少） | OLTP/OLAP | OLTP | 嵌入式 |

DuckDB 的设计目标是分析型工作负载，不是高并发 OLTP 系统。如果业务需要大量并发写入，PostgreSQL 或 MySQL 更合适。

## 最佳实践总结

### 什么时候用事务

- 多个操作需要原子性（要么全做，要么全不做）
- 需要保证数据一致性（约束检查）
- 需要隔离性（防止脏读、幻读）

### 什么时候不用事务

- 单条 SQL 语句（DuckDB 自动作为独立事务）
- 只读查询（不需要事务开销）
- 大批量数据导入（用 `COPY` 命令更高效）

### 并发写入策略

- 低并发（<10 线程）：直接使用，冲突率低
- 中并发（10-100 线程）：实现重试机制
- 高并发（>100 线程）：考虑队列串行化或使用其他数据库

### 性能调优清单

- [ ] 使用批量事务，减少事务数量
- [ ] 使用 Prepared Statement
- [ ] 定期执行 `CHECKPOINT`
- [ ] 避免长事务（减少 undo buffer 占用）
- [ ] 读多写少场景发挥 DuckDB 优势

## 参考资源

1. [DuckDB 官方文档 - Transaction Management](https://duckdb.org/docs/lts/sql/statements/transactions.html)
2. [DuckDB 官方博客 - Analytics-Optimized Concurrent Transactions](https://duckdb.org/2024/10/30/analytics-optimized-concurrent-transactions.html)
3. [DuckDB 并发文档](https://duckdb.org/docs/current/connect/concurrency.html)
4. [MotherDuck - ACID Transactions Explained](https://motherduck.com/learn/acid-transactions-sql)
5. [DuckDB Internals - The Transaction Lifecycle](https://www.alibabacloud.com/blog/duckdb-internals---part-5-the-transaction-lifecycle_602860)
6. [Fast Serializable Multi-Version Concurrency Control (VLDB 2015)](https://db.in.tum.de/~neumann/vldb2015.pdf)

---

这篇文章的代码都在腾讯云 DuckDB 测试环境跑过，性能数据都是真实测试结果。如果有任何疑问，欢迎留言讨论。