如何用Python动态规划实现Join顺序优化：解决高成本及预计算问题

项目背景 #

我正在开发一个通过Python动态规划确定最优Join顺序来优化SQL查询的项目，连接PostgreSQL数据库，需要计算不同表组合的Join成本，已编写表扫描成本和两表集合间Join成本的计算函数。

当前代码与运行结果 #

代码节选 #

import psycopg2
import json

conn = psycopg2.connect(
    dbname="dbname",
    user="postgres",
    password="password",
    host="localhost",
    port="port"
)

def calculate_cost(solution, conn):
    total_cost = 0
    try:
        with conn.cursor() as cur:
            for table in solution:
                cur.execute(f"EXPLAIN (ANALYZE, FORMAT JSON) SELECT * FROM {table};")
                explain_result = cur.fetchone()[0]
                explain_json = json.loads(explain_result) if isinstance(explain_result, str) else explain_result
                total_cost += float(explain_json['Plan']['Total Cost'])
        conn.commit()
    except Exception as e:
        print(f"Error calculating cost: {e}")
        conn.rollback()
    return total_cost

def remove_lower_cost_equivalents(partial_solutions, conn):
    min_cost = float('inf')
    best_solutions = set()
    for solution in partial_solutions:
        total_cost = calculate_cost(solution, conn)
        if total_cost < min_cost:
            min_cost = total_cost
            best_solutions = {solution}
        elif total_cost == min_cost:
            best_solutions.add(solution)
    return best_solutions

def DynamicProgramming(Rels, conn):
    partialsolutions = {frozenset({rel}) for rel in Rels}
    for i in range(1, len(Rels)):
        new_partial_solutions = set()
        for partial_tree in partialsolutions:
            for rel in Rels:
                if rel not in partial_tree:
                    new_partial_tree = partial_tree.union({rel})
                    new_partial_solutions.add(new_partial_tree)
        partialsolutions = remove_lower_cost_equivalents(new_partial_solutions, conn)
    optimal_solution = next(iter(partialsolutions))
    optimal_cost = calculate_cost(optimal_solution, conn)
    optimal_order = list(optimal_solution)
    return len(Rels), optimal_cost, optimal_order

# Running the code
relations = ['title', 'movie_info', 'complete_cast']
nb_relations, cost, optimal_order = DynamicProgramming(relations, conn)
print(f"Number of relations: {nb_relations}")
print(f"Optimal cost: {cost}")
print(f"Optimal order: {optimal_order}")
conn.close()

运行结果 #

nb_Relation,DP_Cost,Runtime,Optimal_Order
3,253206.22,30.894135236740112,"['complete_cast', 'title', 'movie_info']"

现存问题 #

当前实现存在两个核心问题：

• Join成本计算失真，结果过高：现有calculate_cost仅累加单表扫描成本，未考虑真实的Join逻辑，甚至可能触发笛卡尔积，导致成本完全不符合实际。
• 无预计算存储机制：每次计算都重复调用数据库，效率低下，也无法复用历史计算结果。

改进方案 #

一、修正成本计算逻辑，获取真实Join成本 #

核心问题点

现有代码的成本计算是错误的：Join的成本不是单表扫描成本的简单累加，而是多表按关联条件Join后的实际执行成本，且使用EXPLAIN ANALYZE会实际执行查询，耗时极长。

改进后的成本计算函数

def calculate_cost(table_set, conn, join_conditions):
    # 单表直接计算扫描成本
    if len(table_set) == 1:
        table = next(iter(table_set))
        with conn.cursor() as cur:
            cur.execute(f"EXPLAIN (FORMAT JSON) SELECT * FROM {table};")
            # PostgreSQL的EXPLAIN JSON返回是数组结构，注意索引
            explain_json = cur.fetchone()[0][0]
            return float(explain_json['Plan']['Total Cost'])
    
    # 多表生成带关联条件的Join查询
    tables = list(table_set)
    sql = f"SELECT * FROM {tables[0]}"
    # 遍历后续表，拼接Join条件
    for idx in range(1, len(tables)):
        current_table = tables[idx]
        prev_table = tables[idx-1]
        # 从预设的关联条件字典中获取对应规则
        join_pair = frozenset({prev_table, current_table})
        if join_pair not in join_conditions:
            raise ValueError(f"未找到{prev_table}与{current_table}的关联条件")
        sql += f" JOIN {current_table} ON {join_conditions[join_pair]}"
    
    with conn.cursor() as cur:
        cur.execute(f"EXPLAIN (FORMAT JSON) {sql};")
        explain_json = cur.fetchone()[0][0]
        return float(explain_json['Plan']['Total Cost'])

