post_deploy.sql

-- =============================================================================
-- post_deploy.sql — Objects not supported by DEFINE (streams, file formats, tasks)
-- =============================================================================
-- Run after: snow dcm deploy
-- Execution: snow sql -f post_deploy.sql -c <connection>
-- =============================================================================

USE DATABASE {{ db }};
USE SCHEMA {{ raw_schema }};

-- File Formats
CREATE OR REPLACE FILE FORMAT CRMI_RAW_FF_CUSTOMER_A TYPE='CSV' FIELD_DELIMITER=',' SKIP_HEADER=1 FIELD_OPTIONALLY_ENCLOSED_BY='"' NULL_IF=('','NULL','null') EMPTY_FIELD_AS_NULL=TRUE ENCODING='UTF8';
CREATE OR REPLACE FILE FORMAT CRMI_RAW_FF_CUSTOMER_B TYPE='CSV' FIELD_DELIMITER=',' SKIP_HEADER=1 FIELD_OPTIONALLY_ENCLOSED_BY='"' NULL_IF=('','NULL','null') EMPTY_FIELD_AS_NULL=TRUE ENCODING='UTF8';
CREATE OR REPLACE FILE FORMAT CRMI_RAW_FF_CUSTOMER_C TYPE='CSV' FIELD_DELIMITER=',' SKIP_HEADER=1 FIELD_OPTIONALLY_ENCLOSED_BY='"' NULL_IF=('','NULL','null','N/A') EMPTY_FIELD_AS_NULL=TRUE ENCODING='UTF8';
CREATE OR REPLACE FILE FORMAT CRMI_RAW_FF_ADDRESSES_A TYPE='CSV' FIELD_DELIMITER=',' SKIP_HEADER=1 FIELD_OPTIONALLY_ENCLOSED_BY='"' NULL_IF=('','NULL','null') EMPTY_FIELD_AS_NULL=TRUE ENCODING='UTF8';
CREATE OR REPLACE FILE FORMAT CRMI_RAW_FF_ADDRESSES_B TYPE='CSV' FIELD_DELIMITER=',' SKIP_HEADER=1 FIELD_OPTIONALLY_ENCLOSED_BY='"' NULL_IF=('','NULL','null') EMPTY_FIELD_AS_NULL=TRUE ENCODING='UTF8';
CREATE OR REPLACE FILE FORMAT CRMI_RAW_FF_ADDRESSES_C TYPE='CSV' FIELD_DELIMITER=',' SKIP_HEADER=1 FIELD_OPTIONALLY_ENCLOSED_BY='"' NULL_IF=('','NULL','null','N/A') EMPTY_FIELD_AS_NULL=TRUE ENCODING='UTF8';

-- Streams on directory tables
CREATE OR REPLACE STREAM CRMI_RAW_SM_CUSTOMER_A ON STAGE CRMI_RAW_ST_CUSTOMER_A;
CREATE OR REPLACE STREAM CRMI_RAW_SM_CUSTOMER_B ON STAGE CRMI_RAW_ST_CUSTOMER_B;
CREATE OR REPLACE STREAM CRMI_RAW_SM_CUSTOMER_C ON STAGE CRMI_RAW_ST_CUSTOMER_C;
CREATE OR REPLACE STREAM CRMI_RAW_SM_ADDRESSES_A ON STAGE CRMI_RAW_ST_ADDRESSES_A;
CREATE OR REPLACE STREAM CRMI_RAW_SM_ADDRESSES_B ON STAGE CRMI_RAW_ST_ADDRESSES_B;
CREATE OR REPLACE STREAM CRMI_RAW_SM_ADDRESSES_C ON STAGE CRMI_RAW_ST_ADDRESSES_C;

-- Serverless tasks (event-driven ingestion)
CREATE OR REPLACE TASK CRMI_RAW_TS_LOAD_CUSTOMER_A
    SCHEDULE = '60 MINUTE' USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'XSMALL'
    WHEN SYSTEM$STREAM_HAS_DATA('CRMI_RAW_SM_CUSTOMER_A')
AS COPY INTO CRMI_RAW_TB_CUSTOMER_A (src_customer_id, first_name, last_name, email, phone, _SOURCE_FILE)
    FROM (SELECT $1::VARCHAR(50), $2::VARCHAR(100), $3::VARCHAR(100), $4::VARCHAR(255), $5::VARCHAR(50), METADATA$FILENAME FROM @CRMI_RAW_ST_CUSTOMER_A)
    FILE_FORMAT = CRMI_RAW_FF_CUSTOMER_A ON_ERROR = 'CONTINUE';

