如何在数据表中追踪加工分支链路并计算pass字段值

这个需求本质是有向加工链路的分支溯源+时序递推计数问题，核心是识别不同物料编码的上下游依赖关系，按加工先后顺序对链路层级做递增计数，同时兼容单工序多输出的分支拆分场景，以下是三种可直接落地的实现方案：

方案1：SQL递归实现（适配关系型数据库/数仓场景，生产环境最常用） #

前置准备 #

• 确认Record#是严格按加工时间升序排列的唯一序号，同输入下的记录按该字段排序即为加工先后顺序
• 提前清洗脏数据：过滤存在循环依赖的异常记录（如A输出B、B又输出A的逻辑矛盾数据），避免递归死循环

核心实现逻辑 #

用递归CTE（公共表表达式）完成链路遍历：

1. 锚定层：识别所有根输入节点（即input值从未在更早记录的output字段出现过的物料编码），根节点下最早的加工记录pass值从1开始计数
2. 递归层：按加工时序向后匹配，若后续记录的input等于已遍历记录的output，则归属到对应分支递推计数：
   • 同输入下的多条同批次加工记录（如同设备同一次加工产生的多个输出）共享同一个pass值
   • 若某条记录的output与input不一致，说明生成了新的分支物料，后续所有以该新物料为输入的记录，在当前pass值基础上继续向后递推，不影响原主分支的计数逻辑
     核心代码参考：

WITH RECURSIVE process_trace AS (
    -- 锚定所有根链路的起始记录
    SELECT 
        Record#, machine, input, output, material,
        1 AS pass,
        input AS branch_key,
        1 AS next_pass
    FROM process_record r1
    WHERE NOT EXISTS (
        SELECT 1 FROM process_record r2 
        WHERE r2.output = r1.input AND r2.Record# < r1.Record#
    )
    AND Record# = (SELECT MIN(Record#) FROM process_record WHERE input = r1.input)

    UNION ALL

    SELECT 
        curr.Record#, curr.machine, curr.input, curr.output, curr.material,
        p.next_pass AS pass,
        CASE WHEN curr.output != curr.input THEN curr.output ELSE p.branch_key END AS branch_key,
        p.next_pass + 1 AS next_pass
    FROM process_record curr
    INNER JOIN process_trace p 
        ON curr.input = p.branch_key 
        AND curr.Record# > p.Record#
    -- 同分支下按Record顺序匹配，避免跳号
    WHERE curr.Record# = (
        SELECT MIN(Record#) FROM process_record 
        WHERE input = p.branch_key AND Record# > p.Record#
    )
)
-- 补全同批次同输入共享pass的记录
SELECT r.Record#, r.machine, r.input, r.output, t.pass, r.material
FROM process_record r
LEFT JOIN process_trace t 
ON r.input = t.branch_key 
AND r.Record# >= (SELECT MIN(Record#) FROM process_record WHERE input = r.input AND pass = t.pass)
AND r.Record# < (SELECT MIN(Record#) FROM process_record WHERE input = r.input AND pass = t.pass + 1)
ORDER BY r.Record#;

跑完递归后校验是否存在未匹配pass的记录，若存在则补全遗漏的根节点分支即可。

方案2：Python离线脚本实现（适配中小数据量、快速迭代场景） #

逻辑更直观，不需要适配不同数据库的递归语法差异，实现步骤如下：

1. 将所有加工记录按Record#升序排序，初始化两个字典：
   • pass_tracker：key为物料编码，value为该物料作为输入时，下一条加工记录对应的pass值
   • res：存储每条记录的最终计算结果
2. 逐行遍历排序后的记录：
   • 若当前记录的input不在pass_tracker中，说明是新根链路的起始，当前记录pass值设为1，同时更新pass_tracker[input] = 2
   • 若当前记录的input已在pass_tracker中，当前记录pass值直接取pass_tracker[input]，同输入下连续的同批次加工记录（如同设备一次加工出多个输出）复用该pass值，待该批次所有输出处理完成后，将pass_tracker[input]的值加1
   • 若当前记录的output与input不一致，说明生成新分支物料，将pass_tracker[output]设为当前pass值 + 1，后续以该新物料为输入的记录从这个值开始递推
3. 遍历完成后将pass值批量写回原表即可，按该逻辑计算示例数据可完全匹配目标结果。

方案3：图数据库实现（适配千万级以上大流量、多层级链路场景） #

如果日常加工记录量级超千万、链路嵌套层级超过10层，用Neo4j等图数据库存储加工关系计算效率更高：

• 将每个唯一的物料编码（input/output值）创建为节点，每条加工记录创建为从输入物料指向输出物料的有向边，边属性绑定machine、material、加工时序、Record#等信息
• 从根物料节点出发做广度优先遍历，沿加工时序方向遍历所有出边，同节点下的出边按时序顺序递增计数，遇到新生成的物料节点自动开新分支，继承上一条边的pass值继续递推即可，不存在递归深度超限的问题，计算性能远高于传统关系型数据库的递归查询。

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