PostgreSQL超大单行列文本读取内存不足，如何用psycopg2流式处理？

方案1：直接用PostgreSQL COPY命令导出到服务器文件 #

这是最直接的方式，跳过数据传输到Python的环节，让PostgreSQL直接将动态生成的SQL写入服务器本地文件：

-- 将动态生成的SQL导出到指定文件
COPY (SELECT <你的动态生成SQL的完整语句>) TO '/path/to/your/large_query.sql' WITH (FORMAT TEXT);

注意事项：

• PostgreSQL运行用户（通常为postgres）需要拥有目标路径的写入权限
• 如果是远程服务器，后续可通过SCP、SFTP等工具将文件下载到Python客户端，或直接在服务器端处理

Python端可分块读取文件避免内存溢出：

# 示例：分块读取本地大文件
with open('/path/to/your/large_query.sql', 'r', encoding='utf-8') as f:
    while chunk := f.read(1024*1024):  # 每次读取1MB
        # 按需处理片段，比如写入目标位置或执行
        pass

方案2：将大SQL拆分为多行，Python端流式拼接 #

把生成的大SQL字符串拆分为固定大小的片段（如1MB），返回多行结果，再通过Python游标逐行读取并拼接：

PostgreSQL拆分语句 #

SELECT regexp_split_to_table(
    <你的动态生成SQL的语句>,
    E'(?<=.{1048576})'  -- 正向断言，每1MB拆分一次，不截断字符
) AS sql_chunk;

Python端流式读取拼接 #

with self.getConnection() as conn:
    with conn.cursor(name='stream_cursor') as cursor:
        cursor.execute("""
            SELECT regexp_split_to_table(
                <你的动态生成SQL的语句>,
                E'(?<=.{1048576})'
            ) AS sql_chunk;
        """)
        # 直接写入文件避免内存占用
        with open('final_sql.sql', 'w', encoding='utf-8') as f:
            for chunk_row in cursor:
                f.write(chunk_row[0])

这种方式无需服务器额外权限，拆分与拼接逻辑完全在SQL和Python中完成，适合无法直接操作服务器文件的场景。

方案3：流式写入/读取PostgreSQL大对象（LO） #

此前lo_from_bytea因一次性转换全量字符串导致内存溢出，改用分块写入大对象的方式：

1. 创建分块写入大对象的PL/pgSQL函数 #

CREATE OR REPLACE FUNCTION write_large_sql_to_lo(p_sql text) RETURNS oid AS $$
DECLARE
    lo_oid oid;
    pos integer := 1;
    chunk_size integer := 1048576;  -- 每次写入1MB
    chunk text;
BEGIN
    lo_oid := lo_create(0);  -- 创建新大对象
    PERFORM lo_open(lo_oid, 131072);  -- 以写模式打开（O_WRONLY=0x20000）
    
    WHILE pos <= length(p_sql) LOOP
        chunk := substring(p_sql FROM pos FOR chunk_size);
        PERFORM lo_write(lo_oid, chunk::bytea);
        pos := pos + chunk_size;
    END LOOP;
    
    PERFORM lo_close(lo_oid);
    RETURN lo_oid;
END;
$$ LANGUAGE plpgsql;

2. 插入大对象OID到表中 #

INSERT INTO lo_sql_data (valuation_date, dynamic_sql)
SELECT '2023-08-31'::date, write_large_sql_to_lo(<你的动态生成SQL的语句>);

3. Python端分块读取大对象 #

with self.getConnection() as conn:
    with conn.cursor() as cursor:
        # 获取目标大对象的OID
        cursor.execute("SELECT dynamic_sql FROM lo_sql_data WHERE valuation_date = '2023-08-31';")
        lo_oid = cursor.fetchone()[0]
        
        # 流式读取大对象
        with conn.lobject(lo_oid, 'rb') as lo:
            with open('retrieved_sql.sql', 'wb') as f:
                while chunk := lo.read(1048576):  # 每次读取1MB
                    f.write(chunk)

这种方式适合需要将大SQL持久化存储在数据库中的场景，后续可随时通过OID读取。

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