Union capability

.gitignore

@@ -71,3 +71,5 @@ env/
 env.bak/
 venv/
 venv.bak/
+
+CLAUDE.md

CHANGELOG.md

@@ -1,3 +1,25 @@
+# dbt_marketo v1.1.0
+
+[PR #52](https://github.com/fivetran/dbt_marketo/pull/52) includes the following updates:
+
+## Schema/Data Change
+**5 total changes • 4 possible breaking changes**
+
+| Data Model(s) | Change type | Old | New | Notes |
+| ------------- | ----------- | ----| --- | ----- |
+| All models | New column | | `source_relation` | Identifies the source connection when using multiple Marketo connections |
+| `marketo__lead_history` | Updated surrogate key | `lead_history_id` = `lead_id` + `date_day` | `lead_history_id` = `source_relation` + `lead_id` + `date_day` |  |
+| `marketo__email_templates`<br>`stg_marketo__email_template_history` | Updated surrogate key | `email_template_history_id` = `email_template_id` + `inferred_version` | `email_template_history_id` = `source_relation` + `email_template_id` + `inferred_version` |  |
+|`marketo__email_sends`<br>`stg_marketo__activity_click_email`<br>`stg_marketo__activity_email_bounced`<br>`stg_marketo__activity_email_delivered`<br>`stg_marketo__activity_open_email`<br>`stg_marketo__activity_send_email`<br>`stg_marketo__activity_unsubscribe_email` | Updated surrogate key | `email_send_id` = `primary_attribute_value_id` + `campaign_id` + `campaign_run_id` + `lead_id` | `email_send_id` = `source_relation` + `primary_attribute_value_id` + `campaign_id` + `campaign_run_id` + `lead_id` |  |
+| `marketo__change_data_details`<br>`marketo__change_data_pivot`<br>`marketo__change_data_scd`<br>`marketo__lead_calendar_spine` | Updated surrogate key | `lead_day_id` = `lead_id` + `date_day` | `lead_day_id` = `source_relation` + `lead_id` + `date_day` |  |
+
+## Feature Update
+- **Union Data Functionality**: This release supports running the package on multiple Marketo source connections. See the [README](https://github.com/fivetran/dbt_marketo/tree/main?tab=readme-ov-file#step-3-define-database-and-schema-variables) for details on how to leverage this feature.
+
+## Tests Update
+- Removes uniqueness tests. The new unioning feature requires combination-of-column tests to consider the new `source_relation` column in addition to the existing primary key, but this is not supported across dbt versions.
+- These tests will be reintroduced once a version-agnostic solution is available.
+
 # dbt_marketo v1.0.1
 
 [PR #51](https://github.com/fivetran/dbt_marketo/pull/51) includes the following updates:

README.md

@@ -62,17 +62,77 @@ Include the following Marketo package version in your `packages.yml` file.
 ```yml
 packages:
   - package: fivetran/marketo
-    version: [">=1.0.0", "<1.1.0"]
+    version: [">=1.1.0", "<1.2.0"]
 ```
 > All required sources and staging models are now bundled into this transformation package. Do not include `fivetran/marketo_source` in your `packages.yml` since this package has been deprecated.
 
 ### Step 3: Define database and schema variables
-By default, this package runs using your destination and the `marketo` 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 Marketo data is (for example, if your Marketo schema is named `marketo_fivetran`), add the following configuration to your root `dbt_project.yml` file:
+
+#### Option A: Single connection
+By default, this package runs using your [destination](https://docs.getdbt.com/docs/running-a-dbt-project/using-the-command-line-interface/configure-your-profile) and the `marketo` schema. If this is not where your Marketo data is (for example, if your Marketo schema is named `marketo_fivetran`), add the following configuration to your root `dbt_project.yml` file:
+
+```yml
+vars:
+  marketo:
+    marketo_database: your_database_name
+    marketo_schema: your_schema_name
+```
+
+#### Option B: Union multiple connections
+If you have multiple Marketo connections in Fivetran and would like to use this package on all of them simultaneously, we have provided functionality to do so. For each source table, the package will union all of the data together and pass the unioned table into the transformations. The `source_relation` column in each model indicates the origin of each record.
+
+To use this functionality, you will need to set the `marketo_sources` variable in your root `dbt_project.yml` file:
+
+```yml
+# dbt_project.yml
+
+vars:
+  marketo:
+    marketo_sources:
+      - database: connection_1_destination_name # Required
+        schema: connection_1_schema_name # Required
+        name: connection_1_source_name # Required only if following the step in the following subsection
+
+      - database: connection_2_destination_name
+        schema: connection_2_schema_name
+        name: connection_2_source_name
+```
+
+##### Recommended: Incorporate unioned sources into DAG
+> *If you are running the package through [Fivetran Transformations for dbt Core™](https://fivetran.com/docs/transformations/dbt#transformationsfordbtcore), the below step is necessary in order to synchronize model runs with your Marketo connections. Alternatively, you may choose to run the package through Fivetran [Quickstart](https://fivetran.com/docs/transformations/quickstart), which would create separate sets of models for each Marketo source rather than one set of unioned models.*
+
+By default, this package defines one single-connection source, called `marketo`, which will be disabled if you are unioning multiple connections. This means that your DAG will not include your Marketo sources, though the package will run successfully.
+
+To properly incorporate all of your Marketo connections into your project's DAG:
+1. Define each of your sources in a `.yml` file in the models directory of your project. Utilize the following template for the `source`-level configurations, and, **most importantly**, copy and paste the table and column-level definitions from the package's `src_marketo.yml` [file](https://github.com/fivetran/dbt_marketo/blob/main/models/staging/src_marketo.yml).
 
 ```yml
+# a .yml file in your root project
+
+version: 2
+
+sources:
+  - name: <name> # ex: Should match name in marketo_sources
+    schema: <schema_name>
+    database: <database_name>
+    loader: fivetran
+    config:
+      loaded_at_field: _fivetran_synced
+      freshness: # feel free to adjust to your liking
+        warn_after: {count: 72, period: hour}
+        error_after: {count: 168, period: hour}
+
+    tables: # copy and paste from marketo/models/staging/src_marketo.yml - see https://support.atlassian.com/bitbucket-cloud/docs/yaml-anchors/ for how to use anchors to only do so once
+```
+
+> **Note**: If there are source tables you do not have (see [Step 4](https://github.com/fivetran/dbt_marketo?tab=readme-ov-file#step-4-enablingdisabling-models)), you may still include them, as long as you have set the right variables to `False`.
+
+2. Set the `has_defined_sources` variable (scoped to the `marketo` package) to `True`, like such:
+```yml
+# dbt_project.yml
 vars:
-  marketo_database: your_database_name
-  marketo_schema: your_schema_name 
+  marketo:
+    has_defined_sources: true
 ```
 
 ### Step 4: Enabling/Disabling Models

dbt_project.yml

@@ -1,5 +1,5 @@
 name: 'marketo'
-version: '1.0.1'
+version: '1.1.0'
 config-version: 2
 require-dbt-version: [">=1.3.0", "<2.0.0"]
 on-run-start: "{{ lead_history_columns_warning() }}"

integration_tests/dbt_project.yml

@@ -1,5 +1,5 @@
 name: 'marketo_integration_tests'
-version: '1.0.1'
+version: '1.1.0'
 profile: 'integration_tests'
 config-version: 2
 
@@ -8,6 +8,7 @@ models:
 
 vars:
     marketo_schema: marketo_integration_tests
+    consistency_test_exclude_columns: ['activity_id', 'source_relation', 'email_template_id', 'email_send_id', 'email_template_history_id', 'lead_history_id', 'merged_into_lead_id']
     # marketo__enable_campaigns: false ## Uncomment for testing to disable campaigns
     # marketo__enable_programs: false ## Uncomment for testing to disable programs
     marketo:

integration_tests/tests/consistency/consistency_campaigns.sql

@@ -3,23 +3,44 @@
     enabled=var('fivetran_validation_tests_enabled', false) and var('marketo__enable_campaigns', true)
 ) }}
 
-
+-- this test ensures the marketo__campaigns end model matches the prior version
 with prod as (
-    select count(*) as prod_row_count
+    select {{ dbt_utils.star(from=ref('marketo__campaigns'), except=var('consistency_test_exclude_columns', [])) }}
     from {{ target.schema }}_marketo_prod.marketo__campaigns
 ),
 
 dev as (
-    select count(*) as dev_row_count
+    select {{ dbt_utils.star(from=ref('marketo__campaigns'), except=var('consistency_test_exclude_columns', [])) }}
     from {{ target.schema }}_marketo_dev.marketo__campaigns
-), 
+),
+
+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
 
-count_check as (
-    select *
-    from prod
-    join dev
-        on prod.prod_row_count != dev.dev_row_count
+    select
+        *,
+        'from dev' as source
+    from dev_not_in_prod
 )
 
 select *
-from count_check
+from final

integration_tests/tests/consistency/consistency_campaigns_count.sql

@@ -0,0 +1,25 @@
+{{ config(
+    tags="fivetran_validations",
+    enabled=var('fivetran_validation_tests_enabled', false) and var('marketo__enable_campaigns', true)
+) }}
+
+
+with prod as (
+    select count(*) as prod_row_count
+    from {{ target.schema }}_marketo_prod.marketo__campaigns
+),
+
+dev as (
+    select count(*) as dev_row_count
+    from {{ target.schema }}_marketo_dev.marketo__campaigns
+), 
+
+count_check as (
+    select *
+    from prod
+    join dev
+        on prod.prod_row_count != dev.dev_row_count
+)
+
+select *
+from count_check

integration_tests/tests/consistency/consistency_email_sends.sql

@@ -3,14 +3,12 @@
     enabled=var('fivetran_validation_tests_enabled', false)
 ) }}
 
