Data Modeling for Analytics.

# modeling/star_schema.py

"""
Star schema implementation for e-commerce analytics.

This model supports queries like:
- Revenue by product category by month
- Top customers by segment by region
- Sales trends by day of week
"""

from dataclasses import dataclass
from datetime import date, datetime
from typing import Optional, List
from decimal import Decimal


# =============================================================================
# Dimension Tables
# =============================================================================

@dataclass
class DimDate:
    """
    Date dimension with pre-computed attributes.
    
    Generated once, rarely changes.
    Enables fast date-based filtering and grouping.
    """
    date_key: int  # Surrogate key (YYYYMMDD format)
    date: date
    day: int
    month: int
    month_name: str
    quarter: int
    year: int
    day_of_week: int  # 0=Monday, 6=Sunday
    day_name: str
    week_of_year: int
    is_weekend: bool
    is_holiday: bool
    is_last_day_of_month: bool
    fiscal_year: int
    fiscal_quarter: int


@dataclass
class DimProduct:
    """
    Product dimension with hierarchy.
    
    Category → Subcategory → Product
    """
    product_key: int  # Surrogate key
    product_id: str  # Business key
    product_name: str
    category: str
    subcategory: str
    brand: str
    supplier: str
    unit_price: Decimal
    unit_cost: Decimal
    is_active: bool
    
    # SCD Type 2 fields (for tracking changes)
    valid_from: date
    valid_to: Optional[date]
    is_current: bool


@dataclass  
class DimCustomer:
    """
    Customer dimension with segmentation.
    """
    customer_key: int  # Surrogate key
    customer_id: str  # Business key
    first_name: str
    last_name: str
    email: str
    phone: Optional[str]
    
    # Demographics
    age_group: str  # "18-25", "26-35", etc.
    gender: Optional[str]
    
    # Segmentation
    segment: str  # "Premium", "Standard", "Budget"
    tier: str  # "Gold", "Silver", "Bronze"
    lifetime_value_band: str  # "High", "Medium", "Low"
    
    # Geography
    city: str
    state: str
    country: str
    region: str  # "North America", "Europe", etc.
    postal_code: str
    
    # Dates
    first_purchase_date: date
    signup_date: date
    
    # SCD fields
    valid_from: date
    valid_to: Optional[date]
    is_current: bool


@dataclass
class DimStore:
    """
    Store/channel dimension.
    """
    store_key: int
    store_id: str
    store_name: str
    store_type: str  # "Online", "Retail", "Warehouse"
    city: str
    state: str
    country: str
    region: str
    timezone: str
    opened_date: date
    manager_name: str
    square_footage: Optional[int]


# =============================================================================
# Fact Table
# =============================================================================

@dataclass
class FactSales:
    """
    Sales fact table.
    
    Grain: One row per line item in an order.
    
    Contains:
    - Foreign keys to all dimensions
    - Measures (numeric values to aggregate)
    - Degenerate dimensions (order_id - no separate table needed)
    """
    # Surrogate key
    sale_key: int
    
    # Foreign keys to dimensions
    date_key: int
    product_key: int
    customer_key: int
    store_key: int
    
    # Degenerate dimension (stored in fact, no lookup needed)
    order_id: str
    line_item_number: int
    
    # Measures (things we SUM, AVG, COUNT)
    quantity: int
    unit_price: Decimal
    unit_cost: Decimal
    discount_amount: Decimal
    tax_amount: Decimal
    
    # Derived measures (pre-calculated for performance)
    gross_revenue: Decimal  # quantity * unit_price
    net_revenue: Decimal    # gross_revenue - discount_amount
    profit: Decimal         # net_revenue - (quantity * unit_cost)
    
    # Timestamps
    order_timestamp: datetime
    ship_timestamp: Optional[datetime]


# =============================================================================
# Schema Creation SQL
# =============================================================================

