SnowConvert AI - SSIS¶

-- BEGIN Sequence Container: MySequence
-- Task 1 within sequence
EXECUTE DBT PROJECT public.DataFlow1 ARGS='build --target dev';
-- Task 2 within sequence  
EXECUTE DBT PROJECT public.DataFlow2 ARGS='build --target dev';
-- END Sequence Container: MySequence

-- List files from Snowflake stage
LIST @<STAGE_PLACEHOLDER>/FolderPath PATTERN = '.*/file_pattern\.csv';

-- Create cursor for iteration
LET file_cursor CURSOR FOR
   SELECT REGEXP_SUBSTR($1, '[^/]+$') AS FILE_VALUE
   FROM TABLE(RESULT_SCAN(LAST_QUERY_ID()))
   WHERE $1 NOT LIKE '%FolderPath/%/%';

-- Iterate through files
FOR file_row IN file_cursor DO
   User_CurrentFileName := :file_row.FILE_VALUE;
   EXECUTE DBT PROJECT public.My_DataFlow_Project ARGS='build --target dev';
END FOR;

CREATE OR REPLACE TASK public.my_package_send_mail_task
WAREHOUSE=DUMMY_WAREHOUSE
AFTER public.my_package
AS
BEGIN
   -- Step 1: Create Notification Integration dynamically
   BEGIN
      LET my_package_Send_Mail_Task_integration_sql STRING := 'CREATE OR REPLACE NOTIFICATION INTEGRATION my_package_Send_Mail_Task
  TYPE=EMAIL
  ENABLED=TRUE
  ALLOWED_RECIPIENTS=("admin@example.com", "team@example.com")';
      EXECUTE IMMEDIATE :my_package_Send_Mail_Task_integration_sql;
   END;
   
   -- Step 2: Send the email
   CALL SYSTEM$SEND_EMAIL('my_package_Send_Mail_Task', 'admin@example.com,team@example.com', 'Subject', 'Message body');
END;

BEGIN
   BEGIN
      LET my_package_Send_Mail_Task_integration_sql STRING := 'CREATE OR REPLACE NOTIFICATION INTEGRATION my_package_Send_Mail_Task
  TYPE=EMAIL
  ENABLED=TRUE
  ALLOWED_RECIPIENTS=("admin@example.com")';
      EXECUTE IMMEDIATE :my_package_Send_Mail_Task_integration_sql;
   END;
   CALL SYSTEM$SEND_EMAIL('my_package_Send_Mail_Task', 'admin@example.com', 'Daily Report', 'The daily report is ready.');
END;

