-- Snowflake Streams and Tasks — guided SQL phases
-- Documentation reviewed; not executed in a connected Snowflake account.
-- Read each matching article section first. Optional drills have dependencies.
-- Never execute this entire file as one batch.

-- PHASE 1
USE WAREHOUSE BA_LAB_WH;
CREATE SCHEMA BA_CDC_LAB;
USE SCHEMA BA_CDC_LAB;

CREATE TABLE CUSTOMER_EVENTS (
  event_id NUMBER, customer_id VARCHAR, source_version NUMBER,
  operation VARCHAR, customer_name VARCHAR
);
CREATE STREAM CUSTOMER_EVENT_STREAM
  ON TABLE CUSTOMER_EVENTS APPEND_ONLY=TRUE;

CREATE TABLE CUSTOMER_STATE (
  customer_id VARCHAR, customer_name VARCHAR,
  source_version NUMBER, last_event_id NUMBER, is_deleted BOOLEAN
);
CREATE VIEW ACTIVE_CUSTOMERS AS
SELECT customer_id, customer_name, source_version
FROM CUSTOMER_STATE WHERE NOT is_deleted;

INSERT INTO CUSTOMER_EVENTS VALUES
 (101,'C01',1,'U','Asha'),
 (102,'C02',1,'U','Ravi'),
 (103,'C01',2,'U','Asha Rao');

SELECT event_id, customer_id, source_version, METADATA$ACTION
FROM CUSTOMER_EVENT_STREAM ORDER BY event_id;


-- PHASE 2
SELECT CURRENT_ROLE(),CURRENT_DATABASE(),CURRENT_SCHEMA(),CURRENT_WAREHOUSE();
SELECT COUNT(*) AS seed_events FROM CUSTOMER_EVENTS;
SELECT event_id,COUNT(*) AS copies FROM CUSTOMER_EVENTS
GROUP BY event_id HAVING COUNT(*)>1;
SELECT * FROM CUSTOMER_EVENTS
WHERE customer_id IS NULL OR source_version IS NULL OR event_id IS NULL
 OR operation IS NULL OR operation NOT IN ('U','D');

-- PHASE 3
MERGE INTO CUSTOMER_STATE t
USING (
  SELECT event_id, customer_id, source_version, operation, customer_name
  FROM CUSTOMER_EVENT_STREAM
  WHERE METADATA$ACTION='INSERT'
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY customer_id
    ORDER BY source_version DESC, event_id DESC
  )=1
) s
ON t.customer_id=s.customer_id
WHEN MATCHED AND (
  s.source_version > t.source_version OR
  (s.source_version=t.source_version AND s.event_id > t.last_event_id)
) THEN UPDATE SET
  customer_name=s.customer_name,
  source_version=s.source_version,
  last_event_id=s.event_id,
  is_deleted=(s.operation='D')
WHEN NOT MATCHED THEN INSERT
  (customer_id,customer_name,source_version,last_event_id,is_deleted)
VALUES
  (s.customer_id,s.customer_name,s.source_version,s.event_id,s.operation='D');

SELECT * FROM CUSTOMER_STATE ORDER BY customer_id;
SELECT COUNT(*) AS remaining_stream_rows FROM CUSTOMER_EVENT_STREAM;


-- PHASE 4
-- Expected: no duplicate target keys.
SELECT customer_id, COUNT(*) AS copies
FROM CUSTOMER_STATE GROUP BY customer_id HAVING COUNT(*)>1;

-- Save the state, rerun the same MERGE with an empty stream,
-- and compare against this snapshot.
CREATE TEMP TABLE STATE_BEFORE_EMPTY_RERUN AS
SELECT * FROM CUSTOMER_STATE;

MERGE INTO CUSTOMER_STATE t
USING (
  SELECT event_id, customer_id, source_version, operation, customer_name
  FROM CUSTOMER_EVENT_STREAM
  WHERE METADATA$ACTION='INSERT'
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY customer_id
    ORDER BY source_version DESC, event_id DESC
  )=1
) s
ON t.customer_id=s.customer_id
WHEN MATCHED AND (
  s.source_version > t.source_version OR
  (s.source_version=t.source_version AND s.event_id > t.last_event_id)
) THEN UPDATE SET
  customer_name=s.customer_name,
  source_version=s.source_version,
  last_event_id=s.event_id,
  is_deleted=(s.operation='D')