STAR_SCHEMA_DDL = """
-- Date Dimension (generate for all dates needed)
CREATE TABLE dim_date (
    date_key        INTEGER PRIMARY KEY,  -- YYYYMMDD
    date            DATE NOT NULL,
    day             SMALLINT NOT NULL,
    month           SMALLINT NOT NULL,
    month_name      VARCHAR(10) NOT NULL,
    quarter         SMALLINT NOT NULL,
    year            SMALLINT NOT NULL,
    day_of_week     SMALLINT NOT NULL,
    day_name        VARCHAR(10) NOT NULL,
    week_of_year    SMALLINT NOT NULL,
    is_weekend      BOOLEAN NOT NULL,
    is_holiday      BOOLEAN NOT NULL,
    fiscal_year     SMALLINT NOT NULL,
    fiscal_quarter  SMALLINT NOT NULL
);

-- Product Dimension
CREATE TABLE dim_product (
    product_key     SERIAL PRIMARY KEY,
    product_id      VARCHAR(50) NOT NULL,
    product_name    VARCHAR(200) NOT NULL,
    category        VARCHAR(100) NOT NULL,
    subcategory     VARCHAR(100) NOT NULL,
    brand           VARCHAR(100),
    supplier        VARCHAR(100),
    unit_price      DECIMAL(10,2) NOT NULL,
    unit_cost       DECIMAL(10,2) NOT NULL,
    is_active       BOOLEAN NOT NULL DEFAULT TRUE,
    valid_from      DATE NOT NULL,
    valid_to        DATE,
    is_current      BOOLEAN NOT NULL DEFAULT TRUE
);

CREATE INDEX idx_product_id ON dim_product(product_id);
CREATE INDEX idx_product_category ON dim_product(category, subcategory);

-- Customer Dimension  
CREATE TABLE dim_customer (
    customer_key    SERIAL PRIMARY KEY,
    customer_id     VARCHAR(50) NOT NULL,
    first_name      VARCHAR(100),
    last_name       VARCHAR(100),
    email           VARCHAR(200),
    segment         VARCHAR(50) NOT NULL,
    tier            VARCHAR(20) NOT NULL,
    city            VARCHAR(100),
    state           VARCHAR(100),
    country         VARCHAR(100) NOT NULL,
    region          VARCHAR(50) NOT NULL,
    first_purchase_date DATE,
    valid_from      DATE NOT NULL,
    valid_to        DATE,
    is_current      BOOLEAN NOT NULL DEFAULT TRUE
);

CREATE INDEX idx_customer_id ON dim_customer(customer_id);
CREATE INDEX idx_customer_segment ON dim_customer(segment, tier);
CREATE INDEX idx_customer_region ON dim_customer(region, country);

-- Store Dimension
CREATE TABLE dim_store (
    store_key       SERIAL PRIMARY KEY,
    store_id        VARCHAR(50) NOT NULL,
    store_name      VARCHAR(200) NOT NULL,
    store_type      VARCHAR(50) NOT NULL,
    city            VARCHAR(100),
    state           VARCHAR(100),
    country         VARCHAR(100) NOT NULL,
    region          VARCHAR(50) NOT NULL,
    timezone        VARCHAR(50),
    opened_date     DATE
);

-- Sales Fact Table
CREATE TABLE fact_sales (
    sale_key        BIGSERIAL PRIMARY KEY,
    date_key        INTEGER NOT NULL REFERENCES dim_date(date_key),
    product_key     INTEGER NOT NULL REFERENCES dim_product(product_key),
    customer_key    INTEGER NOT NULL REFERENCES dim_customer(customer_key),
    store_key       INTEGER NOT NULL REFERENCES dim_store(store_key),
    order_id        VARCHAR(50) NOT NULL,
    line_item_number SMALLINT NOT NULL,
    quantity        INTEGER NOT NULL,
    unit_price      DECIMAL(10,2) NOT NULL,
    discount_amount DECIMAL(10,2) NOT NULL DEFAULT 0,
    tax_amount      DECIMAL(10,2) NOT NULL DEFAULT 0,
    gross_revenue   DECIMAL(12,2) NOT NULL,
    net_revenue     DECIMAL(12,2) NOT NULL,
    profit          DECIMAL(12,2) NOT NULL,
    order_timestamp TIMESTAMP NOT NULL
);

-- Critical indexes for fact table
CREATE INDEX idx_fact_date ON fact_sales(date_key);
CREATE INDEX idx_fact_product ON fact_sales(product_key);
CREATE INDEX idx_fact_customer ON fact_sales(customer_key);
CREATE INDEX idx_fact_order ON fact_sales(order_id);

-- Composite indexes for common query patterns
CREATE INDEX idx_fact_date_product ON fact_sales(date_key, product_key);
CREATE INDEX idx_fact_date_store ON fact_sales(date_key, store_key);
"""

# modeling/star_queries.py

"""
Example queries on star schema.

These demonstrate why star schema is efficient for analytics.
"""

