빅쿼리 데이터폼

-- tables/dim_customers.sqlx
config {
  type: "table",
  schema: "analytics",
  description: "고객 차원 테이블"
}

SELECT 
  customer_id,
  customer_name,
  email,
  registration_date,
  last_order_date
FROM ${ref("raw_customers")}
WHERE customer_id IS NOT NULL

-- definitions/vw_sales_summary.sqlx
config {
  type: "view",
  description: "매출 요약 뷰"
}

SELECT 
  DATE(order_date) as order_date,
  SUM(total_amount) as daily_sales,
  COUNT(DISTINCT order_id) as order_count
FROM ${ref("fact_orders")}
GROUP BY DATE(order_date)

-- tables/fact_orders_incremental.sqlx
config {
  type: "incremental",
  uniqueKey: ["order_id"],
  updatePartitionFilter: "order_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)"
}

SELECT 
  order_id,
  customer_id,
  order_date,
  total_amount,
  CURRENT_TIMESTAMP() as processed_at
FROM ${ref("raw_orders")}

${ when(incremental(), `WHERE order_date >= (SELECT MAX(order_date) FROM ${self()})`) }

-- definitions/assertions/assert_positive_amounts.sqlx
config {
  type: "assertion",
  description: "주문 금액이 음수가 아닌지 확인"
}

SELECT *
FROM ${ref("fact_orders")}
WHERE total_amount < 0

-- ref() 함수: 동일 프로젝트 내 테이블 참조
SELECT * FROM ${ref("source_table")}

-- resolve() 함수: 외부 테이블 참조  
SELECT * FROM ${resolve("external_dataset.external_table")}

-- self() 함수: 증분 테이블에서 자기 자신 참조
WHERE date > (SELECT MAX(date) FROM ${self()})

-- 스테이징: stg_[source]_[entity].sqlx
-- definitions/staging/stg_ecommerce_orders.sqlx

-- 차원 테이블: dim_[entity].sqlx  
-- definitions/marts/dim_customers.sqlx

-- 팩트 테이블: fact_[process].sqlx
-- definitions/marts/fact_sales.sqlx

-- 어서션: assert_[condition].sqlx
-- definitions/assertions/assert_unique_order_ids.sqlx

-- definitions/marts/dim_products.sqlx
config {
  type: "table",
  schema: "analytics",
  description: "제품 차원 테이블 - 모든 제품 정보를 포함",
  columns: {
    product_id: "제품 고유 식별자",
    product_name: "제품명",
    category: "제품 카테고리",
    price: "제품 가격",
    created_at: "레코드 생성 시간"
  },
  bigquery: {
    partitionBy: "DATE(created_at)",
    clusterBy: ["category", "product_id"]
  }
}

SELECT 
  p.product_id,
  p.product_name,
  c.category_name as category,
  p.price,
  p.created_at,
  CURRENT_TIMESTAMP() as processed_at
FROM ${ref("raw_products")} p
JOIN ${ref("raw_categories")} c
  ON p.category_id = c.category_id
WHERE p.is_active = true

-- definitions/marts/fact_sales_daily.sqlx
config {
  type: "table",
  bigquery: {
    partitionBy: "DATE(order_date)",
    partitionExpirationDays: 365,
    clusterBy: ["customer_id", "product_id"],
    requirePartitionFilter: true
  }
}

SELECT 
  DATE(order_date) as order_date,
  customer_id,
  product_id,
  SUM(quantity) as total_quantity,
  SUM(total_amount) as total_sales,
  COUNT(DISTINCT order_id) as order_count
FROM ${ref("stg_orders")}
WHERE DATE(order_date) >= '2023-01-01'
GROUP BY 1, 2, 3

-- definitions/staging/stg_events_incremental.sqlx
config {
  type: "incremental",
  uniqueKey: ["event_id"],
  bigquery: {
    partitionBy: "DATE(event_timestamp)",
    updatePartitionFilter: "DATE(event_timestamp) >= DATE_SUB(CURRENT_DATE(), INTERVAL 3 DAY)"
  }
}

SELECT 
  event_id,
  user_id,
  event_type,
  event_timestamp,
  event_properties,
  CURRENT_TIMESTAMP() as processed_at
FROM ${ref("raw_events")}

${ when(incremental(), `
WHERE event_timestamp > (
  SELECT COALESCE(MAX(event_timestamp), TIMESTAMP('1900-01-01'))
  FROM ${self()}
)
`) }

-- definitions/marts/dim_customers_scd2.sqlx
config {
  type: "incremental",
  uniqueKey: ["customer_id", "effective_date"],
  bigquery: {
    updatePartitionFilter: "effective_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)"
  }
}

WITH source_data AS (
  SELECT 
    customer_id,
    customer_name,
    email,
    address,
    phone,
    updated_at as effective_date,
    LEAD(updated_at) OVER (PARTITION BY customer_id ORDER BY updated_at) as end_date,
    ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY updated_at DESC) as rn
  FROM ${ref("stg_customers")}
  ${ when(incremental(), `
  WHERE updated_at > (
    SELECT COALESCE(MAX(effective_date), DATE('1900-01-01'))
    FROM ${self()}
  )
  `) }
)

SELECT 
  customer_id,
  customer_name,
  email,
  address,
  phone,
  effective_date,
  COALESCE(end_date, DATE('9999-12-31')) as end_date,
  CASE WHEN end_date IS NULL THEN true ELSE false END as is_current
FROM source_data

-- definitions/marts/vw_customer_metrics.sqlx
config {
  type: "view",
  description: "고객별 주요 지표 뷰"
}

SELECT 
  c.customer_id,
  c.customer_name,
  COUNT(DISTINCT o.order_id) as total_orders,
  SUM(o.total_amount) as total_spent,
  AVG(o.total_amount) as avg_order_value,
  MIN(o.order_date) as first_order_date,
  MAX(o.order_date) as last_order_date,
  DATE_DIFF(CURRENT_DATE(), MAX(o.order_date), DAY) as days_since_last_order
FROM ${ref("dim_customers")} c
LEFT JOIN ${ref("fact_orders")} o
  ON c.customer_id = o.customer_id
GROUP BY 1, 2

-- definitions/marts/mv_daily_sales.sqlx
config {
  type: "view",
  materialized: true,
  bigquery: {
    partitionBy: "DATE(order_date)",
    clusterBy: ["region", "category"]
  }
}

SELECT 
  DATE(order_date) as order_date,
  p.category,
  c.region,
  COUNT(DISTINCT o.order_id) as order_count,
  COUNT(DISTINCT o.customer_id) as customer_count,
  SUM(o.total_amount) as total_sales,
  AVG(o.total_amount) as avg_order_value
FROM ${ref("fact_orders")} o
JOIN ${ref("dim_products")} p ON o.product_id = p.product_id
JOIN ${ref("dim_customers")} c ON o.customer_id = c.customer_id
WHERE DATE(order_date) >= DATE_SUB(CURRENT_DATE(), INTERVAL 90 DAY)
GROUP BY 1, 2, 3

-- definitions/marts/fact_sales_pivot.sqlx
config { type: "table" }

