PostgreSQL表迁移后自增序列起始值错误致插入重复键问题求助

你当前的SQLAlchemy复制脚本能成功迁移表结构和数据，但自增列关联的序列起始值被重置为1，导致新插入数据时触发duplicate key value violates unique constraint错误。以下是具体解决方案：

问题复现 #

你的复制脚本及目标库序列创建语句如下：

Python复制脚本 #

from sqlalchemy import create_engine, MetaData, event
from sqlalchemy.sql import sqltypes
import traceback


src_engine = create_engine("postgresql://user1:mypass@myip1:2025/mydb?options=-c search_path=public")
src_metadata = MetaData(bind=src_engine)


tgt_engine = create_engine("postgresql://user2:mypass@myip2:2025/newdb?options=-c search_path=public")
tgt_metadata = MetaData(bind=tgt_engine)


@event.listens_for(src_metadata, "column_reflect")
def genericize_datatypes(inspector, tablename, column_dict):
    column_dict["type"] = column_dict["type"].as_generic(allow_nulltype=True)


tgt_conn = tgt_engine.connect()
tgt_metadata.reflect()


src_metadata.reflect()


for table in src_metadata.sorted_tables:
   table.create(bind=tgt_engine)


# refresh metadata before you can copy data
tgt_metadata.clear()
tgt_metadata.reflect()

# # Copy all data from src to target
for table in tgt_metadata.sorted_tables:
    src_table = src_metadata.tables[table.name]
    stmt = table.insert()
    temp_list = []
    source_table_count = src_engine.connect().execute(f"select count(*) from {table.name}").fetchall()
    current_row_count = 0

    for index, row in enumerate(src_table.select().execute()):
        temp_list.append(row._asdict())
        if len(temp_list) == 2500:
            stmt.execute(temp_list)
            current_row_count += 2500
            print(f"table = {table.name}, inserted {current_row_count} out of {source_table_count[0][0]}")
            temp_list = []
    if len(temp_list) > 0:
        stmt.execute(temp_list)
        current_row_count += len(temp_list)
        print(f"table = {table.name}, inserted {current_row_count} out of {source_table_count[0][0]}")
    print(f'source table "{table.name}": {source_table_count}')
    print(f'target table "{table.name}": {tgt_engine.connect().execute(f"select count(*) from {table.name}").fetchall()}')

目标库序列创建语句 #

CREATE SEQUENCE public.my_table_request_id_seq
    INCREMENT 1
    START 1
    MINVALUE 1
    MAXVALUE 2147483647
    CACHE 1;

ALTER SEQUENCE public.my_table_request_id_seq
    OWNER TO my_user;

解决方案 #

核心逻辑是：数据复制完成后，从源库获取每个自增序列的当前值，将目标库对应序列的起始值重置为源序列当前值+步长（确保下一个生成的ID不与已有数据冲突）。

1. 在Python脚本中添加序列同步逻辑 #

在数据复制的循环之后，添加以下代码块：

# 同步自增序列的当前值
with src_engine.connect() as src_conn, tgt_engine.connect() as tgt_conn:
    # 查询源库所有序列、关联表、当前值及步长
    seq_query = """
        SELECT 
            seq.relname AS sequence_name,
            tbl.relname AS table_name,
            seq.last_value,
            seq.increment_by
        FROM 
            pg_class seq
        JOIN 
            pg_depend dep ON seq.oid = dep.objid
        JOIN 
            pg_class tbl ON dep.refobjid = tbl.oid
        WHERE 
            seq.relkind = 'S'
            AND tbl.relkind = 'r'
            AND seq.schemaname = 'public';
    """
    seq_results = src_conn.execute(seq_query).fetchall()
    
    for seq_name, table_name, last_value, increment_by in seq_results:
        # 计算新的起始值：最后使用的值 + 步长
        restart_value = last_value + increment_by
        # 重置目标库序列
        alter_stmt = f"""
            ALTER SEQUENCE public.{seq_name}
            RESTART WITH {restart_value};
        """
        tgt_conn.execute(alter_stmt)
        print(f"序列 {seq_name}（关联表 {table_name}）已重置为从 {restart_value} 开始")

2. 验证同步结果 #

执行脚本后，在目标库执行以下SQL验证序列状态：

-- 查看序列当前值和步长
SELECT last_value, increment_by FROM public.my_table_request_id_seq;

-- 对比表中最大主键值
SELECT MAX(request_id) FROM public.my_table;

确保序列的last_value大于表中最大主键值。

关键说明 #

• 采用last_value + increment_by的原因：PostgreSQL序列的last_value是最后一次生成的ID，下一次生成的ID是last_value + increment_by，直接重置为该值可完全避免重复冲突。
• 该逻辑兼容自定义步长的序列，无需额外修改。

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