CREATE OR REPLACE TASK CRMI_RAW_TS_LOAD_CUSTOMER_B
    SCHEDULE = '60 MINUTE' USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'XSMALL'
    WHEN SYSTEM$STREAM_HAS_DATA('CRMI_RAW_SM_CUSTOMER_B')
AS COPY INTO CRMI_RAW_TB_CUSTOMER_B (customer_key, name, email_address, mobile, _SOURCE_FILE)
    FROM (SELECT $1::VARCHAR(50), $2::VARCHAR(200), $3::VARCHAR(255), $4::VARCHAR(50), METADATA$FILENAME FROM @CRMI_RAW_ST_CUSTOMER_B)
    FILE_FORMAT = CRMI_RAW_FF_CUSTOMER_B ON_ERROR = 'CONTINUE';

CREATE OR REPLACE TASK CRMI_RAW_TS_LOAD_CUSTOMER_C
    SCHEDULE = '60 MINUTE' USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'XSMALL'
    WHEN SYSTEM$STREAM_HAS_DATA('CRMI_RAW_SM_CUSTOMER_C')
AS COPY INTO CRMI_RAW_TB_CUSTOMER_C (ticket_customer_id, caller_name, callback_email, callback_phone, _SOURCE_FILE)
    FROM (SELECT $1::VARCHAR(50), $2::VARCHAR(200), $3::VARCHAR(255), $4::VARCHAR(50), METADATA$FILENAME FROM @CRMI_RAW_ST_CUSTOMER_C)
    FILE_FORMAT = CRMI_RAW_FF_CUSTOMER_C ON_ERROR = 'CONTINUE';

CREATE OR REPLACE TASK CRMI_RAW_TS_LOAD_ADDRESSES_A
    SCHEDULE = '60 MINUTE' USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'XSMALL'
    WHEN SYSTEM$STREAM_HAS_DATA('CRMI_RAW_SM_ADDRESSES_A')
AS COPY INTO CRMI_RAW_TB_ADDRESSES_A (src_address_id, src_customer_id, street, city, postal_code, country, _SOURCE_FILE)
    FROM (SELECT $1::VARCHAR(50), $2::VARCHAR(50), $3::VARCHAR(255), $4::VARCHAR(100), $5::VARCHAR(20), $6::VARCHAR(10), METADATA$FILENAME FROM @CRMI_RAW_ST_ADDRESSES_A)
    FILE_FORMAT = CRMI_RAW_FF_ADDRESSES_A ON_ERROR = 'CONTINUE';

CREATE OR REPLACE TASK CRMI_RAW_TS_LOAD_ADDRESSES_B
    SCHEDULE = '60 MINUTE' USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'XSMALL'
    WHEN SYSTEM$STREAM_HAS_DATA('CRMI_RAW_SM_ADDRESSES_B')
AS COPY INTO CRMI_RAW_TB_ADDRESSES_B (addr_id, customer_key, address_line, city, zip, country_code, _SOURCE_FILE)
    FROM (SELECT $1::VARCHAR(50), $2::VARCHAR(50), $3::VARCHAR(255), $4::VARCHAR(100), $5::VARCHAR(20), $6::VARCHAR(10), METADATA$FILENAME FROM @CRMI_RAW_ST_ADDRESSES_B)
    FILE_FORMAT = CRMI_RAW_FF_ADDRESSES_B ON_ERROR = 'CONTINUE';

CREATE OR REPLACE TASK CRMI_RAW_TS_LOAD_ADDRESSES_C
    SCHEDULE = '60 MINUTE' USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'XSMALL'
    WHEN SYSTEM$STREAM_HAS_DATA('CRMI_RAW_SM_ADDRESSES_C')
AS COPY INTO CRMI_RAW_TB_ADDRESSES_C (addr_ref, ticket_customer_id, location, town, postcode, country, _SOURCE_FILE)
    FROM (SELECT $1::VARCHAR(50), $2::VARCHAR(50), $3::VARCHAR(255), $4::VARCHAR(100), $5::VARCHAR(20), $6::VARCHAR(10), METADATA$FILENAME FROM @CRMI_RAW_ST_ADDRESSES_C)
    FILE_FORMAT = CRMI_RAW_FF_ADDRESSES_C ON_ERROR = 'CONTINUE';

-- Resume tasks
ALTER TASK CRMI_RAW_TS_LOAD_CUSTOMER_A RESUME;
ALTER TASK CRMI_RAW_TS_LOAD_CUSTOMER_B RESUME;
ALTER TASK CRMI_RAW_TS_LOAD_CUSTOMER_C RESUME;
ALTER TASK CRMI_RAW_TS_LOAD_ADDRESSES_A RESUME;
ALTER TASK CRMI_RAW_TS_LOAD_ADDRESSES_B RESUME;
ALTER TASK CRMI_RAW_TS_LOAD_ADDRESSES_C RESUME;