写偏斜

工作原理

假设规则要求至少一名医生保持值班。Alice 和 Bob 都在值班。事务 A 数到两名医生，判断可以移除 Alice，并更新 Alice 的行；事务 B 数到两名医生，判断可以移除 Bob，并更新 Bob 的行。两个写入作用于不同行，快照隔离看不到冲突，两个事务都提交，结果零名医生值班。

异常之所以存在，是因为决策依赖跨多行的谓词，而写入互不相交。可重复读给每个事务一个稳定快照，却不会串行化谓词。可串行化隔离会跟踪读写依赖，并以序列化失败中止其中一个事务，应用必须重试。

修复方法

使用可串行化隔离并重试序列化失败。或者物化冲突：用 SELECT ... FOR UPDATE 锁定代表值班组的共享行，让第二个事务等待并重新读取。数据库约束只有在不变量能表达在单行或唯一索引上时才有帮助。应用层锁可行，但增加协调和故障模式。

何时关注

凡是事务读取多行以执行规则、然后写入其中一部分的地方，都要检查写偏斜：座位分配、库存预留、带透支限额的账户余额、权限计数和排班。单行计数器和唯一键插入通常安全；多行不变量才是危险区。

权衡与示例

可串行化隔离是最稳健的修复，但增加中止，因此每个事务都需要有界重试循环和幂等副作用。显式锁减少中止，却带来争用和死锁风险。约束最便宜，但只覆盖模式能表达的不变量。

示例：订票系统允许每个活动最多 100 个有效预订。两个事务各自数到 99，然后插入不同的预订。如果没有谓词锁或可串行化检查，两者都会提交，活动超卖。在计数前锁定活动行，可以串行化决策并保住不变量。

代码示例

from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

engine = create_engine(
    'postgresql+psycopg://user:pass@localhost/reporting',
    isolation_level='READ COMMITTED',
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,
)

reporting_engine = engine.execution_options(
    isolation_level='REPEATABLE READ'
)


def transfer_funds(session: Session, src: int, dst: int, amount: int) -> None:
    session.execute(
        text('UPDATE accounts SET balance = balance - :amt WHERE id = :id'),
        {'amt': amount, 'id': src},
    )
    row = session.execute(
        text('SELECT balance FROM accounts WHERE id = :id'),
        {'id': src},
    ).scalar_one()
    if row < 0:
        raise ValueError('overdraft rejected')
    session.execute(
        text('UPDATE accounts SET balance = balance + :amt WHERE id = :id'),
        {'amt': amount, 'id': dst},
    )


with Session(engine) as session:
    with session.begin():
        transfer_funds(session, src=1, dst=2, amount=75)

with Session(reporting_engine) as session:
    snapshot = session.execute(
        text('SELECT sum(balance) FROM accounts')
    ).scalar_one()
    print('consistent snapshot total:', snapshot)

import csv
import io


def detect_dialect(sample: str) -> dict[str, object]:
    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t|")
    return {
        "delimiter": dialect.delimiter,
        "quotechar": dialect.quotechar,
        "doublequote": dialect.doublequote,
        "skipinitialspace": dialect.skipinitialspace,
    }


def read_rows(text: str) -> list[dict[str, str]]:
    dialect = csv.Sniffer().sniff(text, delimiters=",;\t|")
    reader = csv.DictReader(io.StringIO(text), dialect=dialect)
    return [dict(row) for row in reader]


if __name__ == "__main__":
    sample = "name;role;city\nAda;engineer;London\nGrace;admiral;New York\n"
    print(detect_dialect(sample))
    print(read_rows(sample))