Support changes to SFMC Push APIs

README.md

@@ -22,7 +22,7 @@ After creating an Omnata Push integration, you will be provided with External Fu
 
 packages:
   - git: "https://github.com/omnata-labs/dbt-omnata-push.git"
-    revision: 0.4.0
+    revision: 0.5.0
 
 ```
 

dbt_project.yml

@@ -1,5 +1,5 @@
 name: 'omnata_push'
-version: '0.3.0'
+version: '0.5.0'
 config-version: 2
 
 # This setting configures which "profile" dbt uses for this project.

integration_tests/dbt_project.yml

@@ -25,7 +25,7 @@ vars:
   full-refresh-marketing-cloud: false
   drop-omnata-task-tables: false
 
-# These functions are mocks for the Omnata Salesforce APIs, which are normally External Functions provided by the Omnata setup process
+# These functions are mocks for the Omnata Salesforce/SFMC APIs, which are normally External Functions provided by the Omnata setup process
 on-run-start:
   - "create or replace function SFDC_BULK_API_CREATE_JOB(operation varchar, table_name varchar, use_serial boolean, external_id_field_name varchar)
 returns variant
@@ -70,7 +70,19 @@ return {
   \"data_extension_existed\": true,
   \"success\": true
 }';"
-  - "create or replace function SFMC_DATA_EXTENSION_IMPORT(parameters variant,record variant)
+  - "create or replace function SFMC_STAGE_DATA(record variant)
+returns varchar
+language javascript
+strict
+as '
+return '''abcd'''' ';"
+  - "create or replace function SFMC_DE_IMPORT(configuration variant,stage_data_query_id varchar)
+returns varchar
+language javascript
+strict
+as '
+return '''abcd'''' ';"
+  - "create or replace function SFMC_FETCH_RESULTS(stage_data_query_id varchar,row_index int)
 returns variant
 language javascript
 strict

macros/apps/operations/marketing_cloud_data_extension_upload.sql

@@ -29,15 +29,36 @@
         with load_source as (
             {{ sql }}
         )
+        ,data_indexed as(
+        select row_number() over (partition by null order by null) as row_index,
+        (row_index/100)::int as batch,
+        record
+        from load_source
+        )
+        ,data_staged as(
+        select "{{ omnata_functions_database }}"."{{ omnata_functions_schema }}".SFMC_STAGE_DATA(array_agg(array_construct(row_index,record))) as staged_result
+        from data_indexed
+        group by batch
+        )
+        ,data_imported as(
+            select "{{ omnata_functions_database }}"."{{ omnata_functions_schema }}".SFMC_DE_IMPORT(
+                    PARSE_JSON('{"name":"{{ data_extension_name }}","operation":"{{ import_type }}"}'),
+                    any_value(staged_result)
+                ) as import_result
+            from data_staged
+        ),
+        data_import_result as(
+        select any_value(import_result) as import_result_output
+        from data_imported
+        )
         select '{{ job_id }}' as job_id,
                 UUID_STRING() as job_log_entry_id,
                 '{{ load_task_name }}',
                 '{{ data_extension_name }}',
                 '{{ operation }}',
-                load_source.record,
-                "{{ omnata_functions_database }}"."{{ omnata_functions_schema }}".SFMC_DATA_EXTENSION_IMPORT(PARSE_JSON('{"name":"{{ data_extension_name }}","operation":"{{ import_type }}"}'),
-                                           load_source.record) as result 
-        from load_source
+                record,
+                "{{ omnata_functions_database }}"."{{ omnata_functions_schema }}".SFMC_FETCH_RESULTS(import_result_output,row_index) as result
+        from data_indexed,data_import_result
     {%- endcall %}
 
     {{ adapter.commit() }}