Retail Etl Pipeline Medallion

-- Example: Bronze Products TableCREATE TABLE bronze.products (    product_id INT,    product_name NVARCHAR(255),    category NVARCHAR(100),    unit_price DECIMAL(10,2),    supplier_id INT,    _loaded_at DATETIME2 DEFAULT GETDATE(),    _source_file NVARCHAR(500));
-- Bulk insert from CSVBULK INSERT bronze.productsFROM '/data_source/000.Hypermarket Products.csv'WITH (    FIELDTERMINATOR = ',',    ROWTERMINATOR = '\n',    FIRSTROW = 2,    TABLOCK);

-- Example: Silver Products with Data Quality RulesCREATE PROCEDURE silver.usp_transform_productsASBEGIN    TRUNCATE TABLE silver.products;        INSERT INTO silver.products (        product_id,        product_name,        category,        unit_price,        supplier_id,        is_active,        processed_at    )    SELECT DISTINCT        product_id,        UPPER(TRIM(product_name)) AS product_name,        COALESCE(category, 'UNCATEGORIZED') AS category,        CASE             WHEN unit_price < 0 THEN 0             ELSE unit_price         END AS unit_price,        supplier_id,        1 AS is_active,        GETDATE() AS processed_at    FROM bronze.products    WHERE product_id IS NOT NULL      AND product_name IS NOT NULL;END;

-- Example: Sales Summary by Branch and ProductCREATE PROCEDURE gold.usp_sales_summaryASBEGIN    TRUNCATE TABLE gold.sales_summary;        INSERT INTO gold.sales_summary (        branch_name,        product_category,        total_quantity_sold,        total_revenue,        avg_unit_price,        transaction_count,        report_date    )    SELECT         s.branch_name,        p.category AS product_category,        SUM(s.quantity) AS total_quantity_sold,        SUM(s.quantity * s.unit_price) AS total_revenue,        AVG(s.unit_price) AS avg_unit_price,        COUNT(DISTINCT s.transaction_id) AS transaction_count,        CAST(GETDATE() AS DATE) AS report_date    FROM silver.sales s    INNER JOIN silver.products p ON s.product_id = p.product_id    GROUP BY s.branch_name, p.category;END;

-- Complex business logic: Detect inventory discrepanciesCREATE PROCEDURE gold.usp_inventory_shrinkageASBEGIN    WITH StockLevels AS (        SELECT             product_id,            branch_name,            SUM(quantity_on_hand) AS current_stock        FROM silver.stock        GROUP BY product_id, branch_name    ),    ExpectedStock AS (        SELECT             s.product_id,            s.branch_name,            sl.current_stock - COALESCE(SUM(s.quantity), 0) AS expected_stock        FROM StockLevels sl        LEFT JOIN silver.sales s             ON sl.product_id = s.product_id             AND sl.branch_name = s.branch_name        GROUP BY s.product_id, s.branch_name, sl.current_stock    )    INSERT INTO gold.inventory_shrinkage (        product_id,        branch_name,        current_stock,        expected_stock,        shrinkage_qty,        shrinkage_percent,        calculated_at    )    SELECT         sl.product_id,        sl.branch_name,        sl.current_stock,        es.expected_stock,        sl.current_stock - es.expected_stock AS shrinkage_qty,        CASE             WHEN es.expected_stock > 0             THEN ((sl.current_stock - es.expected_stock) * 100.0 / es.expected_stock)            ELSE 0         END AS shrinkage_percent,        GETDATE() AS calculated_at    FROM StockLevels sl    INNER JOIN ExpectedStock es         ON sl.product_id = es.product_id         AND sl.branch_name = es.branch_name;END;

-- Combine sales from Alex, Cairo, and Giza branchesCREATE VIEW gold.vw_consolidated_sales ASSELECT     'Alexandria' AS branch_name,    * FROM bronze.alex_salesUNION ALLSELECT     'Cairo' AS branch_name,    * FROM bronze.cairo_salesUNION ALLSELECT     'Giza' AS branch_name,    * FROM bronze.giza_sales;