SELECT 
  order_date,
  customer_id,
  ${pivot('payment_method', ['credit_card', 'debit_card', 'cash'], 'SUM', 'total_amount')}
FROM ${ref("fact_orders")}
WHERE ${dateFilter('order_date', 90)}
GROUP BY order_date, customer_id

-- definitions/staging/stg_orders.sqlx
config {
  type: "table",
  schema: dataform.projectConfig.vars.target_schema
}

SELECT 
  order_id,
  customer_id,
  order_date,
  total_amount,
  
  -- 개발환경에서는 데이터 샘플링
  ${ when(dataform.projectConfig.vars.environment === 'dev', `
  FROM ${ref("raw_orders")}
  WHERE MOD(ABS(FARM_FINGERPRINT(CAST(order_id AS STRING))), 100) < 10
  `) }
  
  ${ when(dataform.projectConfig.vars.environment !== 'dev', `
  FROM ${ref("raw_orders")}
  `) }

-- definitions/marts/fact_orders_v2.sqlx
config {
  type: "table",
  disabled: dataform.projectConfig.vars.enable_v2 !== true
}

-- 새로운 버전의 테이블 정의
SELECT 
  order_id,
  customer_id,
  order_date,
  total_amount,
  -- 새 컬럼 추가
  discount_amount,
  tax_amount,
  net_amount
FROM ${ref("stg_orders")}

-- definitions/marts/customer_cohort_analysis.sqlx
config {
  type: "table",
  description: "고객 코호트 분석 테이블"
}

WITH customer_orders AS (
  SELECT 
    customer_id,
    order_date,
    total_amount,
    ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY order_date) as order_number,
    MIN(order_date) OVER (PARTITION BY customer_id) as first_order_date
  FROM ${ref("fact_orders")}
),

cohort_data AS (
  SELECT 
    customer_id,
    DATE_TRUNC(first_order_date, MONTH) as cohort_month,
    DATE_DIFF(DATE_TRUNC(order_date, MONTH), DATE_TRUNC(first_order_date, MONTH), MONTH) as period_number,
    total_amount
  FROM customer_orders
)

SELECT 
  cohort_month,
  period_number,
  COUNT(DISTINCT customer_id) as customers,
  SUM(total_amount) as revenue,
  AVG(total_amount) as avg_revenue_per_customer
FROM cohort_data
GROUP BY cohort_month, period_number
ORDER BY cohort_month, period_number

-- definitions/staging/stg_events_parsed.sqlx
config { type: "table" }

SELECT 
  event_id,
  user_id,
  event_timestamp,
  event_type,
  
  -- JSON 속성 파싱
  JSON_EXTRACT_SCALAR(event_properties, '$.page_url') as page_url,
  JSON_EXTRACT_SCALAR(event_properties, '$.referrer') as referrer,
  CAST(JSON_EXTRACT_SCALAR(event_properties, '$.session_duration') AS INT64) as session_duration,
  
  -- 중첩된 JSON 배열 처리
  ARRAY(
    SELECT JSON_EXTRACT_SCALAR(item, '$.product_id')
    FROM UNNEST(JSON_EXTRACT_ARRAY(event_properties, '$.products')) as item
  ) as product_ids,
  
  -- 조건부 JSON 파싱
  CASE event_type
    WHEN 'purchase' THEN CAST(JSON_EXTRACT_SCALAR(event_properties, '$.amount') AS FLOAT64)
    ELSE NULL
  END as purchase_amount

FROM ${ref("raw_events")}
WHERE JSON_EXTRACT_SCALAR(event_properties, '$.valid') = 'true'

-- definitions/marts/customer_lifetime_value.sqlx
config {
  type: "table",
  dependencies: ["dim_customers", "fact_orders"] -- 명시적 의존성 선언
}

SELECT 
  c.customer_id,
  c.customer_name,
  c.acquisition_date,
  
  -- 의존하는 테이블들 참조
  o.total_orders,
  o.total_spent,
  o.avg_order_value,
  
  -- CLV 계산
  o.total_spent * 1.2 as predicted_ltv
  
FROM ${ref("dim_customers")} c
JOIN (
  SELECT 
    customer_id,
    COUNT(*) as total_orders,
    SUM(total_amount) as total_spent,
    AVG(total_amount) as avg_order_value
  FROM ${ref("fact_orders")}
  GROUP BY customer_id
) o ON c.customer_id = o.customer_id

-- definitions/staging/stg_customer_metrics.sqlx
config {
  type: "table",
  preOps: [`
    CREATE OR REPLACE TABLE ${self()}_temp AS
    SELECT customer_id, COUNT(*) as order_count
    FROM ${ref("raw_orders")}
    GROUP BY customer_id
  `]
}

SELECT 
  c.*,
  COALESCE(t.order_count, 0) as total_orders
FROM ${ref("raw_customers")} c
LEFT JOIN ${self()}_temp t
  ON c.customer_id = t.customer_id

-- definitions/marts/weekend_sales_report.sqlx
config {
  type: "table",
  disabled: "${new Date().getDay() !== 0 && new Date().getDay() !== 6}" -- 주말에만 실행
}

SELECT 
  DATE(order_date) as weekend_date,
  SUM(total_amount) as weekend_sales
FROM ${ref("fact_orders")}
WHERE EXTRACT(DAYOFWEEK FROM order_date) IN (1, 7) -- 일요일, 토요일
  AND DATE(order_date) >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
GROUP BY DATE(order_date)

-- definitions/marts/incremental_report.sqlx
config {
  type: "table",
  disabled: !${tableHasData(ref("stg_daily_updates"))}
}

SELECT * FROM ${ref("stg_daily_updates")}

-- definitions/assertions/assert_unique_customer_ids.sqlx
config {
  type: "assertion",
  description: "고객 ID가 중복되지 않는지 확인"
}

SELECT customer_id
FROM ${ref("dim_customers")}
GROUP BY customer_id
HAVING COUNT(*) > 1

-- definitions/assertions/assert_order_business_rules.sqlx
config {
  type: "assertion",
  description: "주문 데이터 비즈니스 룰 검증"
}

SELECT 
  order_id,
  'negative_amount' as violation_type,
  total_amount
FROM ${ref("fact_orders")}
WHERE total_amount < 0

UNION ALL

SELECT 
  order_id,
  'future_date' as violation_type,
  order_date
FROM ${ref("fact_orders")}
WHERE DATE(order_date) > CURRENT_DATE()

UNION ALL

SELECT 
  order_id,
  'missing_customer' as violation_type,
  customer_id
FROM ${ref("fact_orders")} o
LEFT JOIN ${ref("dim_customers")} c
  ON o.customer_id = c.customer_id
WHERE c.customer_id IS NULL

-- definitions/assertions/data_quality_metrics.sqlx
config {
  type: "assertion",
  description: "데이터 품질 임계값 검증"
}

