Marcin

calculate_largest_expensors.sql

@@ -0,0 +1,15 @@
+USE memory.default;
+
+SELECT 
+    e.employee_id,
+    (e.first_name || ' ' || e.last_name) AS employee_name,
+    e.manager_id,
+    (m.first_name || ' ' || m.last_name) AS manager_name,
+    SUM(exp.unit_price * exp.quantity) AS total_expensed_amount
+FROM expense exp
+JOIN employee e ON exp.employee_id = e.employee_id
+LEFT JOIN employee m ON e.manager_id = m.employee_id
+GROUP BY 
+    e.employee_id, e.first_name, e.last_name, e.manager_id, m.first_name, m.last_name
+HAVING SUM(exp.unit_price * exp.quantity) > 1000
+ORDER BY total_expensed_amount DESC;

create_employees.sql

@@ -0,0 +1,19 @@
+USE memory.default;
+
+CREATE TABLE IF NOT EXISTS employee (
+    employee_id TINYINT,
+    first_name VARCHAR,
+    last_name VARCHAR,
+    job_title VARCHAR,
+    manager_id TINYINT
+);
+
+INSERT INTO employee VALUES (1, 'Ian', 'James', 'CEO', 4);
+INSERT INTO employee VALUES (2, 'Umberto', 'Torrielli', 'CSO', 1);
+INSERT INTO employee VALUES (3, 'Alex', 'Jacobson', 'MD EMEA', 2);
+INSERT INTO employee VALUES (4, 'Darren', 'Poynton', 'CFO', 2);
+INSERT INTO employee VALUES (5, 'Tim', 'Beard', 'MD APAC', 2);
+INSERT INTO employee VALUES (6, 'Gemma', 'Dodd', 'COS', 1);
+INSERT INTO employee VALUES (7, 'Lisa', 'Platten', 'CHR', 6);
+INSERT INTO employee VALUES (8, 'Stefano', 'Camisaca', 'GM Activation', 2);
+INSERT INTO employee VALUES (9, 'Andrea', 'Ghibaudi', 'MD NAM', 2);

create_expenses.sql

@@ -0,0 +1,13 @@
+USE memory.default;
+
+CREATE TABLE IF NOT EXISTS expense (
+    employee_id TINYINT,
+    unit_price DECIMAL(8, 2),
+    quantity TINYINT
+);
+
+INSERT INTO expense VALUES (1, CAST(15.50 AS DECIMAL(8,2)), 2);
+INSERT INTO expense VALUES (2, CAST(120.00 AS DECIMAL(8,2)), 1);
+INSERT INTO expense VALUES (3, CAST(450.00 AS DECIMAL(8,2)), 3);
+INSERT INTO expense VALUES (1, CAST(5.00 AS DECIMAL(8,2)), 10);
+INSERT INTO expense VALUES (5, CAST(800.00 AS DECIMAL(8,2)), 2);

create_invoices.sql

@@ -0,0 +1,27 @@
+USE memory.default;
+
+CREATE TABLE IF NOT EXISTS supplier (
+    supplier_id TINYINT,
+    name VARCHAR
+);
+
+CREATE TABLE IF NOT EXISTS invoice (
+    supplier_id TINYINT,
+    invoice_amount DECIMAL(8, 2),
+    due_date DATE
+);
+
+-- Alphabetical list of unique suppliers
+INSERT INTO supplier VALUES (1, 'Catering Plus');
+INSERT INTO supplier VALUES (2, 'Dave''s Discos');
+INSERT INTO supplier VALUES (3, 'Entertainment tonight');
+INSERT INTO supplier VALUES (4, 'Ice Ice Baby');
+INSERT INTO supplier VALUES (5, 'Party Animals');
+
+-- Distinct individual invoices mapped by supplier_id and date offsets
+INSERT INTO invoice VALUES (5, CAST(6000.00 AS DECIMAL(8,2)), DATE '2026-09-30'); -- Party Animals (3m)
+INSERT INTO invoice VALUES (1, CAST(2000.00 AS DECIMAL(8,2)), DATE '2026-08-31'); -- Catering Plus / Bottles (2m)
+INSERT INTO invoice VALUES (1, CAST(1500.00 AS DECIMAL(8,2)), DATE '2026-09-30'); -- Catering Plus / Food (3m)
+INSERT INTO invoice VALUES (2, CAST(500.00 AS DECIMAL(8,2)), DATE '2026-07-31');  -- Dave's Discos (1m)
+INSERT INTO invoice VALUES (3, CAST(6000.00 AS DECIMAL(8,2)), DATE '2026-09-30'); -- Entertainment tonight (3m)
+INSERT INTO invoice VALUES (4, CAST(4000.00 AS DECIMAL(8,2)), DATE '2026-12-31'); -- Ice Ice Baby (6m)

data_quality_tests.sql

