使用psycopg executemany批量插入时出现SSL连接错误排查

问题概述 #

用psycopg3批量插入带大文本字段的表时出现SSL连接中断，单条插入可成功但速度极慢，PostgreSQL日志提示“incomplete message from client”，升级psycopg版本后问题依旧。

代码与环境 #

原代码 #

type_to_postgres = {"<class 'int'>": 'integer',
                    "<class 'str'>": 'text',
                    "<class 'float'>": 'float',
                    "<class 'bool'>": 'bool',
                    "<class 'datetime.datetime'>": 'timestamp',
                    "<class 'datetime.date'>": 'date',
                    "<class 'pandas._libs.tslibs.timestamps.Timestamp'>": 'timestamp',
                    "<class 'NoneType'>": 'text'}
column_definitions = ''
for i in range(len(value_list[0])):
    column_definitions += keys[i] + ' ' + type_to_postgres.get(str(type(value_list[0][i])), 'text') + ', '
    column_definitions = column_definitions[:-2]
    try:
        cur.execute("DROP TABLE IF EXISTS tebra." + table_name.lower())
        cur.execute("CREATE TABLE tebra." + table_name.lower() + " (" + column_definitions + ");")
        cur.executemany("INSERT INTO tebra." + table_name.lower() + " VALUES (" + ", ".join(["%s"] * len(keys)) + ")", value_list)
        conn.commit()
    except Exception as e:
        print(e, table_name)

环境版本 #

• PostgreSQL 14.12（Ubuntu安装）
• psycopg 3.1.18 → 升级至3.2.1后问题未解决

报错信息 #

初始报错：

error ignored terminating <psycopg.Pipeline [ACTIVE, pipeline=ON] (host=10.1.10.77 database=postgres) at 0x258806e3c50>: flushing failed: No error (0x00000000/0)
SSL SYSCALL error: No error (0x00000000/0)
cannot exit pipeline mode while busy

升级后报错：

error ignored terminating <psycopg.Pipeline [BAD] at 0x2292b42be60>: the connection is lost
sending prepared query failed: SSL SYSCALL error: EOF detected

PostgreSQL日志：

2024-08-08 21:26:40.204 UTC [28654] postgres@postgres LOG:  incomplete message from client

解决方案 #

1. 禁用管道模式 #

psycopg3默认启用的管道模式对大文本字段的批量插入存在兼容性问题，关闭管道模式可避免连接中断：

# 连接时关闭管道
conn = psycopg.connect(
    dbname="postgres",
    host="10.1.10.77",
    user="your_user",
    password="your_pass",
    pipeline=False  # 关键：禁用管道模式
)

# 或创建cursor时指定
cur = conn.cursor(pipeline=False)

2. 拆分批量插入批次 #

将value_list拆分为更小的批次，减少单次发送的数据量：

batch_size = 100  # 根据实际情况调整
for i in range(0, len(value_list), batch_size):
    batch = value_list[i:i+batch_size]
    cur.executemany(
        f"INSERT INTO tebra.{table_name.lower()} VALUES ({', '.join(['%s']*len(keys))})",
        batch
    )
conn.commit()

3. 改用copy_from方法（推荐） #

copy_from是PostgreSQL批量导入的高效方式，对大文本字段支持更稳定：

from io import StringIO
import csv

# 将数据转换为CSV格式的字符串流
output = StringIO()
writer = csv.writer(output)
writer.writerows(value_list)
output.seek(0)

# 使用copy_from导入
cur.copy_from(
    output,
    f"tebra.{table_name.lower()}",
    columns=keys,
    sep=","  # 与CSV分隔符一致
)
conn.commit()

4. 调整PostgreSQL参数（可选） #

若上述方法无效，可尝试调整以下PostgreSQL参数（需重启服务）：

• max_allowed_packet：增大允许的数据包大小
• ssl_renegotiation_limit：调整SSL重协商的字节数，避免频繁重协商导致连接中断

补充说明 #

问题大概率与表中的大文本字段filecontents有关，单条插入时数据量小不会触发连接限制，但批量插入时数据包超过阈值导致SSL连接异常。禁用管道模式或改用copy_from是最直接的解决方式。

内容的提问来源于stack exchange，提问作者Alex Childs