EXAMPLE_QUERIES = """
-- =============================================================================
-- Query 1: Revenue by category by month
-- =============================================================================
-- Star join: fact + product and date dimensions

SELECT 
    p.category,
    d.year,
    d.month,
    SUM(f.net_revenue) as revenue,
    COUNT(DISTINCT f.order_id) as orders
FROM fact_sales f
JOIN dim_product p ON f.product_key = p.product_key
JOIN dim_date d ON f.date_key = d.date_key
WHERE d.year = 2024
  -- No is_current filter: each fact row already points at the
  -- product version valid at sale time (SCD Type 2)
GROUP BY p.category, d.year, d.month
ORDER BY d.year, d.month, revenue DESC;


-- =============================================================================
-- Query 2: Top 10 products by revenue this quarter
-- =============================================================================

SELECT 
    p.product_name,
    p.category,
    SUM(f.net_revenue) as revenue,
    SUM(f.quantity) as units_sold,
    SUM(f.profit) as profit
FROM fact_sales f
JOIN dim_product p ON f.product_key = p.product_key
JOIN dim_date d ON f.date_key = d.date_key
WHERE d.year = 2024 AND d.quarter = 4
GROUP BY p.product_key, p.product_name, p.category
ORDER BY revenue DESC
LIMIT 10;


-- =============================================================================
-- Query 3: Customer segment analysis
-- =============================================================================

SELECT 
    c.segment,
    c.tier,
    c.region,
    COUNT(DISTINCT c.customer_key) as customers,
    SUM(f.net_revenue) as revenue,
    SUM(f.net_revenue) / COUNT(DISTINCT c.customer_key) as revenue_per_customer,
    SUM(f.quantity) * 1.0 / COUNT(DISTINCT f.order_id) as avg_items_per_order
FROM fact_sales f
JOIN dim_customer c ON f.customer_key = c.customer_key
JOIN dim_date d ON f.date_key = d.date_key
WHERE d.year = 2024
GROUP BY c.segment, c.tier, c.region
ORDER BY revenue DESC;


-- =============================================================================
-- Query 4: Year-over-year comparison
-- =============================================================================

WITH current_year AS (
    SELECT 
        p.category,
        SUM(f.net_revenue) as revenue_2024
    FROM fact_sales f
    JOIN dim_product p ON f.product_key = p.product_key
    JOIN dim_date d ON f.date_key = d.date_key
    WHERE d.year = 2024
    GROUP BY p.category
),
previous_year AS (
    SELECT 
        p.category,
        SUM(f.net_revenue) as revenue_2023
    FROM fact_sales f
    JOIN dim_product p ON f.product_key = p.product_key
    JOIN dim_date d ON f.date_key = d.date_key
    WHERE d.year = 2023
    GROUP BY p.category
)
SELECT 
    c.category,
    c.revenue_2024,
    p.revenue_2023,
    (c.revenue_2024 - p.revenue_2023) / p.revenue_2023 * 100 as yoy_growth_pct
FROM current_year c
JOIN previous_year p ON c.category = p.category
ORDER BY yoy_growth_pct DESC;


-- =============================================================================
-- Query 5: Order metrics by store type
-- =============================================================================

SELECT 
    s.store_type,
    COUNT(DISTINCT f.order_id) as orders,
    COUNT(DISTINCT f.customer_key) as unique_customers,
    SUM(f.net_revenue) as revenue,
    SUM(f.net_revenue) / COUNT(DISTINCT f.order_id) as avg_order_value,  -- grain is line item
    SUM(f.discount_amount) / SUM(f.gross_revenue) * 100 as discount_rate
FROM fact_sales f
JOIN dim_store s ON f.store_key = s.store_key
JOIN dim_date d ON f.date_key = d.date_key
WHERE d.year = 2024
GROUP BY s.store_type
ORDER BY revenue DESC;
"""

# modeling/scd_type2.py

"""
Slowly Changing Dimension Type 2 implementation.

Maintains full history of dimension changes.
"""

from dataclasses import dataclass
from datetime import date, datetime
from typing import Optional, Dict, Any, List
import hashlib


@dataclass
class DimensionRecord:
    """A record in an SCD Type 2 dimension."""
    surrogate_key: int
    business_key: str
    attributes: Dict[str, Any]
    valid_from: date
    valid_to: Optional[date]
    is_current: bool
    hash_value: str  # For change detection