WITH quality_metrics AS (
  SELECT 
    'customers' as table_name,
    COUNT(*) as total_rows,
    COUNT(DISTINCT customer_id) as unique_customers,
    COUNTIF(email IS NULL OR email = '') as missing_emails,
    COUNTIF(NOT REGEXP_CONTAINS(email, r'^[^@]+@[^@]+\.[^@]+$')) as invalid_emails
  FROM ${ref("dim_customers")}
)

SELECT *
FROM quality_metrics
WHERE 
  -- 중복률이 5% 초과
  (total_rows - unique_customers) / total_rows > 0.05
  -- 이메일 누락률이 10% 초과  
  OR missing_emails / total_rows > 0.10
  -- 유효하지 않은 이메일이 2% 초과
  OR invalid_emails / total_rows > 0.02

-- definitions/assertions/detect_data_regression.sqlx
config {
  type: "assertion",
  description: "전일 대비 주요 메트릭 변화 감지"
}

WITH current_metrics AS (
  SELECT 
    DATE(CURRENT_DATE()) as metric_date,
    COUNT(DISTINCT customer_id) as active_customers,
    COUNT(DISTINCT order_id) as total_orders,
    SUM(total_amount) as total_revenue
  FROM ${ref("fact_orders")}
  WHERE DATE(order_date) = CURRENT_DATE()
),

previous_metrics AS (
  SELECT 
    DATE(CURRENT_DATE() - 1) as metric_date,
    COUNT(DISTINCT customer_id) as active_customers,
    COUNT(DISTINCT order_id) as total_orders,
    SUM(total_amount) as total_revenue
  FROM ${ref("fact_orders")}
  WHERE DATE(order_date) = DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY)
)

SELECT 
  'regression_detected' as alert_type,
  c.metric_date,
  c.active_customers as current_customers,
  p.active_customers as previous_customers,
  ABS(c.active_customers - p.active_customers) / p.active_customers as customer_change_pct
FROM current_metrics c
CROSS JOIN previous_metrics p
WHERE 
  -- 고객 수가 50% 이상 변화
  ABS(c.active_customers - p.active_customers) / p.active_customers > 0.5
  -- 또는 매출이 70% 이상 변화
  OR ABS(c.total_revenue - p.total_revenue) / p.total_revenue > 0.7

-- definitions/staging/stg_orders_v2.sqlx
config {
  type: "table",
  schema: "staging_v2", -- 새 버전은 별도 스키마에 배포
  disabled: dataform.projectConfig.vars.enable_v2 !== true
}

-- 개선된 스테이징 로직
SELECT 
  order_id,
  customer_id,
  order_date,
  total_amount,
  -- 새로운 데이터 품질 검증
  CASE 
    WHEN total_amount < 0 THEN 0
    ELSE total_amount
  END as cleaned_total_amount
FROM ${ref("raw_orders")}
WHERE order_date >= '2023-01-01'

-- definitions/marts/fact_orders_canary.sqlx
config {
  type: "table",
  disabled: dataform.projectConfig.vars.enable_canary !== true
}

SELECT *
FROM ${ref("stg_orders_v2")}
WHERE ${isCanaryCustomer("customer_id")}

-- 실행 로그 쿼리 예제
SELECT 
  timestamp,
  severity,
  jsonPayload.workflow_id,
  jsonPayload.action_name,
  jsonPayload.status,
  jsonPayload.execution_time_ms
FROM `project.dataset.dataform_logs`
WHERE DATE(timestamp) = CURRENT_DATE()
  AND jsonPayload.status = 'FAILED'
ORDER BY timestamp DESC

-- definitions/monitoring/query_performance_analysis.sqlx
config {
  type: "table",
  description: "Dataform 쿼리 성능 분석"
}

SELECT 
  job_id,
  user_email,
  project_id,
  creation_time,
  start_time,
  end_time,
  TIMESTAMP_DIFF(end_time, start_time, SECOND) as execution_duration_sec,
  total_bytes_processed / 1024 / 1024 / 1024 as gb_processed,
  total_slot_ms / 1000 / 60 as slot_minutes,
  query
FROM `region-us`.INFORMATION_SCHEMA.JOBS_BY_PROJECT
WHERE DATE(creation_time) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
  AND job_type = 'QUERY'
  AND state = 'DONE'
  AND CONTAINS_SUBSTR(query, 'dataform')
ORDER BY execution_duration_sec DESC
LIMIT 100

-- definitions/monitoring/resource_usage_daily.sqlx
config {
  type: "incremental",
  uniqueKey: ["usage_date", "project_id"]
}

SELECT 
  DATE(usage_start_time) as usage_date,
  project.id as project_id,
  service.description as service_name,
  sku.description as sku_description,
  SUM(usage.amount) as total_usage,
  SUM(cost) as total_cost_usd,
  currency
FROM `project.dataset.gcp_billing_export_v1_BILLING_ACCOUNT_ID`
WHERE DATE(usage_start_time) >= CURRENT_DATE() - 7
  AND service.description LIKE '%BigQuery%'
${ when(incremental(), `
  AND DATE(usage_start_time) > (SELECT MAX(usage_date) FROM ${self()})
`) }
GROUP BY 1, 2, 3, 4, 7

-- definitions/monitoring/data_lineage.sqlx
config {
  type: "table",
  description: "데이터 린리지 메타데이터"
}

WITH table_dependencies AS (
  SELECT 
    'fact_orders' as target_table,
    ['stg_orders', 'dim_customers', 'dim_products'] as source_tables
  UNION ALL
  SELECT 
    'dim_customers',
    ['raw_customers', 'raw_customer_addresses']
  -- ... 다른 테이블 의존성들
)

SELECT 
  target_table,
  source_table,
  'dataform' as transformation_tool,
  CURRENT_TIMESTAMP() as last_updated
FROM table_dependencies
CROSS JOIN UNNEST(source_tables) as source_table

-- definitions/monitoring/column_lineage.sqlx
config { type: "table" }

SELECT 
  'fact_orders' as target_table,
  'customer_id' as target_column,
  'stg_orders' as source_table,
  'customer_id' as source_column,
  'direct' as transformation_type

UNION ALL

SELECT 
  'fact_orders',
  'total_amount_with_tax',
  'stg_orders', 
  'total_amount',
  'calculated' -- total_amount * (1 + tax_rate)

-- ... 추가 컬럼 린리지 매핑

-- definitions/staging/stg_ecommerce_orders.sqlx
config {
  type: "table",
  description: "E-commerce 주문 데이터 스테이징",
  bigquery: {
    partitionBy: "DATE(order_timestamp)",
    clusterBy: ["customer_id", "status"]
  }
}

