如何构建动态批次号追溯链？PostgreSQL实现技术求助

核心思路 #

通过递归CTE遍历上下游储罐的批次关联关系，生成完整的储罐-批次链，再通过动态SQL将储罐名转为列、批次号作为对应值，实现横向追溯表的生成。

1. 示例数据准备 #

先创建并填充源数据表：

CREATE TABLE tank_batch_links (
    us_tank VARCHAR(50),  -- 上游储罐
    ds_tank VARCHAR(50),  -- 下游储罐
    us_batch INT,         -- 上游批次号
    ds_batch INT          -- 下游批次号
);

INSERT INTO tank_batch_links VALUES
('citerne', 'B430', 122, 55),
('B430', 'K4', 55, 603);

2. 动态转换方案（适配任意数量储罐） #

使用PostgreSQL的动态SQL实现自动识别所有储罐并生成对应列：

DO $$
DECLARE
    cols TEXT;
BEGIN
    -- 提取所有唯一储罐名，转为SQL安全的列标识符
    SELECT string_agg(DISTINCT quote_ident(tank), ', ') INTO cols
    FROM (
        SELECT us_tank AS tank FROM tank_batch_links
        UNION
        SELECT ds_tank AS tank FROM tank_batch_links
    ) AS all_tanks;

    -- 执行动态生成的追溯链查询
    EXECUTE format('
        WITH recursive tank_chain AS (
            -- 起始节点：所有无上游的储罐
            SELECT 
                us_tank AS tank,
                us_batch AS batch,
                ds_tank AS next_tank,
                ds_batch AS next_batch,
                ARRAY[us_tank] AS tank_path,
                ARRAY[us_batch::TEXT] AS batch_path
            FROM tank_batch_links
            WHERE us_tank NOT IN (SELECT ds_tank FROM tank_batch_links)
            UNION ALL
            -- 递归遍历下游节点
            SELECT 
                t.ds_tank AS tank,
                t.ds_batch AS batch,
                t_next.ds_tank AS next_tank,
                t_next.ds_batch AS next_batch,
                tc.tank_path || t.ds_tank,
                tc.batch_path || t.ds_batch::TEXT
            FROM tank_chain tc
            JOIN tank_batch_links t ON tc.next_tank = t.us_tank
            LEFT JOIN tank_batch_links t_next ON t.ds_tank = t_next.us_tank
        ),
        chain_data AS (
            -- 将储罐路径和批次路径转为JSON键值对
            SELECT jsonb_object(tank_path, batch_path) AS tank_batch_map
            FROM tank_chain
            WHERE next_tank IS NULL  -- 只保留完整的末端链
        )
        -- 动态提取每个储罐对应的批次号作为列
        SELECT %s FROM chain_data;', cols);
END $$;

执行后将直接输出预期的横向追溯表：

3. 静态简化方案（针对已知储罐） #

如果储罐数量固定且已知，可直接使用静态查询避免动态SQL：

WITH recursive tank_chain AS (
    SELECT 
        us_tank AS tank,
        us_batch AS batch,
        ds_tank AS next_tank,
        ds_batch AS next_batch,
        ARRAY[us_tank] AS tank_path,
        ARRAY[us_batch::TEXT] AS batch_path
    FROM tank_batch_links
    WHERE us_tank = 'citerne'
    UNION ALL
    SELECT 
        t.ds_tank AS tank,
        t.ds_batch AS batch,
        t_next.ds_tank AS next_tank,
        t_next.ds_batch AS next_batch,
        tc.tank_path || t.ds_tank,
        tc.batch_path || t.ds_batch::TEXT
    FROM tank_chain tc
    JOIN tank_batch_links t ON tc.next_tank = t.us_tank
    LEFT JOIN tank_batch_links t_next ON t.ds_tank = t_next.us_tank
)
SELECT 
    (batch_path[1])::INT AS citerne,
    (batch_path[2])::INT AS B430,
    (batch_path[3])::INT AS K4
FROM tank_chain
WHERE next_tank IS NULL;

关键说明 #

• 递归CTE自动处理多段上下游关联，支持任意长度的追溯链
• 动态SQL版本可自动适配新增的储罐，无需修改查询逻辑
• 若存在多条独立追溯链，会返回对应数量的行，每行对应一条完整链

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