PostgreSQL CTE 与递归查询

-- 嵌套子查询：难以复用、难以阅读
SELECT c.name, o.total
FROM customers c
JOIN (
    SELECT customer_id, sum(amount) AS total
    FROM orders
    WHERE created_at >= now() - interval '30 days'
    GROUP BY customer_id
) o ON o.customer_id = c.id
WHERE o.total > 1000;

-- CTE：结构清晰，可被多次引用
WITH recent_orders AS (
    SELECT customer_id, sum(amount) AS total
    FROM orders
    WHERE created_at >= now() - interval '30 days'
    GROUP BY customer_id
)
SELECT c.name, r.total
FROM customers c
JOIN recent_orders r ON r.customer_id = c.id
WHERE r.total > 1000;

WITH
base AS (
    SELECT id, customer_id, amount FROM orders WHERE status = 'paid'
),
agg AS (
    -- 合法：agg 可以引用先定义的 base
    SELECT customer_id, count(*) AS cnt, sum(amount) AS total
    FROM base
    GROUP BY customer_id
)
SELECT * FROM agg ORDER BY total DESC LIMIT 20;

WITH RECURSIVE cte_name (col1, col2) AS (
    -- 1. 锚点项（non-recursive term）：起始行
    SELECT 1, 'root'
    UNION ALL
    -- 2. 递归项：引用 cte_name 自身
    SELECT col1 + 1, col2 || '>' || (col1 + 1)
    FROM cte_name
    WHERE col1 < 5          -- 3. 终止条件
)
SELECT * FROM cte_name;

 col1 |    col2
------+-------------
    1 | root
    2 | root>2
    3 | root>2>3
    4 | root>2>3>4
    5 | root>2>3>4>5

1. 求值锚点项 → 结果集 R，工作表 W = R
2. while W 非空:
     a. 求值递归项，其中 cte_name 引用 W
     b. 得到新行集 N
     c. 把 N 追加到最终结果（UNION ALL）或去重后追加（UNION）
     d. W = N
3. 返回最终结果

-- UNION ALL：不去重，快，但要求数据本身无环或已剪枝
WITH RECURSIVE t(n) AS (
    SELECT 1 UNION ALL SELECT n + 1 FROM t WHERE n < 1000
) SELECT count(*) FROM t;

-- UNION：每轮对新行做去重，防止死循环，但代价随规模上升
WITH RECURSIVE t(n) AS (
    SELECT 1 UNION SELECT n + 1 FROM t WHERE n < 1000
) SELECT count(*) FROM t;

CREATE TABLE employees (
    id          int PRIMARY KEY,
    name        text NOT NULL,
    manager_id  int REFERENCES employees(id),
    department  text
);

INSERT INTO employees VALUES
    (1, 'CEO',        NULL, 'exec'),
    (2, 'CTO',           1, 'eng'),
    (3, 'CFO',           1, 'fin'),
    (4, 'Backend Lead',  2, 'eng'),
    (5, 'Frontend Lead', 2, 'eng'),
    (6, 'Engineer A',    4, 'eng'),
    (7, 'Engineer B',    4, 'eng'),
    (8, 'Accountant',    3, 'fin');

WITH RECURSIVE subordinates AS (
    -- 锚点：起点是 CTO(id=2) 的直接下属
    SELECT id, name, manager_id, 1 AS depth,
           ARRAY[id] AS path
    FROM employees
    WHERE manager_id = 2
    UNION ALL
    -- 递归：沿着 manager_id 继续向下
    SELECT e.id, e.name, e.manager_id, s.depth + 1,
           s.path || e.id
    FROM employees e
    JOIN subordinates s ON e.manager_id = s.id
    WHERE NOT e.id = ANY(s.path)      -- 防环剪枝
)
SELECT id, name, depth, path
FROM subordinates
ORDER BY path;

WITH RECURSIVE chain AS (
    SELECT id, name, manager_id, 0 AS level
    FROM employees
    WHERE id = 6                       -- 从 Engineer A 出发
    UNION ALL
    SELECT m.id, m.name, m.manager_id, c.level + 1
    FROM employees m
    JOIN chain c ON m.id = c.manager_id
)
SELECT id, name, level
FROM chain
ORDER BY level;