class SCDType2Manager:
    """
    Manages SCD Type 2 dimension updates.
    
    Handles:
    - New records (INSERT)
    - Changed records (close old, open new)
    - Unchanged records (no action)
    - Deleted records (close, don't delete)
    """
    
    def __init__(self, tracked_columns: List[str]):
        """
        Args:
            tracked_columns: Columns that trigger new version on change
        """
        self.tracked_columns = tracked_columns
    
    def compute_hash(self, attributes: Dict[str, Any]) -> str:
        """
        Compute hash of tracked columns for change detection.
        """
        values = [str(attributes.get(col, "")) for col in self.tracked_columns]
        combined = "|".join(values)
        return hashlib.md5(combined.encode()).hexdigest()
    
    def process_updates(
        self,
        current_records: List[DimensionRecord],
        incoming_records: List[Dict[str, Any]],
        effective_date: date
    ) -> Dict[str, List]:
        """
        Process dimension updates using SCD Type 2 logic.
        
        Returns dict with:
        - 'inserts': New records to insert
        - 'updates': Existing records to update (close)
        - 'unchanged': Records that don't need changes
        """
        
        # Build lookup of current records by business key
        current_by_key = {
            r.business_key: r for r in current_records if r.is_current
        }
        
        inserts = []
        updates = []
        unchanged = []
        
        for incoming in incoming_records:
            business_key = incoming["business_key"]
            new_hash = self.compute_hash(incoming)
            
            if business_key not in current_by_key:
                # New record - INSERT
                inserts.append({
                    **incoming,
                    "valid_from": effective_date,
                    "valid_to": None,
                    "is_current": True,
                    "hash_value": new_hash
                })
            
            else:
                current = current_by_key[business_key]
                
                if current.hash_value != new_hash:
                    # Changed - close old, insert new
                    updates.append({
                        "surrogate_key": current.surrogate_key,
                        "valid_to": effective_date,
                        "is_current": False
                    })
                    
                    inserts.append({
                        **incoming,
                        "valid_from": effective_date,
                        "valid_to": None,
                        "is_current": True,
                        "hash_value": new_hash
                    })
                
                else:
                    # Unchanged
                    unchanged.append(current.surrogate_key)
        
        return {
            "inserts": inserts,
            "updates": updates,
            "unchanged": unchanged
        }


# =============================================================================
# SQL for SCD Type 2 Merge
# =============================================================================

SCD_TYPE2_MERGE_SQL = """
-- SCD Type 2 with plain UPDATE + INSERT steps (works in most modern databases)

-- Step 1: Close changed records
UPDATE dim_customer
SET valid_to = :effective_date,
    is_current = FALSE
WHERE is_current = TRUE
  AND customer_id IN (
    SELECT c.customer_id
    FROM dim_customer c
    JOIN staging_customer s ON c.customer_id = s.customer_id
    WHERE c.is_current = TRUE
      AND c.hash_value != s.hash_value
);

-- Step 2: Insert new versions of changed records
INSERT INTO dim_customer (
    customer_id, name, segment, tier, country,
    valid_from, valid_to, is_current, hash_value
)
SELECT 
    s.customer_id, s.name, s.segment, s.tier, s.country,
    :effective_date, NULL, TRUE, s.hash_value
FROM staging_customer s
JOIN dim_customer c ON s.customer_id = c.customer_id
WHERE c.valid_to = :effective_date;  -- Just closed

-- Step 3: Insert completely new records
INSERT INTO dim_customer (
    customer_id, name, segment, tier, country,
    valid_from, valid_to, is_current, hash_value
)
SELECT 
    s.customer_id, s.name, s.segment, s.tier, s.country,
    :effective_date, NULL, TRUE, s.hash_value
FROM staging_customer s
WHERE NOT EXISTS (
    SELECT 1 FROM dim_customer c 
    WHERE c.customer_id = s.customer_id
);
"""


# =============================================================================
# Querying SCD Type 2 Dimensions
# =============================================================================

SCD_TYPE2_QUERIES = """
-- Query 1: Get current customer data
SELECT * 
FROM dim_customer 
WHERE is_current = TRUE;

-- Query 2: Get customer data as of a specific date
SELECT * 
FROM dim_customer 
WHERE valid_from <= '2024-06-15' 
  AND (valid_to IS NULL OR valid_to > '2024-06-15');

-- Query 3: Revenue by segment with point-in-time accuracy
-- Each sale joins to the customer record that was valid at sale time
SELECT 
    c.segment,
    SUM(f.net_revenue) as revenue
FROM fact_sales f
JOIN dim_customer c ON f.customer_key = c.customer_key
    -- The fact table stores the surrogate key that was valid at transaction time
GROUP BY c.segment;

-- Query 4: Customer history
SELECT 
    customer_id,
    segment,
    tier,
    valid_from,
    valid_to
FROM dim_customer
WHERE customer_id = 'cust_123'
ORDER BY valid_from;
"""

# modeling/parquet_optimization.py

"""
Parquet file optimization for analytics.

Key factors:
1. Row group size
2. Compression codec
3. Column ordering
4. Statistics collection
"""

from typing import List, Dict, Any
import pyarrow as pa
import pyarrow.parquet as pq


