C データエンジニアリング — 価格変更履歴SCD Type2 × point-in-time JOIN（dbt snapshotがtimestamp strategyだがソースDBの同一バッチ内updated_at不変で価格変更を検知できない × dbt_valid_to IS NULL(現在有効行)とBETWEENのNULL比較が常にUNKNOWNとなりJOIN漏れ × BETWEEN閉区間により境界時刻の注文が2つの価格履歴行に二重マッチ × require_partition_filter未設定でフルスキャンが可能 × product_idクラスタリング未設定 × カテゴリ別日別値引き効果サマリを3箇所が個別集計 × dbt testsがunique/not_nullのみでSCD区間の重複を検知できない）（MOps 販促システムチーム「セール価格が反映されない」「同じ注文が2重カウント」不整合 Bad→Good）

概要

問題

ECサイト MOps 販促システムチームでは、商品の価格変更履歴（通常価格 ⇄ セール価格）を追跡し、「キャンペーン期間中の値引きが実際の注文にどれだけ寄与したか」を測定する基盤 mart_campaign_discount_effect を運用している。

商品マスタDB（product_master、updated_atカラムを持つ）から dbt snapshot（snapshot_strategy: timestamp）で snap_product_price_history を作成し、SCD Type2（dbt_valid_from / dbt_valid_to）として価格変更履歴を保持している。これとorders（注文明細、ordered_atを持つ）をpoint-in-time JOIN（「注文した瞬間に有効だった価格」を突き合わせるJOIN）し、注文時点の価格が通常価格かセール価格かを判定した上で、カテゴリ別・日別に値引き効果をLooker Studioダッシュボードで可視化している。dbt Core 1.9+ / BigQuery Standard SQL / Argo Workflows 3.5+で毎晩1回バッチ実行している。

先月起きた出来事: キャンペーン振り返りで、「一部商品でセール価格が反映されないまま通常価格の注文として集計されていた」「同じ注文が値引き効果の集計に2回カウントされていた」という2つの不整合が発覚し、調査したところ、以下の7つの設計・データ品質・コスト上の問題が見つかった。

問題点を全て洗い出し、dbt snapshotの変更検知方式の見直し（check strategy） × point-in-time JOINのNULL境界・重複区間バグの解消 × require_partition_filterによるフルスキャン防止 × product_idクラスタリング × BI Engine / Materialized Viewによるダッシュボードクエリの重複計算解消 × dbt singular testによるSCD区間整合性検証を活用したBad→Goodリファクタリングを行ってください。

制約・前提条件

• BigQuery Standard SQL、dbt Core 1.9+（snapshot、singular test、Unit Testsがいずれも利用可能）、Argo Workflows 3.5+
• product_master: product_id, price, is_on_sale（BOOLEAN）, updated_at。同一バッチジョブ内で複数カラムを更新してもupdated_atが同一タイムスタンプになるソースDB仕様（アプリ側でトランザクション開始時刻を一括セットしている）
• snap_product_price_history（dbt snapshot出力）: product_id, price, is_on_sale, dbt_valid_from, dbt_valid_to（現在有効な行はdbt_valid_to IS NULL）
• orders: order_id, product_id, ordered_at, quantity, order_amount
• 現状のmart_campaign_discount_effectはo.product_id = p.product_id AND o.ordered_at BETWEEN p.dbt_valid_from AND p.dbt_valid_toでJOINしている
• snap_product_price_history・ordersいずれもテーブルレベルでパーティションフィルタを必須化する設定が入っておらず、Looker Studioの自由記述フィルタ画面から日付条件なしでクエリを発行できてしまう
• JOINキーであるproduct_idにクラスタリングが設定されていない
• 「カテゴリ別・日別の値引き効果サマリ」は経営ダッシュボード・現場責任者ダッシュボード・週次レポート用バッチの3箇所が個別に同じ集計クエリを発行しており、キャッシュや事前計算の仕組みが無い
• dbt testsはsnap_product_price_historyに対してunique(product_id, dbt_valid_from)とnot_null(product_id)のみで、同一product_idのSCD区間が互いに重ならないことを検証するテストが無い

悪いコード (Before)

📋 コピーsnapshots:
  - name: snap_product_price_history
    relation: source('ecommerce', 'product_master')
    config:
      unique_key: product_id
      # 問題①: ソースのupdated_atが同一バッチ内で変化しない仕様なのに
      #        timestamp strategyのままで、価格変更が検知されない
      strategy: timestamp
      updated_at: updated_at

📋 コピー-- 問題②: dbt_valid_to IS NULL(現在有効行)に対しBETWEENで比較すると
--        NULLとの比較が常にUNKNOWNとなりJOIN不成立
-- 問題③: BETWEENは閉区間なので境界時刻ちょうどの注文が
--        隣接する2つの価格履歴行の両方にマッチし二重カウントされる
SELECT o.order_id, o.product_id, o.ordered_at,
       p.price, p.is_on_sale
