C データエンジニアリング — セッション別ファネル集計 Bad→Good（相関サブクエリ→LAG × gaps-and-islands セッション化 × ROW_NUMBER→QUALIFY × ファネル自己結合→条件付き集約 × APPROX_COUNT_DISTINCT × incremental insert_overwrite + partition/cluster × dbt model contract）（MOps ユーザー行動ログ → セッションファネルマート）

📋 コピー-- models/marts/mart_session_funnel.sql | dbt Core 1.9+
{{ config(
    materialized='incremental',
    incremental_strategy='insert_overwrite',  -- 修正⑥: 冪等な洗い替え
    partition_by={'field':'event_date',
                  'data_type':'date','granularity':'day'},
    cluster_by=['first_channel'],
    on_schema_change='fail'
) }}

WITH source_events AS (
    SELECT user_id, session_user_key, event_time,
           event_type, channel, event_date
    FROM {{ ref('stg_pageview_events') }}
    WHERE event_type IN ('view','add_to_cart','checkout','purchase')
    {% if is_incremental() %}
      -- 修正⑥: パーティション剪定（対象日のみ読む）
      AND event_date = DATE('{{ var("run_date",
            run_started_at.strftime("%Y-%m-%d")) }}')
    {% endif %}
),
-- 修正①: 相関サブクエリ → LAG で前イベント間隔を1パス
flagged AS (
    SELECT *,
      TIMESTAMP_DIFF(event_time,
        LAG(event_time) OVER (PARTITION BY session_user_key
                              ORDER BY event_time),
        MINUTE) AS gap_min
    FROM source_events
),
-- 修正②: gaps-and-islands でセッション連番を採番
sessionized AS (
    SELECT *,
      SUM(CASE WHEN gap_min IS NULL OR gap_min >= 30
               THEN 1 ELSE 0 END)
        OVER (PARTITION BY session_user_key ORDER BY event_time
              ROWS BETWEEN UNBOUNDED PRECEDING
                       AND CURRENT ROW) AS session_seq
    FROM flagged
),
with_session_id AS (
    SELECT *,
      CONCAT(session_user_key,'-',
             CAST(session_seq AS STRING)) AS session_id  -- 下流 join キー
    FROM sessionized
),
-- 修正④: 4回自己結合 → セッション単位の条件付き集約(1パス)
session_funnel AS (
    SELECT
      session_id, session_user_key,
      MIN(event_date)                     AS event_date,
      MIN(event_time)                     AS session_start_at,
      LOGICAL_OR(event_type='view')       AS reached_view,
      LOGICAL_OR(event_type='add_to_cart')AS reached_cart,
      LOGICAL_OR(event_type='checkout')   AS reached_checkout,
      LOGICAL_OR(event_type='purchase')   AS reached_purchase,
      COUNTIF(event_type='purchase')      AS purchase_events,
      COUNT(*)                            AS event_count
    FROM with_session_id
    GROUP BY session_id, session_user_key
)
-- 修正③: 流入チャネルは QUALIFY で「セッション最初の1行」を取得
SELECT
  f.event_date, c.first_channel, f.session_id,
  f.session_user_key, f.session_start_at,
  f.reached_view, f.reached_cart,
  f.reached_checkout, f.reached_purchase,
  f.purchase_events, f.event_count
FROM session_funnel f
JOIN (
  SELECT session_id, channel AS first_channel
  FROM with_session_id
  QUALIFY ROW_NUMBER() OVER (PARTITION BY session_id
                             ORDER BY event_time) = 1   -- 修正③
) c USING (session_id)