SELECT 
  -- 기본 주문 정보
  CAST(order_id AS STRING) as order_id,
  CAST(customer_id AS STRING) as customer_id,
  order_timestamp,
  DATE(order_timestamp) as order_date,
  
  -- 주문 상태 정규화
  CASE status
    WHEN 'completed' THEN 'COMPLETED'
    WHEN 'cancelled' THEN 'CANCELLED'
    WHEN 'pending' THEN 'PENDING'
    WHEN 'shipped' THEN 'SHIPPED'
    ELSE 'UNKNOWN'
  END as order_status,
  
  -- 금액 정보 정제
  CAST(subtotal AS FLOAT64) as subtotal,
  CAST(tax_amount AS FLOAT64) as tax_amount,
  CAST(shipping_cost AS FLOAT64) as shipping_cost,
  CAST(subtotal AS FLOAT64) + CAST(tax_amount AS FLOAT64) + CAST(shipping_cost AS FLOAT64) as total_amount,
  
  -- 할인 정보
  COALESCE(CAST(discount_amount AS FLOAT64), 0) as discount_amount,
  discount_code,
  
  -- 배송 정보
  shipping_address,
  REGEXP_EXTRACT(shipping_address, r', ([A-Z]{2}) \d{5}') as shipping_state,
  
  -- 메타데이터
  created_at,
  updated_at,
  CURRENT_TIMESTAMP() as processed_at

FROM ${resolve(dataform.projectConfig.vars.source_dataset + '.raw_orders')}
WHERE 
  order_timestamp IS NOT NULL
  AND customer_id IS NOT NULL
  AND CAST(subtotal AS FLOAT64) >= 0

-- definitions/staging/stg_order_items.sqlx
config {
  type: "incremental",
  uniqueKey: ["order_id", "product_id"],
  bigquery: {
    partitionBy: "DATE(order_timestamp)"
  }
}

SELECT 
  CAST(order_id AS STRING) as order_id,
  CAST(product_id AS STRING) as product_id,
  order_timestamp,
  
  -- 수량 및 가격 정보
  CAST(quantity AS INT64) as quantity,
  CAST(unit_price AS FLOAT64) as unit_price,
  CAST(quantity AS INT64) * CAST(unit_price AS FLOAT64) as line_total,
  
  -- 할인 적용
  COALESCE(CAST(item_discount AS FLOAT64), 0) as item_discount,
  (CAST(quantity AS INT64) * CAST(unit_price AS FLOAT64)) - COALESCE(CAST(item_discount AS FLOAT64), 0) as line_total_after_discount,
  
  CURRENT_TIMESTAMP() as processed_at

FROM ${resolve(dataform.projectConfig.vars.source_dataset + '.raw_order_items')}
WHERE 
  order_id IS NOT NULL 
  AND product_id IS NOT NULL
  AND CAST(quantity AS INT64) > 0
  AND CAST(unit_price AS FLOAT64) >= 0

${ when(incremental(), `
  AND order_timestamp > (
    SELECT COALESCE(MAX(order_timestamp), TIMESTAMP('1900-01-01'))
    FROM ${self()}
  )
`) }

-- definitions/marts/dim_customers.sqlx
config {
  type: "table",
  description: "고객 차원 테이블 - SCD Type 2",
  bigquery: {
    clusterBy: ["customer_id", "is_active"]
  }
}

WITH customer_history AS (
  SELECT 
    CAST(customer_id AS STRING) as customer_id,
    email,
    first_name,
    last_name,
    phone,
    date_of_birth,
    gender,
    
    -- 주소 정보
    address_line1,
    address_line2,
    city,
    state,
    zip_code,
    country,
    
    -- 계정 상태
    account_status,
    registration_date,
    last_login_date,
    
    -- 변경 이력 관리
    updated_at,
    LAG(updated_at) OVER (PARTITION BY customer_id ORDER BY updated_at) as previous_update,
    LEAD(updated_at) OVER (PARTITION BY customer_id ORDER BY updated_at) as next_update,
    
    ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY updated_at DESC) as latest_record
  FROM ${resolve(dataform.projectConfig.vars.source_dataset + '.raw_customers')}
  WHERE customer_id IS NOT NULL
)

SELECT 
  -- 고객 식별자
  customer_id,
  
  -- 개인 정보
  email,
  CONCAT(first_name, ' ', last_name) as full_name,
  first_name,
  last_name,
  phone,
  date_of_birth,
  gender,
  
  -- 연령 계산
  DATE_DIFF(CURRENT_DATE(), date_of_birth, YEAR) as age,
  CASE 
    WHEN DATE_DIFF(CURRENT_DATE(), date_of_birth, YEAR) < 25 THEN '18-24'
    WHEN DATE_DIFF(CURRENT_DATE(), date_of_birth, YEAR) < 35 THEN '25-34'
    WHEN DATE_DIFF(CURRENT_DATE(), date_of_birth, YEAR) < 45 THEN '35-44'
    WHEN DATE_DIFF(CURRENT_DATE(), date_of_birth, YEAR) < 55 THEN '45-54'
    ELSE '55+'
  END as age_group,
  
  -- 주소 정보
  address_line1,
  address_line2,
  city,
  state,
  zip_code,
  country,
  
  -- 계정 정보
  account_status,
  registration_date,
  last_login_date,
  
  -- SCD Type 2 컬럼들
  updated_at as effective_date,
  COALESCE(next_update, TIMESTAMP('2999-12-31')) as end_date,
  CASE WHEN latest_record = 1 THEN true ELSE false END as is_active,
  
  -- 메타데이터
  CURRENT_TIMESTAMP() as processed_at

FROM customer_history

-- definitions/marts/dim_products.sqlx
config {
  type: "table",
  description: "제품 차원 테이블",
  bigquery: {
    clusterBy: ["category_id", "is_active"]
  }
}

SELECT 
  -- 제품 식별자
  CAST(p.product_id AS STRING) as product_id,
  p.product_name,
  p.product_description,
  p.sku,
  
  -- 카테고리 정보
  CAST(p.category_id AS STRING) as category_id,
  c.category_name,
  c.parent_category_id,
  c.category_hierarchy,
  
  -- 브랜드 정보
  CAST(p.brand_id AS STRING) as brand_id,
  b.brand_name,
  
  -- 가격 정보
  CAST(p.unit_price AS FLOAT64) as unit_price,
  CAST(p.cost AS FLOAT64) as unit_cost,
  CAST(p.unit_price AS FLOAT64) - CAST(p.cost AS FLOAT64) as unit_margin,
  SAFE_DIVIDE(CAST(p.unit_price AS FLOAT64) - CAST(p.cost AS FLOAT64), CAST(p.unit_price AS FLOAT64)) as margin_percentage,
  
  -- 제품 속성
  p.color,
  p.size,
  p.weight,
  p.dimensions,
  
  -- 재고 정보
  CAST(p.stock_quantity AS INT64) as current_stock,
  CAST(p.reorder_level AS INT64) as reorder_level,
  CASE 
    WHEN CAST(p.stock_quantity AS INT64) <= CAST(p.reorder_level AS INT64) THEN 'LOW_STOCK'
    WHEN CAST(p.stock_quantity AS INT64) = 0 THEN 'OUT_OF_STOCK'
    ELSE 'IN_STOCK'
  END as stock_status,
  
  -- 상태 정보
  p.is_active,
  p.launch_date,
  p.discontinued_date,
  
  -- 메타데이터
  p.created_at,
  p.updated_at,
  CURRENT_TIMESTAMP() as processed_at

FROM ${resolve(dataform.projectConfig.vars.source_dataset + '.raw_products')} p
LEFT JOIN ${resolve(dataform.projectConfig.vars.source_dataset + '.raw_categories')} c
  ON p.category_id = c.category_id