WHEN NOT MATCHED THEN INSERT
  (customer_id,customer_name,source_version,last_event_id,is_deleted)
VALUES
  (s.customer_id,s.customer_name,s.source_version,s.event_id,s.operation='D');
SELECT * FROM CUSTOMER_STATE
MINUS SELECT * FROM STATE_BEFORE_EMPTY_RERUN;
SELECT * FROM STATE_BEFORE_EMPTY_RERUN
MINUS SELECT * FROM CUSTOMER_STATE;


-- PHASE 5
INSERT INTO CUSTOMER_EVENTS VALUES
 (104,'C01',1,'U','Old Asha'),
 (105,'C02',2,'D',NULL),
 (106,'C03',1,'U','Mina');
MERGE INTO CUSTOMER_STATE t
USING (
  SELECT event_id, customer_id, source_version, operation, customer_name
  FROM CUSTOMER_EVENT_STREAM
  WHERE METADATA$ACTION='INSERT'
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY customer_id
    ORDER BY source_version DESC, event_id DESC
  )=1
) s
ON t.customer_id=s.customer_id
WHEN MATCHED AND (
  s.source_version > t.source_version OR
  (s.source_version=t.source_version AND s.event_id > t.last_event_id)
) THEN UPDATE SET
  customer_name=s.customer_name,
  source_version=s.source_version,
  last_event_id=s.event_id,
  is_deleted=(s.operation='D')
WHEN NOT MATCHED THEN INSERT
  (customer_id,customer_name,source_version,last_event_id,is_deleted)
VALUES
  (s.customer_id,s.customer_name,s.source_version,s.event_id,s.operation='D');
SELECT * FROM CUSTOMER_STATE ORDER BY customer_id;
SELECT * FROM ACTIVE_CUSTOMERS ORDER BY customer_id;


-- PHASE 6
WITH expected(customer_id,customer_name,source_version,last_event_id,is_deleted) AS (
 SELECT * FROM VALUES ('C01','Asha Rao',2,103,FALSE),
 ('C02',NULL,2,105,TRUE),('C03','Mina',1,106,FALSE)
)
SELECT 'EXPECTED_NOT_FOUND' AS issue,* FROM (
 SELECT * FROM expected MINUS SELECT * FROM CUSTOMER_STATE
)
UNION ALL
SELECT 'UNEXPECTED_STATE' AS issue,* FROM (
 SELECT * FROM CUSTOMER_STATE MINUS SELECT * FROM expected
);
SELECT COUNT(*) AS state_rows,COUNT_IF(is_deleted) AS tombstones FROM CUSTOMER_STATE;
SELECT customer_id,customer_name,source_version FROM ACTIVE_CUSTOMERS ORDER BY customer_id;

-- PHASE 7
INSERT INTO CUSTOMER_EVENTS VALUES (108,'C02',1,'U','Old Ravi');
MERGE INTO CUSTOMER_STATE t
USING (
  SELECT event_id, customer_id, source_version, operation, customer_name
  FROM CUSTOMER_EVENT_STREAM
  WHERE METADATA$ACTION='INSERT'
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY customer_id
    ORDER BY source_version DESC, event_id DESC
  )=1
) s
ON t.customer_id=s.customer_id
WHEN MATCHED AND (
  s.source_version > t.source_version OR
  (s.source_version=t.source_version AND s.event_id > t.last_event_id)
) THEN UPDATE SET
  customer_name=s.customer_name,
  source_version=s.source_version,
  last_event_id=s.event_id,
  is_deleted=(s.operation='D')
WHEN NOT MATCHED THEN INSERT
  (customer_id,customer_name,source_version,last_event_id,is_deleted)
VALUES
  (s.customer_id,s.customer_name,s.source_version,s.event_id,s.operation='D');
SELECT customer_id,source_version,last_event_id,is_deleted
FROM CUSTOMER_STATE WHERE customer_id='C02';

-- PHASE 8
CREATE TASK APPLY_CUSTOMER_EVENTS
  WAREHOUSE=BA_LAB_WH
  SCHEDULE='5 MINUTES'
  WHEN SYSTEM$STREAM_HAS_DATA('CUSTOMER_EVENT_STREAM')
