SELECT   category, COUNT(*) AS product_count      -- 5. choose & name output columns
FROM     stage.products                            -- 1. start from this table
WHERE    price > 50                                 -- 2. filter individual rows
GROUP BY category                                   -- 3. collapse into groups
HAVING   COUNT(*) > 2                                -- 4. filter the groups
ORDER BY product_count DESC                          -- 6. sort the final output
LIMIT    5;                                          -- 7. cap the row count

SELECT * FROM stage.products;
SELECT product_name, price FROM stage.products;
SELECT DISTINCT category FROM stage.products;
SELECT DISTINCT category, vendor FROM stage.products;

-- a computed column, aliased
SELECT product_name, price, price * 1.18 AS price_with_tax FROM stage.products;

SELECT * FROM stage.products WHERE price = 500.50;
SELECT * FROM stage.products WHERE price != 500.50;
SELECT * FROM stage.products WHERE price > 500.50;
SELECT * FROM stage.products WHERE price <= 500.50;

SELECT * FROM stage.products WHERE price <= 500.50 AND category = 'Electronics';
SELECT * FROM stage.products WHERE price <= 500.50 OR  category = 'Electronics';

SELECT * FROM stage.products WHERE category IN ('Electronics','Furniture');
SELECT * FROM stage.products WHERE category NOT IN ('Electronics','Furniture');

SELECT * FROM stage.products WHERE price BETWEEN 300 AND 600;   -- inclusive both ends

-- % = any number of characters, _ = exactly one character
SELECT * FROM stage.products WHERE LOWER(product_name) LIKE 'mouse%';
SELECT * FROM stage.products WHERE UPPER(product_name) LIKE '%CHAIN';
SELECT * FROM stage.products WHERE UPPER(product_name) LIKE '%A%';
SELECT * FROM stage.products WHERE LOWER(product_name) LIKE '_o%';

-- NULL checks — never use = NULL, it silently matches nothing
SELECT * FROM stage.products WHERE vendor IS NOT NULL;
SELECT * FROM stage.products WHERE vendor IS NULL;

-- arithmetic expressions work directly in WHERE
SELECT * FROM stage.products WHERE price * 87 > 20000;
SELECT *, price * 87 AS price_in_inr FROM stage.products WHERE price > 20000 / 87;

SELECT
  UPPER(customer_name)                          AS upper_name,
  LOWER(customer_name)                          AS lower_name,
  INITCAP(customer_name)                        AS proper_case,
  LENGTH(customer_name)                         AS name_length,
  SUBSTR(customer_name, 1, 5)                   AS first_5_chars,
  POSITION(' ' IN customer_name)                AS space_position,
  TRIM(customer_name)                           AS trimmed,
  CONCAT(customer_name, ' - ', country)         AS name_and_country,
  REPLACE(customer_name, ' ', '_')              AS underscored,
  LPAD(customer_id::VARCHAR, 6, '0')            AS padded_id,
  LEFT(customer_name, 3)                        AS left_3,
  RIGHT(customer_name, 3)                       AS right_3
FROM stage.customers;

-- splitting a full name into first/last using the first space as the delimiter
SELECT
  customer_name,
  TRIM(SUBSTR(customer_name, 1, POSITION(' ' IN customer_name)))   AS first_name,
  TRIM(SUBSTR(customer_name, POSITION(' ' IN customer_name)))      AS last_name
FROM stage.customers;

-- deriving a username from an email
SELECT email, SUBSTR(email, 1, POSITION('@' IN email) - 1) AS username FROM stage.customers;

-- data-quality check: leading/trailing whitespace
SELECT * FROM stage.products WHERE LENGTH(product_name) != LENGTH(TRIM(product_name));

SELECT
  product_name, price,
  CEIL(price)              AS ceil_price,
  FLOOR(price)             AS floor_price,
  ROUND(price)              AS rounded_price,
  ROUND(price, 1)           AS rounded_1dp,
  TRUNC(price, 1)           AS truncated_1dp,
  MOD(ROUND(price), 2)      AS is_odd_flag        -- 0 = even, 1 = odd
FROM stage.products;

SELECT MAX(price) - MIN(price) AS price_range FROM stage.products;

SELECT CURRENT_DATE, CURRENT_TIME, CURRENT_TIMESTAMP;

SELECT
  EXTRACT(YEAR  FROM sale_date) AS sale_year,
  EXTRACT(MONTH FROM sale_date) AS sale_month,
  EXTRACT(DAY   FROM sale_date) AS sale_day
FROM stage.sales;

SELECT
  sale_date,
  DATEADD(DAY,  10, sale_date) AS plus_10_days,
  DATEADD(YEAR, 10, sale_date) AS plus_10_years,
  sale_date + INTERVAL '10 days' AS plus_10_days_alt
FROM stage.sales;

SELECT DATEDIFF(DAY, sale_date, CURRENT_DATE) AS days_since_sale FROM stage.sales;
SELECT DATEDIFF(YEAR, DATE '1995-04-18', CURRENT_DATE) AS age_years;
SELECT TO_CHAR(sale_date, 'DD-MON-YYYY') AS formatted_date FROM stage.sales;

SELECT CAST('123' AS INTEGER);
SELECT CAST('123.45' AS NUMERIC(10,2));
SELECT TRY_CAST('not-a-number' AS INTEGER);   -- returns NULL instead of erroring

SELECT TO_DATE('12/12/2024', 'DD/MM/YYYY');
SELECT TO_CHAR(DATE '2024-01-01', 'YYYY-MM-DD') AS formatted_date;

SELECT * FROM stage.sales WHERE sale_id = CAST('101' AS INT);
SELECT * FROM stage.sales WHERE sale_date = CAST('2024-01-15' AS DATE);

-- IFNULL(a, b) — replace a NULL with a fallback
SELECT product_name, IFNULL(vendor, 'No Vendor') AS vendor_display FROM stage.products;

-- COALESCE(a, b, c, ...) — first non-NULL value, any number of args
SELECT customer_name,
  COALESCE(phone, landline_number, second_number, home_number, 'No contact on file') AS best_contact
FROM stage.customers;

-- NULLIF(a, b) — NULL if a = b, otherwise a (avoids divide-by-zero)
SELECT product_id, price / NULLIF(stock, 0) AS price_per_unit_in_stock FROM stage.products;