@@ -0,0 +1,29 @@
+-- =====================================================================
+-- DATA QUALITY & INTEGRITY ASSURANCE SUITE (CDP INGESTION GATE)
+-- Author: Marcin Siwy
+-- Purpose: Validate data constraints, detect anomalies, and prevent corrupt data ingestion.
+-- =====================================================================
+
+USE memory.default;
+
+-- TEST 1: Referential Integrity Check (Orphaned Expenses)
+-- Verifies if there are any expenses logged for employees who do not exist in the HR registry.
+SELECT COUNT(*) AS orphaned_expenses_count
+FROM expense e
+LEFT JOIN employee ev ON e.employee_id = ev.employee_id
+WHERE ev.employee_id IS NULL;
+
+-- TEST 2: Negative/Zero Value Financial Anomaly Detection
+-- Ensures no invoice or expense possesses zero or negative balances before pipeline processing.
+SELECT 'invoice_amount_anomaly' AS check_type, COUNT(*) AS anomaly_count 
+FROM invoice WHERE invoice_amount <= 0
+UNION ALL
+SELECT 'expense_unit_price_anomaly', COUNT(*) 
+FROM expense WHERE unit_price <= 0;
+
+-- TEST 3: Primary Key Uniqueness & Duplication Check
+-- Validates that core entity keys (supplier_id) remain completely unique after dynamic indexing.
+SELECT supplier_id, COUNT(*) as duplicate_records
+FROM supplier
+GROUP BY supplier_id
+HAVING COUNT(*) > 1;

find_manager_cycles.sql

@@ -0,0 +1,27 @@
+USE memory.default;
+
+WITH RECURSIVE manager_path (start_id, manager_id, path, is_cycle) AS (
+    SELECT 
+        employee_id AS start_id,
+        manager_id,
+        ARRAY[employee_id] AS path,
+        FALSE AS is_cycle
+    FROM employee
+    WHERE manager_id IS NOT NULL
+
+    UNION ALL
+
+    SELECT 
+        mp.start_id,
+        e.manager_id,
+        mp.path || e.employee_id,
+        CASE WHEN e.employee_id = mp.start_id THEN TRUE ELSE FALSE END
+    FROM manager_path mp
+    JOIN employee e ON mp.manager_id = e.employee_id
+    WHERE NOT contains(mp.path, e.employee_id) OR e.employee_id = mp.start_id
+)
+SELECT 
+    start_id AS employee_id, 
+    path AS cycle_path
+FROM manager_path
+WHERE is_cycle = TRUE;

generate_supplier_payment_plans.sql

@@ -0,0 +1,43 @@
+USE memory.default;
+
+WITH invoice_intervals AS (
+    SELECT 
+        i.supplier_id,
+        s.name AS supplier_name,
+        i.invoice_amount,
+        i.due_date,
+        (CAST(EXTRACT(YEAR FROM i.due_date) AS INT) - CAST(EXTRACT(YEAR FROM CURRENT_DATE) AS INT)) * 12 + 
+        (CAST(EXTRACT(MONTH FROM i.due_date) AS INT) - CAST(EXTRACT(MONTH FROM CURRENT_DATE) AS INT)) + 1 AS total_months
+    FROM invoice i
+    JOIN supplier s ON i.supplier_id = s.supplier_id
+),
+expanded_payments AS (
+    SELECT 
+        ii.supplier_id,
+        ii.supplier_name,
+        ii.due_date,
+        CAST(ii.invoice_amount / ii.total_months AS DECIMAL(8,2)) AS payment_amount,
+        m.month_index,
+        CAST(DATE_TRUNC('month', CURRENT_DATE + INTERVAL '1' MONTH * m.month_index) + INTERVAL '1' MONTH - INTERVAL '1' DAY AS DATE) AS payment_date
+    FROM invoice_intervals ii
+    CROSS JOIN UNNEST(sequence(0, 11)) AS m(month_index)
+    WHERE m.month_index < ii.total_months
+),
+monthly_aggregates AS (
+    SELECT 
+        supplier_id,
+        supplier_name,
+        payment_date,
+        SUM(payment_amount) AS monthly_payment_amount
+    FROM expanded_payments
+    GROUP BY supplier_id, supplier_name, payment_date
+)
+SELECT 
+    ma.supplier_id,
+    ma.supplier_name,
+    ma.monthly_payment_amount AS payment_amount,
+    CAST(SUM(ma.monthly_payment_amount) OVER(PARTITION BY ma.supplier_id) - 
+         SUM(ma.monthly_payment_amount) OVER(PARTITION BY ma.supplier_id ORDER BY ma.payment_date) AS DECIMAL(8,2)) AS balance_outstanding,
+    ma.payment_date
+FROM monthly_aggregates ma
+ORDER BY ma.supplier_id, ma.payment_date;