-{% set columns_to_exclude = ['activity_id'] %}
+{% set columns_to_exclude = ['activity_id'] + var('consistency_test_exclude_columns', []) %}
 
 {% if not var('marketo__enable_campaigns', True) %}
     {% do columns_to_exclude.extend(['campaign_type', 'program_id']) %}
 {% endif %}
 
-{% set columns_to_exclude = columns_to_exclude + var('consistency_test_exclude_metrics', []) %}
-
 with prod as (
     select {{ dbt_utils.star(
         from=ref('marketo__email_sends'), 

integration_tests/tests/consistency/consistency_email_templates.sql

@@ -3,13 +3,15 @@
     enabled=var('fivetran_validation_tests_enabled', false)
 ) }}
 
+{% set columns_to_exclude = var('consistency_test_exclude_columns', []) + var('consistency_test_exclude_metrics', []) %}
+
 with prod as (
-    select {{ dbt_utils.star(from=ref('marketo__email_templates'), except=var('consistency_test_exclude_metrics', [])) }}
+    select {{ dbt_utils.star(from=ref('marketo__email_templates'), except=columns_to_exclude) }}
     from {{ target.schema }}_marketo_prod.marketo__email_templates
 ),
 
 dev as (
-    select {{ dbt_utils.star(from=ref('marketo__email_templates'), except=var('consistency_test_exclude_metrics', [])) }}
+    select {{ dbt_utils.star(from=ref('marketo__email_templates'), except=columns_to_exclude) }}
     from {{ target.schema }}_marketo_dev.marketo__email_templates
 ), 
 

integration_tests/tests/consistency/consistency_lead_history.sql

@@ -0,0 +1,46 @@
+{{ config(
+    tags="fivetran_validations",
+    enabled=var('fivetran_validation_tests_enabled', false)
+) }}
+
+-- this test ensures the marketo__lead_history end model matches the prior version
+with prod as (
+    select {{ dbt_utils.star(from=ref('marketo__lead_history'), except=var('consistency_test_exclude_columns', [])) }}
+    from {{ target.schema }}_marketo_prod.marketo__lead_history
+),
+
+dev as (
+    select {{ dbt_utils.star(from=ref('marketo__lead_history'), except=var('consistency_test_exclude_columns', [])) }}
+    from {{ target.schema }}_marketo_dev.marketo__lead_history
+),
+
+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

integration_tests/tests/consistency/consistency_leads.sql

@@ -3,22 +3,44 @@
     enabled=var('fivetran_validation_tests_enabled', false)
 ) }}
 