-- CASE / IFF / DECODE — conditional branching
SELECT product_name, price,
  CASE
    WHEN price > 500 AND category = 'Electronics' THEN 'Premium Electronics'
    WHEN price > 300 THEN 'Mid-Range'
    ELSE 'Budget'
  END AS price_tier
FROM stage.products;

SELECT product_name, price, IFF(price > 500, 'Costly', 'Affordable') AS tier FROM stage.products;

SELECT vendor,
  DECODE(vendor, 'LG','LG Electronics', 'BOAT','BOAT Lifestyle', NULL,'No Vendor Assigned', vendor) AS vendor_display
FROM stage.products;

-- OVER() with no arguments = "the whole result set" — every row keeps its identity
SELECT *, SUM(price) OVER () AS total_all_products FROM stage.products;
SELECT *, MAX(price) OVER () AS max_price, MIN(price) OVER () AS min_price, AVG(price) OVER () AS avg_price
FROM stage.products;

-- PARTITION BY restarts the calculation within each group, like GROUP BY that doesn't collapse rows
SELECT *, SUM(price) OVER (PARTITION BY category) AS category_total FROM stage.products;

SELECT *,
  ROW_NUMBER() OVER (ORDER BY salary DESC) AS row_num,     -- always unique, 1,2,3,4...
  RANK()       OVER (ORDER BY salary DESC) AS rnk,         -- ties share a rank, next rank skips
  DENSE_RANK() OVER (ORDER BY salary DESC) AS dense_rnk    -- ties share a rank, next rank doesn't skip
FROM employees;

-- 2nd highest salary company-wide
SELECT * FROM (SELECT *, DENSE_RANK() OVER (ORDER BY salary DESC) AS r FROM employees) WHERE r = 2;

-- 2nd highest salary PER DEPARTMENT
SELECT * FROM (
  SELECT *, DENSE_RANK() OVER (PARTITION BY department ORDER BY salary DESC) AS r FROM employees
) WHERE r = 2;

SELECT *,
  LAG(salary)  OVER (ORDER BY salary DESC) AS prev_salary,
  LEAD(salary) OVER (ORDER BY salary DESC) AS next_salary
FROM employees;