BEGIN
   BEGIN
      LET my_package_Send_Mail_Task_integration_sql STRING := 'CREATE OR REPLACE NOTIFICATION INTEGRATION my_package_Send_Mail_Task
  TYPE=EMAIL
  ENABLED=TRUE
  ALLOWED_RECIPIENTS=("noreply@company.com", "admin@example.com")';
      EXECUTE IMMEDIATE :my_package_Send_Mail_Task_integration_sql;
   END;
   --** SSC-FDM-SSIS0008 - SNOWFLAKE'S EMAIL INTEGRATION USES A FIXED SENDER ADDRESS. THE ORIGINAL FROM ADDRESS HAS BEEN PREPENDED TO THE MESSAGE BODY FOR REFERENCE. **
   CALL SYSTEM$SEND_EMAIL('my_package_Send_Mail_Task', 'noreply@company.com,admin@example.com', 'Notification', 'Email sent by: noreply@company.com

Package completed successfully.');
END;

BEGIN
   BEGIN
      LET my_package_Send_Mail_Task_integration_sql STRING := 'CREATE OR REPLACE NOTIFICATION INTEGRATION my_package_Send_Mail_Task
  TYPE=EMAIL
  ENABLED=TRUE
  ALLOWED_RECIPIENTS=("noreply@company.com", "admin@example.com", "team@example.com")';
      EXECUTE IMMEDIATE :my_package_Send_Mail_Task_integration_sql;
   END;
   !!!RESOLVE EWI!!! /*** SSC-EWI-SSIS0015 - SNOWFLAKE'S SYSTEM$SEND_EMAIL DOES NOT SUPPORT FILE ATTACHMENTS. CONSIDER USING STAGED FILES WITH SHARED LINKS OR ALTERNATIVE DELIVERY METHODS. ***/!!!
   !!!RESOLVE EWI!!! /*** SSC-EWI-SSIS0016 - EMAIL PRIORITY SETTINGS (HIGH/NORMAL/LOW) ARE NOT SUPPORTED BY SYSTEM$SEND_EMAIL AND WILL BE IGNORED. ***/!!!
   --** SSC-FDM-SSIS0008 - SNOWFLAKE'S EMAIL INTEGRATION USES A FIXED SENDER ADDRESS. THE ORIGINAL FROM ADDRESS HAS BEEN PREPENDED TO THE MESSAGE BODY FOR REFERENCE. **
   --** SSC-FDM-SSIS0009 - SNOWFLAKE'S SYSTEM$SEND_EMAIL DOES NOT SUPPORT CC ADDRESSING. ALL CC RECIPIENTS HAVE BEEN ADDED TO THE MAIN RECIPIENTS LIST. **
   CALL SYSTEM$SEND_EMAIL('my_package_Send_Mail_Task', 'noreply@company.com,admin@example.com,team@example.com', 'Monthly Report', 'Email sent by: noreply@company.com

Please review the attached monthly report.');
END;

-- Upload file to stage
PUT file://report.pdf @my_stage;

-- Get shareable link (valid for 1 hour)
LET file_url STRING := GET_PRESIGNED_URL(@my_stage, 'report.pdf', 3600);

-- Include link in email body
CALL SYSTEM$SEND_EMAIL('my_integration', 'admin@example.com', 'Report Available', 
  'Download the report from: ' || :file_url);

-- Send to main recipients
CALL SYSTEM$SEND_EMAIL('my_integration', 'admin@example.com', 'Subject', 'Message');

-- Send separately to BCC recipients
CALL SYSTEM$SEND_EMAIL('my_integration', 'audit@example.com', 'Subject', 'Message');

CREATE OR REPLACE TASK public.package_bulk_insert_task
WAREHOUSE=DUMMY_WAREHOUSE
AFTER public.package
AS
BEGIN
   ---- Start block 'Package\BulkInsertTask'
   !!!RESOLVE EWI!!! /*** SSC-EWI-SSIS0024 - THE STAGE AND FILE UPLOAD ARE NOT INCLUDED IN THE TRANSLATION. CREATE A SNOWFLAKE STAGE AND UPLOAD THE SOURCE FILE BEFORE EXECUTING THE COPY INTO STATEMENT. REPLACE {STAGE_PLACEHOLDER} WITH YOUR STAGE NAME. ***/!!!
   COPY INTO target_table
   FROM '@{STAGE_PLACEHOLDER}'
   PATTERN = '.*data_file.*'
   FILE_FORMAT = (TYPE = 'CSV', FIELD_DELIMITER = ',', RECORD_DELIMITER = '\r\n', SKIP_HEADER = 1, NULL_IF = ('', 'NULL', 'null'), ERROR_ON_COLUMN_COUNT_MISMATCH = FALSE)
   ON_ERROR = CONTINUE;
   ---- End block 'Package\BulkInsertTask'
END;

CREATE OR REPLACE TASK public.package_load_customers
WAREHOUSE=DUMMY_WAREHOUSE
AS
BEGIN
   !!!RESOLVE EWI!!! /*** SSC-EWI-SSIS0024 - THE STAGE AND FILE UPLOAD ARE NOT INCLUDED IN THE TRANSLATION. CREATE A SNOWFLAKE STAGE AND UPLOAD THE SOURCE FILE BEFORE EXECUTING THE COPY INTO STATEMENT. REPLACE {STAGE_PLACEHOLDER} WITH YOUR STAGE NAME. ***/!!!
   COPY INTO Customers
   FROM '@{STAGE_PLACEHOLDER}'
   PATTERN = '.*customers\.csv.*'
   FILE_FORMAT = (TYPE = 'CSV', FIELD_DELIMITER = ',', RECORD_DELIMITER = '\r\n', SKIP_HEADER = 0, NULL_IF = ('', 'NULL', 'null'), ERROR_ON_COLUMN_COUNT_MISMATCH = FALSE)
   ON_ERROR = CONTINUE;
END;

CREATE OR REPLACE TASK public.package_load_products
WAREHOUSE=DUMMY_WAREHOUSE
AS
BEGIN
   !!!RESOLVE EWI!!! /*** SSC-EWI-SSIS0024 - THE STAGE AND FILE UPLOAD ARE NOT INCLUDED IN THE TRANSLATION. CREATE A SNOWFLAKE STAGE AND UPLOAD THE SOURCE FILE BEFORE EXECUTING THE COPY INTO STATEMENT. REPLACE {STAGE_PLACEHOLDER} WITH YOUR STAGE NAME. ***/!!!
   COPY INTO Products
   FROM '@{STAGE_PLACEHOLDER}'
   PATTERN = '.*products\.txt.*'
   FILE_FORMAT = (TYPE = 'CSV', FIELD_DELIMITER = '\t', RECORD_DELIMITER = '\n', SKIP_HEADER = 1, NULL_IF = ('', 'NULL', 'null'), ERROR_ON_COLUMN_COUNT_MISMATCH = FALSE)
   ON_ERROR = SKIP_FILE_10;
END;

CREATE OR REPLACE TASK public.package_load_orders
WAREHOUSE=DUMMY_WAREHOUSE
AS
BEGIN
   !!!RESOLVE EWI!!! /*** SSC-EWI-SSIS0020 - SSIS BULKINSERTTASK NATIVE OR WIDENATIVE DATA FILE TYPE IS NOT SUPPORTED IN SNOWFLAKE. EXPORT SOURCE DATA TO CSV FORMAT BEFORE MIGRATION. ***/!!!
   !!!RESOLVE EWI!!! /*** SSC-EWI-SSIS0021 - SSIS BULKINSERTTASK LASTROW OPTION IS NOT SUPPORTED IN SNOWFLAKE. USE TEMPORARY TABLE WITH ROW_NUMBER AND LIMIT/OFFSET TO SELECT SPECIFIC ROW RANGE. ***/!!!
   !!!RESOLVE EWI!!! /*** SSC-EWI-SSIS0022 - SSIS BULKINSERTTASK FIRETRIGGERS OPTION IS NOT SUPPORTED IN SNOWFLAKE. CONSIDER USING SNOWFLAKE STREAMS AND TASKS TO IMPLEMENT TRIGGER-LIKE BEHAVIOR. ***/!!!
   !!!RESOLVE EWI!!! /*** SSC-EWI-SSIS0024 - THE STAGE AND FILE UPLOAD ARE NOT INCLUDED IN THE TRANSLATION. CREATE A SNOWFLAKE STAGE AND UPLOAD THE SOURCE FILE BEFORE EXECUTING THE COPY INTO STATEMENT. REPLACE {STAGE_PLACEHOLDER} WITH YOUR STAGE NAME. ***/!!!
   COPY INTO Orders
   FROM '@{STAGE_PLACEHOLDER}'
   PATTERN = '.*orders\.dat.*'
   FILE_FORMAT = (TYPE = 'CSV', FIELD_DELIMITER = ',', RECORD_DELIMITER = '\r\n', SKIP_HEADER = 0, NULL_IF = ('', 'NULL', 'null'), ERROR_ON_COLUMN_COUNT_MISMATCH = FALSE)
   ON_ERROR = CONTINUE;
END;

CREATE OR REPLACE STAGE my_bulk_stage;

-- Change this:
FROM '@{STAGE_PLACEHOLDER}'

-- To this:
FROM '@my_bulk_stage'

LIST @my_bulk_stage;

-- Load all data
COPY INTO staging_table FROM '@my_stage' ...;

-- Insert only rows up to LastRow value
INSERT INTO target_table
SELECT * FROM (
  SELECT *, ROW_NUMBER() OVER (ORDER BY 1) AS rn
  FROM staging_table
) WHERE rn <= 1000;  -- Original LastRow value

-- Create stream to capture inserts
CREATE OR REPLACE STREAM target_stream ON TABLE target_table;

-- Create task to process inserts (trigger logic)
CREATE OR REPLACE TASK process_inserts
  WAREHOUSE = my_warehouse
  SCHEDULE = '1 minute'
  WHEN SYSTEM$STREAM_HAS_DATA('target_stream')
AS
  INSERT INTO audit_table 
  SELECT *, CURRENT_TIMESTAMP() 
  FROM target_stream 
  WHERE METADATA$ACTION = 'INSERT';