AS
MERGE INTO CUSTOMER_STATE t
USING (
 SELECT event_id,customer_id,source_version,operation,customer_name
 FROM CUSTOMER_EVENT_STREAM
 WHERE METADATA$ACTION='INSERT'
 QUALIFY ROW_NUMBER() OVER (
   PARTITION BY customer_id ORDER BY source_version DESC,event_id DESC
 )=1
) s
ON t.customer_id=s.customer_id
WHEN MATCHED AND (
 s.source_version>t.source_version OR
 (s.source_version=t.source_version AND s.event_id>t.last_event_id)
) THEN UPDATE SET customer_name=s.customer_name,
 source_version=s.source_version,last_event_id=s.event_id,
 is_deleted=(s.operation='D')
WHEN NOT MATCHED THEN INSERT
 (customer_id,customer_name,source_version,last_event_id,is_deleted)
VALUES(s.customer_id,s.customer_name,s.source_version,s.event_id,s.operation='D');

EXECUTE TASK APPLY_CUSTOMER_EVENTS;
-- Execution is asynchronous; inspect history before checking its result.
-- Resume only when you want ongoing scheduled runs:
-- ALTER TASK APPLY_CUSTOMER_EVENTS RESUME;


-- PHASE 9
ALTER TASK APPLY_CUSTOMER_EVENTS SUSPEND;
INSERT INTO CUSTOMER_EVENTS VALUES (107,'C03',2,'U','Mina Das');

BEGIN;
MERGE INTO CUSTOMER_STATE t
USING (
  SELECT event_id, customer_id, source_version, operation, customer_name
  FROM CUSTOMER_EVENT_STREAM
  WHERE METADATA$ACTION='INSERT'
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY customer_id
    ORDER BY source_version DESC, event_id DESC
  )=1
) s
ON t.customer_id=s.customer_id
WHEN MATCHED AND (
  s.source_version > t.source_version OR
  (s.source_version=t.source_version AND s.event_id > t.last_event_id)
) THEN UPDATE SET
  customer_name=s.customer_name,
  source_version=s.source_version,
  last_event_id=s.event_id,
  is_deleted=(s.operation='D')
WHEN NOT MATCHED THEN INSERT
  (customer_id,customer_name,source_version,last_event_id,is_deleted)
VALUES
  (s.customer_id,s.customer_name,s.source_version,s.event_id,s.operation='D');
SELECT * FROM CUSTOMER_STATE WHERE customer_id='C03';
ROLLBACK;

SELECT * FROM CUSTOMER_STATE WHERE customer_id='C03';
SELECT event_id FROM CUSTOMER_EVENT_STREAM WHERE event_id=107;
-- Expected after rollback: target still at version 1; event 107 still visible.
MERGE INTO CUSTOMER_STATE t
USING (
  SELECT event_id, customer_id, source_version, operation, customer_name
  FROM CUSTOMER_EVENT_STREAM
  WHERE METADATA$ACTION='INSERT'
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY customer_id
    ORDER BY source_version DESC, event_id DESC
  )=1
) s
ON t.customer_id=s.customer_id
WHEN MATCHED AND (
  s.source_version > t.source_version OR
  (s.source_version=t.source_version AND s.event_id > t.last_event_id)
) THEN UPDATE SET
  customer_name=s.customer_name,
  source_version=s.source_version,
  last_event_id=s.event_id,
  is_deleted=(s.operation='D')
WHEN NOT MATCHED THEN INSERT
  (customer_id,customer_name,source_version,last_event_id,is_deleted)
VALUES
  (s.customer_id,s.customer_name,s.source_version,s.event_id,s.operation='D');


-- PHASE 10
SELECT name, state, scheduled_time, completed_time,
       query_id, error_code, error_message
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
  TASK_NAME=>'APPLY_CUSTOMER_EVENTS',RESULT_LIMIT=>20
))
ORDER BY scheduled_time DESC;

SHOW STREAMS LIKE 'CUSTOMER_EVENT_STREAM';
SELECT customer_id, source_version, last_event_id, is_deleted
FROM CUSTOMER_STATE ORDER BY customer_id;