DuckDB 递归查询高级实战：图遍历、循环检测与最短路径

WITH RECURSIVE cte_name AS (
    -- 锚点查询（Anchor）：递归的起始点
    SELECT ...
    
    UNION ALL
    
    -- 递归查询（Recursive Member）：基于上一轮结果继续递归
    SELECT ...
    FROM cte_name
    WHERE 终止条件
)
SELECT * FROM cte_name;

-- 创建供应链关系表
CREATE TABLE supply_chain (
    id INTEGER,
    from_node VARCHAR,
    to_node VARCHAR,
    node_type VARCHAR,  -- 'raw_material', 'component', 'product', 'warehouse', 'store'
    lead_time_days INTEGER
);

-- 插入供应链数据
INSERT INTO supply_chain VALUES
(1,  '钢材',    '发动机',  'component',  3),
(2,  '钢材',    ' chassis', 'component',  5),
(3,  '橡胶',    '轮胎',   'component',  2),
(4,  '发动机', '整车A',  'product',    7),
(5,  'chassis', '整车A',  'product',    6),
(6,  '轮胎',   '整车A',  'product',    4),
(7,  '整车A',  '华东仓',  'warehouse',  1),
(8,  '整车A',  '华南仓',  'warehouse',  1),
(9,  '华东仓', '上海店', 'store',      1),
(10, '华东仓', '杭州店', 'store',      1),
(11, '华南仓', '广州店', 'store',      1),
(12, '华南仓', '深圳店', 'store',      1);

WITH RECURSIVE path_enum AS (
    -- 锚点：所有原材料
    SELECT 
        from_node AS start_node,
        to_node AS current_node,
        CAST(from_node || ' → ' || to_node AS VARCHAR) AS path,
        lead_time_days AS total_lead_time,
        1 AS depth
    FROM supply_chain
    WHERE from_node IN ('钢材', '橡胶')
    
    UNION ALL
    
    -- 递归：沿着供应链向下追踪
    SELECT 
        pe.start_node,
        sc.to_node,
        pe.path || ' → ' || sc.to_node,
        pe.total_lead_time + sc.lead_time_days,
        pe.depth + 1
    FROM path_enum pe
    JOIN supply_chain sc ON sc.from_node = pe.current_node
    WHERE pe.depth < 10  -- 防止无限递归
      AND pe.path NOT LIKE '%' || sc.to_node || '%'  -- 防止循环
)
SELECT 
    start_node AS 原材料,
    path AS 完整路径,
    total_lead_time AS 总交付周期(天),
    depth AS 层级数
FROM path_enum
WHERE current_node LIKE '%店%'
ORDER BY start_node, total_lead_time;

-- 只想知道"哪些门店受钢材影响"（不需要路径细节）
-- 用 UNION 自动去重，性能更好
WITH RECURSIVE affected_stores AS (
    SELECT to_node AS store FROM supply_chain WHERE from_node = '钢材'
    UNION
    SELECT sc.to_node
    FROM affected_stores a
    JOIN supply_chain sc ON sc.from_node = a.store
    WHERE sc.node_type = 'store'
)
SELECT * FROM affected_stores;
-- 结果：上海店, 杭州店, 广州店, 深圳店

CREATE TABLE approval_chain (
    approver VARCHAR,
    approvee VARCHAR,
    level INTEGER
);

INSERT INTO approval_chain VALUES
('Alice', 'Bob', 1),
('Bob', 'Charlie', 2),
('Charlie', 'Alice', 3),  -- 循环！
('Dave', 'Eve', 1),
('Eve', 'Frank', 2);

WITH RECURSIVE chain AS (
    -- 锚点：所有起点
    SELECT 
        approver AS start_person,
        approvee AS current_person,
        approver || ' → ' || approvee AS path,
        1 AS depth
    FROM approval_chain
    
    UNION ALL
    
    -- 递归：沿着审批链追踪
    SELECT 
        c.start_person,
        ac.approvee,
        c.path || ' → ' || ac.approvee,
        c.depth + 1
    FROM chain c
    JOIN approval_chain ac ON ac.approver = c.current_person
    WHERE c.depth < 10
)
-- 检测循环：当路径中已经包含当前节点时，说明有循环
SELECT 
    start_person,
    path AS 循环路径,
    depth AS 循环深度
FROM chain
WHERE path LIKE '%' || current_person || '%';

WITH RECURSIVE chain AS (
    SELECT 
        approver AS start_person,
        approvee AS current_person,
        ',' || approver || ',' || approvee || ',' AS visited,
        1 AS depth
    FROM approval_chain
    
    UNION ALL
    
    SELECT 
        c.start_person,
        ac.approvee,
        c.visited || ac.approvee || ',',
        c.depth + 1
    FROM chain c
    JOIN approval_chain ac ON ac.approver = c.current_person
    WHERE c.depth < 10
      AND c.visited NOT LIKE '%,' || ac.approvee || ',%'  -- 关键：检测已访问
)
SELECT * FROM chain WHERE depth > 1 AND visited LIKE '%,' || current_person || '%';

CREATE TABLE roads (
    from_city VARCHAR,
    to_city VARCHAR,
    distance_km INTEGER
);

INSERT INTO roads VALUES
('北京', '天津', 137),
('北京', '济南', 400),
('天津', '济南', 360),
('天津', '上海', 1050),
('济南', '上海', 680),
('济南', '南京', 580),
('上海', '南京', 270),
('南京', '杭州', 250),
('杭州', '宁波', 140);