FROM {{ ref('stg_orders') }} o
JOIN {{ ref('snap_product_price_history') }} p
  ON o.product_id = p.product_id
 AND o.ordered_at BETWEEN p.dbt_valid_from AND p.dbt_valid_to

ヒント（段階的開示）

-- point-in-time JOINの骨格（半開区間 + 番兵値）
WITH price_history AS (
  SELECT product_id, price, is_on_sale, dbt_valid_from,
    COALESCE(dbt_valid_to, TIMESTAMP('9999-12-31')) AS dbt_valid_to_sentinel
  FROM {{ ref('snap_product_price_history') }}
)
SELECT o.order_id, o.product_id, o.ordered_at, p.price, p.is_on_sale
FROM {{ ref('stg_orders') }} o
JOIN price_history p
  ON o.product_id = p.product_id
 AND o.ordered_at >= p.dbt_valid_from          -- 開始は含む
 AND o.ordered_at <  p.dbt_valid_to_sentinel   -- 終了は含まない

-- SCD区間整合性の singular test の骨格
SELECT product_id, dbt_valid_from
FROM {{ ref('snap_product_price_history') }} a
JOIN {{ ref('snap_product_price_history') }} b USING (product_id)
WHERE a.dbt_valid_from < b.dbt_valid_from
  AND a.dbt_valid_to > b.dbt_valid_from   -- 区間が重複

問題点分析（7点）

SCD区間比較図 — Bad vs Good の比較

模範解答

📋 コピーsnapshots:
  - name: snap_product_price_history
    relation: source('ecommerce', 'product_master')
    config:
      unique_key: product_id
      strategy: timestamp
      updated_at: updated_at
      # 問題①: 同一バッチ内でupdated_atが変化しない仕様のため
      #        価格・is_on_sale変更を検知できない

📋 コピーsnapshots:
  - name: snap_product_price_history
    relation: source('ecommerce', 'product_master')
    config:
      schema: snapshots
      unique_key: product_id
      # 修正①: updated_atが同一バッチ内で変化しない仕様のため、
      #        タイムスタンプではなく値そのものの変化で検知する
      strategy: check
      check_cols: [price, is_on_sale]

📋 コピー-- 修正②: dbt_valid_toがNULL(現在有効な行)を番兵値に置き換え、NULL比較によるJOIN漏れを防ぐ
-- 修正③: BETWEEN(閉区間)から半開区間([開始,終了))へ変更し、境界時刻の重複マッチを防ぐ

WITH price_history AS (
  SELECT
    product_id,
    price,
    is_on_sale,
    dbt_valid_from,
    -- 修正②: NULL(現在有効)を遠い未来の番兵値に置き換える
    COALESCE(dbt_valid_to, TIMESTAMP('9999-12-31')) AS dbt_valid_to_sentinel
  FROM {{ ref('snap_product_price_history') }}
),

orders_priced AS (
  SELECT
    o.order_id,
    o.product_id,
    o.ordered_at,
    o.quantity,
    o.order_amount,
    p.price AS price_at_order_time,
    p.is_on_sale AS was_on_sale
  FROM {{ ref('stg_orders') }} o
  JOIN price_history p
    ON o.product_id = p.product_id
   -- 修正③: 開始時刻は含み(>=)、終了時刻は含まない(<)半開区間にすることで
   --        隣接する2つの履歴行の境界時刻ちょうどの注文が二重にマッチしないようにする
   AND o.ordered_at >= p.dbt_valid_from
   AND o.ordered_at <  p.dbt_valid_to_sentinel
)

SELECT * FROM orders_priced

📋 コピーCREATE TABLE `mops-project.snapshots.snap_product_price_history`
(
  product_id STRING NOT NULL,
  price NUMERIC NOT NULL,
  is_on_sale BOOL NOT NULL,
  dbt_valid_from TIMESTAMP NOT NULL,
  dbt_valid_to TIMESTAMP
)
PARTITION BY DATE(dbt_valid_from);
-- 問題④⑤: require_partition_filter無し、CLUSTER BY無し

📋 コピー-- 修正④: パーティションフィルタなしのクエリをBigQuery側でエラーにし、
--        意図しないフルスキャンを構造的に防ぐ
-- 修正⑤: product_idでのJOINが主要アクセスパターンのためクラスタリングを設定
CREATE TABLE IF NOT EXISTS `mops-project.snapshots.snap_product_price_history`
(
  product_id STRING NOT NULL,
  price NUMERIC NOT NULL,
  is_on_sale BOOL NOT NULL,
  dbt_valid_from TIMESTAMP NOT NULL,
  dbt_valid_to TIMESTAMP
)
PARTITION BY DATE(dbt_valid_from)
CLUSTER BY product_id
OPTIONS (require_partition_filter = true);   -- 修正④

