PostgreSQL递归广度遍历查询：实现嵌套JSON与深度限制

问题背景 #

我在PostgreSQL中有items和connections两张表构成稀疏图结构，需要通过node-pg从指定种子节点（如ExpressJS的URL参数）出发进行广度优先递归遍历，支持最大深度限制，最终返回符合要求的嵌套JSON结果。目前已实现基础遍历和种子过滤，但无法处理深度限制（循环检测困难），也无法生成目标嵌套JSON，希望通过jsonb_build_object或jsonb_agg等函数解决。

表结构 #

CREATE TABLE IF NOT EXISTS items (
    id    UUID         NOT NULL DEFAULT uuid_generate_v4(),
    title VARCHAR(300) NOT NULL,
    CONSTRAINT items_pk PRIMARY KEY (id)
);

CREATE TABLE IF NOT EXISTS connections (
    id                  UUID         NOT NULL DEFAULT uuid_generate_v4(),
    origin_item_id      UUID         NOT NULL,
    destination_item_id UUID         NOT NULL,
    title               VARCHAR(300) NOT NULL,
    CONSTRAINT origin_item_fk FOREIGN KEY (origin_item_id) REFERENCES items (id),
    CONSTRAINT destination_item_fk FOREIGN KEY (destination_item_id) REFERENCES items (id),
    CONSTRAINT connections_pk PRIMARY KEY (id)
);

期望输出 #

当种子节点为Pyongyang、maxDepth=2时，返回：

{
  "title": "Pyongyang",
  "connections": [
    {
      "title": "Same Author - Guy Delisle - 1",
      "destination_item": {
        "title": "Shenzhen",
        "connections": [
          {
            "title": "Same Author - Guy Delisle - 2",
            "origin_item": {
              "title": "Jerusalem"
            }
          }
        ]
      }
    },
    {
      "title": "Same Author - Guy Delisle - 2",
      "origin_item": {
        "title": "Burma"
      }
    }
  ]
}

解决方案：递归CTE + JSON构建函数 #

以下是封装好的SQL函数，支持传入种子标题和最大深度，返回嵌套JSON：

CREATE OR REPLACE FUNCTION get_nested_graph(p_start_title VARCHAR(300), p_max_depth INT)
RETURNS JSONB AS $$
WITH RECURSIVE graph_walk AS (
    -- 初始节点：获取种子节点基础信息，深度设为1，路径记录自身ID防循环
    SELECT
        i.id AS node_id,
        i.title AS node_title,
        1 AS current_depth,
        ARRAY[i.id] AS visited_path,
        '[]'::JSONB AS connections_json
    FROM items i
    WHERE i.title = p_start_title

    UNION ALL

    -- 递归遍历当前节点的双向连接
    SELECT
        CASE
            WHEN c.origin_item_id = gw.node_id THEN c.destination_item_id
            ELSE c.origin_item_id
        END AS node_id,
        i.title AS node_title,
        gw.current_depth + 1 AS current_depth,
        gw.visited_path || CASE
            WHEN c.origin_item_id = gw.node_id THEN c.destination_item_id
            ELSE c.origin_item_id
        END AS visited_path,
        -- 构建当前节点的连接JSON
        (
            SELECT jsonb_agg(
                CASE
                    WHEN c2.origin_item_id = i.id THEN
                        jsonb_build_object(
                            'title', c2.title,
                            'destination_item', jsonb_build_object('title', i2.title)
                        )
                    ELSE
                        jsonb_build_object(
                            'title', c2.title,
                            'origin_item', jsonb_build_object('title', i2.title)
                        )
                END
            )
            FROM connections c2
            JOIN items i2 ON 
                (c2.origin_item_id = i.id AND c2.destination_item_id = i2.id) 
                OR (c2.destination_item_id = i.id AND c2.origin_item_id = i2.id)
            WHERE i2.id <> ALL(gw.visited_path)
        ) AS connections_json
    FROM graph_walk gw
    JOIN connections c ON 
        gw.node_id = c.origin_item_id OR gw.node_id = c.destination_item_id
    JOIN items i ON 
        (c.origin_item_id = gw.node_id AND c.destination_item_id = i.id) 
        OR (c.destination_item_id = gw.node_id AND c.origin_item_id = i.id)
    WHERE gw.current_depth < p_max_depth
        AND i.id <> ALL(gw.visited_path)
),
-- 从最深层向上聚合嵌套结构
nested_agg AS (
    SELECT
        node_id,
        node_title,
        current_depth,
        jsonb_build_object(
            'title', node_title,
            'connections', COALESCE(connections_json, '[]'::JSONB)
        ) AS full_json
    FROM graph_walk
    WHERE current_depth = p_max_depth

    UNION ALL

    SELECT
        gw.node_id,
        gw.node_title,
        gw.current_depth,
        jsonb_build_object(
            'title', gw.node_title,
            'connections', (
                SELECT jsonb_agg(
                    CASE
                        WHEN c.origin_item_id = gw.node_id THEN
                            jsonb_set(
                                jsonb_build_object('title', c.title),
                                '{destination_item}',
                                na.full_json
                            )
                        ELSE
                            jsonb_set(
                                jsonb_build_object('title', c.title),
                                '{origin_item}',
                                na.full_json
                            )
                    END
                )
                FROM connections c
                JOIN nested_agg na ON 
                    (c.origin_item_id = gw.node_id AND c.destination_item_id = na.node_id)
                    OR (c.destination_item_id = gw.node_id AND c.origin_item_id = na.node_id)
            )
        ) AS full_json
    FROM graph_walk gw
    JOIN nested_agg na ON 
        (gw.node_id = c.origin_item_id AND na.node_id = c.destination_item_id)
        OR (gw.node_id = c.destination_item_id AND na.node_id = c.origin_item_id)
    WHERE gw.current_depth = na.current_depth - 1
)
-- 返回根节点的完整嵌套JSON
SELECT full_json FROM nested_agg WHERE node_title = p_start_title AND current_depth = 1;
$$ LANGUAGE sql STABLE;

使用方法 #

通过node-pg调用该函数：

const { Pool } = require('pg');
const pool = new Pool();

async function getGraph(startTitle, maxDepth) {
  const result = await pool.query(
    'SELECT get_nested_graph($1, $2) AS graph',
    [startTitle, maxDepth]
  );
  return result.rows[0].graph;
}

// 调用示例
getGraph('Pyongyang', 2).then(graph => console.log(JSON.stringify(graph, null, 2)));

关键说明 #

1. 循环检测：通过visited_path数组记录已访问节点ID，递归时排除路径内的节点，避免循环遍历。
2. 深度控制：递归CTE中用current_depth跟踪层级，达到p_max_depth时停止递归。
3. 嵌套JSON构建：用jsonb_build_object生成基础节点结构，jsonb_agg聚合连接列表，jsonb_set将子节点的完整嵌套JSON合并到父节点的连接中，实现层级嵌套。

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