fix for events incremental

CHANGELOG.md

@@ -1,3 +1,16 @@
+# dbt_iterable v0.13.0
+[PR #51](https://github.com/fivetran/dbt_iterable/pull/51) includes the following updates:
+
+## Breaking Changes
+- Adds a field called `first_open_or_click_event_at` in the `iterable__user_campaign` model. This timestamp shows the first time a user interacted with a campaign, recording the earliest occurring event out of 'emailOpen', 'emailClick', and 'pushOpen'. ([PR #50](https://github.com/fivetran/dbt_iterable/pull/50))
+- Updates the incremental logic in `iterable__events` to use the `created_on` date field instead of the `created_at` timestamp. Previously, this would potentially exclude late-arriving new records from populating in the end models if they had an older `created_at` value than what was present in the model. Switching to `created_on` widens the criteria. Additionally, we've updated the unique key used in the incremental strategy to `created_at` in addition to `unique_event_id`. Previously this could incorrectly exclude records with the same unique_event_id occuring at different times. We recommend running a `--full-refresh`.
+
+## Under the Hood
+- In addition to using `created_on` in the incremental logic in `iterable__events`, we introduced a `lookback_window` to increase the window for accommodating potential late-arriving records. The default is 7 days prior to the maximum created_on value present in the `iterable__events` model, but you may customize this by setting the var `lookback_window` in your dbt_project.yml. See the [Lookback Window section of the README](https://github.com/fivetran/dbt_iterable/blob/main/README.md#lookback-window) for more details.
+
+## Contributors
+- [@justin-fundrise](https://github.com/justin-fundrise) ([Issue #48](https://github.com/fivetran/dbt_iterable/issues/48), [PR #49](https://github.com/fivetran/dbt_iterable/pull/49), [PR #50](https://github.com/fivetran/dbt_iterable/pull/50))
+
 # dbt_iterable v0.12.0
 [PR #44](https://github.com/fivetran/dbt_iterable/pull/44) includes the following updates:
 

README.md

@@ -76,7 +76,7 @@ Include the following Iterable package version in your `packages.yml` file.
 ```yaml
 packages:
   - package: fivetran/iterable
-    version: [">=0.12.0", "<0.13.0"]
+    version: [">=0.13.0", "<0.14.0"]
 ```
 ### Step 3: Define database and schema variables
 By default, this package runs using your destination and the `iterable` schema of your [target database](https://docs.getdbt.com/docs/running-a-dbt-project/using-the-command-line-interface/configure-your-profile). If this is not where your Iterable data is located (for example, if your Iterable schema is named `iterable_fivetran`), add the following configuration to your root `dbt_project.yml` file:
@@ -158,6 +158,28 @@ If an individual source table has a different name than what the package expects
 vars:
     iterable_<default_source_table_name>_identifier: "your_table_name"
 ```
+
+#### Pivoting out event metrics 
+In the `iterable__user_campaign` model, there are metrics calculated based on Iterable events. By default, all event metrics are enabled as shown in the [`dbt_project.yml`](https://github.com/fivetran/dbt_iterable/blob/2c0c1764f55af255726397a374b48004de20cf51/dbt_project.yml#L34). If not all metrics apply to your use case, you can specify which event metrics to include by configuring the variable `iterable__event_metrics` in your own `dbt_project.yml` as shown below.  
+
+```yml
+vars:
+  iterable__event_metrics:
+    - "emailSend"
+    - "emailOpen"
+```
+
+#### Lookback Window
+Records from the source can sometimes arrive late. Since several of the models in this package are incremental, by default we look back 7 days to ensure late arrivals are captured while avoiding the need for frequent full refreshes. While the frequency can be reduced, we still recommend running `dbt --full-refresh` periodically to maintain data quality of the models.
+
+To change the default lookback window, add the following variable to your `dbt_project.yml` file:
+
+```yml
+vars:
+  iterable:
+    lookback_window: number_of_days # default is 7
+```
+
 #### Deprecated `CAMPAIGN_SUPRESSION_LIST_HISTORY` table
 
 The Iterable connector schema originally misspelled the `CAMPAIGN_SUPPRESSION_LIST_HISTORY` table as `CAMPAIGN_SUPRESSION_LIST_HISTORY` (note the singular `P`). As of August 2021, Fivetran has deprecated the misspelled table and will only continue syncing the correctly named `CAMPAIGN_SUPPRESSION_LIST_HISTORY` table.
@@ -170,11 +192,8 @@ vars:
 ```
 
 ### (Optional) Step 6: Orchestrate your models with Fivetran Transformations for dbt Core™
-<details><summary>Expand for details</summary>
-<br>
-
+
 Fivetran offers the ability for you to orchestrate your dbt project through [Fivetran Transformations for dbt Core™](https://fivetran.com/docs/transformations/dbt). Learn how to set up your project for orchestration through Fivetran in our [Transformations for dbt Core setup guides](https://fivetran.com/docs/transformations/dbt#setupguide).
-</details>
 
 ## Does this package have dependencies?
 This dbt package is dependent on the following dbt packages. These dependencies are installed by default within this package. For more information on the following packages, refer to the [dbt hub](https://hub.getdbt.com/) site.
@@ -189,7 +208,7 @@ packages:
       version: [">=1.0.0", "<2.0.0"]
 
     - package: fivetran/iterable_source
-      version: [">=0.9.0", "<0.10.0"]
+      version: [">=0.10.0", "<0.11.0"]
 ```
 
 ## How is this package maintained and can I contribute?

dbt_project.yml

@@ -1,5 +1,6 @@
 name: 'iterable'
-version: '0.12.0'
+
+version: '0.13.0'
 config-version: 2
 require-dbt-version: [">=1.3.0", "<2.0.0"]
 models:

integration_tests/ci/sample.profiles.yml

@@ -16,13 +16,13 @@ integration_tests:
       pass: "{{ env_var('CI_REDSHIFT_DBT_PASS') }}"
       dbname: "{{ env_var('CI_REDSHIFT_DBT_DBNAME') }}"
       port: 5439
-      schema: iterable_integration_tests_13
+      schema: iterable_integration_tests_16
       threads: 8
     bigquery:
       type: bigquery
       method: service-account-json
       project: 'dbt-package-testing'
-      schema: iterable_integration_tests_13
+      schema: iterable_integration_tests_16
       threads: 8
       keyfile_json: "{{ env_var('GCLOUD_SERVICE_KEY') | as_native }}"
     snowflake:
@@ -33,7 +33,7 @@ integration_tests:
       role: "{{ env_var('CI_SNOWFLAKE_DBT_ROLE') }}"
       database: "{{ env_var('CI_SNOWFLAKE_DBT_DATABASE') }}"
       warehouse: "{{ env_var('CI_SNOWFLAKE_DBT_WAREHOUSE') }}"
-      schema: iterable_integration_tests_13
+      schema: iterable_integration_tests_16
       threads: 8
     postgres:
       type: postgres
@@ -42,13 +42,13 @@ integration_tests:
       pass: "{{ env_var('CI_POSTGRES_DBT_PASS') }}"
       dbname: "{{ env_var('CI_POSTGRES_DBT_DBNAME') }}"
       port: 5432
-      schema: iterable_integration_tests_13
+      schema: iterable_integration_tests_16
       threads: 8
     databricks:
       catalog: "{{ env_var('CI_DATABRICKS_DBT_CATALOG') }}"
       host: "{{ env_var('CI_DATABRICKS_DBT_HOST') }}"
       http_path: "{{ env_var('CI_DATABRICKS_DBT_HTTP_PATH') }}"
-      schema: iterable_integration_tests_13
+      schema: iterable_integration_tests_16
       threads: 8
       token: "{{ env_var('CI_DATABRICKS_DBT_TOKEN') }}"
       type: databricks

integration_tests/dbt_project.yml

@@ -1,10 +1,10 @@
 config-version: 2
 name: 'iterable_integration_tests'
-version: '0.12.0'
+version: '0.13.0'
 profile: 'integration_tests'
 vars:
   iterable_source:
-    iterable_schema: iterable_integration_tests_13
+    iterable_schema: iterable_integration_tests_16
     iterable_campaign_history_identifier: "campaign_history_data"
     iterable_campaign_label_history_identifier: "campaign_label_history_data"
     iterable_campaign_list_history_identifier: "campaign_list_history_data"
@@ -20,6 +20,7 @@ vars:
     iterable_user_unsubscribed_message_type_identifier: "user_unsubscribed_message_type_data"
     iterable_user_unsubscribed_channel_history_identifier: "user_unsubscribed_channel_history_data"
     iterable_user_unsubscribed_message_type_history_identifier: "user_unsubscribed_message_type_history_data"
+  # lookback_window: 8
 
 models:
   +schema: "iterable_{{ var('directed_schema','dev') }}"

integration_tests/seeds/event_data.csv

@@ -3,4 +3,5 @@ HFNB4mJWJeXKk7AwJAEPCCmvpkM=,nobody@gmail.co.uk,,,,2020-06-12 14:42:41.000,email
 8nxrQcLoFB96e+oOBqzBDsXnMBU=,myson@gmail.com,,,,2020-03-26 12:30:43.000,emailSubscribe,,,,,,,,,null,"{""channelIds"": ""[]"",""emailListIds"": ""[501079]"",""profileUpdatedAt"": ""2020-03-26 12:30:42 +00:00"",""signupSource"": ""Import""}",2020-08-04 14:10:39.711,12345
 8nxrQcLoFB96e+oOBqzBDsXnMBU=,myson@gmail.com,,,,2020-06-12 18:30:43.000,newEvent,,,,,,,,,null,"{""channelIds"": ""[]"",""emailListIds"": ""[501079,101]"",""profileUpdatedAt"": ""2020-04-21 12:30:42 +00:00"",""signupSource"": ""Import""}",2020-08-07 02:10:19.243,12345
 USA101,nobody@gmail.co.uk,,,,2020-06-12 18:30:43.000,newEvent,,,,,,,,,null,,2020-08-07 02:10:19.243,987654
-8nxrQcLoFB96e+oOBqzBDsXnMBU=,myson@gmail.com,,,,2020-07-12 18:30:43.000,newEvent,,,,,,,,,null,"{""channelIds"": ""[]"",""emailListIds"": ""[501079,101]"",""profileUpdatedAt"": ""2020-04-21 12:30:42 +00:00"",""signupSource"": ""Import""}",2020-08-07 02:10:19.243,12345
+8nxrQcLoFB96e+oOBqzBDsXnMBU=,myson@gmail.com,,,,2020-07-12 18:30:43.000,newEvent,,,,,,,,,null,"{""channelIds"": ""[]"",""emailListIds"": ""[501079,101]"",""profileUpdatedAt"": ""2020-04-21 12:30:42 +00:00"",""signupSource"": ""Import""}",2020-08-07 02:10:19.243,12345
+djjfhdjkdagS3r5fd,abc@gmail.com,,,,2020-07-12 22:30:43.000,newEvent,,,,,,,,,null,"{""channelIds"": ""[]"",""emailListIds"": ""[501079,101]"",""profileUpdatedAt"": ""2020-04-21 12:30:42 +00:00"",""signupSource"": ""Import""}",2020-08-07 02:10:19.243,5678

macros/iterable_lookback.sql

@@ -0,0 +1,18 @@
+{% macro iterable_lookback(from_date, datepart, interval, safety_date='2020-01-01') %}
+
+{{ adapter.dispatch('iterable_lookback', 'iterable') (from_date, datepart, interval, safety_date='2020-01-01') }}
+
+{%- endmacro %}
+
+{% macro default__iterable_lookback(from_date, datepart, interval, safety_date='2020-01-01')  %}
+
+    {% set sql_statement %}
+        select coalesce({{ from_date }}, {{ "'" ~ safety_date ~ "'" }})
+        from {{ this }}
+    {%- endset -%}
+
+    {%- set result = dbt_utils.get_single_value(sql_statement) %}
+
+    {{ dbt.dateadd(datepart=datepart, interval=-interval, from_date_or_timestamp="cast('" ~ result ~ "' as date)") }}
+
+{% endmacro %}

models/intermediate/int_iterable__campaign_event_metrics.sql

@@ -15,7 +15,7 @@ with user_campaign as (
         , count(distinct unique_user_key) as count_unique_users
 
         {% for col in user_campaign_columns %}
-            {% if col.name|lower not in ['unique_user_key', 'user_id', '_fivetran_user_id', 'user_email', 'user_full_name', 'campaign_id', 'campaign_name', 'recurring_campaign_id', 'recurring_campaign_name', 'first_event_at', 'last_event_at', 'template_id', 'template_name', 'experiment_id', 'unique_user_campaign_id'] %}
+            {% if col.name|lower not in ['unique_user_key', 'user_id', '_fivetran_user_id', 'user_email', 'user_full_name', 'campaign_id', 'campaign_name', 'recurring_campaign_id', 'recurring_campaign_name', 'first_event_at', 'last_event_at', 'template_id', 'template_name', 'experiment_id', 'unique_user_campaign_id','first_open_or_click_event_at'] %}
                             , sum({{ col.name }}) as {{ col.name }}
                             , count(distinct case when {{ col.name }} > 0 then user_email else null end) as unique_{{ col.name }}
             {% endif %}

models/intermediate/int_iterable__user_event_metrics.sql

@@ -17,7 +17,7 @@ with user_campaign as (
         max(last_event_at) as last_event_at
 
         {% for col in user_campaign_columns %}
-            {% if col.name|lower not in ['unique_user_key', 'user_id', '_fivetran_user_id', 'user_email', 'user_full_name', 'campaign_id', 'campaign_name', 'recurring_campaign_id', 'recurring_campaign_name', 'first_event_at', 'last_event_at', 'template_id', 'template_name', 'experiment_id', 'unique_user_campaign_id'] %}
+            {% if col.name|lower not in ['unique_user_key', 'user_id', '_fivetran_user_id', 'user_email', 'user_full_name', 'campaign_id', 'campaign_name', 'recurring_campaign_id', 'recurring_campaign_name', 'first_event_at', 'last_event_at', 'template_id', 'template_name', 'experiment_id', 'unique_user_campaign_id','first_open_or_click_event_at'] %}
                             , sum({{ col.name }}) as {{ col.name }}
             {% endif %}
         {% endfor %}

models/iterable.yml

@@ -16,19 +16,19 @@ models:
       - name: campaign_name
         description: User defined name of the campaign.
       - name: updated_at
-        description: Last update timestamp as epoch time in milliseconds
+        description: Last update timestamp
       - name: campaign_state
         description: State of the campaign. Can be Draft, Ready, Scheduled, Running, Finished, Starting, Aborted or Recurring
       - name: campaign_type
         description: The campaign type. Can be Blast or Triggered
       - name: send_size
         description: Size of the campaign. Number of individuals included in the campaign
       - name: start_at
-        description: Start timestamp as epoch time in milliseconds
+        description: Start timestamp
       - name: ended_at
-        description: Ended timestamp as epoch time in milliseconds
+        description: Ended timestamp
       - name: created_at
-        description: Creation timestamp as epoch time in milliseconds
+        description: Creation timestamp
       - name: created_by_user_id
         description: Reference to the user who created the campaign
       - name: template_id
@@ -156,7 +156,7 @@ models:
       - dbt_utils.unique_combination_of_columns:
           combination_of_columns:
             - unique_event_id
-            - created_on
+            - created_at
     columns:
       - name: event_id
         description: Unique id of the event generated by Fivetran
@@ -173,7 +173,7 @@ models:
       - name: content_id
         description: Reference to the content the event is associated with
       - name: created_at
-        description: Creation timestamp as epoch time in milliseconds
+        description: Creation timestamp
       - name: email
         description: The user's email. User identifier, for email-based projects. Previously was the unique identifier for user records, for the old Iterable schema before the August 2023 updates. 
       - name: additional_properties
@@ -279,7 +279,7 @@ models:
       - name: template_creator_user_id
         description: Reference to the user who created the template
       - name: created_on
-        description: Timestamp of when the event was created      
+        description: Day when the event was created      
       - name: city 
         description: The city associated with the event, which could be the location of the recipient.
       - name: clicked_url 
@@ -331,11 +331,11 @@ models:
       - name: user_id
         description:  User identifier, for userID-based projects.
       - name: user_signup_date
-        description: Timestamp as epoch time in milliseconds of when the user signed up
+        description: Timestamp of when the user signed up
       - name: user_signup_source
         description: Source where the user signed up
       - name: user_updated_at
-        description: Last update timestamp as epoch time in milliseconds
+        description: Last update timestamp
       - name: list_id
         description: The list associated with the user
       - name: is_current
@@ -422,6 +422,8 @@ models:
         description: Total in app click events associated with the user within the campaign/experiment
       - name: total_pushuninstall
         description: Total push uninstall events associated with the user within the campaign/experiment
+      - name: first_open_or_click_event_at
+        description: The time of the first open or click time. This is the first time a user interacts with a campaign, recording the earliest occurring event out of 'emailOpen', 'emailClick', and 'pushOpen'.
 
   - name: iterable__user_unsubscriptions
     description: Table that generates the total channel and message types a user has chosen to unsubscribe.
@@ -477,11 +479,11 @@ models:
       - name: user_id
         description:  User identifier, for userID-based projects.
       - name: signup_date
-        description: Timestamp as epoch time in milliseconds of when the user signed up
+        description: Timestamp of when the user signed up
       - name: signup_source
         description: Source where the user signed up
       - name: updated_at
-        description: Last update timestamp as epoch time in milliseconds
+        description: Last update timestamp
       - name: phone_number
         description: Phone number associated with the user
       - name: email_list_ids

models/iterable__events.sql

@@ -1,6 +1,6 @@
 {{ config(
         materialized='incremental',
-        unique_key=['unique_event_id'],
+        unique_key=['unique_event_id','created_at'],
         incremental_strategy='insert_overwrite' if target.type in ('bigquery', 'spark', 'databricks') else 'delete+insert',
         partition_by={"field": "created_on", "data_type": "date"} if target.type not in ('spark','databricks') else ['created_on'],
         file_format='parquet',
@@ -14,7 +14,10 @@ with events as (
     from {{ var('event') }}
 
     {% if is_incremental() %}
-    where created_at >= (select max(created_at) from {{ this }} )
+    where created_on >= {{ iterable.iterable_lookback(
+        from_date="max(created_on)",
+        datepart='day', 
+        interval=var('lookback_window', 7)) }}
     {% endif %}
 
 ), campaign as (

models/iterable__user_campaign.sql

@@ -28,7 +28,9 @@ with events as (
         recurring_campaign_name,
 
         min(created_at) as first_event_at,
-        max(created_at) as last_event_at
+        max(created_at) as last_event_at,
+        min(case when event_name in ('emailOpen', 'emailClick', 'pushOpen') then created_at end) as first_open_or_click_event_at
+
 
         -- count up the number of instances of each metric
         -- `iterable__event_metrics` is set by default to all events brought in by fivetran+iterable

packages.yml

@@ -1,3 +1,7 @@
 packages:
-  - package: fivetran/iterable_source
-    version: [">=0.9.0", "<0.10.0"]
+  # - package: fivetran/iterable_source
+  #   version: [">=0.10.0", "<0.11.0"]
+
+  - git: https://github.com/fivetran/dbt_iterable_source.git
+    revision: rm_src_freshness
+    warn-unpinned: false