class ParquetWriter:
    """
    Optimized Parquet writer for analytics workloads.
    """
    
    def __init__(
        self,
        row_group_size: int = 1_000_000,  # ROWS: pyarrow counts rows, not bytes
        compression: str = "snappy",
        enable_statistics: bool = True
    ):
        self.row_group_size = row_group_size
        self.compression = compression
        self.enable_statistics = enable_statistics
    
    def write_table(
        self,
        table: pa.Table,
        output_path: str,
        partition_cols: List[str] = None
    ):
        """
        Write table to Parquet with optimizations.
        """
        
        # Sort by partition columns for better compression
        if partition_cols:
            sort_keys = [(col, "ascending") for col in partition_cols]
            table = table.sort_by(sort_keys)
        
        pq.write_table(
            table,
            output_path,
            row_group_size=self.row_group_size,
            compression=self.compression,
            write_statistics=self.enable_statistics,
            # Use dictionary encoding for low-cardinality columns
            use_dictionary=True,
            # Write column indexes for predicate pushdown
            write_page_index=True
        )
    
    def write_partitioned(
        self,
        table: pa.Table,
        output_path: str,
        partition_cols: List[str]
    ):
        """
        Write partitioned Parquet dataset.
        
        Partitioning enables partition pruning:
        WHERE date = '2024-01-15' reads only that partition.
        """
        
        pq.write_to_dataset(
            table,
            output_path,
            partition_cols=partition_cols,
            compression=self.compression,
            row_group_size=self.row_group_size
        )

# modeling/pre_aggregation.py

"""
Pre-aggregation (Rollups) for fast queries.

Instead of aggregating at query time:
  SELECT date, SUM(revenue) FROM fact_sales GROUP BY date
  → Scans 1B rows

Query from pre-aggregated table:
  SELECT date, revenue FROM daily_revenue
  → Scans 365 rows
"""

from dataclasses import dataclass
from typing import List, Dict, Any
from datetime import date


@dataclass
class RollupDefinition:
    """Defines a pre-aggregation rollup."""
    name: str
    source_table: str
    target_table: str
    dimensions: List[str]  # GROUP BY columns
    measures: Dict[str, str]  # column: aggregation
    refresh_schedule: str  # Cron expression


# Define common rollups
ROLLUPS = [
    RollupDefinition(
        name="daily_revenue_by_category",
        source_table="fact_sales",
        target_table="agg_daily_category",
        dimensions=["date_key", "category"],
        measures={
            "net_revenue": "SUM",
            "quantity": "SUM",
            "order_count": "COUNT DISTINCT order_id",
            "customer_count": "COUNT DISTINCT customer_key"
        },
        refresh_schedule="0 1 * * *"  # Daily at 1 AM
    ),
    
    RollupDefinition(
        name="hourly_revenue_by_store",
        source_table="fact_sales",
        target_table="agg_hourly_store",
        dimensions=["date_key", "hour", "store_key"],
        measures={
            "net_revenue": "SUM",
            "quantity": "SUM",
            "order_count": "COUNT DISTINCT order_id"
        },
        refresh_schedule="0 * * * *"  # Hourly
    ),
    
    RollupDefinition(
        name="monthly_customer_metrics",
        source_table="fact_sales",
        target_table="agg_monthly_customer",
        dimensions=["year", "month", "customer_key", "segment"],
        measures={
            "net_revenue": "SUM",
            "order_count": "COUNT DISTINCT order_id",
            "avg_order_value": "SUM(net_revenue) / COUNT(DISTINCT order_id)"
        },
        refresh_schedule="0 2 1 * *"  # Monthly on 1st at 2 AM
    )
]


ROLLUP_DDL = """
-- Pre-aggregated table for daily revenue by category
CREATE TABLE agg_daily_category (
    date_key        INTEGER NOT NULL,
    category        VARCHAR(100) NOT NULL,
    net_revenue     DECIMAL(15,2) NOT NULL,
    quantity        BIGINT NOT NULL,
    order_count     BIGINT NOT NULL,
    customer_count  BIGINT NOT NULL,
    last_updated    TIMESTAMP NOT NULL DEFAULT NOW(),
    PRIMARY KEY (date_key, category)
);

-- Refresh query
INSERT INTO agg_daily_category
SELECT 
    f.date_key,
    p.category,
    SUM(f.net_revenue) as net_revenue,
    SUM(f.quantity) as quantity,
    COUNT(DISTINCT f.order_id) as order_count,
    COUNT(DISTINCT f.customer_key) as customer_count,
    NOW() as last_updated
FROM fact_sales f
JOIN dim_product p ON f.product_key = p.product_key
WHERE f.date_key = :target_date_key
GROUP BY f.date_key, p.category
ON CONFLICT (date_key, category) 
DO UPDATE SET
    net_revenue = EXCLUDED.net_revenue,
    quantity = EXCLUDED.quantity,
    order_count = EXCLUDED.order_count,
    customer_count = EXCLUDED.customer_count,
    last_updated = NOW();


-- Query from rollup (fast!)
SELECT category, SUM(net_revenue)
FROM agg_daily_category
WHERE date_key BETWEEN 20240101 AND 20241231
GROUP BY category;
-- Scans 365 * ~100 categories = 36,500 rows
-- vs 1B rows in fact_sales
"""

