-- card_id: SQL-PG-09
-- title: PostgreSQL 14-18 VACUUM 积压与 dead tuples 压力快照（只读、聚合、脱敏）
-- engine: PostgreSQL
-- versions: 14, 15, 16, 17, 18
-- risk: R2
-- purpose: 在数据库变慢、表体积增长或 autovacuum 疑似积压时，形成单行实例级聚合证据，供负责人继续核对长事务、维护窗口、磁盘、写入节奏和表级证据。
-- input: 无；只读取用户表统计聚合。
-- output: 表数量、live/dead tuple估算、最大dead ratio、维护与分析新鲜度和压力桶。
-- permission: pg_read_all_stats 成员或经负责人批准的等价可见性。
-- resource_limit: 单行结果，建议客户端超时 5 秒。
-- validation: 已在 PostgreSQL 14-18 隔离合成删除压力与清理恢复验证；生产及客户环境未验证。
-- failure: 缺少可见性时返回 insufficient_visibility；不得为获取对象身份扩大权限。
-- prohibited: 不得输出schema、表名、用户、SQL、PID、地址或业务对象；不得VACUUM、ANALYZE、写入、改参数、kill、切换或自动修复。

WITH visibility AS (
    SELECT pg_has_role(current_user, 'pg_read_all_stats', 'MEMBER') AS can_observe
),
table_snapshot AS (
    SELECT
        n_live_tup::numeric AS live_rows,
        n_dead_tup::numeric AS dead_rows,
        n_mod_since_analyze::numeric AS modified_since_analyze,
        coalesce(last_autovacuum, last_vacuum) AS last_vacuum_at
    FROM pg_stat_user_tables
    CROSS JOIN visibility
    WHERE visibility.can_observe
),
aggregate_snapshot AS (
    SELECT
        count(*)::integer AS tracked_tables,
        count(*) FILTER (WHERE dead_rows > 0)::integer AS tables_with_dead_rows,
        coalesce(sum(live_rows), 0)::numeric AS live_rows_total,
        coalesce(sum(dead_rows), 0)::numeric AS dead_rows_total,
        coalesce(max(dead_rows), 0)::numeric AS max_dead_rows,
        coalesce(
            max(
                CASE
                    WHEN live_rows + dead_rows = 0 THEN 0
                    ELSE round(dead_rows * 100.0 / (live_rows + dead_rows), 2)
                END
            ),
            0
        )::numeric AS max_dead_ratio_pct,
        count(*) FILTER (
            WHERE live_rows + dead_rows > 0
              AND dead_rows * 100.0 / (live_rows + dead_rows) >= 20
        )::integer AS tables_dead_ratio_20pct_or_more,
        count(*) FILTER (WHERE last_vacuum_at IS NULL)::integer AS tables_never_vacuumed,
        count(*) FILTER (
            WHERE last_vacuum_at IS NULL
               OR last_vacuum_at < current_timestamp - interval '7 days'
        )::integer AS tables_without_vacuum_7d,
        count(*) FILTER (WHERE modified_since_analyze >= 10000)::integer
            AS tables_mod_since_analyze_10000_or_more,
        coalesce(max(modified_since_analyze), 0)::numeric AS max_mod_since_analyze,
        coalesce(
            max(extract(epoch FROM (current_timestamp - last_vacuum_at)))
                FILTER (WHERE last_vacuum_at IS NOT NULL),
            0
        )::numeric AS oldest_vacuum_age_seconds
    FROM table_snapshot
),
ready AS (
    SELECT
        'ready'::text AS observation_status,
        tracked_tables,
        tables_with_dead_rows,
        live_rows_total,
        dead_rows_total,
        max_dead_rows,
        max_dead_ratio_pct,
        tables_dead_ratio_20pct_or_more,
        tables_never_vacuumed,
        tables_without_vacuum_7d,
        tables_mod_since_analyze_10000_or_more,
        max_mod_since_analyze,
        round(oldest_vacuum_age_seconds, 0) AS oldest_vacuum_age_seconds,
        CASE
            WHEN tracked_tables = 0 THEN 'no_user_tables'
            WHEN dead_rows_total >= 1000000 OR tables_dead_ratio_20pct_or_more >= 3
                THEN 'high_evidence_pressure'
            WHEN dead_rows_total >= 100000 OR tables_dead_ratio_20pct_or_more >= 1
                THEN 'review_evidence_pressure'
            ELSE 'low_observed_pressure'
        END::text AS pressure_bucket
    FROM aggregate_snapshot
    CROSS JOIN visibility
    WHERE visibility.can_observe
),
visibility_status AS (
    SELECT
        'insufficient_visibility'::text AS observation_status,
        NULL::integer AS tracked_tables,
        NULL::integer AS tables_with_dead_rows,
        NULL::numeric AS live_rows_total,
        NULL::numeric AS dead_rows_total,
        NULL::numeric AS max_dead_rows,
        NULL::numeric AS max_dead_ratio_pct,
        NULL::integer AS tables_dead_ratio_20pct_or_more,
        NULL::integer AS tables_never_vacuumed,
        NULL::integer AS tables_without_vacuum_7d,
        NULL::integer AS tables_mod_since_analyze_10000_or_more,
        NULL::numeric AS max_mod_since_analyze,
        NULL::numeric AS oldest_vacuum_age_seconds,
        'unknown'::text AS pressure_bucket
    FROM visibility
    WHERE NOT can_observe
)
SELECT * FROM ready
UNION ALL
SELECT * FROM visibility_status
LIMIT 1;
