如何用Python(Psycopg2)实现Postgres多列分区及已有分区数据插入

我需要对Postgres表进行双列分区，并使用Python（Psycopg2）向已存在的分区插入数据。作为新手，遇到了Postgres不支持基于多列的List分区的问题。

现有两张结构相同的表：cust_details_curr和cust_details_hist，其中cust_details_hist需要按area_code和eff_date两列分区，表结构如下：

CREATE TABLE cust_details_curr
(
 cust_id int,
 area_code varchar(5),
 cust_name varchar(20),
 cust_age int,
 eff_date date
);

CREATE TABLE cust_details_hist
(
 cust_id int,
 area_code varchar(5),
 cust_name varchar(20),
 cust_age int,
 eff_date date
); -- 需要按area_code和eff_date分区

需求说明 #

• area_code作为程序参数传入，eff_date为当前程序运行日期；
• 程序需依次处理多个area_code（如A501、A502等），同一日期运行的所有任务eff_date相同；
• 每次加载curr表对应area_code的数据前，需先将该area_code在curr表的现有数据迁移到hist表对应area_code和eff_date的分区，再删除curr表对应数据并加载新数据（新数据的eff_date为当前运行日期）。

疑问 #

1. 如何为cust_details_hist表实现基于area_code和eff_date的双列分区？
2. 当同一eff_date的分区已被某area_code（如A501）创建并加载数据后，如何将另一area_code（如A502）的数据插入该已存在的分区？

之前实现过基于eff_date单列分区的方案，但无法扩展到双列分区场景，相关简化代码如下：

CREATE TABLE cust_details_curr
(
 cust_id int,
 area_code varchar(5),
 cust_name varchar(20),
 cust_age int,
 eff_date date
);

CREATE TABLE cust_details_hist
(
 cust_id int,
 area_code varchar(5),
 cust_name varchar(20),
 cust_age int,
 eff_date date
) PARTITIONED BY LIST (eff_dt); -- 按List分区

table_name = "cust_details_curr"
table_name_hist = table_name + '_hist'

e = datetime.now()
eff_date = e.strftime("%Y-%m-%d")
dttime = e.strftime("%Y%m%d_%H%M%S")

table_name_curr_part = table_name_part + '_' + str(dttime)

query_count = f"SELECT count(*) as cnt from {table_name} where area_code = '{area_code}'; "
query_date = f"SELECT distinct eff_date as eff_dt from {table_name} where area_code = '{area_code}';"

cur.execute(quey_date)
eff_date = cur.fetchone()[0]

query_crt = f"CREATE TABLE {table_name_curr_part} LIKE {table_name_part} INCLUDING DEFAULTS);"
query_ins_part = f"INSERT INTO {table_name_curr_part} SELECT * FROM {table_name} where area_code = '{area_code}' AND eff_dt = '{eff_date}';"
query_add_part = f"ALTER TABLE {table_name_part} ATTACH PARTITION {table_name_curr_part} FOR VALUES IN (DATE '{eff_date}') ;"
query_del = f"DELETE FROM {table_name} WHERE area_code = '{area_code}';"
query_ins_curr = f"INSERT INTO {table_name} (cust_id, area_code, cust_name, cust_age, eff_dt) VALUES %s"

cur.execute(....)
# 代码已简化

1. 双列分区的实现方案 #

Postgres不支持多列List分区，但可以通过嵌套分区（先按eff_date做List分区，再在每个日期分区下按area_code做List分区）实现，逻辑清晰且符合Postgres分区规则：

步骤1：创建主分区表 #

以eff_date为顶级List分区键创建主表：

CREATE TABLE cust_details_hist
(
 cust_id int,
 area_code varchar(5),
 cust_name varchar(20),
 cust_age int,
 eff_date date
) PARTITIONED BY LIST (eff_date);

步骤2：创建日期级分区表 #

针对每个eff_date创建分区表，且该表再按area_code做List分区：

-- 示例：针对2024-05-20创建日期分区
CREATE TABLE cust_details_hist_20240520
PARTITION OF cust_details_hist
FOR VALUES IN ('2024-05-20')
PARTITIONED BY LIST (area_code);