CREATE TABLE IF NOT EXISTS `mops-project.raw_events.orders`
(
  order_id STRING NOT NULL,
  product_id STRING NOT NULL,
  ordered_at TIMESTAMP NOT NULL,
  quantity INT64,
  order_amount NUMERIC
)
PARTITION BY DATE(ordered_at)
CLUSTER BY product_id
OPTIONS (require_partition_filter = true);   -- 修正④

📋 コピー-- 修正⑥: 経営ダッシュボード・現場ダッシュボード・週次バッチの3箇所が
--        個別に発行していた集計を1つのdbtモデルに集約し、全員がここを参照する。
-- BI Engineの予約容量を有効化しておくことで、このテーブルへの
-- 繰り返しクエリはインメモリで高速応答され、追加のスキャン課金も発生しにくくなる。
{{ config(materialized='table', partition_by={'field': 'order_date', 'data_type': 'date'}, cluster_by=['category']) }}

SELECT
  DATE(o.ordered_at) AS order_date,
  c.category,
  o.was_on_sale,
  COUNT(*) AS order_count,
  SUM(o.order_amount) AS total_amount,
  AVG(o.order_amount) AS avg_order_amount   -- 客単価
FROM {{ ref('mart_campaign_discount_effect') }} o
JOIN {{ ref('dim_product') }} c USING (product_id)
GROUP BY order_date, category, was_on_sale

📋 コピー-- 以後この事前計算済みテーブルのみを参照する（require_partition_filter適用）
SELECT category, was_on_sale, SUM(total_amount) AS total_amount
FROM `mops-project.marts.mart_campaign_discount_effect_daily_summary`
WHERE order_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
GROUP BY category, was_on_sale

📋 コピー-- 修正⑦: 同一product_id内でSCD区間が重複していないことを検証する。
--        1行でも返れば test failure(=区間の重なりがある)。
SELECT
  a.product_id,
  a.dbt_valid_from AS interval_a_start,
  b.dbt_valid_from AS interval_b_start
FROM {{ ref('snap_product_price_history') }} a
JOIN {{ ref('snap_product_price_history') }} b
  ON a.product_id = b.product_id
 AND a.dbt_valid_from < b.dbt_valid_from
WHERE a.dbt_valid_to IS NOT NULL          -- 現在有効な行(NULL)は比較対象から除外
  AND a.dbt_valid_to > b.dbt_valid_from   -- 前の区間の終了が次の区間の開始より後ろ＝重複

実行例（input → output）

input（改善前のシナリオ）:

商品Aは同一バッチジョブ内でpriceが1,000円→800円に変わったがupdated_atは変化しなかった。また商品Bは価格変更履歴がdbt_valid_from=2026-09-01 00:00:00→dbt_valid_to=2026-09-05 00:00:00（旧価格）、dbt_valid_from=2026-09-05 00:00:00→dbt_valid_to=NULL（新価格）という2行を持つ。ordered_at=2026-09-05 00:00:00（境界ちょうど）の注文と、ordered_at=2026-09-10（現在有効な行が対象）の注文が存在する。

output（Before・問題のある実装）:

商品Aはsnap_product_price_historyに新しい行が作られず、800円の注文が1,000円時点の価格として集計され続ける。商品Bの2026-09-05 00:00:00ちょうどの注文はBETWEENの閉区間により旧価格行・新価格行の両方にマッチし2重カウントされる。2026-09-10の注文はdbt_valid_to IS NULLとのBETWEEN評価がUNKNOWNになりJOIN結果から消える。

output（After・改善後）:

check strategyにより商品Aの価格変更（1,000円→800円）が正しく新しいsnapshot行として記録される。商品Bの2026-09-05 00:00:00の注文は半開区間[dbt_valid_from, dbt_valid_to_sentinel)により新価格行のみにマッチし2重カウントされない。2026-09-10の注文は番兵値9999-12-31により現在有効な行と正しくJOINされる。singular testをdbt buildに含めることで、今後同様の区間重複がSCDデータに混入した場合はCIで検知される。

ポイント解説

実務への応用

MOpsチームが扱う「価格変更履歴 × 注文」のようなSCD Type2×point-in-time JOINの構造は、価格に限らず会員ランク・配送料率・クーポン適用条件など「ある時点で有効だった設定値」を後から突き合わせるあらゆる分析基盤で繰り返し登場する。今回のNULL番兵値化・半開区間化は、dbt snapshotを使うプロジェクトで再利用できる汎用パターンであり、既存の他のSCDモデル（会員ランク履歴など）に同じ観点でレビューをかけることで、同種のバグを未然に洗い出せる。

今日のまとめ

次のステップ

• 発展問題: 今回は価格が1系統だが、「地域別価格」「会員ランク別価格」のように同時に複数のSCDディメンションを持つ場合、point-in-time JOINはどう設計を変えるべきか（複合キーでのSCD管理・区間整合性テストの拡張）を設計せよ。
• 参考: dbt snapshot strategy（timestamp / check） / SCD Type2 point-in-time join / BigQuery require_partition_filter / BI Engine / dbt singular test

自己評価（あとで記入）