LEFT JOIN ${resolve(dataform.projectConfig.vars.source_dataset + '.raw_brands')} b
  ON p.brand_id = b.brand_id
WHERE p.product_id IS NOT NULL

-- definitions/marts/fact_orders.sqlx
config {
  type: "incremental",
  uniqueKey: ["order_id"],
  bigquery: {
    partitionBy: "order_date",
    clusterBy: ["customer_id", "order_status"]
  }
}

SELECT 
  -- 주문 식별자
  o.order_id,
  
  -- 고객 외래키
  o.customer_id,
  
  -- 날짜 정보
  o.order_timestamp,
  o.order_date,
  EXTRACT(YEAR FROM o.order_date) as order_year,
  EXTRACT(MONTH FROM o.order_date) as order_month,
  EXTRACT(DAYOFWEEK FROM o.order_date) as day_of_week,
  CASE EXTRACT(DAYOFWEEK FROM o.order_date)
    WHEN 1 THEN 'Sunday'
    WHEN 2 THEN 'Monday'
    WHEN 3 THEN 'Tuesday'
    WHEN 4 THEN 'Wednesday'
    WHEN 5 THEN 'Thursday'
    WHEN 6 THEN 'Friday'
    WHEN 7 THEN 'Saturday'
  END as day_name,
  
  -- 주문 상태
  o.order_status,
  
  -- 금액 정보
  o.subtotal,
  o.tax_amount,
  o.shipping_cost,
  o.discount_amount,
  o.total_amount,
  o.total_amount - o.discount_amount as net_amount,
  
  -- 할인 정보
  o.discount_code,
  CASE WHEN o.discount_amount > 0 THEN true ELSE false END as has_discount,
  SAFE_DIVIDE(o.discount_amount, o.subtotal) as discount_percentage,
  
  -- 배송 정보
  o.shipping_address,
  o.shipping_state,
  
  -- 주문 상품 집계
  oi.total_items,
  oi.total_quantity,
  oi.avg_item_price,
  
  -- 고객 정보 (스냅샷)
  c.customer_name,
  c.customer_email,
  c.customer_age_group,
  c.customer_state,
  c.customer_registration_date,
  
  -- 계산된 메트릭
  DATE_DIFF(o.order_date, c.customer_registration_date, DAY) as days_since_registration,
  CASE 
    WHEN DATE_DIFF(o.order_date, c.customer_registration_date, DAY) <= 30 THEN 'NEW_CUSTOMER'
    WHEN DATE_DIFF(o.order_date, c.customer_registration_date, DAY) <= 365 THEN 'RETURNING_CUSTOMER'
    ELSE 'LOYAL_CUSTOMER'
  END as customer_segment,
  
  -- 메타데이터
  o.processed_at

FROM ${ref("stg_ecommerce_orders")} o

-- 주문 상품 집계 조인
LEFT JOIN (
  SELECT 
    order_id,
    COUNT(DISTINCT product_id) as total_items,
    SUM(quantity) as total_quantity,
    AVG(unit_price) as avg_item_price
  FROM ${ref("stg_order_items")}
  GROUP BY order_id
) oi ON o.order_id = oi.order_id

-- 고객 정보 조인 (현재 활성 레코드)
LEFT JOIN (
  SELECT 
    customer_id,
    full_name as customer_name,
    email as customer_email,
    age_group as customer_age_group,
    state as customer_state,
    registration_date as customer_registration_date
  FROM ${ref("dim_customers")}
  WHERE is_active = true
) c ON o.customer_id = c.customer_id

${ when(incremental(), `
WHERE o.order_timestamp > (
  SELECT COALESCE(MAX(order_timestamp), TIMESTAMP('1900-01-01'))
  FROM ${self()}
)
`) }

-- definitions/marts/marketing_campaign_performance.sqlx
config {
  type: "table",
  description: "마케팅 캠페인별 성과 분석",
  bigquery: {
    partitionBy: "DATE(campaign_date)",
    clusterBy: ["campaign_id", "channel"]
  }
}

WITH campaign_metrics AS (
  SELECT 
    c.campaign_id,
    c.campaign_name,
    c.channel,
    c.campaign_type,
    DATE(c.start_date) as campaign_date,
    c.budget_amount,
    
    -- 광고 지출
    SUM(ad.spend_amount) as total_spend,
    SUM(ad.impressions) as total_impressions,
    SUM(ad.clicks) as total_clicks,
    SAFE_DIVIDE(SUM(ad.clicks), SUM(ad.impressions)) as ctr,
    SAFE_DIVIDE(SUM(ad.spend_amount), SUM(ad.clicks)) as cpc,
    
    -- 전환 지표
    COUNT(DISTINCT CASE WHEN o.order_id IS NOT NULL THEN c.customer_id END) as converted_customers,
    COUNT(DISTINCT o.order_id) as total_orders,
    SUM(o.total_amount) as total_revenue,
    
    -- 계산된 지표
    SAFE_DIVIDE(COUNT(DISTINCT CASE WHEN o.order_id IS NOT NULL THEN c.customer_id END), 
                COUNT(DISTINCT c.customer_id)) as conversion_rate,
    SAFE_DIVIDE(SUM(o.total_amount), SUM(ad.spend_amount)) as roas,
    SAFE_DIVIDE(SUM(ad.spend_amount), COUNT(DISTINCT CASE WHEN o.order_id IS NOT NULL THEN c.customer_id END)) as cac
    
  FROM ${resolve(dataform.projectConfig.vars.source_dataset + '.campaigns')} c
  LEFT JOIN ${resolve(dataform.projectConfig.vars.source_dataset + '.ad_spend')} ad
    ON c.campaign_id = ad.campaign_id
  LEFT JOIN ${resolve(dataform.projectConfig.vars.source_dataset + '.campaign_customers')} cc
    ON c.campaign_id = cc.campaign_id
  LEFT JOIN ${ref("fact_orders")} o
    ON cc.customer_id = o.customer_id
    AND DATE(o.order_date) BETWEEN DATE(c.start_date) AND DATE(c.end_date)
  
  GROUP BY 1, 2, 3, 4, 5, 6
)

SELECT 
  *,
  -- 성과 등급
  CASE 
    WHEN roas >= 4.0 THEN 'EXCELLENT'
    WHEN roas >= 2.0 THEN 'GOOD' 
    WHEN roas >= 1.0 THEN 'BREAK_EVEN'
    ELSE 'POOR'
  END as performance_grade,
  
  -- 예산 효율성
  SAFE_DIVIDE(total_spend, budget_amount) as budget_utilization,
  budget_amount - total_spend as remaining_budget

FROM campaign_metrics

-- definitions/marts/customer_segmentation.sqlx
config {
  type: "table",
  description: "RFM 기반 고객 세분화",
  bigquery: {
    clusterBy: ["segment", "customer_id"]
  }
}