# modeling/materialized_views.py

"""
Materialized Views for automatic pre-aggregation.

Better than manual rollups because:
1. Automatic refresh (BigQuery, Snowflake; PostgreSQL needs REFRESH)
2. Query optimizer can use them transparently (BigQuery, Snowflake;
   not PostgreSQL)
3. Built-in freshness tracking
"""

from datetime import datetime, timezone
from typing import Dict, List

MATERIALIZED_VIEW_PATTERNS = """
-- =============================================================================
-- Pattern 1: Simple aggregation view
-- =============================================================================

CREATE MATERIALIZED VIEW mv_daily_revenue AS
SELECT 
    date_key,
    SUM(net_revenue) as revenue,
    COUNT(*) as transaction_count
FROM fact_sales
GROUP BY date_key
WITH DATA;

-- Refresh options
REFRESH MATERIALIZED VIEW mv_daily_revenue;  -- Full refresh, blocks reads
-- CONCURRENTLY doesn't block readers but needs a UNIQUE index
CREATE UNIQUE INDEX ON mv_daily_revenue (date_key);
REFRESH MATERIALIZED VIEW CONCURRENTLY mv_daily_revenue;


-- =============================================================================
-- Pattern 2: Join materialized into view
-- =============================================================================

CREATE MATERIALIZED VIEW mv_product_performance AS
SELECT 
    p.category,
    p.subcategory,
    p.brand,
    d.year,
    d.month,
    SUM(f.net_revenue) as revenue,
    SUM(f.quantity) as units,
    SUM(f.profit) as profit
FROM fact_sales f
JOIN dim_product p ON f.product_key = p.product_key
JOIN dim_date d ON f.date_key = d.date_key
GROUP BY p.category, p.subcategory, p.brand, d.year, d.month
WITH DATA;

-- Index for fast queries
CREATE INDEX idx_mv_product_year 
ON mv_product_performance(year, month);


-- =============================================================================
-- Pattern 3: Incremental materialized view (database-specific)
-- =============================================================================

-- In BigQuery: 
CREATE MATERIALIZED VIEW mv_hourly_revenue
OPTIONS (
    enable_refresh = true,
    refresh_interval_minutes = 60
)
AS
SELECT 
    TIMESTAMP_TRUNC(order_timestamp, HOUR) as hour,
    SUM(net_revenue) as revenue
FROM fact_sales
GROUP BY 1;

-- BigQuery automatically updates incrementally

-- In Snowflake:
CREATE MATERIALIZED VIEW mv_hourly_revenue
AS
SELECT 
    DATE_TRUNC('HOUR', order_timestamp) as hour,
    SUM(net_revenue) as revenue
FROM fact_sales
GROUP BY 1;

-- Snowflake maintains incrementally if possible
"""


class MaterializedViewManager:
    """
    Manages materialized view lifecycle.
    """
    
    def __init__(self, db_connection):
        self.db = db_connection
        self.views: Dict[str, Dict] = {}
    
    def create_view(
        self,
        name: str,
        query: str,
        refresh_schedule: str = None,
        indexes: List[str] = None
    ):
        """Create a materialized view."""
        
        sql = f"CREATE MATERIALIZED VIEW {name} AS {query} WITH DATA"
        self.db.execute(sql)
        
        if indexes:
            for idx_cols in indexes:
                idx_name = f"idx_{name}_{idx_cols.replace(', ', '_')}"
                self.db.execute(
                    f"CREATE INDEX {idx_name} ON {name}({idx_cols})"
                )
        
        self.views[name] = {
            "query": query,
            "refresh_schedule": refresh_schedule,
            "last_refresh": None
        }
    
    def refresh_view(self, name: str, concurrent: bool = True):
        """Refresh a materialized view."""
        
        if concurrent:
            sql = f"REFRESH MATERIALIZED VIEW CONCURRENTLY {name}"
        else:
            sql = f"REFRESH MATERIALIZED VIEW {name}"
        
        self.db.execute(sql)
        self.views[name]["last_refresh"] = datetime.now(timezone.utc)
    
    def get_freshness(self, name: str) -> Dict:
        """Get view freshness information."""
        
        view = self.views.get(name)
        if not view:
            return {"error": "View not found"}
        
        last_refresh = view.get("last_refresh")
        if not last_refresh:
            return {"status": "never_refreshed"}
        
        age = datetime.now(timezone.utc) - last_refresh
        
        return {
            "last_refresh": last_refresh.isoformat(),
            "age_seconds": age.total_seconds(),
            "status": "fresh" if age.total_seconds() < 3600 else "stale"
        }

