MySQL递归查询实现仓库多层级复合产品可生产数量计算

问题背景 #

• 开发仓库管理系统，需管理物品的可用库存数量
• 支持复合产品：由一种或多种其他产品组成，且组件本身也可以是复合产品
• 示例：库存有100升水和100个1升瓶，「瓶装水」由1升水+1个瓶子组成，虽无直接库存，但理论可生产100份

数据结构 #

已创建两张MySQL表：

1. 基础产品表：存储产品基本信息
2. 复合产品关联表：记录复合产品所需的组件及数量

JSON数据示例：

{"products": [
    {"id": 1, "name": "Water", "quantity": 120, "compound": false, "decimal": true},
    {"id": 2, "name": "Bottles", "quantity": 240, "compound": false, "decimal": false},
    {"id": 3, "name": "Small Water Bottle", "quantity": null, "compound": true, "decimal": false},
    {"id": 4, "name": "Water Box", "quantity": null, "compound": true, "decimal": false},
    {"id": 5, "name": "Water Shelf", "quantity": null, "compound": true, "decimal": false}
],
"compound_list": [
    {"id": 1, "compound_id": 3, "requirement_id": 1, "requirement_quantity": 1},
    {"id": 2, "compound_id": 3, "requirement_id": 2, "requirement_quantity": 1},
    {"id": 3, "compound_id": 4, "requirement_id": 3, "requirement_quantity": 6},
    {"id": 4, "compound_id": 5, "requirement_id": 4, "requirement_quantity": 10}
]}

现有问题 #

当前查询仅支持由基础产品直接组成的复合产品，无法处理多层级嵌套的复合产品：

SELECT 
    art.id AS product_id,
    art.name AS product_name,
    art.qty AS product_requirement_qty,
    art.compound 
FROM articles art 
WHERE art.compound = 0 

UNION ALL 

SELECT 
    art.id AS product_id,
    art.name AS product_name,
    MIN(CASE WHEN art.decimal = 1 THEN ROUND( (art_calc.qty/lc.requirement_qty), 2) ELSE FLOOR( (art_calc.qty/lc.requirement_qty) ) END ) AS product_requirement_qty,
    art.compound 
FROM articles art 
INNER JOIN compound_list lc ON lc.compound_id = art.id 
INNER JOIN articles art_calc ON art_calc.id = lc.requirement_id 
WHERE art.compound = 1 
GROUP BY art.id;

Node.js递归实现参考 #

已在Node.js中实现正确的递归逻辑，可计算多层级复合产品的可生产数量：

const { performance } = require('perf_hooks');
const { exit } = require('process');

const products = [
    {"id": 1, "name": "Water",  "quantity": 100, "compound": false, "decimal": true},
    {"id": 2, "name": "Bottles", "quantity": 100, "compound": false, "decimal": false},
    {"id": 3, "name": "Small Water Bottle", "quantity": null, "compound": true, "decimal": false},
    {"id": 4, "name": "Water Box", "quantity": null, "compound": true, "decimal": false},
    {"id": 5, "name": "Water Shelf", "quantity": null, "compound": true, "decimal": false}
];

const compound_list = [
    {"id": 1, "compound_id": 3, "requirement_id": 1, "requirement_quantity": 1},
    {"id": 2, "compound_id": 3, "requirement_id": 2, "requirement_quantity": 1},
    {"id": 3, "compound_id": 4, "requirement_id": 3, "requirement_quantity": 6},
    {"id": 4, "compound_id": 5, "requirement_id": 4, "requirement_quantity": 10}
];

function ViewQty(products, compound_list){
    const startTime = performance.now();
    var tableQTY = [];
    products.forEach(product => {
        tableQTY.push(recursiveFunction(products, product, compound_list));
    });
    const endTime = performance.now();
    console.table(tableQTY);
    const elapsedTime = endTime - startTime;
    console.log(`Execution Time: ${elapsedTime}`);
}

function recursiveFunction(products, product, compound_list){
    if(!product.compound){
        return {
            "id": product.id,
            "name": product.name,
            "quantity": product.quantity
        };
    } else {
        var components = getComponents(product.id, compound_list);
        var minQty;
        if(components.length > 0){
            var possibleQtyList = [];
            components.forEach(component => {
                var x = products.find(art => art.id === component.requirement_id);
                var obj = recursiveFunction(products, x, compound_list);
                obj.quantity /= component.requirement_quantity;
                possibleQtyList.push(obj);
            });
            minQty = Math.min(...possibleQtyList.map(element => element.quantity));
        } else {
            minQty = 0;
        }
        if(product.decimal){
            minQty = minQty.toFixed(2);
        } else {
            minQty = Math.floor(minQty);
        }
        return {
            "id": product.id,
            "name": product.name,
            "quantity": minQty
        };
    }
}

