Presto SQL场景下多级BOM物料总用量计算方案咨询

针对Presto不支持递归查询的场景，以下提供两种可落地的实现方案：

方案1：Presto SQL 固定层级展开 #

适用场景：BOM最大层级可预估（大多数生产场景BOM层级不超过10级）

• 实现思路：通过多层CTE自关联逐层传递用量，最终合并所有层级的计算结果
• 示例代码：

-- 假设BOM表名为bom_table
WITH 
-- 提取顶层父件：无上级父件的物料，总用量直接取bom_qty
top_level AS (
    SELECT 
        parent_number,
        part_number,
        bom_qty,
        bom_qty AS total_bom_qty,
        1 AS bom_level
    FROM bom_table
    WHERE parent_number NOT IN (SELECT DISTINCT part_number FROM bom_table WHERE lowest_level = 'N')
),
-- 计算二级BOM用量
level_2 AS (
    SELECT 
        b.parent_number,
        b.part_number,
        b.bom_qty,
        t.total_bom_qty * b.bom_qty AS total_bom_qty,
        2 AS bom_level
    FROM bom_table b
    INNER JOIN top_level t ON b.parent_number = t.part_number
),
-- 计算三级BOM用量，以此类推直到覆盖最大层级
level_3 AS (
    SELECT 
        b.parent_number,
        b.part_number,
        b.bom_qty,
        l2.total_bom_qty * b.bom_qty AS total_bom_qty,
        3 AS bom_level
    FROM bom_table b
    INNER JOIN level_2 l2 ON b.parent_number = l2.part_number
)
-- 合并所有层级结果
SELECT * FROM top_level
UNION ALL
SELECT * FROM level_2
UNION ALL
SELECT * FROM level_3
-- 后续新增层级只需继续添加UNION ALL对应层级的CTE即可

• 注意事项：可先测试当前使用的Presto版本是否支持递归CTE（部分高版本Presto/Trino已支持递归语法），如果支持可直接用递归CTE实现动态层级计算，无需固定层级展开。

方案2：Python 拓扑排序实现 #

适用场景：BOM层级不固定、层级数较多的场景

• 实现思路：将BOM视为有向无环图，通过拓扑排序按从顶层到底层的顺序遍历节点，自动传递父件总用量到子件，子件转为父件时直接读取已计算好的总用量即可。
• 示例代码：

import pandas as pd
from collections import defaultdict, deque

# 读取BOM数据，可替换为从数据库/数仓读取的逻辑
bom_df = pd.read_csv("bom_data.csv")

# 初始化数据结构
adjacency = defaultdict(list)  # BOM邻接表：key=父件编码，value=子件列表(子件编码, 单级用量)
in_degree = defaultdict(int)   # 节点入度，用于拓扑排序
part_is_parent = dict()        # 存储物料是否可作为父件
top_parent_qty = dict()        # 存储顶层父件的初始用量

# 遍历原始数据构建结构
for _, row in bom_df.iterrows():
    parent = row["parent_number"]
    part = row["part_number"]
    qty = row["bom_qty"]
    is_parent = row["lowest_level"] == "N"
    
    adjacency[parent].append((part, qty))
    in_degree[part] += 1
    part_is_parent[part] = is_parent
    # 记录顶层父件的初始用量
    if parent not in part_is_parent:
        top_parent_qty[parent] = qty

# 初始化拓扑队列：加入所有顶层父件
calc_queue = deque()
part_total_qty = dict()
for top_parent, init_qty in top_parent_qty.items():
    part_total_qty[top_parent] = init_qty
    calc_queue.append(top_parent)

# 遍历计算所有物料总用量
result = []
while calc_queue:
    current_parent = calc_queue.popleft()
    current_parent_total = part_total_qty[current_parent]
    # 遍历当前父件的所有子件
    for child_part, bom_qty in adjacency.get(current_parent, []):
        child_total = current_parent_total * bom_qty
        # 保存计算结果
        result.append({
            "parent_number": current_parent,
            "part_number": child_part,
            "bom_qty": bom_qty,
            "total_bom_qty": child_total
        })
        # 如果子件可作为父件，保存其总用量供后续计算使用
        if part_is_parent.get(child_part, False):
            part_total_qty[child_part] = child_total
        # 更新入度，入度为0表示当前子件的所有父件都已计算完成，加入队列
        in_degree[child_part] -= 1
        if in_degree[child_part] == 0:
            calc_queue.append(child_part)

# 输出结果
result_df = pd.DataFrame(result)
print(result_df)

• 注意事项：可自行添加BOM循环依赖校验逻辑，避免遍历死循环。

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