WITH RECURSIVE tree AS (
    SELECT id, name, 0 AS depth, name::text AS indented
    FROM employees
    WHERE manager_id IS NULL
    UNION ALL
    SELECT e.id, e.name, t.depth + 1,
           repeat('    ', t.depth + 1) || e.name
    FROM employees e
    JOIN tree t ON e.manager_id = t.id
)
SELECT indented FROM tree ORDER BY indented;

CREATE TABLE edges (src int, dst int);
INSERT INTO edges VALUES (1,2),(2,3),(3,4),(1,4),(4,5);

WITH RECURSIVE bfs AS (
    SELECT 1 AS node, ARRAY[1] AS path, 0 AS hops
    UNION ALL
    SELECT e.dst, b.path || e.dst, b.hops + 1
    FROM edges e
    JOIN bfs b ON e.src = b.node
    WHERE NOT e.dst = ANY(b.path)      -- 不重复访问
      AND b.hops < 10                  -- 深度上限兜底
)
SELECT path, hops FROM bfs WHERE node = 5 ORDER BY hops LIMIT 1;

-- PG12+：CTE 只被引用一次 → 内联，谓词可下推
WITH big AS (
    SELECT * FROM orders
)
SELECT * FROM big WHERE customer_id = 42;   -- customer_id 谓词会被下推到基表

-- CTE 被引用两次 → 物化，避免重复计算
WITH big AS (SELECT * FROM orders)
SELECT (SELECT count(*) FROM big), (SELECT count(*) FROM big WHERE amount > 0);

-- 强制物化：适合 CTE 结果被多次使用且计算昂贵
WITH heavy AS MATERIALIZED (
    SELECT customer_id, sum(amount) AS total
    FROM orders GROUP BY customer_id
)
SELECT * FROM heavy WHERE total > 10000;

-- 强制内联：适合 CTE 结果被过滤后行数很少
WITH filtered AS NOT MATERIALIZED (
    SELECT * FROM orders WHERE status = 'paid'
)
SELECT * FROM filtered WHERE customer_id = 42;

EXPLAIN (ANALYZE, BUFFERS)
WITH c AS MATERIALIZED (SELECT * FROM orders WHERE status = 'paid')
SELECT * FROM c WHERE customer_id = 42;

 CTE Scan on c  (cost=... rows=... width=...)
   Filter: (customer_id = 42)
   CTE c
     ->  Seq Scan on orders  (rows=120000)
           Filter: (status = 'paid')

 Index Scan using orders_customer_id_idx on orders
   Index Cond: (customer_id = 42)
   Filter: (status = 'paid')

-- 反例：CTE 被引用一次却写成物化，导致全表扫描
WITH paid AS MATERIALIZED (SELECT * FROM orders WHERE status = 'paid')
SELECT count(*) FROM paid WHERE created_at > now() - interval '1 day';

WITH moved AS (
    DELETE FROM orders
    WHERE status = 'expired'
    RETURNING id, customer_id, amount
)
INSERT INTO orders_archive (id, customer_id, amount, archived_at)
SELECT id, customer_id, amount, now() FROM moved;

- 数据修改型 CTE 的各个分支看到的是同一个快照
- 任一子语句失败，整条语句回滚
- 不能在同一个 CTE 里对同一行既 UPDATE 又 DELETE

-- 反例：同一条语句里对同一行做两次 UPDATE，结果不可预测
WITH a AS (
    UPDATE counters SET value = value + 1 WHERE id = 1 RETURNING id
)
UPDATE counters SET value = value + 10 WHERE id = 1;

WITH incoming (id, amount) AS (
    VALUES (1, 100), (2, 200), (3, 300)
),
updated AS (
    UPDATE accounts a
    SET balance = a.balance + i.amount
    FROM incoming i
    WHERE a.id = i.id
    RETURNING a.id
)
INSERT INTO accounts (id, balance)
SELECT i.id, i.amount
FROM incoming i
LEFT JOIN updated u ON u.id = i.id
WHERE u.id IS NULL;

-- 查看当前 work_mem
SHOW work_mem;   -- 默认 4MB

-- 会话级临时放大，用于一次性深递归任务
SET work_mem = '64MB';

递归行数 × 每行宽度 × 去重开销 > work_mem → 落盘 → 性能骤降

