-- Unity Catalog voorbeeld: Catalog aanmaken en gebruiken
CREATE CATALOG IF NOT EXISTS production_catalog;

-- Grant permissions aan gebruikers
GRANT USAGE ON CATALOG production_catalog TO `data-engineers`;

-- Schema aanmaken in catalog
CREATE SCHEMA IF NOT EXISTS production_catalog.sales_data;

-- Table aanmaken met Unity Catalog
CREATE TABLE production_catalog.sales_data.customers (
  customer_id STRING,
  name STRING,
  email STRING,
  created_date TIMESTAMP age INT
)
COMMENT "Customer dimension table";

-- Fine-grained access control
GRANT SELECT ON TABLE production_catalog.sales_data.customers TO `analysts`;

-- Medallion Architecture pipeline voorbeeld
-- BRONZE: Raw data ingestie
CREATE OR REFRESH STREAMING LIVE TABLE sales_bronze
COMMENT "Raw sales data from source"
AS SELECT * FROM cloud_files(
  "s3://raw-data/sales/",
  "json"
);

-- SILVER: Cleaned & validated data
CREATE OR REFRESH STREAMING LIVE TABLE sales_silver
COMMENT "Cleaned sales data with quality checks"
AS SELECT
  customer_id,
  TRIM(product_name) AS product_name,
  CAST(amount AS DECIMAL(10,2)) AS amount,
  DATE(timestamp) AS sale_date,
  -- Data quality checks
  CASE 
    WHEN amount > 0 THEN 'valid'
    ELSE 'invalid'
  END AS data_quality
FROM STREAM(live.sales_bronze)
WHERE timestamp IS NOT NULL;

-- GOLD: Business aggregates
CREATE OR REFRESH LIVE TABLE daily_sales_gold
COMMENT "Daily sales aggregates for reporting"
AS SELECT
  sale_date,
  COUNT(*) AS total_transactions,
  SUM(amount) AS total_revenue,
  AVG(amount) AS avg_transaction_value
FROM live.sales_silver
WHERE data_quality = 'valid'
GROUP BY sale_date;

-- Complete SCD Type 2 implementatie
CREATE OR REPLACE TABLE customer_dimension_scd2 (
  customer_key BIGINT GENERATED ALWAYS AS IDENTITY,
  customer_id STRING,
  customer_name STRING,
  email STRING,
  address STRING,
  start_date TIMESTAMP,
  end_date TIMESTAMP,
  is_current BOOLEAN,
  version_number INT
)
USING DELTA
PARTITIONED BY (is_current);

-- Initial load
INSERT INTO customer_dimension_scd2
SELECT
  customer_id,
  customer_name,
  email,
  address,
  CURRENT_TIMESTAMP() AS start_date,
  NULL AS end_date,
  TRUE AS is_current,
  1 AS version_number
FROM initial_customer_data;

-- Incremental update met SCD Type 2
MERGE INTO customer_dimension_scd2 AS target
USING (
  SELECT 
    *,
    ROW_NUMBER() OVER (
      PARTITION BY customer_id 
      ORDER BY updated_at DESC
    ) AS rn
  FROM customer_updates
  WHERE updated_at > (SELECT MAX(start_date) FROM customer_dimension_scd2)
) AS source
ON target.customer_id = source.customer_id 
  AND target.is_current = TRUE
WHEN MATCHED AND (
  target.customer_name != source.customer_name
  OR target.email != source.email
  OR target.address != source.address
) THEN
  -- Expire old record
  UPDATE SET 
    target.end_date = CURRENT_TIMESTAMP(),
    target.is_current = FALSE
WHEN NOT MATCHED AND source.rn = 1 THEN
  -- Insert new version
  INSERT (
    customer_id, customer_name, email, address, 
    start_date, end_date, is_current, version_number
  )
  VALUES (
    source.customer_id,
    source.customer_name,
    source.email,
    source.address,
    CURRENT_TIMESTAMP(),
    NULL,
    TRUE,
    COALESCE(
      (SELECT MAX(version_number) 
       FROM customer_dimension_scd2 
       WHERE customer_id = source.customer_id), 
      0
    ) + 1
  );