SELECT *,
  FIRST_VALUE(salary) OVER (ORDER BY salary DESC) AS highest_salary,
  LAST_VALUE(salary)  OVER (ORDER BY salary DESC
                             ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS lowest_salary,
  NTH_VALUE(salary, 3) OVER (ORDER BY salary DESC
                              ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS third_highest
FROM employees;


-- 1. Create the Batsmen Table
CREATE TABLE batsmen (
    player_id INT PRIMARY KEY,
    player_name VARCHAR(100),
    team VARCHAR(50),
    runs_scored INT,
    matches_played INT
);

-- 2. Populate the Batsmen Table
INSERT INTO batsmen (player_id, player_name, team, runs_scored, matches_played) VALUES
(1, 'Virat Kohli', 'India', 13848, 292),
(2, 'Rohit Sharma', 'India', 10709, 262),
(3, 'Babar Azam', 'Pakistan', 5729, 117),
(4, 'Kane Williamson', 'New Zealand', 6810, 165),
(5, 'Joe Root', 'England', 6522, 171),
(6, 'Steve Smith', 'Australia', 5065, 145);

-- 3. Create the Bowlers Table
CREATE TABLE bowlers (
    player_id INT PRIMARY KEY,
    player_name VARCHAR(100),
    team VARCHAR(50),
    wickets_taken INT,
    economy_rate DECIMAL(4,2)
);

-- 4. Populate the Bowlers Table
INSERT INTO bowlers (player_id, player_name, team, wickets_taken, economy_rate) VALUES
(101, 'Jasprit Bumrah', 'India', 149, 4.59),
(102, 'Mitchell Starc', 'Australia', 236, 5.11),
(103, 'Trent Boult', 'New Zealand', 211, 4.93),
(104, 'Shaheen Afridi', 'Pakistan', 104, 5.42),
(105, 'Adil Rashid', 'England', 199, 5.67),
(106, 'Rashid Khan', 'Afghanistan', 183, 4.21);


-- Query 1: Finding previous and next highest run-scorers
SELECT *, 
       LAG(runs_scored) OVER (ORDER BY runs_scored DESC) AS prev_higher_runs, 
       LEAD(runs_scored) OVER (ORDER BY runs_scored DESC) AS next_lower_runs 
FROM batsmen;

-- Query 2: Finding highest, lowest, and 3rd highest run-scorers
SELECT *, 
       FIRST_VALUE(runs_scored) OVER (ORDER BY runs_scored DESC) AS highest_runs, 
       LAST_VALUE(runs_scored) OVER (ORDER BY runs_scored DESC ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS lowest_runs, 
       NTH_VALUE(runs_scored, 3) OVER (ORDER BY runs_scored DESC ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS third_highest_runs 
FROM batsmen;


-- Query 1: Finding previous and next highest wicket-takers
SELECT *, 
       LAG(wickets_taken) OVER (ORDER BY wickets_taken DESC) AS prev_higher_wickets, 
       LEAD(wickets_taken) OVER (ORDER BY wickets_taken DESC) AS next_lower_wickets 
FROM bowlers;

-- Query 2: Finding highest, lowest, and 3rd highest wicket-takers
SELECT *, 
       FIRST_VALUE(wickets_taken) OVER (ORDER BY wickets_taken DESC) AS highest_wickets, 
       LAST_VALUE(wickets_taken) OVER (ORDER BY wickets_taken DESC ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS lowest_wickets, 
       NTH_VALUE(wickets_taken, 3) OVER (ORDER BY wickets_taken DESC ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS third_highest_wickets 
FROM bowlers;

SELECT *,
  SUM(quantity_sold) OVER (ORDER BY sale_date ASC ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS running_total
FROM stage.sales;

SELECT *,
  SUM(quantity_sold) OVER (ORDER BY sale_date ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING) AS moving_3row_total
FROM stage.sales;

SELECT *,
  AVG(quantity_sold) OVER (ORDER BY sale_date ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS moving_avg_3
FROM stage.sales;

-- With QUALIFY — same result as the "2nd highest per department" query above, no subquery needed
SELECT *, DENSE_RANK() OVER (PARTITION BY department ORDER BY salary DESC) AS r
FROM employees
QUALIFY r = 2;

-- Top earner per department, in one clean pass
SELECT * FROM employees
QUALIFY ROW_NUMBER() OVER (PARTITION BY department ORDER BY salary DESC) = 1;

-- INNER JOIN: only rows that match on both sides
SELECT e.emp_id, e.emp_name, e.salary, d.dept_name
FROM employee e INNER JOIN department d ON e.dept_id = d.dept_id;

-- LEFT JOIN: every employee, department columns NULL if there's no match
SELECT e.emp_id, e.emp_name, e.salary, d.dept_name
FROM employee e LEFT JOIN department d ON e.dept_id = d.dept_id;

-- RIGHT JOIN: every department, employee columns NULL if nobody's in it
SELECT e.emp_id, e.emp_name, e.salary, d.dept_name
FROM department d RIGHT JOIN employee e ON e.dept_id = d.dept_id;

-- FULL JOIN: everything from both sides, matched where possible
SELECT e.emp_id, e.emp_name, e.salary, d.dept_name
FROM department d FULL JOIN employee e ON e.dept_id = d.dept_id;

-- anti-join: employees whose department doesn't exist in department
SELECT e.* FROM employee e
LEFT JOIN department d ON e.dept_id = d.dept_id
WHERE d.dept_name IS NULL;

-- CROSS JOIN: every combination, no ON clause
SELECT e.emp_id, e.emp_name, d.dept_name FROM department d CROSS JOIN employee e;

-- NATURAL JOIN: auto-matches on same-named columns (dept_id in both)
SELECT e.emp_id, e.emp_name, e.salary, d.dept_name FROM department d NATURAL JOIN employee e;

-- SELF JOIN: workers joined to itself to resolve each manager's name
SELECT emp.w_name AS employee_name, mgr.w_name AS manager_name
FROM workers emp LEFT JOIN workers mgr ON emp.mgr_id = mgr.wid;

-- who has no manager at all? (self join + anti-join)
SELECT emp.w_name FROM workers emp LEFT JOIN workers mgr ON emp.mgr_id = mgr.wid WHERE mgr.w_name IS NULL;

-- UNION ALL: stack both, keep every row including duplicates
SELECT emp_name AS person FROM employee UNION ALL SELECT contractor_name FROM contractor;

-- UNION: stack and de-duplicate
SELECT emp_name AS person FROM employee UNION SELECT contractor_name FROM contractor;

-- INTERSECT: only names appearing in BOTH tables
SELECT emp_name FROM employee INTERSECT SELECT contractor_name FROM contractor;

-- MINUS (Snowflake also accepts EXCEPT): in the first, not in the second
SELECT emp_name FROM employee MINUS SELECT contractor_name FROM contractor;

-- scalar subquery: returns exactly one value, used like a constant
SELECT * FROM stage.products WHERE price > (SELECT AVG(price) FROM stage.products);

-- non-scalar (multi-row) subquery with IN
SELECT * FROM stage.products
WHERE product_id IN (SELECT product_id FROM stage.products WHERE category = 'Furniture');

-- EXISTS: stops at the first match, often faster than IN for large subqueries
SELECT * FROM stage.products a
WHERE EXISTS (SELECT 1 FROM stage.products b WHERE a.product_id = b.product_id AND b.category = 'Furniture');

-- a correlated subquery references a column from the OUTER query (a.dept_id) —
-- conceptually re-evaluated once per outer row, unlike a plain subquery which runs once
SELECT a.* FROM employee a
WHERE a.salary > (SELECT AVG(b.salary) FROM employee b WHERE b.dept_id = a.dept_id);

-- in SELECT: a constant column, computed once, repeated on every row
SELECT *, (SELECT SUM(salary) FROM workers) AS company_total_salary FROM workers;

-- in FROM (a "derived table"): pre-aggregate before joining
SELECT d.dept_name, s.total_salary
FROM department d
INNER JOIN (SELECT dept_id, SUM(salary) AS total_salary FROM employee GROUP BY dept_id) s
        ON d.dept_id = s.dept_id;

-- in JOIN: filter BEFORE joining instead of after
SELECT a.*, b.*
FROM (SELECT * FROM stage.products WHERE category = 'Electronics') a
INNER JOIN stage.sales b ON a.product_id = b.product_id;

WITH high_earners AS (
  SELECT * FROM workers WHERE salary > (SELECT AVG(salary) FROM workers)
)
SELECT * FROM high_earners ORDER BY salary DESC;

-- multiple CTEs, chained together
WITH cte_employees AS (
  SELECT dept_id, emp_id, emp_name, salary FROM employee
),
cte_hr_dept AS (
  SELECT dept_id, dept_name FROM department WHERE dept_name = 'HR'
)
SELECT a.*, b.* FROM cte_employees a INNER JOIN cte_hr_dept b ON a.dept_id = b.dept_id;

-- every level of the org chart under Alice (wid = 1), with a computed depth
WITH RECURSIVE org_chart AS (
  -- anchor: the starting row
  SELECT wid, w_name, mgr_id, designation, 0 AS depth
  FROM workers WHERE wid = 1

  UNION ALL

  -- recursive step: joins back to org_chart itself, one level down each pass
  SELECT w.wid, w.w_name, w.mgr_id, w.designation, oc.depth + 1
  FROM workers w INNER JOIN org_chart oc ON w.mgr_id = oc.wid
)
SELECT * FROM org_chart ORDER BY depth, wid;

-- building an indented reporting-line string as it recurses
WITH RECURSIVE reporting_line AS (
  SELECT wid, w_name, mgr_id, w_name::VARCHAR AS chain
  FROM workers WHERE mgr_id IS NULL

  UNION ALL

  SELECT w.wid, w.w_name, w.mgr_id, rl.chain || ' -> ' || w.w_name
  FROM workers w INNER JOIN reporting_line rl ON w.mgr_id = rl.wid
)
SELECT * FROM reporting_line ORDER BY wid;

-- PIVOT: rows to columns — one row per company, one column per stock symbol
SELECT * FROM share_market_data
PIVOT (SUM(volume) FOR stock_symbol IN ('PYPL', 'GOOGL', 'MSFT'))
AS p (company_name, trade_date, open_price, close_price, high_price, low_price, pypl_vol, googl_vol, msft_vol);

-- UNPIVOT: columns to rows
CREATE OR REPLACE TABLE scratch_prices AS
SELECT stock_symbol, trade_date, open_price, close_price, high_price, low_price FROM share_market_data;

SELECT stock_symbol, trade_date, price_type, price_value
FROM scratch_prices
UNPIVOT (price_value FOR price_type IN (open_price, close_price, high_price, low_price));

SELECT category, LISTAGG(product_name, ', ') WITHIN GROUP (ORDER BY product_name) AS products
FROM stage.products GROUP BY category;

SELECT mgr.w_name AS manager, LISTAGG(emp.w_name, ', ') WITHIN GROUP (ORDER BY emp.w_name) AS direct_reports
FROM workers emp INNER JOIN workers mgr ON emp.mgr_id = mgr.wid
GROUP BY mgr.w_name;

MERGE INTO employees_2023 AS target
USING employees_2024 AS source
   ON target.emp_id = source.emp_id
WHEN MATCHED THEN
   UPDATE SET target.emp_name = source.emp_name, target.department = source.department
WHEN NOT MATCHED THEN
   INSERT (emp_id, emp_name, department) VALUES (source.emp_id, source.emp_name, source.department);

SELECT * FROM employees_2023 ORDER BY emp_id;

MERGE INTO dwh.dim_customer AS target
USING (
    SELECT customer_id, customer_name, email, address, country, gender
    FROM stage.customers
    QUALIFY ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY created_at DESC) = 1
) AS source
ON target.customer_id = source.customer_id
WHEN MATCHED THEN
    UPDATE SET target.first_name = SPLIT_PART(source.customer_name, ' ', 1),
               target.last_name  = SPLIT_PART(source.customer_name, ' ', 2),
               target.email = TRIM(LOWER(source.email)), target.address = source.address,
               target.country = source.country, target.gender = source.gender
WHEN NOT MATCHED THEN
    INSERT (customer_id, first_name, last_name, email, address, country, gender, is_current, effective_date)
    VALUES (source.customer_id, SPLIT_PART(source.customer_name, ' ', 1), SPLIT_PART(source.customer_name, ' ', 2),
            TRIM(LOWER(source.email)), source.address, source.country, source.gender, TRUE, CURRENT_DATE());

-- Step 1: expire any current row whose details changed
MERGE INTO dwh.dim_customer AS target
USING (
    SELECT customer_id, customer_name, email, address FROM stage.customers
    QUALIFY ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY created_at DESC) = 1
) AS source
ON target.customer_id = source.customer_id AND target.is_current = TRUE
WHEN MATCHED AND (target.email != TRIM(LOWER(source.email)) OR target.address != source.address) THEN
    UPDATE SET target.is_current = FALSE, target.expiry_date = CURRENT_DATE();

-- Step 2: insert a fresh "current" row for anyone new, or just-expired
INSERT INTO dwh.dim_customer (customer_id, first_name, last_name, email, address, country, gender, is_current, effective_date)
SELECT s.customer_id, SPLIT_PART(s.customer_name,' ',1), SPLIT_PART(s.customer_name,' ',2),
       TRIM(LOWER(s.email)), s.address, s.country, s.gender, TRUE, CURRENT_DATE()
FROM (
    SELECT customer_id, customer_name, email, address, country, gender FROM stage.customers
    QUALIFY ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY created_at DESC) = 1
) s
LEFT JOIN dwh.dim_customer d ON d.customer_id = s.customer_id AND d.is_current = TRUE
WHERE d.customer_id IS NULL;

CREATE OR REPLACE PROCEDURE sp_hello_world()
RETURNS STRING
LANGUAGE SQL
AS
$$
BEGIN
    RETURN 'Hello from Snowflake Scripting!';
END;
$$;

CALL sp_hello_world();

-- the same idea using real data: how many products are in the catalog right now?
CREATE OR REPLACE PROCEDURE sp_product_count()
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
    total INT;
BEGIN
    SELECT COUNT(*) INTO total FROM stage.products;
    RETURN 'Product catalog currently has ' || total::STRING || ' item(s)';
END;
$$;

CALL sp_product_count();

CREATE OR REPLACE FUNCTION fn_product_cost_status(exchange_rate NUMBER)
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
    total  NUMBER;
    result STRING;
BEGIN
    SELECT SUM(price) * exchange_rate INTO total FROM stage.products;

    IF (total > 200000) THEN
        result := 'Costly Catalog';
    ELSEIF (total > 100000) THEN
        result := 'Normal Catalog';
    ELSE
        result := 'Cheap Catalog';
    END IF;

    RETURN result;
END;
$$;

SELECT fn_product_cost_status(83);

-- counting FOR loop
CREATE OR REPLACE PROCEDURE sp_generate_test_rows(n INT)
RETURNS STRING
LANGUAGE SQL
AS
$$
BEGIN
    FOR i IN 1 TO n DO
        INSERT INTO stage.vendors (vendor_id, vendor_name, country)
        VALUES (900 + i, 'Test Vendor ' || i::STRING, 'Testland');
    END FOR;
    RETURN n::STRING || ' test vendor rows inserted';
END;
$$;
CALL sp_generate_test_rows(3);

-- FOR loop over a query result (an implicit cursor)
CREATE OR REPLACE TABLE stage.low_stock_log (
    product_id INT, product_name VARCHAR(100), stock_at_flag INT, flagged_at TIMESTAMP_NTZ
);

CREATE OR REPLACE PROCEDURE sp_flag_low_stock(threshold INT)
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
    flagged_count INT DEFAULT 0;
BEGIN
    FOR record IN (SELECT product_id, product_name, stock FROM stage.products WHERE stock < threshold) DO
        INSERT INTO stage.low_stock_log VALUES (record.product_id, record.product_name, record.stock, CURRENT_TIMESTAMP());
        flagged_count := flagged_count + 1;
    END FOR;
    RETURN flagged_count::STRING || ' low-stock product(s) flagged';
END;
$$;
CALL sp_flag_low_stock(20);

CREATE OR REPLACE PROCEDURE sp_countdown(start_at INT)
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
    counter INT DEFAULT start_at;
    log_msg STRING DEFAULT '';
BEGIN
    WHILE (counter > 0) DO
        log_msg := log_msg || counter::STRING || ' ';
        counter := counter - 1;
    END WHILE;
    RETURN 'Countdown: ' || log_msg;
END;
$$;
CALL sp_countdown(5);

-- CONTINUE: skip the rest of this iteration, move to the next
-- BREAK: exit the loop immediately, regardless of how many iterations remain
CREATE OR REPLACE PROCEDURE sp_process_products_demo()
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
    processed INT DEFAULT 0;
    stopped_early BOOLEAN DEFAULT FALSE;
BEGIN
    FOR record IN (SELECT product_id, category, price FROM stage.products ORDER BY product_id) DO

        IF (record.category IS NULL) THEN
            CONTINUE;   -- skip uncategorized rows, but keep looping
        END IF;

        IF (record.price > 1000) THEN
            stopped_early := TRUE;
            BREAK;       -- stop entirely once we hit an unusually expensive item
        END IF;

        processed := processed + 1;
    END FOR;

    RETURN 'Processed ' || processed::STRING || ' product(s), stopped early: ' || stopped_early::STRING;
END;
$$;
CALL sp_process_products_demo();

CREATE OR REPLACE FUNCTION fn_sales_amount(qty INT, unit_price NUMBER)
RETURNS NUMBER(12,2)
LANGUAGE SQL
AS
$$
  qty * unit_price
$$;

SELECT s.sale_id, fn_sales_amount(s.quantity_sold, p.price) AS sales_amount
FROM stage.sales s JOIN stage.products p ON s.product_id = p.product_id;

CREATE OR REPLACE FUNCTION fn_price_tier(price NUMBER)
RETURNS STRING
LANGUAGE SQL
AS
$$
BEGIN
    IF (price > 500) THEN
        RETURN 'Premium';
    ELSEIF (price > 200) THEN
        RETURN 'Standard';
    ELSE
        RETURN 'Budget';
    END IF;
END;
$$;

SELECT product_name, price, fn_price_tier(price) AS tier FROM stage.products;

CREATE OR REPLACE FUNCTION fn_top_products_by_category(cat STRING, n INT)
RETURNS TABLE (product_id INT, product_name STRING, price NUMBER)
LANGUAGE SQL
AS
$$
  SELECT product_id, product_name, price
  FROM stage.products
  WHERE category = cat
  ORDER BY price DESC
  LIMIT n
$$;

SELECT * FROM TABLE(fn_top_products_by_category('Electronics', 3));

CREATE OR REPLACE FILE FORMAT stage.ff_csv_standard
    TYPE = CSV
    FIELD_DELIMITER = ','
    SKIP_HEADER = 1
    FIELD_OPTIONALLY_ENCLOSED_BY = '"'
    NULL_IF = ('NULL', 'null', '')
    EMPTY_FIELD_AS_NULL = TRUE
    DATE_FORMAT = 'YYYY-MM-DD'
    COMPRESSION = AUTO;

CREATE OR REPLACE FILE FORMAT stage.ff_json_standard
    TYPE = JSON
    STRIP_OUTER_ARRAY = TRUE;

CREATE OR REPLACE FILE FORMAT stage.ff_parquet_standard
    TYPE = PARQUET;

SHOW FILE FORMATS IN SCHEMA stage;
DESC FILE FORMAT stage.ff_csv_standard;

-- 1. USER STAGE — every user gets one automatically, referenced as @~. No setup needed.
LIST @~;

-- 2. TABLE STAGE — every table gets one automatically, referenced as @%table_name.
LIST @%stage.products;

-- 3. NAMED INTERNAL STAGE — Snowflake-managed storage, files uploaded via PUT (Section 24)
CREATE OR REPLACE STAGE stage.internal_csv_stage
    FILE_FORMAT = stage.ff_csv_standard;

-- 4. NAMED EXTERNAL STAGE — points at YOUR cloud storage (S3/GCS/Azure Blob) — see Sections 25-27
CREATE OR REPLACE STAGE stage.s3_external_stage
    URL = 's3://my-retail-bucket/incoming/'
    STORAGE_INTEGRATION = s3_retail_integration
    FILE_FORMAT = stage.ff_csv_standard;

SHOW STAGES IN SCHEMA stage;
LIST @stage.internal_csv_stage;

-- Step 1: create the integration object, pointing at the IAM role you set up in AWS
CREATE OR REPLACE STORAGE INTEGRATION s3_retail_integration
    TYPE = EXTERNAL_STAGE
    STORAGE_PROVIDER = 'S3'
    ENABLED = TRUE
    STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::123456789012:role/snowflake-retail-role'
    STORAGE_ALLOWED_LOCATIONS = ('s3://my-retail-bucket/incoming/', 's3://my-retail-bucket/archive/');

-- Step 2: Snowflake generates its own AWS IAM user + external ID for the trust relationship —
-- copy these into the IAM role's trust policy back in AWS
DESC INTEGRATION s3_retail_integration;
-- look at STORAGE_AWS_IAM_USER_ARN and STORAGE_AWS_EXTERNAL_ID in the output

-- Step 3: create the external stage using the integration (no AWS keys anywhere in this SQL)
CREATE OR REPLACE STAGE stage.s3_external_stage
    URL = 's3://my-retail-bucket/incoming/'
    STORAGE_INTEGRATION = s3_retail_integration
    FILE_FORMAT = stage.ff_csv_standard;

LIST @stage.s3_external_stage;

COPY INTO stage.products
FROM @stage.s3_external_stage
PATTERN = '.*products.*\\.csv'
ON_ERROR = 'CONTINUE';

-- works, but embeds a secret key in the stage definition — fine for a scratch/demo,
-- avoid in anything that resembles a real pipeline
CREATE OR REPLACE STAGE stage.s3_quick_stage
    URL = 's3://my-retail-bucket/incoming/'
    CREDENTIALS = (AWS_KEY_ID = 'AKIA...' AWS_SECRET_KEY = '...')
    FILE_FORMAT = stage.ff_csv_standard;

-- ABORT_STATEMENT (the default): one bad row fails the entire load, nothing is committed
COPY INTO stage.products FROM @stage.s3_external_stage
FILE_FORMAT = (FORMAT_NAME = stage.ff_csv_standard) ON_ERROR = 'ABORT_STATEMENT';

-- CONTINUE: skip bad rows, load everything else, report the errors afterward
COPY INTO stage.products FROM @stage.s3_external_stage
FILE_FORMAT = (FORMAT_NAME = stage.ff_csv_standard) ON_ERROR = 'CONTINUE';

-- SKIP_FILE: if a file has even one bad row, skip that WHOLE file, but still load other files
COPY INTO stage.products FROM @stage.s3_external_stage
FILE_FORMAT = (FORMAT_NAME = stage.ff_csv_standard) ON_ERROR = 'SKIP_FILE';

-- SKIP_FILE_N / SKIP_FILE_N%: tolerate up to N (or N%) bad rows per file before skipping it
COPY INTO stage.products FROM @stage.s3_external_stage
FILE_FORMAT = (FORMAT_NAME = stage.ff_csv_standard) ON_ERROR = 'SKIP_FILE_10';

-- FORCE: reload files even if Snowflake's load metadata thinks they were already loaded
-- (by default, COPY INTO silently skips files it has already loaded from this stage/table pair)
COPY INTO stage.products FROM @stage.s3_external_stage
FILE_FORMAT = (FORMAT_NAME = stage.ff_csv_standard) FORCE = TRUE;

-- PATTERN: only load files matching a regex
COPY INTO stage.products FROM @stage.s3_external_stage
FILE_FORMAT = (FORMAT_NAME = stage.ff_csv_standard) PATTERN = '.*products_2026.*\\.csv';

-- PURGE: delete source files from the stage after a successful load (careful with this one)
COPY INTO stage.products FROM @stage.s3_external_stage
FILE_FORMAT = (FORMAT_NAME = stage.ff_csv_standard) PURGE = TRUE;

-- VALIDATION_MODE: dry run — parses and validates without loading a single row
COPY INTO stage.products FROM @stage.s3_external_stage
FILE_FORMAT = (FORMAT_NAME = stage.ff_csv_standard) VALIDATION_MODE = 'RETURN_ERRORS';

-- MATCH_BY_COLUMN_NAME: map source columns to target columns by name instead of position —
-- essential when loading semi-structured formats (JSON/Parquet/Avro) into a relational table
COPY INTO stage.products FROM @stage.s3_external_stage
FILE_FORMAT = (FORMAT_NAME = stage.ff_parquet_standard) MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

-- checking what actually happened after a load
SELECT * FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
    TABLE_NAME => 'stage.products', START_TIME => DATEADD(HOUR, -24, CURRENT_TIMESTAMP())
));

CREATE OR REPLACE TABLE stage.products_raw_json (raw_data VARIANT);

-- PARSE_JSON turns a JSON-formatted string into a queryable VARIANT value —
-- this simulates what COPY INTO does automatically when loading a .json file
INSERT INTO stage.products_raw_json
SELECT PARSE_JSON('{
  "product_id": 1,
  "product_name": "Dell Laptop",
  "category": "Electronics",
  "price": 800.50,
  "tags": ["computers", "electronics", "work-from-home"],
  "vendor": {"id": 101, "name": "LG Electronics", "country": "South Korea"}
}');

-- the real load, once a JSON file is staged (see Sections 24-27 for the stage part)
COPY INTO stage.products_raw_json
FROM @stage.s3_external_stage
FILE_FORMAT = (FORMAT_NAME = stage.ff_json_standard)
ON_ERROR = 'CONTINUE';

-- colon notation reaches into an object; :: casts the VARIANT out to a real type
SELECT
    raw_data:product_id::INT           AS product_id,
    raw_data:product_name::STRING      AS product_name,
    raw_data:price::NUMBER(10,2)       AS price,
    raw_data:vendor:name::STRING       AS vendor_name,          -- nested object
    raw_data:tags[0]::STRING           AS first_tag              -- array element by index
FROM stage.products_raw_json;

-- without a :: cast, you get the raw VARIANT back (still JSON-quoted) — usually not what you want
SELECT raw_data:product_name FROM stage.products_raw_json;         -- "Dell Laptop"  (VARIANT)
SELECT raw_data:product_name::STRING FROM stage.products_raw_json; -- Dell Laptop    (STRING)

-- one row per tag, per product — this is how you turn a nested array into a proper relational shape
SELECT
    raw_data:product_id::INT      AS product_id,
    raw_data:product_name::STRING AS product_name,
    tag.value::STRING             AS tag
FROM stage.products_raw_json,
     LATERAL FLATTEN(input => raw_data:tags) tag;

-- construct a JSON object from relational columns — handy when unloading for an API consumer
SELECT OBJECT_CONSTRUCT(
    'product_id', product_id,
    'product_name', product_name,
    'category', category,
    'price', price
) AS product_json
FROM stage.products;

-- aggregate a whole table into one JSON array
SELECT ARRAY_AGG(OBJECT_CONSTRUCT('product_id', product_id, 'product_name', product_name)) AS all_products
FROM stage.products;

-- Parquet loaded directly into typed columns, matched by name instead of landing in VARIANT
COPY INTO stage.products (product_id, product_name, category, price)
FROM @stage.s3_external_stage
FILE_FORMAT = (TYPE = PARQUET)
MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

CREATE OR REPLACE WAREHOUSE etl_wh
    WAREHOUSE_SIZE = 'SMALL'         -- XSMALL, SMALL, MEDIUM, LARGE, XLARGE, ... 6XLARGE
    AUTO_SUSPEND = 60                -- seconds idle before it suspends (stop paying for idle compute)
    AUTO_RESUME = TRUE               -- automatically wake up when a query needs it
    INITIALLY_SUSPENDED = TRUE;

USE WAREHOUSE etl_wh;

-- resize up for a heavy batch job, then back down afterward
ALTER WAREHOUSE etl_wh SET WAREHOUSE_SIZE = 'LARGE';
-- ... run the heavy ETL ...
ALTER WAREHOUSE etl_wh SET WAREHOUSE_SIZE = 'X-SMALL';

SHOW WAREHOUSES;
SELECT * FROM TABLE(INFORMATION_SCHEMA.WAREHOUSE_LOAD_HISTORY(
    DATE_RANGE_START => DATEADD('hour', -24, CURRENT_TIMESTAMP())
));

ALTER WAREHOUSE etl_wh SET
    MIN_CLUSTER_COUNT = 1,
    MAX_CLUSTER_COUNT = 3,
    SCALING_POLICY = 'STANDARD';       -- or 'ECONOMY' to favor cost over instant scale-out

-- create roles for the retail_dwh project
CREATE ROLE IF NOT EXISTS retail_etl_role;
CREATE ROLE IF NOT EXISTS retail_analyst_role;

-- roles form a hierarchy too — SYSADMIN should sit above your custom roles
-- so admins can still manage objects created under them
GRANT ROLE retail_etl_role TO ROLE SYSADMIN;
GRANT ROLE retail_analyst_role TO ROLE SYSADMIN;

-- grant object privileges to the ETL role: full read/write on stage & dwh
GRANT USAGE ON DATABASE retail_dwh TO ROLE retail_etl_role;
GRANT USAGE ON SCHEMA retail_dwh.stage TO ROLE retail_etl_role;
GRANT USAGE ON SCHEMA retail_dwh.dwh TO ROLE retail_etl_role;
GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA retail_dwh.stage TO ROLE retail_etl_role;
GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA retail_dwh.dwh TO ROLE retail_etl_role;
GRANT USAGE ON WAREHOUSE etl_wh TO ROLE retail_etl_role;

-- the analyst role only needs read access to the reporting layer
GRANT USAGE ON SCHEMA retail_dwh.data_mart TO ROLE retail_analyst_role;
GRANT SELECT ON ALL TABLES IN SCHEMA retail_dwh.data_mart TO ROLE retail_analyst_role;
GRANT USAGE ON WAREHOUSE etl_wh TO ROLE retail_analyst_role;

-- FUTURE grants: automatically extend the same privilege to tables created LATER, not just today
GRANT SELECT ON FUTURE TABLES IN SCHEMA retail_dwh.data_mart TO ROLE retail_analyst_role;

-- create users and assign roles
CREATE USER IF NOT EXISTS etl_service_user
    PASSWORD = 'TemporaryStrongPassword123!'
    DEFAULT_ROLE = retail_etl_role
    DEFAULT_WAREHOUSE = etl_wh
    MUST_CHANGE_PASSWORD = TRUE;

GRANT ROLE retail_etl_role TO USER etl_service_user;

-- inspecting the setup
SHOW ROLES;
SHOW GRANTS TO ROLE retail_analyst_role;
SHOW GRANTS OF ROLE retail_etl_role;

-- switching role mid-session (a user can hold several)
USE ROLE retail_analyst_role;

-- on a large fact table, queries that always filter/group by sale_date benefit from
-- clustering on that column — it keeps rows with similar dates physically co-located
ALTER TABLE dwh.fact_sales CLUSTER BY (sale_date);

-- a compound clustering key, when queries commonly filter on both together
ALTER TABLE dwh.fact_sales CLUSTER BY (sale_date, warehouse_id);

-- check how well-clustered a table currently is (0 = perfectly clustered, higher = worse)
SELECT SYSTEM$CLUSTERING_INFORMATION('dwh.fact_sales', '(sale_date)');

-- drop clustering if it's no longer earning its automatic-reclustering cost
ALTER TABLE dwh.fact_sales DROP CLUSTERING KEY;

-- PERMANENT (the default): full Time Travel (up to 90 days on Enterprise+) + 7-day Fail-safe.
-- Fail-safe is Snowflake-managed disaster recovery AFTER Time Travel expires — you can't query it
-- yourself, only Snowflake support can restore from it. This is what you want for real DWH tables.
CREATE OR REPLACE TABLE dwh.fact_sales_permanent (sale_id INT, sales_amount DECIMAL(12,2));

-- TRANSIENT: Time Travel capped at 1 day, NO Fail-safe at all — cheaper storage, but truly gone
-- once Time Travel expires. Good fit for staging tables you can always reload from source.
CREATE OR REPLACE TRANSIENT TABLE stage.products_transient (
    product_id INT, product_name VARCHAR(100), price DECIMAL(10,2)
);

-- TEMPORARY: exists only for the current session — gone the moment you disconnect, not visible
-- to any other session even while it exists. Perfect for scratch work during ETL development.
CREATE OR REPLACE TEMPORARY TABLE scratch_calculation (id INT, computed_value NUMBER);

-- an EXTERNAL TABLE reads directly from files sitting in cloud storage — no COPY INTO,
-- no data actually stored inside Snowflake, just metadata pointing at the files
CREATE OR REPLACE EXTERNAL TABLE stage.products_external (
    product_id   INT AS (VALUE:product_id::INT),
    product_name STRING AS (VALUE:product_name::STRING),
    price        DECIMAL(10,2) AS (VALUE:price::DECIMAL(10,2))
)
LOCATION = @stage.s3_external_stage
FILE_FORMAT = (TYPE = PARQUET)
AUTO_REFRESH = TRUE;                 -- automatically picks up new files landing in that location

SELECT * FROM stage.products_external WHERE price > 500;

-- Snowflake as the catalog + write engine, storage lives in your own external volume
CREATE OR REPLACE ICEBERG TABLE dwh.fact_sales_iceberg (
    sale_id INT, product_id INT, customer_id INT, sale_date DATE, sales_amount DECIMAL(12,2)
)
CATALOG = 'SNOWFLAKE'
EXTERNAL_VOLUME = 'retail_iceberg_volume'
BASE_LOCATION = 'fact_sales/';

INSERT INTO dwh.fact_sales_iceberg
SELECT sale_id, product_id, customer_id, sale_date, sales_amount FROM dwh.fact_sales;

-- looks and queries exactly like a normal table from inside Snowflake
SELECT * FROM dwh.fact_sales_iceberg WHERE sale_date >= '2026-01-01';

CREATE OR REPLACE PROCEDURE dwh.sp_load_dim_product()
RETURNS STRING LANGUAGE SQL AS
$$
BEGIN
    MERGE INTO dwh.dim_product AS target
    USING (
        SELECT product_id, TRIM(product_name) AS product_name, category, sub_category, price, stock, wid, vendor_id
        FROM stage.products
        QUALIFY ROW_NUMBER() OVER (PARTITION BY product_id ORDER BY created_at DESC) = 1
    ) AS source
    ON target.product_id = source.product_id
    WHEN MATCHED THEN
        UPDATE SET target.product_name = source.product_name, target.category = source.category,
                   target.sub_category = source.sub_category, target.price = source.price,
                   target.stock = source.stock, target.warehouse_id = source.wid, target.vendor_id = source.vendor_id
    WHEN NOT MATCHED THEN
        INSERT (product_id, product_name, category, sub_category, price, stock, warehouse_id, vendor_id)
        VALUES (source.product_id, source.product_name, source.category, source.sub_category,
                source.price, source.stock, source.wid, source.vendor_id);
    RETURN 'dim_product loaded: ' || SQLROWCOUNT::STRING || ' row(s) affected';
END;
$$;

CREATE OR REPLACE PROCEDURE dwh.sp_load_dim_vendor()
RETURNS STRING LANGUAGE SQL AS
$$
BEGIN
    MERGE INTO dwh.dim_vendor AS target
    USING stage.vendors AS source
    ON target.vendor_id = source.vendor_id
    WHEN MATCHED THEN
        UPDATE SET target.vendor_name = source.vendor_name, target.country = source.country,
                   target.contact_email = source.contact_email, target.phone = source.phone
    WHEN NOT MATCHED THEN
        INSERT (vendor_id, vendor_name, country, contact_email, phone)
        VALUES (source.vendor_id, source.vendor_name, source.country, source.contact_email, source.phone);
    RETURN 'dim_vendor loaded';
END;
$$;

CREATE OR REPLACE PROCEDURE dwh.sp_load_dim_warehouse()
RETURNS STRING LANGUAGE SQL AS
$$
BEGIN
    MERGE INTO dwh.dim_warehouse AS target
    USING stage.warehouse AS source
    ON target.warehouse_id = source.wid
    WHEN MATCHED THEN
        UPDATE SET target.warehouse_name = source.warehouse_name, target.city = source.city,
                   target.state = source.state, target.country = source.country, target.capacity = source.capacity
    WHEN NOT MATCHED THEN
        INSERT (warehouse_id, warehouse_name, location, city, state, country, pincode, capacity)
        VALUES (source.wid, source.warehouse_name, source.location, source.city, source.state,
                source.country, source.pincode, source.capacity);
    RETURN 'dim_warehouse loaded';
END;
$$;

CREATE OR REPLACE PROCEDURE dwh.sp_load_dim_customer()
RETURNS STRING LANGUAGE SQL AS
$$
BEGIN
    MERGE INTO dwh.dim_customer AS target
    USING (
        SELECT customer_id, customer_name, email, address, country, gender
        FROM stage.customers
        QUALIFY ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY created_at DESC) = 1
    ) AS source
    ON target.customer_id = source.customer_id
    WHEN MATCHED THEN
        UPDATE SET target.first_name = SPLIT_PART(source.customer_name, ' ', 1),
                   target.last_name  = SPLIT_PART(source.customer_name, ' ', 2),
                   target.email = TRIM(LOWER(source.email)), target.address = source.address,
                   target.country = source.country, target.gender = source.gender
    WHEN NOT MATCHED THEN
        INSERT (customer_id, first_name, last_name, email, address, country, gender, is_current, effective_date)
        VALUES (source.customer_id, SPLIT_PART(source.customer_name, ' ', 1), SPLIT_PART(source.customer_name, ' ', 2),
                TRIM(LOWER(source.email)), source.address, source.country, source.gender, TRUE, CURRENT_DATE());
    RETURN 'dim_customer loaded';
END;
$$;

CREATE OR REPLACE PROCEDURE dwh.sp_load_fact_sales()
RETURNS STRING LANGUAGE SQL AS
$$
BEGIN
    MERGE INTO dwh.fact_sales AS target
    USING (
        SELECT
            s.sale_id, s.customer_id, s.product_id, p.vendor_id, p.wid AS warehouse_id, s.address_id,
            s.sale_date, s.quantity_sold, s.quantity_sold * p.price AS sales_amount
        FROM stage.sales s
        JOIN stage.products p ON s.product_id = p.product_id
        WHERE s.quantity_sold > 0                     -- drop the negative-quantity bad row
          AND s.sale_date <= CURRENT_DATE()            -- drop the future-dated bad row
    ) AS source
    ON target.sale_id = source.sale_id
    WHEN MATCHED THEN
        UPDATE SET target.quantity_sold = source.quantity_sold, target.sales_amount = source.sales_amount
    WHEN NOT MATCHED THEN
        INSERT (sale_id, customer_id, product_id, vendor_id, warehouse_id, address_id, sale_date, quantity_sold, sales_amount)
        VALUES (source.sale_id, source.customer_id, source.product_id, source.vendor_id, source.warehouse_id,
                source.address_id, source.sale_date, source.quantity_sold, source.sales_amount);
    RETURN 'fact_sales loaded: ' || SQLROWCOUNT::STRING || ' row(s) affected';
END;
$$;

CREATE OR REPLACE PROCEDURE dwh.sp_refresh_data_mart()
RETURNS STRING LANGUAGE SQL AS
$$
BEGIN
    TRUNCATE TABLE data_mart.dm_sales_summary;
    INSERT INTO data_mart.dm_sales_summary
    SELECT sale_date, COUNT(DISTINCT sale_id), SUM(quantity_sold), SUM(sales_amount)
    FROM dwh.fact_sales GROUP BY sale_date;

    TRUNCATE TABLE data_mart.dm_customer_sales;
    INSERT INTO data_mart.dm_customer_sales
    SELECT c.customer_id, c.first_name || ' ' || c.last_name, COUNT(DISTINCT f.sale_id),
           SUM(f.quantity_sold), SUM(f.sales_amount)
    FROM dwh.fact_sales f
    JOIN dwh.dim_customer c ON f.customer_id = c.customer_id AND c.is_current = TRUE
    GROUP BY c.customer_id, c.first_name, c.last_name;

    TRUNCATE TABLE data_mart.dm_product_sales;
    INSERT INTO data_mart.dm_product_sales
    SELECT p.product_id, p.product_name, p.category, COUNT(DISTINCT f.sale_id), SUM(f.quantity_sold), SUM(f.sales_amount)
    FROM dwh.fact_sales f
    JOIN dwh.dim_product p ON f.product_id = p.product_id
    GROUP BY p.product_id, p.product_name, p.category;

    TRUNCATE TABLE data_mart.dm_top_products;
    INSERT INTO data_mart.dm_top_products
    SELECT RANK() OVER (ORDER BY SUM(f.sales_amount) DESC), p.product_name, SUM(f.sales_amount)
    FROM dwh.fact_sales f JOIN dwh.dim_product p ON f.product_id = p.product_id
    GROUP BY p.product_name
    QUALIFY RANK() OVER (ORDER BY SUM(f.sales_amount) DESC) <= 10;

    RETURN 'data_mart refreshed';
END;
$$;

CREATE OR REPLACE PROCEDURE dwh.sp_run_etl_pipeline()
RETURNS STRING LANGUAGE SQL AS
$$
BEGIN
    CALL dwh.sp_load_dim_vendor();
    CALL dwh.sp_load_dim_warehouse();
    CALL dwh.sp_load_dim_customer();
    CALL dwh.sp_load_dim_product();
    CALL dwh.sp_load_fact_sales();
    CALL dwh.sp_refresh_data_mart();
    RETURN 'Full retail_dwh pipeline completed at ' || CURRENT_TIMESTAMP()::STRING;
END;
$$;

-- one call runs the entire Stage -> DWH -> Data Mart pipeline
CALL dwh.sp_run_etl_pipeline();

SELECT * FROM data_mart.dm_top_products ORDER BY rank_no;