+-- this test ensures the marketo__leads end model matches the prior version
 with prod as (
-    select count(*) as prod_row_count
+    select {{ dbt_utils.star(from=ref('marketo__leads'), except=var('consistency_test_exclude_columns', [])) }}
     from {{ target.schema }}_marketo_prod.marketo__leads
 ),
 
 dev as (
-    select count(*) as dev_row_count
+    select {{ dbt_utils.star(from=ref('marketo__leads'), except=var('consistency_test_exclude_columns', [])) }}
     from {{ target.schema }}_marketo_dev.marketo__leads
-), 
+),
+
+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
 
-count_check as (
-    select *
-    from prod
-    join dev
-        on prod.prod_row_count != dev.dev_row_count
+    select
+        *,
+        'from dev' as source
+    from dev_not_in_prod
 )
 
 select *
-from count_check
+from final

integration_tests/tests/consistency/consistency_leads_count.sql

@@ -0,0 +1,24 @@
+{{ config(
+    tags="fivetran_validations",
+    enabled=var('fivetran_validation_tests_enabled', false)
+) }}
+
+with prod as (
+    select count(*) as prod_row_count
+    from {{ target.schema }}_marketo_prod.marketo__leads
+),
+
+dev as (
+    select count(*) as dev_row_count
+    from {{ target.schema }}_marketo_dev.marketo__leads
+), 
+
+count_check as (
+    select *
+    from prod
+    join dev
+        on prod.prod_row_count != dev.dev_row_count
+)
+
+select *
+from count_check