# modeling/etl_pipeline.py

"""
ETL pipeline to build star schema from source data.

Flow:
1. Extract from source systems
2. Transform into dimension and fact tables
3. Load into data warehouse
"""

from dataclasses import dataclass
from datetime import date, datetime, timedelta
from typing import Dict, Any, List, Optional
import logging

logger = logging.getLogger(__name__)


@dataclass
class ETLConfig:
    """ETL job configuration."""
    source_db_url: str
    target_db_url: str
    batch_date: date
    full_refresh: bool = False


class DimensionLoader:
    """
    Loads dimension tables with SCD Type 2 handling.
    """
    
    def __init__(self, target_db, tracked_columns: Dict[str, List[str]]):
        self.db = target_db
        self.tracked_columns = tracked_columns
    
    def load_product_dimension(self, products_df, batch_date: date):
        """
        Load product dimension with Type 2 SCD.
        """
        
        logger.info(f"Loading product dimension for {batch_date}")
        
        # Get current records
        current = self.db.read_table("dim_product", "is_current = TRUE")
        
        # Compute changes
        scd_manager = SCDType2Manager(
            self.tracked_columns.get("product", ["name", "category", "price"])
        )
        
        changes = scd_manager.process_updates(
            current_records=current,
            incoming_records=products_df.to_records(),
            effective_date=batch_date
        )
        
        # Apply changes
        if changes["updates"]:
            self.db.update_batch("dim_product", changes["updates"])
            logger.info(f"Closed {len(changes['updates'])} changed products")
        
        if changes["inserts"]:
            self.db.insert_batch("dim_product", changes["inserts"])
            logger.info(f"Inserted {len(changes['inserts'])} product versions")
        
        return {
            "inserted": len(changes["inserts"]),
            "updated": len(changes["updates"]),
            "unchanged": len(changes["unchanged"])
        }
    
    def load_customer_dimension(self, customers_df, batch_date: date):
        """Load customer dimension."""
        # Similar pattern to product
        pass
    
    def load_date_dimension(self, start_date: date, end_date: date):
        """
        Generate and load date dimension.
        
        Usually done once, extended as needed.
        """
        
        dates = []
        current = start_date
        
        while current <= end_date:
            dates.append({
                "date_key": int(current.strftime("%Y%m%d")),
                "date": current,
                "day": current.day,
                "month": current.month,
                "month_name": current.strftime("%B"),
                "quarter": (current.month - 1) // 3 + 1,
                "year": current.year,
                "day_of_week": current.weekday(),
                "day_name": current.strftime("%A"),
                "week_of_year": current.isocalendar()[1],
                "is_weekend": current.weekday() >= 5,
                "is_holiday": False,  # Add holiday logic
                "fiscal_year": current.year,  # Adjust for fiscal calendar
                "fiscal_quarter": (current.month - 1) // 3 + 1
            })
            current = current + timedelta(days=1)
        
        self.db.upsert_batch("dim_date", dates, key_columns=["date_key"])
        
        return {"dates_loaded": len(dates)}


