Simplify intermediate conversation model logic

CHANGELOG.md

@@ -1,3 +1,24 @@
+# dbt_intercom v1.1.0-a2
+[PR #68](https://github.com/fivetran/dbt_intercom/pull/67) is a pre-release that includes the following updates:
+
+## Bug Fixes
+- Consolidated CTEs and joins to reduce the compute load of `int_intercom__conversation_part_events` for customers with large `intercom__conversation_part_history` source tables. 
+- Corrected `last_close_by_author_id` logic that was incorrectly referencing` first_close_by_author_id` values.
+
+## Under the Hood
+- Created consistency test for `intercom__conversation_enhanced` to validate that the above changes do not impact end model values. 
+
+# dbt_intercom v1.1.0-a1
+[PR #67](https://github.com/fivetran/dbt_intercom/pull/67) is a pre-release that includes the following updates:
+
+## Schema Updates (`--full-refresh` encouraged after upgrading)
+
+| Data Model                                                                                                                                               | Change Type | Old Behavior                     | New Behavior                                             | Notes                                                                                    |
+| -------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------- | ---------------------------- | ---------------------------------------------------- | ---------------------------------------------------------------------------------------- |
+| `int_intercom__conversation_part_events`             | Updated materialization  |  Ephemeral |   Table    |  Reduces compute load of `intercom__conversation_enhanced` to prevent timeout run issues.  |
+| `int_intercom__conversation_string_aggregates`                                | Updated materialization | Ephemeral  |  Table   |    Reduces compute load of `intercom__conversation_enhanced` to prevent timeout run issues.  |
+| `int_intercom__conversation_part_aggregates`                                | Updated materialization | Ephemeral  |  Table   |    Reduces compute load of `intercom__conversation_metrics` to prevent timeout run issues.  |
+
 # dbt_intercom v1.0.0
 
 [PR #66](https://github.com/fivetran/dbt_intercom/pull/66) includes the following updates:

README.md

@@ -37,7 +37,7 @@ The following table provides a detailed list of all tables materialized within t
 | [intercom__conversation_metrics](https://github.com/fivetran/dbt_intercom/blob/main/models/intercom__conversation_metrics.sql)                                 | Each record represents a single row from `intercom__conversation_enhanced`, enriched with data like time to first response, time to first close, and time to last close. |
 
 ### Materialized Models
-Each Quickstart transformation job run materializes 32 models if all components of this data model are enabled. This count includes all staging, intermediate, and final models materialized as `view`, `table`, or `incremental`.
+Each Quickstart transformation job run materializes 35 models if all components of this data model are enabled. This count includes all staging, intermediate, and final models materialized as `view`, `table`, or `incremental`.
 <!--section-end-->
 
 ## How do I use the dbt package?
@@ -54,7 +54,7 @@ Include the following intercom package version in your `packages.yml` file:
 ```yaml
 packages:
   - package: fivetran/intercom
-    version: [">=1.0.0", "<1.1.0"]
+    version: 1.1.0-a2
 ```
 ### Step 3: Define database and schema variables
 By default, this package runs using your destination and the `intercom` schema. If this is not where your Intercom data is (for example, if your Intercom schema is named `intercom_fivetran`), add the following configuration to your root `dbt_project.yml` file:

dbt_project.yml

@@ -1,13 +1,13 @@
 config-version: 2
 name: 'intercom'
-version: '1.0.0'
+version: '1.1.0'
 require-dbt-version: [">=1.3.0", "<2.0.0"]
 models:
   intercom:
     +schema: intercom
     +materialized: table
     intermediate:
-      +materialized: ephemeral
+      +materialized: table
     staging:
       +schema: stg_intercom
       +materialized: table

integration_tests/ci/sample.profiles.yml

@@ -12,13 +12,13 @@ integration_tests:
       pass: "{{ env_var('CI_REDSHIFT_DBT_PASS') }}"
       dbname: "{{ env_var('CI_REDSHIFT_DBT_DBNAME') }}"
       port: 5439
-      schema: intercom_integration_tests
+      schema: intercom_integration_tests_2
       threads: 8
     bigquery:
       type: bigquery
       method: service-account-json
       project: 'dbt-package-testing'
-      schema: intercom_integration_tests
+      schema: intercom_integration_tests_2
       threads: 8
       keyfile_json: "{{ env_var('GCLOUD_SERVICE_KEY') | as_native }}"
     snowflake:
@@ -29,7 +29,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: intercom_integration_tests
+      schema: intercom_integration_tests_2
       threads: 8
     postgres:
       type: postgres
@@ -38,13 +38,13 @@ integration_tests:
       pass: "{{ env_var('CI_POSTGRES_DBT_PASS') }}"
       dbname: "{{ env_var('CI_POSTGRES_DBT_DBNAME') }}"
       port: 5432
-      schema: intercom_integration_tests
+      schema: intercom_integration_tests_2
       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: intercom_integration_tests
+      schema: intercom_integration_tests_2
       threads: 8
       token: "{{ env_var('CI_DATABRICKS_DBT_TOKEN') }}"
       type: databricks

integration_tests/dbt_project.yml

@@ -1,13 +1,13 @@
 name: 'intercom_integration_tests'
-version: '1.0.0'
+version: '1.1.0'
 profile: 'integration_tests'
 config-version: 2
 
 models:
   +schema: "intercom_{{ var('directed_schema','dev') }}"
 
 vars:
-  intercom_schema: intercom_integration_tests
+  intercom_schema: intercom_integration_tests_2
   intercom:
     intercom_admin_identifier: "admin_data"
     intercom_company_history_identifier: "company_history_data"

integration_tests/packages.yml

@@ -1,2 +1,2 @@
 packages:
-    - local: ../
+  - local: ../

integration_tests/requirements.txt

@@ -4,5 +4,5 @@ dbt-redshift>=1.3.0,<2.0.0
 dbt-postgres>=1.3.0,<2.0.0
 dbt-spark>=1.3.0,<2.0.0
 dbt-spark[PyHive]>=1.3.0,<2.0.0
-dbt-databricks>=1.6.0,<2.0.0
+dbt-databricks>=1.3.0,<1.10.10
 certifi==2025.1.31

integration_tests/tests/consistency/consistency_events.sql

@@ -0,0 +1,46 @@
+{{ config(
+    tags="fivetran_validations",
+    enabled=(var('fivetran_validation_tests_enabled', false))
+) }}
+
+-- this test ensures the contacts end model matches the prior version
+with prod as (
+    select *
+    from {{ target.schema }}_intercom_prod.intercom__conversation_enhanced
+),
+
+dev as (
+    select *
+    from {{ target.schema }}_intercom_dev.intercom__conversation_enhanced
+),
+
+prod_not_in_dev as (
+    -- rows from prod not found in dev
+    select * from prod
+    except distinct
+    select * from dev
+),
+
+dev_not_in_prod as (
+    -- rows from dev not found in prod
+    select * from dev
+    except distinct
+    select * from prod
+),
+
+final as (
+    select
+        *,
+        'from prod' as source
+    from prod_not_in_dev
+
+    union all -- union since we only care if rows are produced
+
+    select
+        *,
+        'from dev' as source
+    from dev_not_in_prod
+)
+
+select *
+from final

models/intermediate/int_intercom__conversation_part_events.sql

@@ -1,90 +1,73 @@
 with conversation_part_history as (
+
   select *
   from {{ ref('stg_intercom__conversation_part_history') }}
 ),
 
---Obtains the first and last values for conversations where the part type was close and part was authored by an admin.
-conversation_admin_events as (
-  select
-    conversation_id,
-    first_value(author_id) over (partition by conversation_id order by created_at asc, conversation_id rows unbounded preceding) as first_close_by_admin_id,
-    first_value(author_id) over (partition by conversation_id order by created_at desc, conversation_id rows unbounded preceding) as last_close_by_admin_id,
-    first_value(created_at) over (partition by conversation_id order by created_at asc, conversation_id rows unbounded preceding) as first_admin_close_at,
-    first_value(created_at) over (partition by conversation_id order by created_at desc, conversation_id rows unbounded preceding) as last_admin_close_at
-  from conversation_part_history
-
-  where part_type = 'close' and author_type = 'admin'
-
-), 
+conversation_events as (
 
-conversation_all_close_events as (
-
-  select
+  select distinct
     conversation_id,
-    first_value(author_id) over (partition by conversation_id order by created_at asc, conversation_id rows unbounded preceding) as first_close_by_author_id,
-    first_value(author_id) over (partition by conversation_id order by created_at desc, conversation_id rows unbounded preceding) as last_close_by_author_id,
-    first_value(created_at) over (partition by conversation_id order by created_at asc, conversation_id rows unbounded preceding) as first_close_at,
-    first_value(created_at) over (partition by conversation_id order by created_at desc, conversation_id rows unbounded preceding) as last_close_at
-  from conversation_part_history
-
-  where part_type = 'close'
-
-),
-
---Obtains the first and last values for conversations where the part type was authored by a contact (which is either a user or lead).
-conversation_contact_events as ( 
-  select
-    conversation_id,
-    first_value(author_id) over (partition by conversation_id order by created_at asc, conversation_id rows unbounded preceding) as first_contact_author_id,
-    first_value(author_id) over (partition by conversation_id order by created_at desc, conversation_id rows unbounded preceding) as last_contact_author_id
-  from conversation_part_history
-
-  where author_type in ('user','lead')
-
-), 
-
---Obtains the first and last values for conversations where the part type was authored by a team
-conversation_team_events as (
-  select
-    conversation_id,
-    first_value(assigned_to_id) over (partition by conversation_id order by created_at asc, conversation_id rows unbounded preceding) as first_team_id,
-    first_value(assigned_to_id) over (partition by conversation_id order by created_at desc, conversation_id rows unbounded preceding) as last_team_id
-  from conversation_part_history
-
-  where assigned_to_type = 'team'
+    --Obtains the first and last values for conversations where the part type was close and part was authored by an admin.
+    first_value(case when part_type = 'close' and author_type = 'admin' then author_id end)
+      over (partition by conversation_id order by case when part_type = 'close' and author_type = 'admin' then 0 else 1 end,
+        created_at asc rows between unbounded preceding and unbounded following) as first_close_by_admin_id,
+    first_value(case when part_type = 'close' and author_type = 'admin' then author_id end)
+      over (partition by conversation_id order by case when part_type = 'close' and author_type = 'admin' then 0 else 1 end,
+        created_at desc rows between unbounded preceding and unbounded following) as last_close_by_admin_id,
+    first_value(case when part_type = 'close' and author_type = 'admin' then created_at end)
+      over (partition by conversation_id order by case when part_type = 'close' and author_type = 'admin' then 0 else 1 end,
+        created_at asc rows between unbounded preceding and unbounded following) as first_admin_close_at,
+    first_value(case when part_type = 'close' and author_type = 'admin' then created_at end)
+      over (partition by conversation_id order by case when part_type = 'close' and author_type = 'admin' then 0 else 1 end,
+        created_at desc rows between unbounded preceding and unbounded following) as last_admin_close_at,           
+    --Obtains the first and last values for conversations where the part type was close.
+    first_value(case when part_type = 'close' then author_id end)
+      over (partition by conversation_id  order by case when part_type = 'close' then 0 else 1 end, 
+        created_at asc rows between unbounded preceding and unbounded following) as first_close_by_author_id,
+    first_value(case when part_type = 'close' then author_id end)
+      over (partition by conversation_id order by case when part_type = 'close' then 0 else 1 end, 
+        created_at desc rows between unbounded preceding and unbounded following) as last_close_by_author_id,
+    first_value(case when part_type = 'close' then created_at end)
+      over (partition by conversation_id order by case when part_type = 'close' then 0 else 1 end, 
+        created_at asc rows between unbounded preceding and unbounded following) as first_close_at,
+    first_value(case when part_type = 'close' then created_at end)
+      over (partition by conversation_id order by case when part_type = 'close' then 0 else 1 end, 
+        created_at desc rows between unbounded preceding and unbounded following) as last_close_at,   
+    --Obtains the first and last values for conversations where the part type was authored by a contact (which is either a user or lead).
+    first_value(case when author_type in ('user','lead') then author_id end)
+      over (partition by conversation_id order by case when author_type in ('user','lead') then 0 else 1 end, 
+        created_at asc rows between unbounded preceding and unbounded following) as first_contact_author_id,
+    first_value(case when author_type in ('user','lead') then author_id end)
+      over (partition by conversation_id order by case when author_type in ('user','lead') then 0 else 1 end,
+        created_at desc rows between unbounded preceding and unbounded following) as last_contact_author_id,
+    --Obtains the first and last values for conversations where the part type was authored by a team
+    first_value(case when assigned_to_type = 'team' then assigned_to_id end)
+      over (partition by conversation_id order by case when assigned_to_type = 'team' then 0 else 1 end, 
+        created_at asc rows between unbounded preceding and unbounded following) as first_team_id,
+    first_value(case when assigned_to_type = 'team' then assigned_to_id end)
+      over (partition by conversation_id order by case when assigned_to_type = 'team' then 0 else 1 end, 
+        created_at desc rows between unbounded preceding and unbounded following) as last_team_id  
+    from conversation_part_history
 ),
 
---Joins the above two CTEs with conversation part history. Distinct was necessary to ensure only one first/last value was returned for each individual conversation.
 final as (
-    select distinct
-        conversation_part_history.conversation_id,
-        cast(conversation_admin_events.first_close_by_admin_id as {{ dbt.type_string() }}) as first_close_by_admin_id,
-        cast(conversation_admin_events.last_close_by_admin_id as {{ dbt.type_string() }}) as last_close_by_admin_id,
-        cast(conversation_all_close_events.first_close_by_author_id as {{ dbt.type_string() }}) as first_close_by_author_id,
-        cast(conversation_all_close_events.first_close_by_author_id as {{ dbt.type_string() }}) as last_close_by_author_id,
-        cast(conversation_contact_events.first_contact_author_id as {{ dbt.type_string() }}) as first_contact_author_id,
-        cast(conversation_contact_events.last_contact_author_id as {{ dbt.type_string() }}) as last_contact_author_id,
-        cast(conversation_team_events.first_team_id as {{ dbt.type_string() }}) as first_team_id,
-        cast(conversation_team_events.last_team_id as {{ dbt.type_string() }}) as last_team_id,
-        conversation_admin_events.first_admin_close_at,
-        conversation_admin_events.last_admin_close_at,
-        conversation_all_close_events.first_close_at,
-        conversation_all_close_events.last_close_at
-
-    from conversation_part_history
-
-    left join conversation_admin_events
-        on conversation_admin_events.conversation_id = conversation_part_history.conversation_id
-
-    left join conversation_all_close_events
-        on conversation_all_close_events.conversation_id = conversation_part_history.conversation_id
-
-    left join conversation_contact_events
-        on conversation_contact_events.conversation_id = conversation_part_history.conversation_id
-
-    left join conversation_team_events
-        on conversation_team_events.conversation_id = conversation_part_history.conversation_id
+    select
+        conversation_id,
+        cast(first_close_by_admin_id as {{ dbt.type_string() }}) as first_close_by_admin_id,
+        cast(last_close_by_admin_id as {{ dbt.type_string() }}) as last_close_by_admin_id,
+        cast(first_close_by_author_id as {{ dbt.type_string() }}) as first_close_by_author_id,
+        cast(last_close_by_author_id as {{ dbt.type_string() }}) as last_close_by_author_id,
+        cast(first_contact_author_id as {{ dbt.type_string() }}) as first_contact_author_id,
+        cast(last_contact_author_id as {{ dbt.type_string() }}) as last_contact_author_id,
+        cast(first_team_id as {{ dbt.type_string() }}) as first_team_id,
+        cast(last_team_id as {{ dbt.type_string() }}) as last_team_id,
+        first_admin_close_at,
+        last_admin_close_at,
+        first_close_at,
+        last_close_at
+    from conversation_events
 )
 
 select *
-from final
+from final