-- Sales Data Mart: Star Schema
CREATE TABLE fact_sales (
    sale_id         BIGINT PRIMARY KEY,
    date_key        INT REFERENCES dim_date(date_key),
    product_key     INT REFERENCES dim_product(product_key),
    customer_key    INT REFERENCES dim_customer(customer_key),
    store_key       INT REFERENCES dim_store(store_key),
    quantity        INT,
    unit_price      DECIMAL(10,2),
    discount        DECIMAL(10,2),
    total_amount    DECIMAL(12,2),
    tax_amount      DECIMAL(10,2),
    sale_timestamp  TIMESTAMP
);

CREATE TABLE dim_date (
    date_key        INT PRIMARY KEY,
    date            DATE,
    year            INT,
    quarter         INT,
    month           INT,
    day_of_week     INT,
    is_weekend      BOOLEAN,
    fiscal_year     INT
);

CREATE TABLE dim_product (
    product_key     INT PRIMARY KEY,
    product_id      VARCHAR(50),
    product_name    VARCHAR(200),
    category        VARCHAR(100),
    subcategory     VARCHAR(100),
    brand           VARCHAR(100),
    unit_cost       DECIMAL(10,2)
);

CREATE TABLE dim_customer (
    customer_key    INT PRIMARY KEY,
    customer_id     VARCHAR(50),
    customer_name   VARCHAR(200),
    email           VARCHAR(200),
    segment         VARCHAR(50),
    city            VARCHAR(100),
    state           VARCHAR(50),
    country         VARCHAR(50)
);

-- Pre-aggregated monthly sales by product category
CREATE TABLE agg_monthly_sales_by_category (
    year            INT,
    month           INT,
    category        VARCHAR(100),
    total_revenue   DECIMAL(12,2),
    total_units     INT,
    avg_order_value DECIMAL(10,2),
    unique_customers INT,
    PRIMARY KEY (year, month, category)
);

-- Refresh strategy: Daily incremental
INSERT INTO agg_monthly_sales_by_category
SELECT 
    EXTRACT(YEAR FROM sale_timestamp) as year,
    EXTRACT(MONTH FROM sale_timestamp) as month,
    p.category,
    SUM(s.total_amount) as total_revenue,
    SUM(s.quantity) as total_units,
    AVG(s.total_amount) as avg_order_value,
    COUNT(DISTINCT s.customer_key) as unique_customers
FROM fact_sales s
JOIN dim_product p ON s.product_key = p.product_key
WHERE sale_timestamp >= CURRENT_DATE - INTERVAL '1 day'
GROUP BY 1, 2, 3;

-- Incremental refresh: only process new/changed records
MERGE INTO fact_sales target
USING (
    SELECT *
    FROM stg_sales
    WHERE updated_at > (SELECT MAX(last_refresh) FROM refresh_log WHERE table_name = 'fact_sales')
) source
ON target.sale_id = source.sale_id
WHEN MATCHED THEN UPDATE SET
    quantity = source.quantity,
    unit_price = source.unit_price,
    total_amount = source.total_amount,
    updated_at = source.updated_at
WHEN NOT MATCHED THEN INSERT VALUES (
    source.sale_id, source.date_key, source.product_key,
    source.customer_key, source.quantity, source.unit_price,
    source.total_amount, source.updated_at
);

-- Log refresh
INSERT INTO refresh_log (table_name, last_refresh, records_processed)
VALUES ('fact_sales', CURRENT_TIMESTAMP, (SELECT COUNT(*) FROM stg_sales WHERE updated_at > (SELECT MAX(last_refresh) FROM refresh_log WHERE table_name = 'fact_sales')));

# AWS Glue job to create a Sales Data Mart from warehouse data
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)

# Read from data warehouse (Redshift)
fact_sales = spark.read.format("jdbc") \
    .option("url", "jdbc:redshift://warehouse-cluster:5439/dev") \
    .option("dbtable", "public.fact_sales") \
    .option("user", "admin") \
    .option("password", secrets_manager.get_secret("redshift-password")) \
    .load()

dim_product = spark.read.format("jdbc") \
    .option("url", "jdbc:redshift://warehouse-cluster:5439/dev") \
    .option("dbtable", "public.dim_product") \
    .option("user", "admin") \
    .option("password", secrets_manager.get_secret("redshift-password")) \
    .load()

dim_customer = spark.read.format("jdbc") \
    .option("url", "jdbc:redshift://warehouse-cluster:5439/dev") \
    .option("dbtable", "public.dim_customer") \
    .option("user", "admin") \
    .option("password", secrets_manager.get_secret("redshift-password")) \
    .load()

# Create Sales Data Mart: join fact with dimensions
sales_mart = fact_sales \
    .join(dim_product, "product_key") \
    .join(dim_customer, "customer_key") \
    .select(
        "sale_id", "date_key", "product_name", "category",
        "customer_name", "segment", "quantity", "unit_price",
        "total_amount", "sale_timestamp"
    )

# Write to S3 as Parquet (data mart layer)
sales_mart.write \
    .mode("overwrite") \
    .partitionBy("category") \
    .parquet("s3://data-lake/sales-mart/")

# Load into Redshift Sales Mart table
sales_mart.write.format("jdbc") \
    .option("url", "jdbc:redshift://mart-cluster:5439/sales_mart") \
    .option("dbtable", "public.fact_sales") \
    .mode("overwrite") \
    .save()