WITH customer_rfm AS (
  SELECT 
    c.customer_id,
    c.customer_name,
    c.registration_date,
    
    -- Recency: 마지막 구매 이후 일수
    DATE_DIFF(CURRENT_DATE(), MAX(o.order_date), DAY) as recency_days,
    
    -- Frequency: 총 주문 횟수
    COUNT(DISTINCT o.order_id) as frequency,
    
    -- Monetary: 총 구매 금액
    SUM(o.total_amount) as monetary,
    
    -- 추가 메트릭
    AVG(o.total_amount) as avg_order_value,
    MIN(o.order_date) as first_order_date,
    MAX(o.order_date) as last_order_date,
    DATE_DIFF(MAX(o.order_date), MIN(o.order_date), DAY) + 1 as customer_lifetime_days
    
  FROM ${ref("dim_customers")} c
  LEFT JOIN ${ref("fact_orders")} o
    ON c.customer_id = o.customer_id
    AND o.order_status = 'COMPLETED'
  WHERE c.is_active = true
  GROUP BY 1, 2, 3
),

rfm_scores AS (
  SELECT 
    *,
    -- RFM 점수 계산 (1-5 척도)
    CASE 
      WHEN recency_days <= 30 THEN 5
      WHEN recency_days <= 90 THEN 4
      WHEN recency_days <= 180 THEN 3
      WHEN recency_days <= 365 THEN 2
      ELSE 1
    END as r_score,
    
    CASE 
      WHEN frequency >= 10 THEN 5
      WHEN frequency >= 5 THEN 4
      WHEN frequency >= 3 THEN 3
      WHEN frequency >= 2 THEN 2
      ELSE 1
    END as f_score,
    
    CASE 
      WHEN monetary >= 1000 THEN 5
      WHEN monetary >= 500 THEN 4
      WHEN monetary >= 250 THEN 3
      WHEN monetary >= 100 THEN 2
      ELSE 1
    END as m_score
    
  FROM customer_rfm
)

SELECT 
  customer_id,
  customer_name,
  registration_date,
  recency_days,
  frequency,
  monetary,
  avg_order_value,
  first_order_date,
  last_order_date,
  customer_lifetime_days,
  
  -- RFM 점수
  r_score,
  f_score, 
  m_score,
  CONCAT(CAST(r_score AS STRING), CAST(f_score AS STRING), CAST(m_score AS STRING)) as rfm_code,
  
  -- 세분화 결과
  CASE 
    WHEN r_score >= 4 AND f_score >= 4 AND m_score >= 4 THEN 'CHAMPIONS'
    WHEN r_score >= 3 AND f_score >= 4 AND m_score >= 4 THEN 'LOYAL_CUSTOMERS'
    WHEN r_score >= 4 AND f_score <= 2 AND m_score >= 3 THEN 'POTENTIAL_LOYALISTS'
    WHEN r_score >= 4 AND f_score <= 2 AND m_score <= 2 THEN 'NEW_CUSTOMERS'
    WHEN r_score >= 3 AND f_score >= 3 AND m_score >= 3 THEN 'PROMISING'
    WHEN r_score >= 3 AND f_score <= 2 AND m_score <= 2 THEN 'NEED_ATTENTION'
    WHEN r_score <= 2 AND f_score >= 3 AND m_score >= 3 THEN 'AT_RISK'
    WHEN r_score <= 2 AND f_score >= 4 AND m_score >= 4 THEN 'CANT_LOSE_THEM'
    WHEN r_score <= 2 AND f_score <= 2 AND m_score >= 3 THEN 'HIBERNATING'
    ELSE 'LOST'
  END as segment,
  
  -- 예상 생애가치 (간단한 모델)
  CASE 
    WHEN customer_lifetime_days > 0 
    THEN (monetary / customer_lifetime_days) * 365 * 2 -- 향후 2년 예상
    ELSE avg_order_value * 4 -- 신규 고객 추정
  END as predicted_clv,
  
  CURRENT_TIMESTAMP() as processed_at

FROM rfm_scores

-- 파일명 규칙
-- [layer]_[business_area]_[entity].sqlx

-- staging: stg_[source]_[entity].sqlx
-- definitions/staging/ecommerce/stg_shopify_orders.sqlx

-- intermediate: int_[description].sqlx  
-- definitions/intermediate/int_orders_with_customers.sqlx

-- marts: [entity_type]_[entity].sqlx
-- definitions/marts/core/dim_customers.sqlx
-- definitions/marts/core/fact_orders.sqlx

-- assertions: assert_[rule_description].sqlx
-- definitions/assertions/core/assert_unique_customer_ids.sqlx

-- 날짜 기반 파티셔닝 (가장 일반적)
config {
  type: "incremental",
  bigquery: {
    partitionBy: "DATE(created_date)",
    requirePartitionFilter: true,
    partitionExpirationDays: 1095 // 3년
  }
}

-- 정수 범위 파티셔닝
config {
  type: "table", 
  bigquery: {
    partitionBy: "RANGE_BUCKET(customer_id, GENERATE_ARRAY(0, 1000000, 1000))"
  }
}

-- 쿼리 패턴에 맞는 클러스터링
config {
  type: "table",
  bigquery: {
    partitionBy: "DATE(order_date)",
    clusterBy: ["customer_id", "product_category", "region"] // 자주 필터링/조인되는 컬럼 순서로
  }
}

-- 효율적인 증분 처리 패턴
config {
  type: "incremental",
  uniqueKey: ["order_id"],
  bigquery: {
    partitionBy: "DATE(order_date)",
    updatePartitionFilter: "DATE(order_date) >= DATE_SUB(CURRENT_DATE(), INTERVAL 3 DAY)" // 최근 3일만 업데이트
  }
}

SELECT *
FROM ${ref("source_table")}
${ when(incremental(), `
WHERE 
  updated_at > (SELECT MAX(updated_at) FROM ${self()})
  OR order_id IN (
    SELECT order_id FROM ${ref("updated_orders")} -- 특정 업데이트 목록
  )
`) }

-- definitions/assertions/comprehensive_order_validation.sqlx
config {
  type: "assertion",
  description: "주문 데이터 종합 품질 검증"
}

-- 1. 필수 필드 누락 검사
SELECT 'missing_required_fields' as check_type, order_id
FROM ${ref("fact_orders")}
WHERE customer_id IS NULL OR order_date IS NULL OR total_amount IS NULL

UNION ALL

-- 2. 비즈니스 룰 검증
SELECT 'negative_amounts' as check_type, order_id
FROM ${ref("fact_orders")}  
WHERE total_amount < 0 OR subtotal < 0

UNION ALL

-- 3. 참조 무결성 검증
SELECT 'orphaned_orders' as check_type, o.order_id
FROM ${ref("fact_orders")} o
LEFT JOIN ${ref("dim_customers")} c ON o.customer_id = c.customer_id
WHERE c.customer_id IS NULL

UNION ALL

-- 4. 중복 데이터 검사
SELECT 'duplicate_orders' as check_type, order_id
FROM ${ref("fact_orders")}
GROUP BY order_id
HAVING COUNT(*) > 1

UNION ALL

