SQLite 合并流水后金额变少：UNION 隐式去重的定位与修复

场景与现象

离线对账任务把柜台、网店两张流水表合并，按客户统计净入账金额。源表各有一笔 10000 分的入账，合并后却只剩 10000 分；任务没有报错，单渠道报表也正常。

本文用 Python 标准库 sqlite3 复现，适用于两个来源分别保存独立业务事件的场景。金额使用整数分，退款用负数，不涉及汇率。源表内部主键唯一，不代表两个来源的编号全局唯一。

先检查合并行数，再检查 SUM

问题查询通常类似：把两张表都投影成客户号、金额，然后用 UNION 拼起来。若两笔独立入账刚好属于同一客户、金额也相同，投影结果完全一致，就只留下了一行。

按下面顺序定位更直接：

固定同一对账批次与一致的数据快照，分别记录两张源表的行数、金额。
暂时去掉外层聚合，直接看合并后的行及其数量。
检查 UNION 前是否已经丢掉来源和事件编号，检查其他 DISTINCT、GROUP BY 是否也在消除合法记录。
用同一数据分别执行 UNION 与 UNION ALL，对照行数和金额。不要只挑金额各不相同的样本。

SQLite 的 UNION 会消除最终结果中的重复行，UNION ALL 保留各分支的全部行；这里比较的是投影后的整行，不是业务事件身份。参见 SQLite 复合查询官方说明。因此，即使重复投影来自同一分支，UNION 也可能将它消除。

可直接运行的复现与修复

保存下面完整代码为 test_ledger_merge.py，执行 python test_ledger_merge.py。只创建内存数据库，不访问网络或生产文件。

import sqlite3
import unittest


SCHEMA = """
CREATE TABLE counter_entries (
    entry_id INTEGER PRIMARY KEY,
    customer_id INTEGER NOT NULL,
    amount_cents INTEGER NOT NULL
);
CREATE TABLE web_entries (
    entry_id INTEGER PRIMARY KEY,
    customer_id INTEGER NOT NULL,
    amount_cents INTEGER NOT NULL
);
"""

MERGED = """
SELECT customer_id, amount_cents FROM counter_entries
UNION ALL
SELECT customer_id, amount_cents FROM web_entries
"""


def totals(connection: sqlite3.Connection) -> list[tuple[int, int, int]]:
    """按独立流水累加，返回客户号、流水数与整数分净额。"""
    return connection.execute(
        f"""
        SELECT customer_id, COUNT(*), SUM(amount_cents)
        FROM ({MERGED})
        GROUP BY customer_id
        ORDER BY customer_id
        """
    ).fetchall()


class LedgerMergeTests(unittest.TestCase):
    """同值流水、退款、空输入与源表约束的回归检查。"""

    def setUp(self) -> None:
        self.connection = sqlite3.connect(":memory:")
        self.addCleanup(self.connection.close)
        self.connection.executescript(SCHEMA)

    def seed(
        self,
        counter_rows: list[tuple[int, int, int]],
        web_rows: list[tuple[int, int, int]],
    ) -> None:
        """来源固定，数据通过参数绑定写入。"""
        self.connection.executemany(
            "INSERT INTO counter_entries VALUES (?, ?, ?)", counter_rows
        )
        self.connection.executemany(
            "INSERT INTO web_entries VALUES (?, ?, ?)", web_rows
        )

    def test_equal_rows_across_sources(self) -> None:
        self.seed([(1, 7, 10000)], [(1, 7, 10000)])
        broken = MERGED.replace("UNION ALL", "UNION")
        self.assertEqual(self.connection.execute(broken).fetchall(), [(7, 10000)])
        self.assertEqual(totals(self.connection), [(7, 2, 20000)])

    def test_equal_rows_within_source(self) -> None:
        self.seed([(1, 7, 10000), (2, 7, 10000)], [])
        self.assertEqual(totals(self.connection), [(7, 2, 20000)])

    def test_refund_and_multiple_customers(self) -> None:
        self.seed([(1, 7, 10000), (2, 8, 3000)], [(1, 7, -2000)])
        self.assertEqual(totals(self.connection), [(7, 2, 8000), (8, 1, 3000)])

    def test_zero_amount_is_still_an_event(self) -> None:
        self.seed([(1, 7, 0)], [(1, 7, 0)])
        self.assertEqual(totals(self.connection), [(7, 2, 0)])

    def test_empty_sources(self) -> None:
        self.assertEqual(totals(self.connection), [])

    def test_duplicate_identity_in_source_is_rejected(self) -> None:
        self.seed([(1, 7, 10000)], [])
        with self.assertRaises(sqlite3.IntegrityError):
            self.seed([(1, 7, 10000)], [])
        self.assertEqual(totals(self.connection), [(7, 1, 10000)])


