SQL开发需求：统计每小时刷新表的库存、新增及关闭案件数

嘿，这个需求我之前处理过类似的，咱们一步步拆解来做！首先得明确核心逻辑：要对比当前小时快照和上一小时快照里的案件存在情况，才能算出新增和关闭的数量，同时还要保证所有时间+WorkLevel的组合都能显示（哪怕数据为0，就像示例里的Case Setup 2那样）。

假设前提 #

我先假设你的业务表叫hourly_case_snapshots，包含以下字段：

• case_id：唯一案件ID（主键）
• work_level：案件所属的WorkLevel（比如示例里的Intake、Case Setup 1等）
• snapshot_datetime：每小时快照的时间戳（格式类似2018-01-30 07:00:00，精确到小时）

核心思路 #

1. 生成所有需要统计的时间点+WorkLevel组合：避免遗漏那些没有案件的维度组合，确保输出像示例里那样出现0值的行。
2. 分别统计当前小时、上一小时每个WorkLevel下的案件集合。
3. 通过集合对比，计算新增（当前有、上一小时无）、关闭（上一小时有、当前无）的数量，同时统计当前库存数。

完整SQL代码（以PostgreSQL为例） #

WITH all_snapshot_times AS (
    -- 提取所有存在的小时快照时间点
    SELECT DISTINCT DATE_TRUNC('hour', snapshot_datetime) AS snapshot_hour
    FROM hourly_case_snapshots
),
all_work_levels AS (
    -- 提取所有存在的WorkLevel值
    SELECT DISTINCT work_level
    FROM hourly_case_snapshots
),
time_level_combinations AS (
    -- 生成时间点和WorkLevel的全组合，确保所有维度都覆盖
    SELECT 
        t.snapshot_hour,
        w.work_level
    FROM all_snapshot_times t
    CROSS JOIN all_work_levels w
),
current_hour_cases AS (
    -- 统计当前小时每个WorkLevel的案件数和案件ID集合
    SELECT
        DATE_TRUNC('hour', snapshot_datetime) AS snapshot_hour,
        work_level,
        COUNT(case_id) AS inventory,
        ARRAY_AGG(case_id) AS case_ids
    FROM hourly_case_snapshots
    GROUP BY snapshot_hour, work_level
),
prev_hour_cases AS (
    -- 统计上一小时每个WorkLevel的案件ID集合（关联当前小时的时间点）
    SELECT
        DATE_TRUNC('hour', snapshot_datetime) + INTERVAL '1 hour' AS current_snapshot_hour,
        work_level,
        ARRAY_AGG(case_id) AS prev_case_ids
    FROM hourly_case_snapshots
    GROUP BY DATE_TRUNC('hour', snapshot_datetime), work_level
)
-- 最终统计结果
SELECT
    -- 格式化日期为示例要求的格式（如01/30/2018 7:00 AM）
    TO_CHAR(tlc.snapshot_hour, 'MM/DD/YYYY HH:MI AM') AS "Date",
    tlc.work_level AS "WorkLevel",
    COALESCE(chc.inventory, 0) AS "Inventory",
    -- 计算新增案件数：当前有但上一小时没有的案件
    COALESCE(
        (SELECT COUNT(*) FROM UNNEST(chc.case_ids) c 
         WHERE c NOT IN (SELECT UNNEST(phc.prev_case_ids))),
        0
    ) AS "New",
    -- 计算关闭案件数：上一小时有但当前没有的案件
    COALESCE(
        (SELECT COUNT(*) FROM UNNEST(phc.prev_case_ids) p 
         WHERE p NOT IN (SELECT UNNEST(chc.case_ids))),
        0
    ) AS "Closed"
FROM time_level_combinations tlc
LEFT JOIN current_hour_cases chc 
    ON tlc.snapshot_hour = chc.snapshot_hour 
    AND tlc.work_level = chc.work_level
LEFT JOIN prev_hour_cases phc 
    ON tlc.snapshot_hour = phc.current_snapshot_hour 
    AND tlc.work_level = phc.work_level
ORDER BY tlc.snapshot_hour DESC, tlc.work_level;

关键细节说明 #

• 全组合生成：用CROSS JOIN生成时间和WorkLevel的所有可能组合，保证哪怕某个维度没有案件，也会输出0值行，和示例格式完全匹配。
• 集合对比：用ARRAY_AGG把每个维度的案件ID存成数组，再通过UNNEST展开对比，计算新增和关闭的数量。如果你的数据库不支持数组操作，也可以用EXISTS子查询替代。
• 日期格式化：TO_CHAR是PostgreSQL的写法，MySQL可以用DATE_FORMAT，SQL Server用CONVERT，根据你的数据库类型调整即可。
• 空值处理：用COALESCE把空值转换成0，避免结果出现NULL。

适配MySQL的替代写法（无数组支持） #

如果你的数据库是MySQL，把新增和关闭的计算逻辑改成以下形式即可：

-- 新增数
COALESCE(
    (SELECT COUNT(*) FROM hourly_case_snapshots curr
     WHERE DATE_FORMAT(curr.snapshot_datetime, '%Y-%m-%d %H:00:00') = DATE_FORMAT(tlc.snapshot_hour, '%Y-%m-%d %H:00:00')
       AND curr.work_level = tlc.work_level
       AND NOT EXISTS (
           SELECT 1 FROM hourly_case_snapshots prev
           WHERE DATE_FORMAT(prev.snapshot_datetime, '%Y-%m-%d %H:00:00') = DATE_FORMAT(tlc.snapshot_hour - INTERVAL 1 HOUR, '%Y-%m-%d %H:00:00')
             AND prev.work_level = tlc.work_level
             AND prev.case_id = curr.case_id
       )),
    0
) AS "New",
-- 关闭数
COALESCE(
    (SELECT COUNT(*) FROM hourly_case_snapshots prev
     WHERE DATE_FORMAT(prev.snapshot_datetime, '%Y-%m-%d %H:00:00') = DATE_FORMAT(tlc.snapshot_hour - INTERVAL 1 HOUR, '%Y-%m-%d %H:00:00')
       AND prev.work_level = tlc.work_level
       AND NOT EXISTS (
           SELECT 1 FROM hourly_case_snapshots curr
           WHERE DATE_FORMAT(curr.snapshot_datetime, '%Y-%m-%d %H:00:00') = DATE_FORMAT(tlc.snapshot_hour, '%Y-%m-%d %H:00:00')
             AND curr.work_level = tlc.work_level
             AND curr.case_id = prev.case_id
       )),
    0
) AS "Closed"

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