-- 5. 데이터 분포 이상 감지
SELECT 'suspicious_outliers' as check_type, order_id
FROM ${ref("fact_orders")}
WHERE total_amount > (
  SELECT PERCENTILE_CONT(total_amount, 0.99) OVER()
  FROM ${ref("fact_orders")}
  LIMIT 1
) * 10 -- 99%ile의 10배를 초과하는 경우

-- definitions/monitoring/data_quality_dashboard.sqlx
config {
  type: "table",
  description: "일일 데이터 품질 메트릭"
}

WITH quality_checks AS (
  SELECT 
    'customers' as table_name,
    CURRENT_DATE() as check_date,
    COUNT(*) as total_records,
    COUNTIF(customer_id IS NULL) as null_ids,
    COUNTIF(email IS NULL OR email = '') as missing_emails,
    COUNTIF(NOT REGEXP_CONTAINS(email, r'^[^@]+@[^@]+\.[^@]+$')) as invalid_emails,
    COUNT(DISTINCT customer_id) as unique_customers
  FROM ${ref("dim_customers")}
  WHERE is_active = true
  
  UNION ALL
  
  SELECT 
    'orders',
    CURRENT_DATE(),
    COUNT(*),
    COUNTIF(order_id IS NULL),
    COUNTIF(customer_id IS NULL), 
    COUNTIF(total_amount IS NULL),
    COUNT(DISTINCT order_id)
  FROM ${ref("fact_orders")}
  WHERE order_date = CURRENT_DATE()
)

SELECT 
  *,
  -- 품질 점수 계산
  1 - (null_ids + missing_emails + invalid_emails) / total_records as quality_score,
  
  -- 경고 플래그
  CASE 
    WHEN (null_ids + missing_emails + invalid_emails) / total_records > 0.05 THEN 'CRITICAL'
    WHEN (null_ids + missing_emails + invalid_emails) / total_records > 0.01 THEN 'WARNING'
    ELSE 'OK'
  END as quality_status

FROM quality_checks

-- definitions/marts/core/dim_customers.sqlx
config {
  type: "table",
  description: "고객 차원 테이블 - 모든 고객의 마스터 데이터를 포함합니다. SCD Type 2로 구현되어 고객 정보 변경 이력을 추적합니다.",
  columns: {
    customer_id: "고유 고객 식별자 (Primary Key)",
    customer_name: "고객 성명",
    email: "이메일 주소 (고유값, 마케팅 커뮤니케이션에 사용)",
    phone: "전화번호",
    date_of_birth: "생년월일",
    age: "현재 나이 (매일 계산됨)",
    age_group: "연령대 구분 (18-24, 25-34, 35-44, 45-54, 55+)",
    registration_date: "고객 등록일",
    total_orders: "총 주문 횟수",
    total_spent: "총 구매 금액 (USD)",
    avg_order_value: "평균 주문 금액",
    last_order_date: "최근 주문일",
    customer_segment: "고객 세그먼트 (NEW, RETURNING, LOYAL)",
    is_active: "현재 활성 레코드 여부 (SCD Type 2)",
    effective_date: "레코드 유효 시작일",
    end_date: "레코드 유효 종료일 (2999-12-31이면 현재 활성)",
    processed_at: "데이터 처리 타임스탬프"
  },
  bigquery: {
    partitionBy: "DATE(effective_date)",
    clusterBy: ["customer_id", "is_active"],
    labels: {
      team: "data",
      domain: "customer",
      criticality: "high"
    }
  }
}

-- 예상 비용 계산 예제
-- definitions/monitoring/query_cost_estimation.sqlx
config { type: "table" }

WITH query_analysis AS (
  SELECT 
    job_id,
    user_email,
    creation_time,
    total_bytes_processed / 1024 / 1024 / 1024 / 1024 as tb_processed,
    -- 미국 기준 $6.00/TB
    (total_bytes_processed / 1024 / 1024 / 1024 / 1024) * 6.0 as estimated_cost_usd,
    query
  FROM `region-us.INFORMATION_SCHEMA.JOBS_BY_PROJECT`
  WHERE 
    DATE(creation_time) = CURRENT_DATE()
    AND job_type = 'QUERY'
    AND state = 'DONE'
    AND CONTAINS_SUBSTR(query, 'dataform')
)

SELECT 
  DATE(creation_time) as query_date,
  COUNT(*) as total_dataform_queries,
  ROUND(SUM(tb_processed), 3) as total_tb_processed,
  ROUND(SUM(estimated_cost_usd), 2) as estimated_daily_cost_usd
FROM query_analysis
GROUP BY DATE(creation_time)
ORDER BY query_date DESC

-- 스토리지 비용 모니터링
-- definitions/monitoring/storage_cost_analysis.sqlx
config { type: "table" }

SELECT 
  table_schema,
  table_name,
  ROUND(size_bytes / 1024 / 1024 / 1024, 2) as size_gb,
  -- Active storage 기준 $0.020/GB/month
  ROUND((size_bytes / 1024 / 1024 / 1024) * 0.020, 2) as monthly_storage_cost_usd,
  last_modified_time,
  CASE 
    WHEN TIMESTAMP_DIFF(CURRENT_TIMESTAMP(), last_modified_time, DAY) > 90 
    THEN ROUND((size_bytes / 1024 / 1024 / 1024) * 0.010, 2)
    ELSE ROUND((size_bytes / 1024 / 1024 / 1024) * 0.020, 2)
  END as actual_monthly_cost_usd
FROM `region-us.INFORMATION_SCHEMA.TABLE_STORAGE_BY_PROJECT`
WHERE table_schema NOT IN ('INFORMATION_SCHEMA', 'sys')
ORDER BY size_gb DESC

-- 파티션 필터를 통한 스캔 데이터 최소화
config {
  type: "incremental",
  bigquery: {
    partitionBy: "DATE(order_date)",
    requirePartitionFilter: true
  }
}

SELECT *
FROM ${ref("fact_orders")}
WHERE DATE(order_date) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY) -- 파티션 필터 필수

-- 전체 재처리 대신 증분 처리로 비용 90% 절약 가능
config {
  type: "incremental",
  uniqueKey: ["order_id"]
}

SELECT *
FROM ${ref("raw_orders")}
${ when(incremental(), `
WHERE updated_at > (SELECT MAX(updated_at) FROM ${self()})
`) }

-- ❌ 비효율적 - 전체 컬럼 스캔
SELECT * FROM large_table

-- ✅ 효율적 - 필요한 컬럼만 선택
SELECT 
  order_id,
  customer_id,
  total_amount,
  order_date
FROM large_table

-- 테스트용 샘플 데이터 사용으로 비용 절약
-- definitions/staging/stg_orders_sample.sqlx
config {
  type: "table",
  disabled: dataform.projectConfig.vars.environment !== 'test'
}

SELECT *
FROM ${ref("raw_orders")}
WHERE MOD(ABS(FARM_FINGERPRINT(order_id)), 100) < 1 -- 1% 샘플링

-- definitions/monitoring/daily_cost_tracking.sqlx
config {
  type: "incremental",
  uniqueKey: ["cost_date", "project_id"]
}