function getComponents(id, compound_list){
    var miniList = [];
    compound_list.forEach(element => {
        if(element.compound_id === id){
            miniList.push(element);
        }
    });
    if(miniList.length === 0){
        return {"requirement_id": null, "requirement_quantity": 0};
    } else {
        return miniList;
    }
}

ViewQty(products, compound_list);

核心需求 #

1. 理解MySQL递归CTE的工作原理
2. 编写支持多层级复合产品的递归查询，并创建视图

MySQL递归CTE工作原理 #

递归CTE由两部分组成，用UNION ALL连接：

1. 锚点成员：定义递归的起始点，通常是基础数据（比如非复合产品的库存）
2. 递归成员：引用CTE自身，用于遍历嵌套层级（比如复合产品的组件）

执行流程：

• 先执行锚点成员，得到初始数据集
• 再将递归成员应用到上一轮的结果集，生成新的数据集
• 重复执行递归成员，直到没有新数据生成
• 最终合并所有轮次的结果

多层级递归查询实现 #

以下查询可计算所有产品（包括多层级复合产品）的可生产数量，可直接创建为视图：

WITH RECURSIVE product_calc AS (
    -- 锚点成员：基础产品的库存数量
    SELECT
        p.id AS product_id,
        p.name AS product_name,
        p.quantity AS available_qty,
        p.compound,
        p.decimal,
        0 AS level  -- 层级标记，基础产品为0
    FROM products p
    WHERE p.compound = FALSE

    UNION ALL

    -- 递归成员：计算复合产品的可生产数量
    SELECT
        cp.id AS product_id,
        cp.name AS product_name,
        -- 根据产品类型计算可生产数量：小数类型取两位小数，整数类型取地板值
        CASE 
            WHEN cp.decimal THEN ROUND(MIN(pc.available_qty / cl.requirement_quantity), 2)
            ELSE FLOOR(MIN(pc.available_qty / cl.requirement_quantity))
        END AS available_qty,
        cp.compound,
        cp.decimal,
        pc.level + 1 AS level
    FROM products cp
    JOIN compound_list cl ON cp.id = cl.compound_id
    JOIN product_calc pc ON cl.requirement_id = pc.product_id
    WHERE cp.compound = TRUE
    GROUP BY cp.id, cp.name, cp.decimal, pc.level + 1
)
-- 最终查询：每个产品取最深层级的计算结果（避免重复）
SELECT
    product_id,
    product_name,
    available_qty,
    compound,
    decimal
FROM product_calc pc
WHERE level = (
    SELECT MAX(level) 
    FROM product_calc 
    WHERE product_id = pc.product_id
);

创建视图 #

将上述查询封装为视图：

CREATE VIEW product_available_quantity AS
WITH RECURSIVE product_calc AS (
    SELECT
        p.id AS product_id,
        p.name AS product_name,
        p.quantity AS available_qty,
        p.compound,
        p.decimal,
        0 AS level
    FROM products p
    WHERE p.compound = FALSE

    UNION ALL

    SELECT
        cp.id AS product_id,
        cp.name AS product_name,
        CASE 
            WHEN cp.decimal THEN ROUND(MIN(pc.available_qty / cl.requirement_quantity), 2)
            ELSE FLOOR(MIN(pc.available_qty / cl.requirement_quantity))
        END AS available_qty,
        cp.compound,
        cp.decimal,
        pc.level + 1 AS level
    FROM products cp
    JOIN compound_list cl ON cp.id = cl.compound_id
    JOIN product_calc pc ON cl.requirement_id = pc.product_id
    WHERE cp.compound = TRUE
    GROUP BY cp.id, cp.name, cp.decimal, pc.level + 1
)
SELECT
    product_id,
    product_name,
    available_qty,
    compound,
    decimal
FROM product_calc pc
WHERE level = (
    SELECT MAX(level) 
    FROM product_calc 
    WHERE product_id = pc.product_id
);

逻辑说明 #

• 锚点成员先获取所有基础产品的库存，作为递归的起点
• 递归成员遍历每个复合产品，关联其所有组件的计算结果，通过MIN(available_qty / requirement_quantity)得到该复合产品的最大可生产数量（受限于最紧缺的组件）
• 最终查询通过层级筛选，确保每个产品只返回最深层级的计算结果（复合产品可能在递归过程中被多次计算，取最终的多层级结果）

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