Postgres任意嵌套JSONB对象的指定条件更新问题

实现思路 #

要完成任意层级嵌套JSONB中目标对象的替换，核心是先精准定位所有符合条件的节点路径，再通过jsonb_set逐个替换节点。以下是具体实现方案：

完整SQL代码（PostgreSQL 12+） #

WITH RECURSIVE json_tree AS (
    -- 遍历顶层JSON数据，初始化节点信息
    SELECT
        id,
        data AS original_data,
        data AS current_node,
        ''::text AS path,
        jsonb_typeof(data) AS node_type
    FROM my_table  -- 替换为你的表名
    UNION ALL
    -- 遍历对象类型的子节点，拼接路径
    SELECT
        j.id,
        j.original_data,
        v AS current_node,
        CASE 
            WHEN j.path = '' THEN concat('$.', k) 
            ELSE concat(j.path, '.', k) 
        END AS path,
        jsonb_typeof(v) AS node_type
    FROM json_tree j
    JOIN LATERAL jsonb_each(j.current_node) AS t(k, v)
        ON j.node_type = 'object'
    UNION ALL
    -- 遍历数组类型的子节点，拼接路径（带索引）
    SELECT
        j.id,
        j.original_data,
        v AS current_node,
        CASE 
            WHEN j.path = '' THEN concat('$[', i, ']') 
            ELSE concat(j.path, '[', i, ']') 
        END AS path,
        jsonb_typeof(v) AS node_type
    FROM json_tree j
    JOIN LATERAL jsonb_array_elements(j.current_node) WITH ORDINALITY AS t(v, i)
        ON j.node_type = 'array'
),
target_paths AS (
    -- 筛选出包含"key_a":"foo"的节点路径，去重避免重复处理
    SELECT DISTINCT id, path
    FROM json_tree
    WHERE current_node @> '{"key_a": "foo"}'::jsonb
),
updated_data AS (
    -- 对每个ID的所有路径，依次执行节点替换
    SELECT
        id,
        reduce(
            array_agg(path ORDER BY path),
            (SELECT data FROM my_table WHERE id = tp.id),
            (acc, p) -> jsonb_set(
                acc,
                -- 将路径字符串转换为jsonb_set所需的text数组格式
                string_to_array(
                    regexp_replace(regexp_replace(p, '^\$', ''), '\[(\d+)\]', '.\1', 'g'),
                    '.'
                ),
                '{"or": [{"key_a":"foo"}, {"key_a":"bar"}, {"key_a":"bla"}]}'::jsonb
            )
        ) AS new_data
    FROM target_paths tp
    GROUP BY id
)
-- 执行最终更新
UPDATE my_table
SET data = ud.new_data
FROM updated_data ud
WHERE my_table.id = ud.id;

代码说明 #

1. 递归CTE json_tree：
   遍历JSONB的每一层节点，无论是对象还是数组，都记录节点的完整路径（例如$[0].a.d）和节点值，确保不遗漏任何嵌套层级。

2. target_paths 筛选：
   通过current_node @> '{"key_a": "foo"}'精准匹配包含目标属性的节点，并用DISTINCT避免重复处理同一节点。

3. updated_data 批量替换：
   使用PostgreSQL 12+支持的reduce函数，迭代每个ID的路径列表，依次对每个路径执行jsonb_set操作，将目标节点替换为指定的新结构。

4. 路径转换：
   通过正则表达式将路径字符串（如$[0].a.d）转换为jsonb_set所需的数组格式（如['0', 'a', 'd']），确保替换操作能定位到正确节点。

兼容旧版本PostgreSQL（低于12） #

如果你的PostgreSQL版本不支持reduce函数，可使用递归CTE逐个处理路径：

WITH RECURSIVE json_tree AS (
    SELECT
        id,
        data AS original_data,
        data AS current_node,
        ''::text AS path,
        jsonb_typeof(data) AS node_type
    FROM my_table
    UNION ALL
    SELECT
        j.id,
        j.original_data,
        v AS current_node,
        CASE WHEN j.path = '' THEN concat('$.', k) ELSE concat(j.path, '.', k) END AS path,
        jsonb_typeof(v) AS node_type
    FROM json_tree j
    JOIN LATERAL jsonb_each(j.current_node) AS t(k, v)
        ON j.node_type = 'object'
    UNION ALL
    SELECT
        j.id,
        j.original_data,
        v AS current_node,
        CASE WHEN j.path = '' THEN concat('$[', i, ']') ELSE concat(j.path, '[', i, ']') END AS path,
        jsonb_typeof(v) AS node_type
    FROM json_tree j
    JOIN LATERAL jsonb_array_elements(j.current_node) WITH ORDINALITY AS t(v, i)
        ON j.node_type = 'array'
),
target_paths AS (
    SELECT DISTINCT id, path, row_number() OVER (PARTITION BY id ORDER BY path) AS rn
    FROM json_tree
    WHERE current_node @> '{"key_a": "foo"}'::jsonb
),
recursive_update AS (
    -- 初始替换第一条路径
    SELECT
        id,
        jsonb_set(
            data,
            string_to_array(regexp_replace(regexp_replace(path, '^\$', ''), '\[(\d+)\]', '.\1', 'g'), '.'),
            '{"or": [{"key_a":"foo"}, {"key_a":"bar"}, {"key_a":"bla"}]}'::jsonb
        ) AS updated_data,
        rn
    FROM target_paths tp
    JOIN my_table t ON tp.id = t.id
    WHERE rn = 1
    UNION ALL
    -- 递归替换后续路径
    SELECT
        ru.id,
        jsonb_set(
            ru.updated_data,
            string_to_array(regexp_replace(regexp_replace(tp.path, '^\$', ''), '\[(\d+)\]', '.\1', 'g'), '.'),
            '{"or": [{"key_a":"foo"}, {"key_a":"bar"}, {"key_a":"bla"}]}'::jsonb
        ) AS updated_data,
        tp.rn
    FROM recursive_update ru
    JOIN target_paths tp ON ru.id = tp.id AND tp.rn = ru.rn + 1
)
UPDATE my_table
SET data = ru.updated_data
FROM recursive_update ru
WHERE my_table.id = ru.id
AND ru.rn = (SELECT MAX(rn) FROM target_paths WHERE id = ru.id);

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