BROLLY ACADEMY — Snowpipe: Step-by-Step S3 Auto-Ingest and Validation Guide
Documentation-reviewed examples; not executed in a connected account.
Run one labeled phase at a time. Replace prerequisites/placeholders first.
Cleanup commands are optional and must be reviewed.

-- PHASE 1
-- Run using a role authorized to manage integrations.
CREATE STORAGE INTEGRATION BA_SP_S3_INT
 TYPE=EXTERNAL_STAGE
 STORAGE_PROVIDER='S3'
 ENABLED=TRUE
 STORAGE_AWS_ROLE_ARN='arn:aws:iam::YOUR_AWS_ACCOUNT:role/YOUR_READ_ROLE'
 STORAGE_ALLOWED_LOCATIONS=('s3://YOUR_BUCKET/orders/');
DESC INTEGRATION BA_SP_S3_INT;

-- PHASE 2
USE DATABASE YOUR_LAB_DATABASE;
USE WAREHOUSE YOUR_LAB_WAREHOUSE;
CREATE SCHEMA BA_SP_LAB;
USE SCHEMA BA_SP_LAB;
CREATE FILE FORMAT FF_ORDER_JSON TYPE=JSON;
CREATE TABLE RAW_ORDER_EVENTS (payload VARIANT);
CREATE STAGE ORDER_STAGE
 URL='s3://YOUR_BUCKET/orders/'
 STORAGE_INTEGRATION=BA_SP_S3_INT
 FILE_FORMAT=FF_ORDER_JSON;
LIST @ORDER_STAGE;
SELECT COUNT(*) AS initial_rows FROM RAW_ORDER_EVENTS;

-- PHASE 3
CREATE PIPE ORDER_PIPE AUTO_INGEST=TRUE AS
 COPY INTO RAW_ORDER_EVENTS
 FROM @ORDER_STAGE
 FILE_FORMAT=(FORMAT_NAME=FF_ORDER_JSON)
 ON_ERROR='SKIP_FILE';
SHOW PIPES LIKE 'ORDER_PIPE' IN SCHEMA BA_SP_LAB;
SELECT SYSTEM$PIPE_STATUS('BA_SP_LAB.ORDER_PIPE');

-- PHASE 4
SELECT payload:event_id::NUMBER AS event_id,
 payload:amount::NUMBER(12,2) AS amount
FROM RAW_ORDER_EVENTS ORDER BY event_id;
SELECT * FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
 TABLE_NAME=>'BA_SP_LAB.RAW_ORDER_EVENTS',
 START_TIME=>DATEADD('hour',-2,CURRENT_TIMESTAMP())
)) ORDER BY LAST_LOAD_TIME DESC;

-- PHASE 5
SELECT COUNT(*) AS raw_rows,
 SUM(TRY_TO_DECIMAL(payload:amount::STRING,12,2)) AS raw_amount
FROM RAW_ORDER_EVENTS;
SELECT payload:event_id::NUMBER AS event_id,COUNT(*) AS copies
FROM RAW_ORDER_EVENTS GROUP BY 1 HAVING COUNT(*)>1;

-- PHASE 6
CREATE VIEW ORDER_EVENT_QUALITY AS
SELECT payload,
 TRY_TO_NUMBER(payload:event_id::STRING) AS event_id,
 TRY_TO_TIMESTAMP_TZ(payload:event_ts::STRING) AS event_ts,
 TRY_TO_DECIMAL(payload:amount::STRING,12,2) AS amount
FROM RAW_ORDER_EVENTS;

CREATE VIEW ORDERS_CURRENT AS
SELECT event_id,event_ts,amount FROM ORDER_EVENT_QUALITY
WHERE event_id IS NOT NULL AND event_ts IS NOT NULL
 AND amount IS NOT NULL AND amount>=0
QUALIFY ROW_NUMBER() OVER(
 PARTITION BY event_id
 ORDER BY event_ts DESC,TO_JSON(payload) DESC
)=1;

SELECT COUNT(*) AS distinct_orders,SUM(amount) AS revenue
FROM ORDERS_CURRENT;
SELECT * FROM ORDERS_CURRENT ORDER BY event_id;

-- PHASE 7
SELECT payload FROM ORDER_EVENT_QUALITY
WHERE event_id IS NULL OR event_ts IS NULL OR amount IS NULL OR amount<0;
SELECT COUNT(*) AS raw_rows FROM RAW_ORDER_EVENTS;
SELECT COUNT(*) AS report_rows,SUM(amount) AS revenue FROM ORDERS_CURRENT;

-- PHASE 8
SELECT * FROM TABLE(INFORMATION_SCHEMA.VALIDATE_PIPE_LOAD(
 PIPE_NAME=>'BA_SP_LAB.ORDER_PIPE',
 START_TIME=>DATEADD('hour',-2,CURRENT_TIMESTAMP())
));
SELECT SYSTEM$PIPE_STATUS('BA_SP_LAB.ORDER_PIPE');

-- PHASE 9
-- Optional recovery action, only after inspecting the backlog:
ALTER PIPE ORDER_PIPE REFRESH PREFIX='backfill/';

-- PHASE 10
-- Execute only when the disposable lab is no longer needed.
ALTER PIPE ORDER_PIPE SET PIPE_EXECUTION_PAUSED=TRUE;
-- After cloud notification and producer cleanup, and after review:
DROP SCHEMA BA_SP_LAB;