if __name__ == "__main__":
    print("SQLite:", sqlite3.sqlite_version)
    unittest.main(verbosity=2)

第一项测试同时固定了错误查询的表现与正确的业务结果：错误查询只保留一行，修复后为两笔、20000 分。第二项验证同一来源的两笔同额入账也必须保留；零金额事件仍计入流水数。空输入返回空列表，因为此处按客户分组，没有凭空生成客户行。

MERGED 是程序内固定 SQL 常量，插入值使用参数绑定；示例中的字符串插值只组合固定 SQL，不接收外部片段。若增加客户或批次筛选，应继续绑定参数，不能把请求字符串插入 SQL。

修复落地：保留流水身份与业务去重边界

对于“各来源都是真实独立流水”的模型，合并采用 UNION ALL，再按客户聚合。排查用的明细查询应带上来源和源编号，方便追溯。下面查询可在上述建表之后执行：

SELECT 'counter' AS source, entry_id, customer_id, amount_cents
FROM counter_entries
UNION ALL
SELECT 'web' AS source, entry_id, customer_id, amount_cents
FROM web_entries
ORDER BY source, entry_id;

本例的记录身份是 (source, entry_id)，两个来源都出现编号 1 并不冲突。不要改成 SUM(DISTINCT amount_cents)，那仍然会消除合法同额流水。

也不能把所有 UNION 批量替换成 UNION ALL。如果两个来源是同一笔支付的重复副本，必须先明确全局支付编号、租户范围、版本优先级和冲突处理规则，再进行业务去重。仅凭金额相同无法判定重复。当前代码只保证各源主键不重复，没有实现跨来源业务去重。

上线核对与回滚

发布前，用历史只读快照同时运行旧、新查询，核对每个客户的流水数和净额，并抽查差额对应的来源编号。对于当前无过滤模型，合并行数应等于两张源表行数之和；净额可能因退款抵消为零，所以不能只看总额。

将修复查询先用于影子报表，确认差额后切换正式报表。若下游已经按旧结果生成结算，先停止自动补差，保留旧报表、版本与修正明细，按业务审核流程重算，避免直接重复入账。回滚可切回旧查询版本，但旧查询仍存在漏算，回滚只能恢复程序版本，不能替代对账纠正。

这里的测试覆盖本地内存 SQLite 的行集合与聚合结果；不包含生产 ORM、并发读写快照、跨库抽取、海量流水性能和跨来源重复副本处理。实际查询带时间、租户或状态条件时，两个分支必须使用一致口径，不能把过滤不一致归因于 UNION。

总结

合并报表金额减少时，先比较聚合前的行数。UNION 按输出行去重，而合法流水的身份通常不止客户与金额两个字段。对独立事件使用 UNION ALL 保留记录，再单独设计业务去重与追溯规则，才能让修复结果经得起逐笔对账。

场景与现象

先检查合并行数，再检查 SUM

可直接运行的复现与修复

修复落地：保留流水身份与业务去重边界

上线核对与回滚

总结

相关阅读

评论