WITH RECURSIVE shortest_path AS (
    -- 锚点：从起点出发，距离为 0
    SELECT 
        '北京' AS start_city,
        from_city AS current_city,
        distance_km AS distance,
        '北京 → ' || from_city AS path,
        1 AS hops
    FROM roads
    WHERE from_city = '北京'
    
    UNION ALL
    
    -- 递归：寻找更短的路径
    SELECT 
        sp.start_city,
        r.to_city,
        sp.distance + r.distance_km,
        sp.path || ' → ' || r.to_city,
        sp.hops + 1
    FROM shortest_path sp
    JOIN roads r ON r.from_city = sp.current_city
    WHERE sp.distance + r.distance_km < COALESCE(
        (SELECT MIN(distance) FROM shortest_path sp2 
         WHERE sp2.start_city = sp.start_city AND sp2.current_city = r.to_city),
        999999
    )
    AND sp.hops < 10
)
SELECT 
    current_city AS 目的地,
    distance AS 最短距离(km),
    path AS 路径,
    hops AS 中转次数
FROM shortest_path
WHERE current_city != '北京'
  AND NOT EXISTS (
      SELECT 1 FROM shortest_path sp2
      WHERE sp2.start_city = shortest_path.start_city
        AND sp2.current_city = shortest_path.current_city
        AND sp2.distance < shortest_path.distance
  )
ORDER BY distance;

WITH RECURSIVE paths AS (
    SELECT 
        '北京' AS start_city,
        from_city AS current_city,
        distance_km AS total_distance,
        '北京 → ' || from_city AS path,
        1 AS hops
    FROM roads
    WHERE from_city = '北京'
    
    UNION ALL
    
    SELECT 
        p.start_city,
        r.to_city,
        p.total_distance + r.distance_km,
        p.path || ' → ' || r.to_city,
        p.hops + 1
    FROM paths p
    JOIN roads r ON r.from_city = p.current_city
    WHERE p.hops < 10
      AND p.path NOT LIKE '%' || r.to_city || '%'
)
SELECT 
    current_city,
    MIN(total_distance) AS min_distance,
    ARRAY_AGG(path ORDER BY total_distance LIMIT 1)[1] AS best_path
FROM paths
GROUP BY current_city
ORDER BY min_distance;

CREATE TABLE relationships (
    person_a VARCHAR,
    person_b VARCHAR,
    relation_type VARCHAR,
    strength INTEGER  -- 关系强度 1-10
);

INSERT INTO relationships VALUES
('张三', '李四', '同事', 8),
('李四', '王五', '朋友', 7),
('王五', '赵六', '同学', 9),
('张三', '钱七', '邻居', 5),
('钱七', '孙八', '朋友', 6),
('孙八', '周九', '同事', 7);

-- 查询张三的 N 度人脉（N 由参数控制）
CREATE OR REPLACE FUNCTION get_kdegree_connections(
    start_person VARCHAR,
    k_degree INTEGER
) RETURNS TABLE (
    target VARCHAR,
    degree INTEGER,
    path VARCHAR,
    strength_sum INTEGER
) AS $$
WITH RECURSIVE connections AS (
    -- 一度关系
    SELECT 
        CASE WHEN person_a = start_person THEN person_b ELSE person_a END AS target,
        1 AS degree,
        start_person || ' → ' || 
        CASE WHEN person_a = start_person THEN person_b ELSE person_a END AS path,
        CASE WHEN person_a = start_person THEN strength ELSE strength END AS strength_sum
    FROM relationships
    WHERE person_a = start_person OR person_b = start_person
    
    UNION ALL
    
    -- 递归：K 度关系
    SELECT 
        CASE WHEN c.person_a = conn.target THEN c.person_b ELSE c.person_a END,
        conn.degree + 1,
        conn.path || ' → ' || 
        CASE WHEN c.person_a = conn.target THEN c.person_b ELSE c.person_a END,
        conn.strength_sum + c.strength
    FROM connections conn
    JOIN relationships c ON c.person_a = conn.target OR c.person_b = conn.target
    WHERE conn.degree < k_degree
      AND (conn.path || ' → ' || 
           CASE WHEN c.person_a = conn.target THEN c.person_b ELSE c.person_a END
          ) NOT LIKE '%→ ' || CASE WHEN c.person_a = conn.target THEN c.person_b ELSE c.person_a END || '% →%'
)
SELECT * FROM connections;
$$ LANGUAGE sql;

-- 使用：查询张三的 3 度人脉
SELECT * FROM get_kdegree_connections('张三', 3);

-- 防止无限递归导致 OOM
WHERE pe.depth < 100

-- 只需判断"是否可达"，用 UNION 去重减少迭代
WITH RECURSIVE reachable AS (
    SELECT start_node FROM edges WHERE id = 1
    UNION  -- 自动去重！
    SELECT e.to_node FROM reachable r JOIN edges e ON e.from_node = r.end_node
)
SELECT * FROM reachable;

-- 设置合适的并行度
PRAGMA threads=8;
PRAGMA memory_limit='4GB';

-- DuckDB 自动为小表建立哈希索引
-- 但对于大图，可以手动物化
CREATE TABLE edges_indexed AS
SELECT * FROM edges ORDER BY from_node;

-- 这样递归时的 JOIN 效率更高