WITH daily_costs AS (
  SELECT 
    DATE(usage_start_time) as cost_date,
    project.id as project_id,
    service.description as service_name,
    SUM(cost) as total_cost_usd,
    SUM(CASE WHEN sku.description LIKE '%Query%' THEN cost ELSE 0 END) as query_cost_usd,
    SUM(CASE WHEN sku.description LIKE '%Storage%' THEN cost ELSE 0 END) as storage_cost_usd
  FROM `project.dataset.gcp_billing_export_v1_BILLING_ACCOUNT_ID`
  WHERE 
    service.description = 'BigQuery'
    AND DATE(usage_start_time) >= CURRENT_DATE() - 30
  GROUP BY 1, 2, 3
)

SELECT 
  *,
  -- 전일 대비 증가율 계산
  LAG(total_cost_usd) OVER (
    PARTITION BY project_id 
    ORDER BY cost_date
  ) as previous_day_cost,
  
  SAFE_DIVIDE(
    total_cost_usd - LAG(total_cost_usd) OVER (PARTITION BY project_id ORDER BY cost_date),
    LAG(total_cost_usd) OVER (PARTITION BY project_id ORDER BY cost_date)
  ) * 100 as cost_change_pct

FROM daily_costs
${ when(incremental(), `
WHERE cost_date > (SELECT MAX(cost_date) FROM ${self()})
`) }

-- definitions/assertions/cost_alert_thresholds.sqlx
config {
  type: "assertion",
  description: "일일 비용 임계값 초과 시 알림"
}

WITH cost_summary AS (
  SELECT 
    DATE(usage_start_time) as cost_date,
    SUM(cost) as daily_cost_usd
  FROM `project.dataset.gcp_billing_export_v1_BILLING_ACCOUNT_ID`
  WHERE 
    service.description = 'BigQuery'
    AND DATE(usage_start_time) = CURRENT_DATE()
  GROUP BY 1
)

SELECT 
  cost_date,
  daily_cost_usd,
  'DAILY_COST_THRESHOLD_EXCEEDED' as alert_type
FROM cost_summary
WHERE daily_cost_usd > 100 -- $100 임계값 설정

-- 환경별 데이터 샘플링
-- definitions/staging/stg_orders_optimized.sqlx
config {
  type: "table",
  disabled: dataform.projectConfig.vars.environment === 'dev' 
           && dataform.projectConfig.vars.sample_percentage < 1.0
}

SELECT *
FROM ${ref("raw_orders")}
${ when(dataform.projectConfig.vars.environment === 'dev', `
WHERE MOD(ABS(FARM_FINGERPRINT(CAST(order_id AS STRING))), 100) 
      < ${dataform.projectConfig.vars.sample_percentage * 100}
`) }

-- definitions/monitoring/expensive_queries_analysis.sqlx
config {
  type: "table",
  description: "비용이 높은 쿼리 분석"
}

SELECT 
  job_id,
  user_email,
  creation_time,
  ROUND(total_bytes_processed / 1024 / 1024 / 1024, 2) as gb_processed,
  ROUND(total_slot_ms / 1000 / 60, 2) as slot_minutes,
  ROUND((total_bytes_processed / 1024 / 1024 / 1024 / 1024) * 6.0, 2) as estimated_cost_usd,
  -- 쿼리 패턴 분석
  CASE 
    WHEN CONTAINS_SUBSTR(query, 'SELECT *') THEN 'FULL_TABLE_SCAN'
    WHEN CONTAINS_SUBSTR(query, 'GROUP BY') AND NOT CONTAINS_SUBSTR(query, 'LIMIT') THEN 'LARGE_AGGREGATION'
    WHEN CONTAINS_SUBSTR(query, 'JOIN') AND NOT CONTAINS_SUBSTR(query, 'WHERE') THEN 'CARTESIAN_JOIN_RISK'
    ELSE 'OPTIMIZED'
  END as query_pattern,
  LEFT(query, 200) as query_preview
FROM `region-us.INFORMATION_SCHEMA.JOBS_BY_PROJECT`
WHERE 
  DATE(creation_time) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
  AND job_type = 'QUERY'
  AND state = 'DONE'
  AND total_bytes_processed > 10 * 1024 * 1024 * 1024 -- 10GB 이상만
ORDER BY estimated_cost_usd DESC
LIMIT 50

-- definitions/monitoring/monthly_cost_forecast.sqlx
config { 
  type: "table",
  description: "월간 예상 비용 계산"
}

WITH usage_trends AS (
  SELECT 
    DATE_TRUNC(DATE(creation_time), MONTH) as month,
    COUNT(*) as total_queries,
    SUM(total_bytes_processed) / 1024 / 1024 / 1024 / 1024 as total_tb_processed,
    SUM(total_bytes_processed) / 1024 / 1024 / 1024 / 1024 * 6.0 as total_query_cost_usd
  FROM `region-us.INFORMATION_SCHEMA.JOBS_BY_PROJECT`
  WHERE 
    DATE(creation_time) >= DATE_SUB(DATE_TRUNC(CURRENT_DATE(), MONTH), INTERVAL 3 MONTH)
    AND job_type = 'QUERY'
    AND state = 'DONE'
  GROUP BY 1
),

storage_costs AS (
  SELECT 
    DATE_TRUNC(CURRENT_DATE(), MONTH) as month,
    SUM(size_bytes) / 1024 / 1024 / 1024 as total_storage_gb,
    SUM(size_bytes) / 1024 / 1024 / 1024 * 0.020 as monthly_storage_cost_usd
  FROM `region-us.INFORMATION_SCHEMA.TABLE_STORAGE_BY_PROJECT`
)

SELECT 
  u.month,
  u.total_queries,
  ROUND(u.total_tb_processed, 3) as tb_processed,
  ROUND(u.total_query_cost_usd, 2) as query_cost_usd,
  ROUND(s.monthly_storage_cost_usd, 2) as storage_cost_usd,
  ROUND(u.total_query_cost_usd + COALESCE(s.monthly_storage_cost_usd, 0), 2) as total_monthly_cost_usd,
  
  -- 다음 달 예상 비용 (최근 3개월 평균 기준)
  ROUND(
    AVG(u.total_query_cost_usd) OVER (
      ORDER BY u.month 
      ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
    ) + COALESCE(s.monthly_storage_cost_usd, 0), 
    2
  ) as forecasted_next_month_usd

FROM usage_trends u
LEFT JOIN storage_costs s ON u.month = s.month
ORDER BY u.month DESC

-- 배치 처리로 대용량 데이터 처리
config {
  type: "incremental",
  bigquery: {
    partitionBy: "DATE(order_date)",
    updatePartitionFilter: "DATE(order_date) = CURRENT_DATE()" // 하루씩 처리
  }
}

-- 또는 WITH문으로 중간 결과 구체화
WITH large_aggregation AS (
  SELECT 
    customer_id,
    SUM(amount) as total_amount
  FROM ${ref("fact_orders")}
  WHERE DATE(order_date) >= DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY)
  GROUP BY customer_id
)
SELECT * FROM large_aggregation