-- Date dimension (conformed - used by all fact tables)
CREATE TABLE dim_date (
  date_sk       INT         PRIMARY KEY,  -- YYYYMMDD integer
  full_date     DATE        NOT NULL,
  day_of_week   VARCHAR(10) NOT NULL,
  month_name    VARCHAR(10) NOT NULL,
  quarter       INT         NOT NULL,
  fiscal_year   INT         NOT NULL,
  is_weekend    BOOLEAN     NOT NULL,
  is_holiday    BOOLEAN     NOT NULL
);

-- Customer dimension
CREATE TABLE dim_customer (
  customer_sk   INT         PRIMARY KEY,  -- surrogate key
  customer_id   VARCHAR(50) NOT NULL,     -- natural key
  name          VARCHAR(200),
  segment       VARCHAR(50),
  region        VARCHAR(100),
  -- SCD Type 2 tracking
  effective_date DATE       NOT NULL,
  expiry_date    DATE       NOT NULL DEFAULT '9999-12-31',
  is_current     BOOLEAN    NOT NULL DEFAULT TRUE
);

-- Product dimension
CREATE TABLE dim_product (
  product_sk    INT         PRIMARY KEY,
  product_id    VARCHAR(50) NOT NULL,
  product_name  VARCHAR(200),
  category      VARCHAR(100),
  subcategory   VARCHAR(100),
  brand         VARCHAR(100),
  unit_cost     DECIMAL(12,2)
);

-- Sales fact table (transaction grain: one row per line item)
CREATE TABLE fct_sales (
  sale_sk       BIGINT      PRIMARY KEY,
  date_sk       INT         NOT NULL REFERENCES dim_date(date_sk),
  customer_sk   INT         NOT NULL REFERENCES dim_customer(customer_sk),
  product_sk    INT         NOT NULL REFERENCES dim_product(product_sk),
  quantity      INT         NOT NULL,
  unit_price    DECIMAL(12,2) NOT NULL,
  discount_amt  DECIMAL(12,2) NOT NULL DEFAULT 0,
  net_amount    DECIMAL(12,2) NOT NULL,
  order_number  VARCHAR(50) NOT NULL  -- degenerate dimension
);

-- Snowflake MERGE for SCD Type 2
MERGE INTO dim_customer AS target
USING staging_customers AS source
  ON target.customer_id = source.customer_id
     AND target.is_current = TRUE

-- Existing row where attributes changed: expire it
WHEN MATCHED
  AND (target.segment  != source.segment
    OR target.region   != source.region)
THEN UPDATE SET
  target.expiry_date = CURRENT_DATE - 1,
  target.is_current  = FALSE

-- No match: brand new customer
WHEN NOT MATCHED THEN INSERT (
  customer_sk, customer_id, name, segment, region,
  effective_date, expiry_date, is_current
) VALUES (
  dim_customer_seq.NEXTVAL,
  source.customer_id, source.name, source.segment, source.region,
  CURRENT_DATE, '9999-12-31', TRUE
);

-- Second pass: insert the new current row for changed records
INSERT INTO dim_customer
SELECT dim_customer_seq.NEXTVAL,
       s.customer_id, s.name, s.segment, s.region,
       CURRENT_DATE, '9999-12-31', TRUE
FROM staging_customers s
JOIN dim_customer d
  ON s.customer_id = d.customer_id
  AND d.expiry_date = CURRENT_DATE - 1
  AND d.is_current = FALSE;

-- BigQuery SCD Type 2 using MERGE
MERGE `project.dataset.dim_customer` AS target
USING `project.dataset.staging_customers` AS source
  ON target.customer_id = source.customer_id
     AND target.is_current = TRUE

WHEN MATCHED
  AND (target.segment != source.segment
    OR target.region  != source.region)
THEN UPDATE SET
  expiry_date = DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY),
  is_current  = FALSE

WHEN NOT MATCHED BY TARGET THEN INSERT (
  customer_sk, customer_id, name, segment, region,
  effective_date, expiry_date, is_current
) VALUES (
  GENERATE_UUID(),
  source.customer_id, source.name, source.segment, source.region,
  CURRENT_DATE(), DATE '9999-12-31', TRUE
);

-- Insert new current rows for changed records
INSERT INTO `project.dataset.dim_customer`
SELECT GENERATE_UUID(), s.customer_id, s.name, s.segment, s.region,
       CURRENT_DATE(), DATE '9999-12-31', TRUE
FROM `project.dataset.staging_customers` s
INNER JOIN `project.dataset.dim_customer` d
  ON s.customer_id = d.customer_id
  AND d.expiry_date = DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY);

-- Cluster keys for large fact tables (order matters: most filtered first)
ALTER TABLE fct_sales CLUSTER BY (date_sk, customer_sk);

-- Monitor clustering depth
SELECT SYSTEM$CLUSTERING_DEPTH('fct_sales');
-- Values close to 1.0 = well clustered; > 5 = recluster needed

-- Use result caching and warehouse sizing
ALTER WAREHOUSE analytics_wh SET
  WAREHOUSE_SIZE = 'MEDIUM'
  AUTO_SUSPEND = 60
  AUTO_RESUME = TRUE;

-- Materialized views for expensive aggregations
CREATE MATERIALIZED VIEW mv_daily_sales AS
SELECT date_sk,
       SUM(net_amount) AS total_sales,
       COUNT(*)        AS transaction_count
FROM fct_sales
GROUP BY date_sk;

-- Partition by date (reduces bytes scanned = lower cost)
CREATE TABLE `project.dataset.fct_sales`
(
  sale_sk       STRING      NOT NULL,
  sale_date     DATE        NOT NULL,
  customer_sk   STRING      NOT NULL,
  product_sk    STRING      NOT NULL,
  quantity      INT64       NOT NULL,
  net_amount    NUMERIC     NOT NULL
)
PARTITION BY sale_date
CLUSTER BY customer_sk, product_sk;

-- Check bytes scanned before running expensive queries
-- Use dry run: bq query --dry_run --use_legacy_sql=false 'SELECT ...'

-- Materialized view with automatic refresh
CREATE MATERIALIZED VIEW `project.dataset.mv_daily_sales`
AS
SELECT sale_date,
       SUM(net_amount) AS total_sales,
       COUNT(*)        AS transaction_count
FROM `project.dataset.fct_sales`
GROUP BY sale_date;

-- BigQuery: generate a date spine
CREATE TABLE `project.dataset.dim_date` AS
WITH date_spine AS (
  SELECT date
  FROM UNNEST(
    GENERATE_DATE_ARRAY('2020-01-01', '2030-12-31', INTERVAL 1 DAY)
  ) AS date
)
SELECT
  CAST(FORMAT_DATE('%Y%m%d', date) AS INT64) AS date_sk,
  date                                        AS full_date,
  FORMAT_DATE('%A', date)                     AS day_of_week,
  FORMAT_DATE('%B', date)                     AS month_name,
  EXTRACT(QUARTER FROM date)                  AS quarter,
  EXTRACT(YEAR FROM date)                     AS fiscal_year,
  EXTRACT(DAYOFWEEK FROM date) IN (1, 7)      AS is_weekend
FROM date_spine;