-- Kimball Approach: Bottom-up, business-process focused
-- Centraal concept: Dimensional modeling met star schemas

-- Star Schema Voorbeeld: Sales Data Mart
-- Fact Table: Feiten en metingen
CREATE TABLE fact_sales (
    sales_key BIGINT PRIMARY KEY,
    date_key INT NOT NULL,
    product_key INT NOT NULL,
    customer_key INT NOT NULL,
    store_key INT NOT NULL,
    promotion_key INT NOT NULL,
    sales_quantity DECIMAL(10, 2) NOT NULL,
    sales_amount DECIMAL(12, 2) NOT NULL,
    cost_amount DECIMAL(12, 2) NOT NULL,
    gross_profit DECIMAL(12, 2) NOT NULL,
    transaction_count INT NOT NULL
);

-- Dimension Tables: Beschrijvende attributen
CREATE TABLE dim_date (
    date_key INT PRIMARY KEY,
    full_date DATE NOT NULL,
    day_number INT NOT NULL,
    day_name VARCHAR(20) NOT NULL,
    week_number INT NOT NULL,
    month_number INT NOT NULL,
    month_name VARCHAR(20) NOT NULL,
    quarter_number INT NOT NULL,
    year_number INT NOT NULL,
    is_weekend BOOLEAN NOT NULL,
    is_holiday BOOLEAN NOT NULL
);

CREATE TABLE dim_product (
    product_key INT PRIMARY KEY,
    product_id VARCHAR(50) NOT NULL,
    product_name VARCHAR(255) NOT NULL,
    category_id INT NOT NULL,
    category_name VARCHAR(100) NOT NULL,
    brand_id INT NOT NULL,
    brand_name VARCHAR(100) NOT NULL,
    supplier_id INT NOT NULL,
    supplier_name VARCHAR(100) NOT NULL,
    unit_price DECIMAL(10, 2) NOT NULL,
    cost_price DECIMAL(10, 2) NOT NULL,
    start_date DATE NOT NULL,
    end_date DATE,
    is_current BOOLEAN NOT NULL DEFAULT TRUE
);

CREATE TABLE dim_customer (
    customer_key INT PRIMARY KEY,
    customer_id VARCHAR(50) NOT NULL,
    first_name VARCHAR(100) NOT NULL,
    last_name VARCHAR(100) NOT NULL,
    email VARCHAR(255) NOT NULL,
    phone VARCHAR(50),
    address_line1 VARCHAR(255),
    address_line2 VARCHAR(255),
    city VARCHAR(100),
    state_province VARCHAR(100),
    postal_code VARCHAR(20),
    country VARCHAR(100),
    customer_segment VARCHAR(50),
    registration_date DATE NOT NULL,
    start_date DATE NOT NULL,
    end_date DATE,
    is_current BOOLEAN NOT NULL DEFAULT TRUE
);

-- Inmon Approach: Top-down, enterprise-wide
-- Centraal concept: Normalized data warehouse (3NF)