class FactLoader:
    """
    Loads fact tables from source transactions.
    """
    
    def __init__(self, target_db):
        self.db = target_db
    
    def load_sales_fact(
        self,
        orders_df,
        batch_date: date,
        dimension_lookups: Dict
    ):
        """
        Load sales fact table.
        
        Steps:
        1. Join with dimensions to get surrogate keys
        2. Calculate derived measures
        3. Insert into fact table
        """
        
        logger.info(f"Loading sales fact for {batch_date}")
        
        facts = []
        
        for order in orders_df.to_records():
            # Look up dimension keys
            date_key = int(order["order_date"].strftime("%Y%m%d"))
            
            product_key = dimension_lookups["product"].get(
                order["product_id"], self._get_unknown_key("product")
            )
            
            customer_key = dimension_lookups["customer"].get(
                order["customer_id"], self._get_unknown_key("customer")
            )
            
            store_key = dimension_lookups["store"].get(
                order["store_id"], self._get_unknown_key("store")
            )
            
            # Calculate measures
            gross_revenue = order["quantity"] * order["unit_price"]
            net_revenue = gross_revenue - order.get("discount", 0)
            cost = order["quantity"] * order.get("unit_cost", 0)
            profit = net_revenue - cost
            
            facts.append({
                "date_key": date_key,
                "product_key": product_key,
                "customer_key": customer_key,
                "store_key": store_key,
                "order_id": order["order_id"],
                "line_item_number": order.get("line_number", 1),
                "quantity": order["quantity"],
                "unit_price": order["unit_price"],
                "discount_amount": order.get("discount", 0),
                "tax_amount": order.get("tax", 0),
                "gross_revenue": gross_revenue,
                "net_revenue": net_revenue,
                "profit": profit,
                "order_timestamp": order["order_timestamp"]
            })
        
        # Batch insert
        self.db.insert_batch("fact_sales", facts)
        
        logger.info(f"Loaded {len(facts)} sales facts")
        
        return {"facts_loaded": len(facts)}
    
    def _get_unknown_key(self, dimension: str) -> int:
        """Get the 'unknown' dimension key for missing lookups."""
        # Convention: -1 or a specific unknown record
        return -1
    
    def build_dimension_lookups(self, batch_date: date) -> Dict:
        """
        Build lookup dictionaries for dimension foreign keys.
        
        Returns mapping from business key → surrogate key
        for each dimension, using the record valid on batch_date.
        """
        
        lookups = {}
        
        # Version valid on batch_date (not is_current: a backfill of an
        # old date must pick the version that was valid back then)
        valid_on = """
            valid_from <= :batch_date
            AND (valid_to IS NULL OR valid_to > :batch_date)
        """
        
        # Product lookup
        products = self.db.query(f"""
            SELECT product_id, product_key 
            FROM dim_product 
            WHERE {valid_on}
        """, {"batch_date": batch_date})
        lookups["product"] = {
            r["product_id"]: r["product_key"] for r in products
        }
        
        # Customer lookup
        customers = self.db.query(f"""
            SELECT customer_id, customer_key 
            FROM dim_customer 
            WHERE {valid_on}
        """, {"batch_date": batch_date})
        lookups["customer"] = {
            r["customer_id"]: r["customer_key"] for r in customers
        }
        
        # Store lookup
        stores = self.db.query("SELECT store_id, store_key FROM dim_store")
        lookups["store"] = {
            r["store_id"]: r["store_key"] for r in stores
        }
        
        return lookups


class ETLOrchestrator:
    """
    Orchestrates the full ETL pipeline.
    """
    
    def __init__(
        self,
        source_db,
        target_db,
        dim_loader: DimensionLoader,
        fact_loader: FactLoader
    ):
        self.source = source_db
        self.target = target_db
        self.dimensions = dim_loader
        self.facts = fact_loader
    
    def run_daily_etl(self, batch_date: date) -> Dict:
        """
        Run complete daily ETL pipeline.
        """
        
        results = {"batch_date": batch_date.isoformat()}
        
        try:
            # Step 1: Load dimensions (order matters for lookups)
            logger.info("Loading dimensions...")
            
            products = self.source.extract_products(batch_date)
            results["products"] = self.dimensions.load_product_dimension(
                products, batch_date
            )
            
            customers = self.source.extract_customers(batch_date)
            results["customers"] = self.dimensions.load_customer_dimension(
                customers, batch_date
            )
            
            # Step 2: Build lookups
            logger.info("Building dimension lookups...")
            lookups = self.facts.build_dimension_lookups(batch_date)
            
            # Step 3: Load facts
            logger.info("Loading facts...")
            orders = self.source.extract_orders(batch_date)
            results["facts"] = self.facts.load_sales_fact(
                orders, batch_date, lookups
            )
            
            # Step 4: Update aggregations
            logger.info("Updating rollups...")
            results["rollups"] = self._refresh_rollups(batch_date)
            
            results["status"] = "success"
            
        except Exception as e:
            logger.error(f"ETL failed: {e}", exc_info=True)
            results["status"] = "failed"
            results["error"] = str(e)
        
        return results
    
    def _refresh_rollups(self, batch_date: date) -> Dict:
        """Refresh materialized views / rollup tables."""
        
        rollups_refreshed = []
        
        for rollup in ROLLUPS:
            self.target.execute(
                rollup.refresh_query,
                {"date_key": int(batch_date.strftime("%Y%m%d"))}
            )
            rollups_refreshed.append(rollup.name)
        
        return {"refreshed": rollups_refreshed}