-- 为递归连接列建索引，通常是性能提升最明显的一步
CREATE INDEX idx_employees_manager_id ON employees(manager_id);

ALTER TABLE employees ADD COLUMN path ltree;
-- 或用物化视图定期刷新层级路径
CREATE MATERIALIZED VIEW employee_tree AS
WITH RECURSIVE t AS (
    SELECT id, name, NULL::int AS parent, 0 AS depth FROM employees WHERE manager_id IS NULL
    UNION ALL
    SELECT e.id, e.name, e.manager_id, t.depth + 1
    FROM employees e JOIN t ON e.manager_id = t.id
)
SELECT * FROM t;

CREATE INDEX ON employee_tree(depth);

-- 观察递归 CTE 是否落盘：查看临时文件使用
SELECT temp_files, temp_bytes, pg_size_pretty(temp_bytes)
FROM pg_stat_database WHERE datname = current_database();

-- 观察 CTE 节点的实际行数与估计行数偏差
EXPLAIN (ANALYZE, BUFFERS, VERBOSE)
WITH RECURSIVE t AS (
    SELECT 1 AS n UNION ALL SELECT n + 1 FROM t WHERE n < 100
) SELECT * FROM t;

-- ========== 1. 建表与样例数据 ==========
CREATE TABLE employees (
    id          int PRIMARY KEY,
    name        text NOT NULL,
    manager_id  int REFERENCES employees(id),
    department  text
);

INSERT INTO employees VALUES
    (1, 'CEO',           NULL, 'exec'),
    (2, 'CTO',              1, 'eng'),
    (3, 'CFO',              1, 'fin'),
    (4, 'Backend Lead',     2, 'eng'),
    (5, 'Frontend Lead',    2, 'eng'),
    (6, 'Engineer A',       4, 'eng'),
    (7, 'Engineer B',       4, 'eng'),
    (8, 'Accountant',       3, 'fin');

CREATE INDEX idx_employees_manager_id ON employees(manager_id);

-- ========== 2. 非递归 CTE ==========
WITH recent_orders AS (
    SELECT customer_id, sum(amount) AS total
    FROM orders
    WHERE created_at >= now() - interval '30 days'
    GROUP BY customer_id
)
SELECT c.name, r.total
FROM customers c
JOIN recent_orders r ON r.customer_id = c.id
WHERE r.total > 1000;

-- ========== 3. 递归：自顶向下求所有下属 ==========
WITH RECURSIVE subordinates AS (
    SELECT id, name, manager_id, 1 AS depth, ARRAY[id] AS path
    FROM employees
    WHERE manager_id = 2
    UNION ALL
    SELECT e.id, e.name, e.manager_id, s.depth + 1, s.path || e.id
    FROM employees e
    JOIN subordinates s ON e.manager_id = s.id
    WHERE NOT e.id = ANY(s.path)
)
SELECT id, name, depth, path FROM subordinates ORDER BY path;

-- ========== 4. 递归：自底向上求汇报链 ==========
WITH RECURSIVE chain AS (
    SELECT id, name, manager_id, 0 AS level FROM employees WHERE id = 6
    UNION ALL
    SELECT m.id, m.name, m.manager_id, c.level + 1
    FROM employees m JOIN chain c ON m.id = c.manager_id
)
SELECT id, name, level FROM chain ORDER BY level;

-- ========== 5. 物化控制对比 ==========
EXPLAIN (ANALYZE, BUFFERS)
WITH c AS MATERIALIZED (SELECT * FROM orders WHERE status = 'paid')
SELECT * FROM c WHERE customer_id = 42;

EXPLAIN (ANALYZE, BUFFERS)
WITH c AS NOT MATERIALIZED (SELECT * FROM orders WHERE status = 'paid')
SELECT * FROM c WHERE customer_id = 42;

-- ========== 6. 数据修改型 CTE：归档 ==========
WITH moved AS (
    DELETE FROM orders WHERE status = 'expired'
    RETURNING id, customer_id, amount
)
INSERT INTO orders_archive (id, customer_id, amount, archived_at)
SELECT id, customer_id, amount, now() FROM moved;

-- ========== 7. 诊断 ==========
SELECT temp_files, pg_size_pretty(temp_bytes) AS temp_used
FROM pg_stat_database WHERE datname = current_database();

SHOW work_mem;