步骤3：创建区域级子分区 #

在日期分区下，为每个area_code创建子分区：

-- 示例：为A501区域创建子分区
CREATE TABLE cust_details_hist_20240520_a501
PARTITION OF cust_details_hist_20240520
FOR VALUES IN ('A501');

-- 示例：为A502区域创建子分区
CREATE TABLE cust_details_hist_20240520_a502
PARTITION OF cust_details_hist_20240520
FOR VALUES IN ('A502');

这种嵌套结构会自动将数据路由到对应的「日期+区域」子分区，满足双列分区需求。

2. 向已存在的分区插入数据 #

当日期分区已存在时，只需检查对应area_code的子分区是否存在，不存在则创建，之后直接向主表或子分区插入数据即可：

Python（Psycopg2）实现代码 #

import psycopg2
import psycopg2.extras
from datetime import datetime

# 数据库连接参数，根据实际情况修改
conn_params = {
    "dbname": "your_db",
    "user": "your_user",
    "password": "your_password",
    "host": "your_host"
}

def migrate_and_load(area_code):
    conn = psycopg2.connect(**conn_params)
    cur = conn.cursor()
    try:
        # 获取当前运行日期，格式化用于分区表命名
        current_date = datetime.now().strftime("%Y-%m-%d")
        date_part_name = f"cust_details_hist_{current_date.replace('-', '')}"
        area_part_name = f"{date_part_name}_{area_code.lower()}"

        # 1. 检查日期分区是否存在，不存在则创建
        cur.execute("""
            SELECT EXISTS (
                SELECT 1 FROM pg_tables 
                WHERE tablename = %s
            )
        """, (date_part_name,))
        if not cur.fetchone()[0]:
            create_date_part_sql = f"""
                CREATE TABLE {date_part_name}
                PARTITION OF cust_details_hist
                FOR VALUES IN ('{current_date}')
                PARTITIONED BY LIST (area_code);
            """
            cur.execute(create_date_part_sql)

        # 2. 检查区域子分区是否存在，不存在则创建
        cur.execute("""
            SELECT EXISTS (
                SELECT 1 FROM pg_tables 
                WHERE tablename = %s
            )
        """, (area_part_name,))
        if not cur.fetchone()[0]:
            create_area_part_sql = f"""
                CREATE TABLE {area_part_name}
                PARTITION OF {date_part_name}
                FOR VALUES IN ('{area_code}');
            """
            cur.execute(create_area_part_sql)

        # 3. 迁移curr表数据到hist表（Postgres自动路由到对应子分区）
        migrate_sql = """
            INSERT INTO cust_details_hist
            SELECT * FROM cust_details_curr
            WHERE area_code = %s;
        """
        cur.execute(migrate_sql, (area_code,))

        # 4. 删除curr表对应区域的数据
        delete_sql = """
            DELETE FROM cust_details_curr
            WHERE area_code = %s;
        """
        cur.execute(delete_sql, (area_code,))

        # 5. 批量插入新数据到curr表（示例数据，实际替换为你的数据源）
        new_data = [
            (1001, area_code, "John Doe", 30, current_date),
            (1002, area_code, "Jane Smith", 28, current_date)
        ]
        insert_curr_sql = """
            INSERT INTO cust_details_curr (cust_id, area_code, cust_name, cust_age, eff_date)
            VALUES %s;
        """
        psycopg2.extras.execute_values(cur, insert_curr_sql, new_data)

        conn.commit()
        print(f"处理完成：区域{area_code}，迁移{cur.rowcount}条历史数据，插入{len(new_data)}条新数据")
    except Exception as e:
        conn.rollback()
        print(f"区域{area_code}处理失败：{str(e)}")
    finally:
        cur.close()
        conn.close()

# 批量处理多个area_code
for area in ["A501", "A502"]:
    migrate_and_load(area)

关键说明 #

• 使用pg_tables系统表检查分区存在性，避免重复创建；
• 插入数据时直接操作主表cust_details_hist，Postgres会自动根据eff_date和area_code路由到对应子分区；
• 使用psycopg2.extras.execute_values批量插入新数据，提升效率；
• 所有操作包裹在事务中，保证数据一致性。

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