关键优化点

• 使用真实的表关联条件：必须传入表之间的Join规则（比如join_conditions = {frozenset({'title','movie_info'}): 'title.movie_id = movie_info.movie_id'}），避免数据库执行笛卡尔积。
• 替换EXPLAIN ANALYZE为EXPLAIN：仅获取数据库的成本估算，无需实际执行查询，大幅提升计算速度。

二、预计算并存储所有表组合的Join成本 #

动态规划优化Join顺序的核心是复用子集的最优计算结果，可以通过字典存储所有表子集的最小成本与拆分方式，实现预计算复用。

改进后的动态规划函数

def DynamicProgramming(Rels, conn, join_conditions):
    # DP字典：key为表的frozenset，value为(最小成本, 最优拆分方式)
    dp = {}
    # 初始化单表成本
    for rel in Rels:
        s = frozenset({rel})
        dp[s] = (calculate_cost(s, conn, join_conditions), None)
    
    total_tables = len(Rels)
    # 遍历子集大小从2到总表数
    for subset_size in range(2, total_tables + 1):
        from itertools import combinations
        # 生成所有size为subset_size的表组合
        for subset_tuple in combinations(Rels, subset_size):
            subset = frozenset(subset_tuple)
            min_total_cost = float('inf')
            best_split = None
            
            # 遍历所有可能的拆分方式：将子集拆分为两个非空子集
            first_rel = next(iter(subset))
            other_rels = subset - {first_rel}
            # 拆分出包含first_rel的子集，大小从1到subset_size-1
            for split_size in range(1, subset_size):
                for split_tuple in combinations(other_rels, split_size - 1):
                    subset_a = frozenset({first_rel}) | frozenset(split_tuple)
                    subset_b = subset - subset_a
                    
                    # 计算当前拆分的总成本
                    current_cost = calculate_cost(subset, conn, join_conditions)
                    if current_cost < min_total_cost:
                        min_total_cost = current_cost
                        best_split = (subset_a, subset_b)
            
            dp[subset] = (min_total_cost, best_split)
    
    # 获取全表组合的最优结果
    full_set = frozenset(Rels)
    optimal_cost = dp[full_set][0]
    optimal_order = get_join_order(full_set, dp)
    return total_tables, optimal_cost, optimal_order

def get_join_order(subset, dp):
    # 递归回溯拆分方式，生成嵌套的Join顺序
    split = dp[subset][1]
    if split is None:
        return list(subset)
    subset_a, subset_b = split
    return [get_join_order(subset_a, dp), get_join_order(subset_b, dp)]

预计算存储方案

可以将dp字典的结果序列化存储到本地文件（如JSON），后续优化查询时直接读取，无需重复调用数据库计算：

import json

# 存储预计算结果
with open('join_cost_cache.json', 'w') as f:
    # 将frozenset转为列表以便序列化
    serializable_dp = {str(list(k)): v for k, v in dp.items()}
    json.dump(serializable_dp, f)

# 读取预计算结果
with open('join_cost_cache.json', 'r') as f:
    loaded_dp = json.load(f)
    # 将列表转回frozenset
    dp = {frozenset(eval(k)): v for k, v in loaded_dp.items()}

总结 #

• 核心改进是使用真实Join条件计算成本，避免笛卡尔积导致的虚假高成本；
• 通过动态规划的DP字典存储所有子集的最优成本，实现预计算与结果复用，提升效率；
• 移除EXPLAIN ANALYZE改用EXPLAIN，大幅降低计算耗时。

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