SQL バッチ処理 — 論理削除・ページネーションの応用

WITH daily_diff AS (
  SELECT
    product_id, snapshot_date, quantity,
    LAG(quantity, 1) OVER (               -- 同一商品内で日付順に並べた際の「1つ前」の在庫数を取得
      PARTITION BY product_id
      ORDER BY     snapshot_date
    ) AS prev_quantity
  FROM stock_snapshots
)

SELECT
  product_id, snapshot_date, quantity, prev_quantity,
  quantity - prev_quantity AS diff           -- 当日と前日の差分を計算
FROM  daily_diff
WHERE prev_quantity IS NOT NULL              -- 初日（前日データなし）を除外
  AND quantity <> prev_quantity              -- 在庫数に変化があった行のみ抽出
ORDER BY product_id, snapshot_date;

/*
  実行順序:
  [CTE: daily_diff]
  1. FROM stock_snapshots  → 行を読み込む
  2. LAG()                 → ウィンドウ関数を評価（行数は保持）

  [メインクエリ]
  3. FROM daily_diff       → CTE を読み込む
  4. WHERE                 → 行を絞り込む
  5. SELECT                → 列を評価
  6. ORDER BY              → 並び替えて出力
*/

WITH paged AS (
  SELECT
    order_id, customer_id, amount, status
  FROM  orders
  WHERE status   = 'pending'
    AND order_id > 1003       -- カーソル条件: 前回取得の最終IDより大きいものを対象
  ORDER BY order_id           -- カーソルと同じ列でソート
  LIMIT    5                  -- 次のページの最大件数
)

SELECT
  *,
  MAX(order_id) OVER () AS next_cursor  -- 取得した5件の末尾を次ページカーソルとして全行に付与
FROM paged
ORDER BY order_id;

/*
  実行順序:
  [CTE: paged]
  1. FROM orders   → 行を読み込む
  2. WHERE         → 行を絞り込む
  3. ORDER BY      → 並び替えて出力
  4. LIMIT         → 件数を制限

  [メインクエリ]
  5. FROM paged    → CTE を読み込む
  6. SELECT        → ウィンドウ関数を評価（行数は保持）
*/

-- ① 前回バッチ実行時刻を取得するCTE
WITH last_checkpoint AS (
  SELECT last_run_at
  FROM   batch_checkpoints
  WHERE  job_name = 'order_sync'
),
diff_summary AS (                         -- ② 差分対象レコードを抽出して集計するCTE
  SELECT
    o.customer_id,
    COUNT(*)          AS order_count,     -- 注文件数
    SUM(o.amount)     AS total_amount,    -- 合計金額
    MAX(o.updated_at) AS last_updated_at  -- グループ内の最新更新日時
  FROM        orders o
  CROSS JOIN  last_checkpoint lc          -- 前回実行時刻(1行)を全行に付与
  WHERE o.updated_at >= lc.last_run_at    -- 前回バッチ実行以降に更新されたデータ
    AND o.updated_at <  TIMESTAMP '2024-06-01 06:30'             -- バッチ実行中の新規更新は次回に回すため除外
  GROUP BY o.customer_id
)

-- 集計結果を出力
SELECT * FROM diff_summary ORDER BY customer_id;

-- ③ バッチ完了後にチェックポイントを更新（別クエリとして実行）
UPDATE batch_checkpoints
SET    last_run_at = TIMESTAMP '2024-06-01 06:30'
WHERE  job_name    = 'order_sync';

/*
  ②の実行順序:
  1. FROM / CROSS JOIN  → 結合（直積）
  2. WHERE              → 行を絞り込む
  3. GROUP BY           → グループ化
  4. SELECT             → 集計関数を評価
*/

WITH recent_runs AS (
  -- ① 各ジョブの直近5件を抽出するため、最新順に連番を振る
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY job_name
      ORDER BY     started_at DESC
    ) AS rn
  FROM job_runs
)

SELECT
  job_name,
  COUNT(*) FILTER (WHERE status = 'success') AS success_count,          -- ② 成功・失敗の回数をそれぞれカウント
  COUNT(*) FILTER (WHERE status = 'failed')  AS fail_count,
  ROUND(                                                                -- ③ 実行時間の平均を秒単位で計算し、小数第1位で丸める
    AVG(EXTRACT(EPOCH FROM (ended_at - started_at)))
  , 1) AS avg_sec,
  BOOL_OR(                                                              -- ④ ストール（停止）検知：1件でも30分以上 running なら TRUE
    status = 'running'
    AND started_at < TIMESTAMP '2024-06-01 06:30' - INTERVAL '30 minutes'
  ) AS is_stalled,
  CASE                                                                  -- ⑤ アラート判定：直近5回中3回以上の失敗でアラート
    WHEN COUNT(*) FILTER (WHERE status = 'failed') >= 3 THEN '! ALERT'
    ELSE 'OK'
  END AS alert

FROM  recent_runs
WHERE rn <= 5
GROUP BY job_name
ORDER BY job_name;

/*
  実行順序:
  1. FROM / ROW_NUMBER  → ウィンドウ関数を評価（行数は保持）
  2. FROM(CTE) / WHERE  → 行を絞り込む
  3. GROUP BY           → グループ化
  4. SELECT             → 集計関数を評価
  5. ORDER BY           → 並び替えて出力
*/