-- Track conversion rates for processed meat productsCREATE TABLE silver.recipe_conversions (    recipe_id INT PRIMARY KEY,    raw_product_id INT,    finished_product_id INT,    conversion_ratio DECIMAL(5,2), -- e.g., 0.85 (15% waste)    effective_date DATE);
-- Calculate actual yield vs. expectedSELECT     rc.recipe_id,    SUM(s.quantity * rc.conversion_ratio) AS expected_yield,    SUM(stock.quantity_on_hand) AS actual_yield,    (SUM(stock.quantity_on_hand) - SUM(s.quantity * rc.conversion_ratio)) AS yield_varianceFROM silver.sales sINNER JOIN silver.recipe_conversions rc ON s.product_id = rc.raw_product_idINNER JOIN silver.stock stock ON rc.finished_product_id = stock.product_idGROUP BY rc.recipe_id;

-- Track purchase volume for supplier rebate calculationsCREATE TABLE gold.supplier_rebate_tiers (    supplier_id INT,    total_purchases DECIMAL(15,2),    rebate_tier VARCHAR(20),    rebate_percentage DECIMAL(5,2),    calculated_at DATETIME2);
INSERT INTO gold.supplier_rebate_tiersSELECT     p.supplier_id,    SUM(s.quantity * s.unit_price) AS total_purchases,    CASE         WHEN SUM(s.quantity * s.unit_price) > 100000 THEN 'Platinum'        WHEN SUM(s.quantity * s.unit_price) > 50000 THEN 'Gold'        WHEN SUM(s.quantity * s.unit_price) > 25000 THEN 'Silver'        ELSE 'Bronze'    END AS rebate_tier,    CASE         WHEN SUM(s.quantity * s.unit_price) > 100000 THEN 5.0        WHEN SUM(s.quantity * s.unit_price) > 50000 THEN 3.5        WHEN SUM(s.quantity * s.unit_price) > 25000 THEN 2.0        ELSE 1.0    END AS rebate_percentage,    GETDATE() AS calculated_atFROM silver.sales sINNER JOIN silver.products p ON s.product_id = p.product_idGROUP BY p.supplier_id;

-- Check file path permissionsEXEC xp_cmdshell 'dir C:\data_source\*.csv';
-- Use ERRORFILE to capture bad rowsBULK INSERT bronze.productsFROM '/data_source/000.Hypermarket Products.csv'WITH (    FIELDTERMINATOR = ',',    ROWTERMINATOR = '\n',    FIRSTROW = 2,    ERRORFILE = '/data_source/errors/products_errors.csv',    MAXERRORS = 10);

-- Add deduplication logic with ROW_NUMBERWITH DeduplicatedSales AS (    SELECT *,        ROW_NUMBER() OVER (            PARTITION BY transaction_id             ORDER BY _loaded_at DESC        ) AS rn    FROM bronze.sales)INSERT INTO silver.salesSELECT * FROM DeduplicatedSalesWHERE rn = 1;

-- Create indexes on frequently joined columnsCREATE NONCLUSTERED INDEX idx_sales_product_id     ON silver.sales(product_id) INCLUDE (quantity, unit_price);
CREATE NONCLUSTERED INDEX idx_stock_product_branch     ON silver.stock(product_id, branch_name) INCLUDE (quantity_on_hand);
-- Partition large tables by dateCREATE PARTITION FUNCTION pf_sales_date (DATE)AS RANGE RIGHT FOR VALUES (    '2025-01-01', '2025-02-01', '2025-03-01', '2025-04-01');

-- Data quality checks for Silver layerSELECT 'Products' AS layer,    COUNT(*) AS total_records,    COUNT(DISTINCT product_id) AS unique_products,    SUM(CASE WHEN unit_price < 0 THEN 1 ELSE 0 END) AS negative_prices,    SUM(CASE WHEN product_name IS NULL THEN 1 ELSE 0 END) AS null_namesFROM silver.products
UNION ALL
SELECT 'Sales' AS layer,    COUNT(*) AS total_records,    COUNT(DISTINCT transaction_id) AS unique_transactions,    SUM(CASE WHEN quantity <= 0 THEN 1 ELSE 0 END) AS invalid_quantity,    SUM(CASE WHEN product_id NOT IN (SELECT product_id FROM silver.products) THEN 1 ELSE 0 END) AS orphaned_productsFROM silver.sales;