-- Normalized Tables Voorbeeld
CREATE TABLE customer (
    customer_id INT PRIMARY KEY,
    customer_code VARCHAR(50) UNIQUE NOT NULL,
    first_name VARCHAR(100) NOT NULL,
    last_name VARCHAR(100) NOT NULL,
    birth_date DATE,
    registration_date DATE NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE address (
    address_id INT PRIMARY KEY,
    customer_id INT NOT NULL,
    address_type VARCHAR(20) NOT NULL,  -- 'billing', 'shipping'
    address_line1 VARCHAR(255) NOT NULL,
    address_line2 VARCHAR(255),
    city VARCHAR(100) NOT NULL,
    state_province VARCHAR(100),
    postal_code VARCHAR(20) NOT NULL,
    country VARCHAR(100) NOT NULL,
    is_primary BOOLEAN DEFAULT FALSE,
    valid_from DATE NOT NULL,
    valid_to DATE,
    FOREIGN KEY (customer_id) REFERENCES customer(customer_id)
);

CREATE TABLE customer_contact (
    contact_id INT PRIMARY KEY,
    customer_id INT NOT NULL,
    contact_type VARCHAR(20) NOT NULL,  -- 'email', 'phone', 'mobile'
    contact_value VARCHAR(255) NOT NULL,
    is_primary BOOLEAN DEFAULT FALSE,
    valid_from DATE NOT NULL,
    valid_to DATE,
    FOREIGN KEY (customer_id) REFERENCES customer(customer_id)
);

CREATE TABLE customer_segment (
    segment_id INT PRIMARY KEY,
    customer_id INT NOT NULL,
    segment_code VARCHAR(50) NOT NULL,
    segment_name VARCHAR(100) NOT NULL,
    segment_type VARCHAR(50) NOT NULL,  -- 'demographic', 'behavioral', 'value'
    assigned_date DATE NOT NULL,
    removed_date DATE,
    FOREIGN KEY (customer_id) REFERENCES customer(customer_id)
);

-- Data Marts worden gebouwd als views op de normalized warehouse
CREATE VIEW sales_data_mart AS
SELECT 
    c.customer_id,
    c.customer_code,
    CONCAT(c.first_name, ' ', c.last_name) AS customer_name,
    cs.segment_name,
    a.city,
    a.country,
    -- Aggregated sales metrics would be joined here
    0 AS total_sales_amount,
    0 AS order_count
FROM customer c
LEFT JOIN customer_segment cs 
    ON c.customer_id = cs.customer_id 
    AND cs.removed_date IS NULL
LEFT JOIN address a 
    ON c.customer_id = a.customer_id 
    AND a.is_primary = TRUE 
    AND a.valid_to IS NULL;

-- STAR SCHEMA: Eenvoudige, denormalized structuur
-- Één fact table omringd door dimension tables

-- Fact Table: Sales Transactions
CREATE TABLE fact_sales (
    sales_id BIGINT PRIMARY KEY,
    transaction_date_key INT NOT NULL,
    product_key INT NOT NULL,
    customer_key INT NOT NULL,
    store_key INT NOT NULL,
    employee_key INT NOT NULL,
    promotion_key INT,
    
    -- Measures (Feiten)
    quantity_sold INT NOT NULL,
    unit_price DECIMAL(10, 2) NOT NULL,
    discount_amount DECIMAL(10, 2) DEFAULT 0,
    tax_amount DECIMAL(10, 2) NOT NULL,
    total_amount DECIMAL(12, 2) NOT NULL,
    
    -- Degenerate Dimensions
    invoice_number VARCHAR(50) NOT NULL,
    line_item_number INT NOT NULL,
    
    -- Audit columns
    created_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    modified_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    -- Foreign keys
    FOREIGN KEY (transaction_date_key) REFERENCES dim_date(date_key),
    FOREIGN KEY (product_key) REFERENCES dim_product(product_key),
    FOREIGN KEY (customer_key) REFERENCES dim_customer(customer_key),
    FOREIGN KEY (store_key) REFERENCES dim_store(store_key),
    FOREIGN KEY (employee_key) REFERENCES dim_employee(employee_key)
);

-- Dimension Table: Product (Type 2 SCD - Slowly Changing Dimension)
CREATE TABLE dim_product (
    product_key INT PRIMARY KEY,
    product_id VARCHAR(50) NOT NULL,  -- Natural key
    product_name VARCHAR(255) NOT NULL,
    product_description TEXT,
    category_id INT NOT NULL,
    category_name VARCHAR(100) NOT NULL,
    subcategory_id INT,
    subcategory_name VARCHAR(100),
    brand_id INT NOT NULL,
    brand_name VARCHAR(100) NOT NULL,
    supplier_id INT NOT NULL,
    supplier_name VARCHAR(100) NOT NULL,
    unit_price DECIMAL(10, 2) NOT NULL,
    cost_price DECIMAL(10, 2) NOT NULL,
    weight_kg DECIMAL(8, 3),
    dimensions VARCHAR(50),
    
    -- SCD Type 2 columns
    version_number INT NOT NULL DEFAULT 1,
    is_current BOOLEAN NOT NULL DEFAULT TRUE,
    valid_from DATE NOT NULL,
    valid_to DATE,
    effective_date DATE NOT NULL,
    expiry_date DATE,
    
    -- Audit columns
    created_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    modified_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    -- Indexes
    INDEX idx_product_id (product_id),
    INDEX idx_product_current (is_current),
    INDEX idx_category (category_id, subcategory_id)
);

-- SNOWFLAKE SCHEMA: Genormaliseerde dimension tables
-- Dimension tables zijn verder genormaliseerd

-- Core Dimension Tables (gesnowflaked)
CREATE TABLE dim_product_core (
    product_key INT PRIMARY KEY,
    product_id VARCHAR(50) NOT NULL,
    product_name VARCHAR(255) NOT NULL,
    product_description TEXT,
    category_key INT NOT NULL,
    brand_key INT NOT NULL,
    supplier_key INT NOT NULL,
    unit_price DECIMAL(10, 2) NOT NULL,
    cost_price DECIMAL(10, 2) NOT NULL,
    weight_kg DECIMAL(8, 3),
    dimensions VARCHAR(50),
    is_current BOOLEAN DEFAULT TRUE,
    valid_from DATE NOT NULL,
    valid_to DATE,
    FOREIGN KEY (category_key) REFERENCES dim_product_category(category_key),
    FOREIGN KEY (brand_key) REFERENCES dim_brand(brand_key),
    FOREIGN KEY (supplier_key) REFERENCES dim_supplier(supplier_key)
);

-- Product Category Dimension
CREATE TABLE dim_product_category (
    category_key INT PRIMARY KEY,
    category_id INT NOT NULL,
    category_name VARCHAR(100) NOT NULL,
    category_description TEXT,
    department_key INT,
    is_active BOOLEAN DEFAULT TRUE,
    FOREIGN KEY (department_key) REFERENCES dim_department(department_key)
);

-- Product Subcategory (indien nodig)
CREATE TABLE dim_product_subcategory (
    subcategory_key INT PRIMARY KEY,
    subcategory_id INT NOT NULL,
    subcategory_name VARCHAR(100) NOT NULL,
    category_key INT NOT NULL,
    FOREIGN KEY (category_key) REFERENCES dim_product_category(category_key)
);

-- Brand Dimension
CREATE TABLE dim_brand (
    brand_key INT PRIMARY KEY,
    brand_id INT NOT NULL,
    brand_name VARCHAR(100) NOT NULL,
    brand_description TEXT,
    manufacturer_key INT,
    country_of_origin VARCHAR(100),
    FOREIGN KEY (manufacturer_key) REFERENCES dim_manufacturer(manufacturer_key)
);

-- Supplier Dimension
CREATE TABLE dim_supplier (
    supplier_key INT PRIMARY KEY,
    supplier_id INT NOT NULL,
    supplier_name VARCHAR(200) NOT NULL,
    contact_name VARCHAR(100),
    contact_title VARCHAR(100),
    address VARCHAR(500),
    city VARCHAR(100),
    region VARCHAR(100),
    postal_code VARCHAR(20),
    country VARCHAR(100),
    phone VARCHAR(50),
    fax VARCHAR(50),
    homepage VARCHAR(500),
    is_active BOOLEAN DEFAULT TRUE
);

-- Department Dimension
CREATE TABLE dim_department (
    department_key INT PRIMARY KEY,
    department_id INT NOT NULL,
    department_name VARCHAR(100) NOT NULL,
    division_key INT,
    FOREIGN KEY (division_key) REFERENCES dim_division(division_key)
);

-- Manufacturer Dimension
CREATE TABLE dim_manufacturer (
    manufacturer_key INT PRIMARY KEY,
    manufacturer_id INT NOT NULL,
    manufacturer_name VARCHAR(200) NOT NULL,
    headquarters_country VARCHAR(100),
    founded_year INT
);

-- Division Dimension
CREATE TABLE dim_division (
    division_key INT PRIMARY KEY,
    division_id INT NOT NULL,
    division_name VARCHAR(100) NOT NULL
);

-- Query voorbeeld met snowflake schema
SELECT 
    fs.transaction_date_key,
    dpc.category_name,
    db.brand_name,
    SUM(fs.total_amount) AS total_sales,
    COUNT(*) AS transaction_count
FROM fact_sales fs
JOIN dim_product_core dp ON fs.product_key = dp.product_key
JOIN dim_product_category dpc ON dp.category_key = dpc.category_key
JOIN dim_brand db ON dp.brand_key = db.brand_key
WHERE fs.transaction_date_key BETWEEN 20240101 AND 20241231
GROUP BY 
    fs.transaction_date_key,
    dpc.category_name,
    db.brand_name
ORDER BY total_sales DESC;

-- TRADITIONALE ETL PIPELINE: Transformatie gebeurt voor loading
-- Gebruikt vaak staging areas voor data transformatie

-- Stap 1: Extract naar Staging Area
CREATE TABLE staging_orders (
    order_id INT,
    customer_id INT,
    order_date TIMESTAMP,
    required_date TIMESTAMP,
    shipped_date TIMESTAMP,
    ship_via INT,
    freight DECIMAL(10, 2),
    ship_name VARCHAR(100),
    ship_address VARCHAR(200),
    ship_city VARCHAR(100),
    ship_region VARCHAR(100),
    ship_postal_code VARCHAR(20),
    ship_country VARCHAR(100),
    extracted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Extract procedure (voorbeeld voor SQL Server)
CREATE PROCEDURE sp_extract_orders
AS
BEGIN
    -- Truncate staging table
    TRUNCATE TABLE staging_orders;
    
    -- Extract data from source system
    INSERT INTO staging_orders (
        order_id, customer_id, order_date, required_date,
        shipped_date, ship_via, freight, ship_name,
        ship_address, ship_city, ship_region,
        ship_postal_code, ship_country
    )
    SELECT 
        o.OrderID,
        o.CustomerID,
        o.OrderDate,
        o.RequiredDate,
        o.ShippedDate,
        o.ShipVia,
        o.Freight,
        o.ShipName,
        o.ShipAddress,
        o.ShipCity,
        o.ShipRegion,
        o.ShipPostalCode,
        o.ShipCountry
    FROM SourceDatabase.dbo.Orders o
    WHERE o.OrderDate >= DATEADD(day, -1, CAST(GETDATE() AS DATE))
        OR o.ShippedDate >= DATEADD(day, -1, CAST(GETDATE() AS DATE))
        OR o.ModifiedDate >= DATEADD(day, -1, CAST(GETDATE() AS DATE));
END;

-- Stap 2: Transform in Staging Area
CREATE PROCEDURE sp_transform_orders
AS
BEGIN
    -- Data cleansing en validation
    UPDATE staging_orders
    SET 
        -- Standardize date formats
        order_date = CASE 
            WHEN ISDATE(order_date) = 1 THEN CAST(order_date AS DATETIME)
            ELSE NULL
        END,
        -- Clean up string data
        ship_name = TRIM(ship_name),
        ship_city = UPPER(TRIM(ship_city)),
        ship_country = CASE 
            WHEN ship_country = 'USA' THEN 'United States'
            WHEN ship_country = 'UK' THEN 'United Kingdom'
            ELSE ship_country
        END,
        -- Handle null values
        freight = ISNULL(freight, 0)
    WHERE extracted_at >= DATEADD(hour, -24, GETDATE());
    
    -- Mark invalid records
    UPDATE staging_orders
    SET validation_status = 'INVALID'
    WHERE order_id IS NULL 
        OR customer_id IS NULL 
        OR order_date IS NULL;
    
    -- Create transformed staging table
    CREATE TABLE staging_orders_transformed AS
    SELECT 
        order_id,
        customer_id,
        order_date,
        required_date,
        shipped_date,
        ship_via,
        freight,
        ship_name,
        ship_address,
        ship_city,
        ship_region,
        ship_postal_code,
        ship_country,
        -- Add derived columns
        DATEDIFF(day, order_date, shipped_date) AS days_to_ship,
        CASE 
            WHEN shipped_date <= required_date THEN 'On Time'
            ELSE 'Late'
        END AS shipping_status
    FROM staging_orders
    WHERE validation_status IS NULL;  -- Only valid records
END;

-- Stap 3: Load naar Data Warehouse
CREATE PROCEDURE sp_load_orders
AS
BEGIN
    BEGIN TRANSACTION;
    
    INSERT INTO fact_orders (
        order_key, date_key, customer_key, shipper_key,
        freight, order_status, days_to_ship, order_amount
    )
    SELECT 
        so.order_id,
        dd.date_key,
        dc.customer_key,
        ds.shipper_key,
        so.freight,
        so.shipping_status,
        so.days_to_ship,
        -- Calculate order amount from order details
        (
            SELECT SUM(UnitPrice * Quantity * (1 - Discount))
            FROM SourceDatabase.dbo.[Order Details] od
            WHERE od.OrderID = so.order_id
        ) AS order_amount
    FROM staging_orders_transformed so
    JOIN dim_date dd ON CAST(so.order_date AS DATE) = dd.full_date
    JOIN dim_customer dc ON so.customer_id = dc.customer_id
    JOIN dim_shipper ds ON so.ship_via = ds.shipper_id
    WHERE NOT EXISTS (
        SELECT 1 FROM fact_orders fo 
        WHERE fo.order_key = so.order_id
    );
    
    COMMIT TRANSACTION;
    
    -- Log successful load
    INSERT INTO etl_log (
        process_name, start_time, end_time, 
        rows_processed, status, error_message
    )
    VALUES (
        'sp_load_orders', 
        @start_time, 
        GETDATE(), 
        @@ROWCOUNT, 
        'SUCCESS', 
        NULL
    );
END;

-- MODERNE ELT PIPELINE: Load raw data eerst, dan transformeren
-- Gebruikt de processing power van het moderne data warehouse

-- Stap 1: Extract en Load raw data naar Data Lake / Bronze Layer
-- Gebruik bijvoorbeeld Azure Data Factory, AWS Glue, of dbt

-- Raw/bronze table (Parquet format voor efficiency)
CREATE EXTERNAL TABLE bronze.orders_raw (
    order_id INT,
    customer_id STRING,
    order_date STRING,  -- Raw string, kan verschillende formats hebben
    required_date STRING,
    shipped_date STRING,
    ship_via INT,
    freight DECIMAL(10, 2),
    ship_name STRING,
    ship_address STRING,
    ship_city STRING,
    ship_region STRING,
    ship_postal_code STRING,
    ship_country STRING,
    source_file STRING,
    ingestion_time TIMESTAMP,
    batch_id STRING
)
STORED AS PARQUET
LOCATION 's3://data-lake/bronze/orders/';

-- Stap 2: Transform in Data Warehouse (Silver Layer)
-- Gebruik data warehouse processing power (Snowflake, BigQuery, etc.)

CREATE OR REPLACE TABLE silver.orders_cleaned 
CLUSTER BY (order_date_parsed, customer_id)
AS
WITH cleaned_orders AS (
    SELECT 
        -- Standardize and parse dates
        order_id,
        customer_id,
        TRY_CAST(order_date AS TIMESTAMP) AS order_date_parsed,
        TRY_CAST(required_date AS TIMESTAMP) AS required_date_parsed,
        TRY_CAST(shipped_date AS TIMESTAMP) AS shipped_date_parsed,
        
        -- Clean and standardize string fields
        ship_via,
        COALESCE(freight, 0) AS freight,
        TRIM(ship_name) AS ship_name_clean,
        TRIM(ship_address) AS ship_address_clean,
        INITCAP(TRIM(ship_city)) AS ship_city_clean,
        UPPER(TRIM(ship_region)) AS ship_region_clean,
        TRIM(ship_postal_code) AS ship_postal_code_clean,
        CASE 
            WHEN UPPER(TRIM(ship_country)) = 'USA' THEN 'United States'
            WHEN UPPER(TRIM(ship_country)) = 'UK' THEN 'United Kingdom'
            WHEN UPPER(TRIM(ship_country)) = 'NL' THEN 'Netherlands'
            ELSE INITCAP(TRIM(ship_country))
        END AS ship_country_clean,
        
        -- Add quality flags
        CASE 
            WHEN order_id IS NULL THEN 'MISSING_ORDER_ID'
            WHEN customer_id IS NULL THEN 'MISSING_CUSTOMER_ID'
            WHEN TRY_CAST(order_date AS TIMESTAMP) IS NULL THEN 'INVALID_ORDER_DATE'
            ELSE NULL
        END AS data_quality_issue,
        
        -- Audit columns
        source_file,
        ingestion_time,
        batch_id,
        CURRENT_TIMESTAMP() AS processed_at
    FROM bronze.orders_raw
    WHERE batch_id = '{{ batch_id }}'  -- Incremental processing
)
SELECT *
FROM cleaned_orders
WHERE data_quality_issue IS NULL;  -- Filter out invalid records

-- Stap 3: Business Transformations (Gold Layer)
CREATE OR REPLACE TABLE gold.fact_orders 
CLUSTER BY (order_date_key, customer_key)
AS
SELECT 
    -- Surrogate keys
    ROW_NUMBER() OVER (ORDER BY so.order_id) AS order_key,
    dd.date_key AS order_date_key,
    dc.customer_key,
    ds.shipper_key,
    
    -- Measures
    so.freight,
    od.total_amount,
    od.total_quantity,
    od.total_discount,
    
    -- Derived business metrics
    CASE 
        WHEN so.shipped_date_parsed <= so.required_date_parsed 
        THEN 'On Time'
        ELSE 'Late'
    END AS shipping_status,
    
    DATEDIFF(day, so.order_date_parsed, so.shipped_date_parsed) AS days_to_ship,
    
    -- Audit columns
    so.order_id AS source_order_id,
    so.ingestion_time,
    CURRENT_TIMESTAMP() AS dw_loaded_at
    
FROM silver.orders_cleaned so
JOIN gold.dim_date dd 
    ON DATE(so.order_date_parsed) = dd.full_date
JOIN gold.dim_customer dc 
    ON so.customer_id = dc.customer_id
    AND dc.is_current = TRUE
LEFT JOIN gold.dim_shipper ds 
    ON so.ship_via = ds.shipper_id
    AND ds.is_current = TRUE
LEFT JOIN (
    -- Aggregate order details
    SELECT 
        OrderID,
        SUM(UnitPrice * Quantity) AS total_amount,
        SUM(Quantity) AS total_quantity,
        SUM(UnitPrice * Quantity * Discount) AS total_discount
    FROM bronze.order_details_raw
    WHERE batch_id = '{{ batch_id }}'
    GROUP BY OrderID
) od ON so.order_id = od.OrderID;

-- DATA VAULT 2.0 MODELING: Agile, scalable data warehouse modeling
-- Gebaseerd op Hub-Link-Satellite structuren

-- HUB TABLES: Business keys
CREATE TABLE hub_customer (
    customer_hash_key CHAR(32) PRIMARY KEY,  -- MD5 hash van business key
    customer_id VARCHAR(50) NOT NULL,           -- Business key
    load_date TIMESTAMP NOT NULL,
    record_source VARCHAR(100) NOT NULL,      -- Bron systeem
    UNIQUE (customer_id)
);

CREATE TABLE hub_product (
    product_hash_key CHAR(32) PRIMARY KEY,
    product_id VARCHAR(50) NOT NULL,
    load_date TIMESTAMP NOT NULL,
    record_source VARCHAR(100) NOT NULL,
    UNIQUE (product_id)
);

-- LINK TABLES: Relaties tussen hubs
CREATE TABLE link_customer_order (
    customer_order_hash_key CHAR(32) PRIMARY KEY,
    customer_hash_key CHAR(32) NOT NULL,
    order_hash_key CHAR(32) NOT NULL,
    load_date TIMESTAMP NOT NULL,
    record_source VARCHAR(100) NOT NULL,
    FOREIGN KEY (customer_hash_key) REFERENCES hub_customer(customer_hash_key),
    FOREIGN KEY (order_hash_key) REFERENCES hub_order(order_hash_key),
    UNIQUE (customer_hash_key, order_hash_key)
);

CREATE TABLE link_order_product (
    order_product_hash_key CHAR(32) PRIMARY KEY,
    order_hash_key CHAR(32) NOT NULL,
    product_hash_key CHAR(32) NOT NULL,
    load_date TIMESTAMP NOT NULL,
    record_source VARCHAR(100) NOT NULL,
    UNIQUE (order_hash_key, product_hash_key)
);

-- SATELLITE TABLES: Beschrijvende attributen met historische tracking
CREATE TABLE sat_customer_details (
    customer_hash_key CHAR(32) NOT NULL,
    load_date TIMESTAMP NOT NULL,
    load_end_date TIMESTAMP,
    record_source VARCHAR(100) NOT NULL,
    hash_diff CHAR(32) NOT NULL,  -- Hash van alle attributen voor change detection
    
    -- Customer attributes
    customer_name VARCHAR(255),
    email VARCHAR(255),
    phone VARCHAR(50),
    address_line VARCHAR(255),
    city VARCHAR(100),
    postal_code VARCHAR(20),
    country VARCHAR(100),
    
    PRIMARY KEY (customer_hash_key, load_date),
    FOREIGN KEY (customer_hash_key) REFERENCES hub_customer(customer_hash_key)
);

CREATE TABLE sat_product_details (
    product_hash_key CHAR(32) NOT NULL,
    load_date TIMESTAMP NOT NULL,
    load_end_date TIMESTAMP,
    record_source VARCHAR(100) NOT NULL,
    hash_diff CHAR(32) NOT NULL,
    
    -- Product attributes
    product_name VARCHAR(255),
    description TEXT,
    category VARCHAR(100),
    brand VARCHAR(100),
    unit_price DECIMAL(10, 2),
    weight_kg DECIMAL(8, 3),
    dimensions VARCHAR(50),
    
    PRIMARY KEY (product_hash_key, load_date),
    FOREIGN KEY (product_hash_key) REFERENCES hub_product(product_hash_key)
);

-- BUSINESS VAULT: Business rules en aggregaties
CREATE VIEW business_vault.customer_orders AS
SELECT 
    hc.customer_id,
    MAX(satc.customer_name) AS customer_name,
    COUNT(DISTINCT lo.order_hash_key) AS total_orders,
    SUM(sato.order_amount) AS total_spent,
    MIN(sato.order_date) AS first_order_date,
    MAX(sato.order_date) AS last_order_date
FROM hub_customer hc
JOIN sat_customer_details satc 
    ON hc.customer_hash_key = satc.customer_hash_key
    AND satc.load_end_date IS NULL  -- Current record
JOIN link_customer_order lco 
    ON hc.customer_hash_key = lco.customer_hash_key
JOIN hub_order ho 
    ON lco.order_hash_key = ho.order_hash_key
JOIN sat_order_details sato 
    ON ho.order_hash_key = sato.order_hash_key
    AND sato.load_end_date IS NULL
GROUP BY hc.customer_id;

-- DATA QUALITY FRAMEWORK: Comprehensive checks en monitoring

-- Data Quality Rules Definition
CREATE TABLE dq_rules (
    rule_id INT PRIMARY KEY,
    rule_name VARCHAR(200) NOT NULL,
    rule_type VARCHAR(50) NOT NULL,  -- 'completeness', 'validity', 'consistency', 'timeliness'
    rule_description TEXT,
    table_name VARCHAR(100) NOT NULL,
    column_name VARCHAR(100),
    rule_condition VARCHAR(1000) NOT NULL,
    severity_level VARCHAR(20) NOT NULL,  -- 'critical', 'high', 'medium', 'low'
    is_active BOOLEAN DEFAULT TRUE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Example Data Quality Rules
INSERT INTO dq_rules (rule_id, rule_name, rule_type, rule_description, table_name, column_name, rule_condition, severity_level) VALUES
(1, 'Customer ID Not Null', 'completeness', 'Customer ID must not be null', 'dim_customer', 'customer_id', 'customer_id IS NOT NULL', 'critical'),
(2, 'Valid Email Format', 'validity', 'Email must be valid format', 'dim_customer', 'email', "email LIKE '%@%.%'", 'high'),
(3, 'Order Date Not Future', 'validity', 'Order date cannot be in the future', 'fact_orders', 'order_date', 'order_date <= CURRENT_DATE', 'high'),
(4, 'Positive Quantity', 'validity', 'Order quantity must be positive', 'fact_order_details', 'quantity', 'quantity > 0', 'critical'),
(5, 'Referential Integrity Customer', 'consistency', 'All orders must reference valid customers', 'fact_orders', NULL, 'customer_key IN (SELECT customer_key FROM dim_customer)', 'critical');

-- Data Quality Results Logging
CREATE TABLE dq_results (
    result_id BIGINT PRIMARY KEY,
    rule_id INT NOT NULL,
    batch_id VARCHAR(100) NOT NULL,
    check_timestamp TIMESTAMP NOT NULL,
    table_name VARCHAR(100) NOT NULL,
    error_count INT NOT NULL,
    total_count INT NOT NULL,
    error_percentage DECIMAL(5, 2) NOT NULL,
    error_details JSONB,  -- Sample error records
    status VARCHAR(20) NOT NULL,  -- 'passed', 'failed', 'warning'
    FOREIGN KEY (rule_id) REFERENCES dq_rules(rule_id),
    INDEX idx_batch (batch_id),
    INDEX idx_timestamp (check_timestamp)
);

-- Automated Data Quality Check Procedure
CREATE PROCEDURE sp_run_data_quality_checks (
    IN p_batch_id VARCHAR(100),
    IN p_table_name VARCHAR(100) DEFAULT NULL
)
AS $$
DECLARE
    v_rule_id INT;
    v_rule_name VARCHAR(200);
    v_rule_condition VARCHAR(1000);
    v_table_name VARCHAR(100);
    v_error_count INT;
    v_total_count INT;
    v_error_percentage DECIMAL(5, 2);
    v_error_details JSONB;
    v_status VARCHAR(20);
    rule_cursor CURSOR FOR
        SELECT rule_id, rule_name, rule_condition, table_name
        FROM dq_rules
        WHERE is_active = TRUE
            AND (p_table_name IS NULL OR table_name = p_table_name);
BEGIN
    OPEN rule_cursor;
    
    LOOP
        FETCH rule_cursor INTO v_rule_id, v_rule_name, v_rule_condition, v_table_name;
        EXIT WHEN NOT FOUND;
        
        -- Execute rule check
        EXECUTE FORMAT(
            'SELECT 
                COUNT(*) FILTER (WHERE NOT (%s)) AS error_count,
                COUNT(*) AS total_count,
                CASE 
                    WHEN COUNT(*) > 0 THEN 
                        ROUND(100.0 * COUNT(*) FILTER (WHERE NOT (%s)) / COUNT(*), 2)
                    ELSE 0 
                END AS error_percentage,
                COALESCE(
                    JSONB_AGG(
                        JSONB_BUILD_OBJECT(
                            ''record_id'', %I,
                            ''failed_column'', %L
                        )
                    ) FILTER (WHERE NOT (%s)),
                    ''[]''::JSONB
                ) AS error_details
            FROM %I',
            v_rule_condition,
            v_rule_condition,
            CASE 
                WHEN v_table_name = 'dim_customer' THEN 'customer_key'
                WHEN v_table_name = 'fact_orders' THEN 'order_key'
                ELSE 'id'
            END,
            v_rule_name,
            v_rule_condition,
            v_table_name
        ) INTO v_error_count, v_total_count, v_error_percentage, v_error_details;
        
        -- Determine status
        v_status := CASE 
            WHEN v_error_count = 0 THEN 'passed'
            WHEN v_error_percentage <= 5.0 THEN 'warning'
            ELSE 'failed'
        END;
        
        -- Log result
        INSERT INTO dq_results (
            rule_id, batch_id, check_timestamp, table_name,
            error_count, total_count, error_percentage,
            error_details, status
        ) VALUES (
            v_rule_id, p_batch_id, CURRENT_TIMESTAMP, v_table_name,
            v_error_count, v_total_count, v_error_percentage,
            v_error_details, v_status
        );
        
        -- Alert on critical failures
        IF v_status = 'failed' THEN
            RAISE NOTICE 'Data quality check failed: % (Rule: %, Table: %, Errors: %/% - %%)',
                v_rule_name, v_rule_id, v_table_name, v_error_count, v_total_count, v_error_percentage;
        END IF;
        
    END LOOP;
    
    CLOSE rule_cursor;
    
    -- Return summary
    SELECT 
        COUNT(*) AS total_checks,
        COUNT(*) FILTER (WHERE status = 'passed') AS passed_checks,
        COUNT(*) FILTER (WHERE status = 'warning') AS warning_checks,
        COUNT(*) FILTER (WHERE status = 'failed') AS failed_checks
    FROM dq_results
    WHERE batch_id = p_batch_id;
    
END;
$$ LANGUAGE plpgsql;

-- DATA WAREHOUSE PERFORMANCE OPTIMIZATION TECHNIQUES

-- 1. Partitionering voor grote tabellen
-- Date-based partitionering voor fact tables
CREATE TABLE fact_sales_partitioned (
    sales_key BIGINT,
    date_key INT,
    customer_key INT,
    product_key INT,
    sales_amount DECIMAL(15, 2),
    sales_quantity INT
)
PARTITION BY RANGE (date_key) (
    PARTITION p_2023_q1 VALUES LESS THAN (20230401),
    PARTITION p_2023_q2 VALUES LESS THAN (20230701),
    PARTITION p_2023_q3 VALUES LESS THAN (20231001),
    PARTITION p_2023_q4 VALUES LESS THAN (20240101),
    PARTITION p_2024_q1 VALUES LESS THAN (20240401),
    PARTITION p_2024_q2 VALUES LESS THAN (20240701),
    PARTITION p_2024_q3 VALUES LESS THAN (20241001),
    PARTITION p_2024_q4 VALUES LESS THAN (20250101),
    PARTITION p_future VALUES LESS THAN (MAXVALUE)
);

-- 2. Clustering voor query performance
-- Snowflake clustering voor dimension tables
CREATE OR REPLACE TABLE dim_customer 
CLUSTER BY (country, city, customer_segment);

-- Fact table clustering op vaak gebruikte join columns
CREATE OR REPLACE TABLE fact_orders 
CLUSTER BY (date_key, customer_key, product_key);

-- 3. Materialized Views voor complexe aggregaties
CREATE MATERIALIZED VIEW mv_daily_sales_summary
AS
SELECT 
    date_key,
    customer_key,
    product_key,
    COUNT(*) AS transaction_count,
    SUM(sales_amount) AS total_sales,
    AVG(sales_amount) AS avg_sale_amount,
    MIN(sales_amount) AS min_sale_amount,
    MAX(sales_amount) AS max_sale_amount
FROM fact_sales
GROUP BY date_key, customer_key, product_key;

-- Automatic refresh van materialized views
CREATE OR REPLACE TASK refresh_mv_daily_sales
WAREHOUSE = analytics_wh
SCHEDULE = 'USING CRON 0 2 * * * UTC'  -- Dagelijks om 2:00 UTC
AS
ALTER MATERIALIZED VIEW mv_daily_sales_summary REFRESH;

-- 4. Query Performance Monitoring
CREATE TABLE query_performance_log (
    query_id VARCHAR(100) PRIMARY KEY,
    query_text TEXT,
    user_name VARCHAR(100),
    start_time TIMESTAMP,
    end_time TIMESTAMP,
    execution_time_ms BIGINT,
    rows_returned BIGINT,
    bytes_scanned BIGINT,
    warehouse_size VARCHAR(50),
    error_message TEXT,
    query_plan JSONB,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 5. Index strategieën voor mixed workloads
-- Bitmap indexes voor low-cardinality columns
CREATE BITMAP INDEX idx_sales_status 
ON fact_sales (order_status);

-- B-tree indexes voor high-cardinality columns
CREATE INDEX idx_sales_customer_date 
ON fact_sales (customer_key, date_key);

-- 6. Data Compression voor storage optimalisatie
ALTER TABLE fact_sales 
REORGANIZE WITH (DATA_COMPRESSION = PAGE);

-- 7. Workload Management voor resource allocation
CREATE WORKLOAD GROUP reporting_wg 
WITH (
    MIN_PERCENTAGE_RESOURCE = 30,
    REQUEST_MIN_RESOURCE_GRANT_PERCENT = 5,
    CAP_PERCENTAGE_RESOURCE = 60
);

CREATE WORKLOAD GROUP adhoc_wg 
WITH (
    MIN_PERCENTAGE_RESOURCE = 10,
    REQUEST_MIN_RESOURCE_GRANT_PERCENT = 1,
    CAP_PERCENTAGE_RESOURCE = 30
);

-- 8. Query Hints voor performance tuning
SELECT /*+ FACT(fs) */
    dc.customer_name,
    dp.product_name,
    SUM(fs.sales_amount) AS total_sales
FROM fact_sales fs
JOIN dim_customer dc ON fs.customer_key = dc.customer_key
JOIN dim_product dp ON fs.product_key = dp.product_key
WHERE fs.date_key BETWEEN 20240101 AND 20241231
GROUP BY dc.customer_name, dp.product_name;

-- 9. Automatic Query Optimization
ALTER DATABASE data_warehouse 
SET AUTO_OPTIMIZE = TRUE;

ALTER TABLE fact_sales 
SET AUTO_CLUSTERING = TRUE;

-- MEDALLION ARCHITECTURE: Data Lakehouse pattern
-- Bronze (Raw) → Silver (Cleaned) → Gold (Business-ready)

-- BRONZE LAYER: Raw, immutable data
CREATE TABLE bronze.orders 
USING DELTA
LOCATION 'abfss://data-lake@storage.dfs.core.windows.net/bronze/orders';

-- Python code voor data ingestion naar Bronze
"""
# Databricks notebook voor data ingestion
from pyspark.sql.functions import *

# Read from source
orders_df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:sqlserver://server.database.windows.net:1433") \
    .option("dbtable", "Sales.Orders") \
    .option("user", "username") \
    .option("password", "password") \
    .load()

# Add metadata columns
orders_augmented = orders_df \
    .withColumn("ingestion_timestamp", current_timestamp()) \
    .withColumn("source_system", lit("SQL_SERVER")) \
    .withColumn("batch_id", lit("2024-12-20-001"))

# Write to Bronze layer (append mode)
orders_augmented.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("bronze.orders")
"""

-- SILVER LAYER: Cleaned, validated, conformed data
CREATE OR REPLACE TABLE silver.orders_cleaned 
USING DELTA
PARTITIONED BY (order_date)
LOCATION 'abfss://data-lake@storage.dfs.core.windows.net/silver/orders_cleaned'
AS
SELECT 
    OrderID AS order_id,
    CustomerID AS customer_id,
    TRY_CAST(OrderDate AS DATE) AS order_date,
    TRY_CAST(RequiredDate AS DATE) AS required_date,
    TRY_CAST(ShippedDate AS DATE) AS shipped_date,
    ShipVia AS ship_via,
    COALESCE(Freight, 0) AS freight_amount,
    TRIM(ShipName) AS ship_to_name,
    TRIM(ShipAddress) AS ship_to_address,
    TRIM(ShipCity) AS ship_to_city,
    TRIM(ShipRegion) AS ship_to_region,
    TRIM(ShipPostalCode) AS ship_to_postal_code,
    TRIM(ShipCountry) AS ship_to_country,
    
    -- Data quality flags
    CASE 
        WHEN OrderID IS NULL THEN TRUE
        ELSE FALSE
    END AS is_order_id_null,
    
    CASE 
        WHEN TRY_CAST(OrderDate AS DATE) IS NULL THEN TRUE
        ELSE FALSE
    END AS is_order_date_invalid,
    
    -- Metadata
    ingestion_timestamp,
    source_system,
    batch_id,
    current_timestamp() AS silver_processed_at
    
FROM bronze.orders
WHERE batch_id = '2024-12-20-001';

-- GOLD LAYER: Business-ready dimensional model
CREATE OR REPLACE TABLE gold.fact_orders 
USING DELTA
PARTITIONED BY (order_year, order_month)
LOCATION 'abfss://data-lake@storage.dfs.core.windows.net/gold/fact_orders'
AS
SELECT 
    -- Surrogate keys
    ROW_NUMBER() OVER (ORDER BY so.order_id) AS order_key,
    dd.date_key,
    dc.customer_key,
    ds.shipper_key,
    
    -- Business measures
    so.order_id AS source_order_id,
    so.freight_amount,
    od.total_order_amount,
    od.total_quantity,
    od.average_unit_price,
    
    -- Derived metrics
    CASE 
        WHEN so.shipped_date IS NOT NULL 
            AND so.shipped_date <= so.required_date 
        THEN 'On Time'
        WHEN so.shipped_date IS NOT NULL 
            AND so.shipped_date > so.required_date 
        THEN 'Late'
        ELSE 'Not Shipped'
    END AS shipping_status,
    
    DATEDIFF(day, so.order_date, so.shipped_date) AS days_to_ship,
    
    -- Audit columns
    so.silver_processed_at,
    current_timestamp() AS gold_loaded_at,
    YEAR(so.order_date) AS order_year,
    MONTH(so.order_date) AS order_month
    
FROM silver.orders_cleaned so
JOIN gold.dim_date dd ON so.order_date = dd.full_date
JOIN gold.dim_customer dc ON so.customer_id = dc.customer_id
    AND dc.is_current = TRUE
LEFT JOIN gold.dim_shipper ds ON so.ship_via = ds.shipper_id
    AND ds.is_current = TRUE
LEFT JOIN (
    -- Aggregate order details from silver layer
    SELECT 
        OrderID,
        SUM(UnitPrice * Quantity) AS total_order_amount,
        SUM(Quantity) AS total_quantity,
        AVG(UnitPrice) AS average_unit_price
    FROM silver.order_details_cleaned
    WHERE batch_id = '2024-12-20-001'
    GROUP BY OrderID
) od ON so.order_id = od.OrderID;

-- Z-Order Optimizatie voor query performance
OPTIMIZE gold.fact_orders
ZORDER BY (customer_key, date_key);

-- IMPLEMENTATIE ROADMAP: Gefaseerde aanpak

-- Fase 1: Foundation & Assessment (Weeks 1-4)
-- Doel: Begrijp requirements en zet foundation op
-- Deliverables:
-- 1. Requirements gathering workshops
-- 2. Technology stack selection
-- 3. Data source inventory
-- 4. Initial architecture design
-- 5. Proof of Concept (PoC)

-- Fase 2: Core Platform Setup (Weeks 5-8)
-- Doel: Zet core platform en eerste data pipeline op
-- SQL voor platform setup:

-- Create core database structure
CREATE DATABASE data_warehouse
WITH (
    MAXSIZE = 100 TB,
    EDITION = 'Datawarehouse',
    SERVICE_OBJECTIVE = 'DW500c'
);

-- Create schemas voor verschillende data layers
CREATE SCHEMA staging;
CREATE SCHEMA edw;  -- Enterprise Data Warehouse
CREATE SCHEMA marts;  -- Data Marts
CREATE SCHEMA reports;
CREATE SCHEMA metadata;

-- Create metadata management tables
CREATE TABLE metadata.data_sources (
    source_id INT PRIMARY KEY,
    source_name VARCHAR(100) NOT NULL,
    source_type VARCHAR(50) NOT NULL,  -- 'Database', 'API', 'File'
    connection_string VARCHAR(500),
    extraction_frequency VARCHAR(50),  -- 'Daily', 'Hourly', 'Real-time'
    last_extraction TIMESTAMP,
    is_active BOOLEAN DEFAULT TRUE
);

CREATE TABLE metadata.etl_jobs (
    job_id INT PRIMARY KEY,
    job_name VARCHAR(200) NOT NULL,
    job_type VARCHAR(50) NOT NULL,  -- 'Extract', 'Transform', 'Load'
    source_id INT,
    target_table VARCHAR(100),
    schedule_cron VARCHAR(50),
    depends_on VARCHAR(500),  -- Comma-separated job IDs
    is_active BOOLEAN DEFAULT TRUE,
    FOREIGN KEY (source_id) REFERENCES metadata.data_sources(source_id)
);

-- Fase 3: First Data Mart (Weeks 9-12)
-- Doel: Implementeer eerste business domain (bijv. Sales)
-- Deliverables:
-- 1. Sales data mart met star schema
-- 2. ETL pipeline voor sales data
-- 3. Basic reporting layer
-- 4. Data quality checks

-- Create sales data mart
CREATE SCHEMA sales_mart;

CREATE TABLE sales_mart.fact_sales (
    sales_key BIGINT IDENTITY(1, 1) PRIMARY KEY,
    date_key INT NOT NULL,
    product_key INT NOT NULL,
    customer_key INT NOT NULL,
    store_key INT NOT NULL,
    sales_quantity INT NOT NULL,
    sales_amount DECIMAL(15, 2) NOT NULL,
    cost_amount DECIMAL(15, 2) NOT NULL,
    profit_amount DECIMAL(15, 2) NOT NULL,
    created_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (date_key) REFERENCES sales_mart.dim_date(date_key),
    FOREIGN KEY (product_key) REFERENCES sales_mart.dim_product(product_key),
    FOREIGN KEY (customer_key) REFERENCES sales_mart.dim_customer(customer_key),
    FOREIGN KEY (store_key) REFERENCES sales_mart.dim_store(store_key)
);

-- Create partitionering voor performance
CREATE PARTITION FUNCTION pf_sales_date (INT)
AS RANGE RIGHT FOR VALUES (
    20230101, 20230401, 20230701, 20231001,
    20240101, 20240401, 20240701, 20241001
);

-- Fase 4: Expand & Scale (Weeks 13-24)
-- Doel: Voeg aanvullende data marts en geavanceerde features toe
-- Deliverables:
-- 1. Finance data mart
-- 2. Marketing data mart
-- 3. Advanced analytics features
-- 4. Performance optimization
-- 5. Data governance framework

-- Fase 5: Mature & Optimize (Ongoing)
-- Doel: Continue verbetering en optimalisatie
-- Deliverables:
-- 1. Advanced monitoring en alerting
-- 2. Cost optimization
-- 3. Automation en DevOps
-- 4. User training en adoption
-- 5. Innovation en nieuwe features