Python导入CSV至PostgreSQL时UniqueViolation异常排查

问题场景 #

• 通过Python将10万+行数据的CSV文件导入PostgreSQL，成功导入的文件归档，异常文件留存
• 对比归档与异常CSV数据未发现明显差异：归档数据可在PgAdmin中查询到，异常数据用相同查询却查不到
• 调试出现矛盾行为：select exists查询返回False，但执行插入时触发UniqueViolation；且该数据可在PgAdmin中手动插入，仅部分文件出现此问题

调试输出 #

query="select exists(select time from history_eurusd_part_table where time='2017-10-03 00:00:00.183000')"
result=False
query="insert into history_eurusd_part_table (time,ask,bid,ask_volume,bid_volume) values ('2017-10-03 00:00:00.183000',1.17321,1.17319,1310000,1000000)"
UniqueViolation: data="'2017-10-03 00:00:00.183000',1.17321,1.17319,1310000,1000000", e=UniqueViolation('duplicate key value violates unique constraint "history_eurusd_2017_pkey"\nDETAIL:  Key ("time")=(2017-10-03 00:00:00.183) already exists.\n')

表结构 #

time TIMESTAMP PRIMARY KEY,
ask DECIMAL(8, 5),
bid DECIMAL(8,5),
ask_volume INT,
bid_volume INT

核心代码 #

导入逻辑片段 #

query = f"select exists(select {fieldnames[0]} from {tableVals['name']} where {fieldnames[0]}={row[0]})"
print(f"{query=}")
success = Attempt_Query(cur, query, [])
# If check if entry exists query executes successfully
if success:
    result = cur.fetchone()[0]
    print(f"{result=}")
    # If entry doesn't exist in db, insert
    if not result:
        data = ",".join(row)
        query = f"insert into {tableVals['name']} ({columns}) values ({data})"
        print(f"{query=}")
        # If insert fails, return false to avoid archiving
        if not Attempt_Query(cur, query, data):
            csvInDB = False

查询执行函数 #

def Attempt_Query(cur, query, data):
    cur.execute(query)

原因分析与解决方法 #

1. 竞态条件（并发问题） #

exists查询和insert是两个独立的SQL语句，中间存在时间间隙。如果有其他进程/线程在这两个操作之间插入了相同的time值，就会出现“检查时不存在，插入时已存在”的冲突。

解决：使用原子化的INSERT ... ON CONFLICT语句，将检查与插入合并为一步操作，彻底避免竞态：

# 替换原有的exists查询+insert逻辑
placeholders = ", ".join(["%s"] * len(row))
query = f"INSERT INTO {tableVals['name']} ({columns}) VALUES ({placeholders}) ON CONFLICT DO NOTHING"
cur.execute(query, row)
# 通过rowcount判断是否成功插入
if cur.rowcount == 0:
    # 数据已存在，无需处理
    pass

2. 时间戳精度不一致 #

报错提示中存在的时间是2017-10-03 00:00:00.183（毫秒级），但查询使用的是2017-10-03 00:00:00.183000（微秒级）。PostgreSQL的TIMESTAMP默认支持6位微秒精度，但如果分区表的时间字段被隐式截断精度，会导致查询匹配不到，但实际存在低精度的重复值。

验证：执行以下查询确认：

SELECT EXISTS(SELECT time FROM history_eurusd_part_table WHERE time = '2017-10-03 00:00:00.183');

如果返回True，则说明是精度问题。

解决：确保查询与插入的时间戳精度一致，或查询时显式截断精度：

# 查询时将时间戳截断到毫秒
query = f"SELECT EXISTS(SELECT {fieldnames[0]} FROM {tableVals['name']} WHERE DATE_TRUNC('milliseconds', {fieldnames[0]}) = DATE_TRUNC('milliseconds', %s))"
cur.execute(query, (row[0],))

3. 字符串拼接导致的SQL问题 #

代码直接用字符串拼接生成SQL，存在SQL注入风险，同时可能因数据格式处理不当导致查询匹配错误（例如特殊字符、未正确加引号等）。

解决：改用参数化查询，避免字符串拼接：

# 检查数据是否存在的参数化查询
query = f"SELECT EXISTS(SELECT {fieldnames[0]} FROM {tableVals['name']} WHERE {fieldnames[0]} = %s)"
cur.execute(query, (row[0],))
result = cur.fetchone()[0]

4. 分区表路由问题 #

从约束名history_eurusd_2017_pkey可判断这是分区表，若查询未正确命中子分区，会出现“查询不到但插入时冲突”的情况——因为插入操作会自动路由到对应子分区，而查询可能未覆盖该子分区。

验证：直接查询对应子分区：

SELECT time FROM history_eurusd_part_table PARTITION (history_eurusd_2017) WHERE time = '2017-10-03 00:00:00.183';

若能查到数据，说明是分区路由问题。

解决：确保查询覆盖所有相关子分区，或直接使用INSERT ... ON CONFLICT自动处理分区路由。

内容的提问来源于stack exchange，提问作者Andrew Powers