Feature/union data

.gitignore

@@ -57,6 +57,9 @@ wheels/
 credentials.json
 service-account.json
 
+# Claude Code instructions
+CLAUDE.md
+
 # Temporary files
 .cache/
 *.temp

CHANGELOG.md

@@ -1,3 +1,27 @@
+# dbt_mailchimp v1.1.0
+[PR #57](https://github.com/fivetran/dbt_mailchimp/pull/57) includes the following updates:
+
+## Schema/Data Change
+**7 total changes • 0 possible breaking changes**
+
+| Data Model(s) | Change type | Old | New | Notes |
+| ------------- | ----------- | ----| --- | ----- |
+| All models | New column | | `source_relation` | Identifies the source connection when using multiple Mailchimp connections |
+| `stg_mailchimp__automation_activities`<br>`mailchimp__automation_activities` | Updated surrogate key | `activity_id` = `action_type` + `automation_email_id` + `member_id` + `activity_timestamp` | `activity_id` = `source_relation` + `action_type` + `automation_email_id` + `member_id` + `activity_timestamp` | Updated to include `source_relation` |
+| `stg_mailchimp__automation_recipients` | Updated surrogate key | `automation_recipient_id` = `member_id` + `automation_email_id` | `automation_recipient_id` = `source_relation` + `member_id` + `automation_email_id` | Updated to include `source_relation` |
+| `stg_mailchimp__campaign_activities`<br>`mailchimp__campaign_activities` | Updated surrogate key | `activity_id` = `action_type` + `campaign_id` + `member_id` + `activity_timestamp` | `activity_id` = `source_relation` + `action_type` + `campaign_id` + `member_id` + `activity_timestamp` | Updated to include `source_relation` |
+| `stg_mailchimp__campaign_activities`<br>`mailchimp__campaign_activities` | Updated surrogate key | `email_id` = `campaign_id` + `member_id` | `email_id` = `source_relation` + `campaign_id` + `member_id` | Updated to include `source_relation` |
+| `stg_mailchimp__campaign_recipients`<br>`mailchimp__campaign_recipients` | Updated surrogate key | `email_id` = `campaign_id` + `member_id` | `email_id` = `source_relation` + `campaign_id` + `member_id` | Updated to include `source_relation` |
+| `stg_mailchimp__unsubscribes` | Updated surrogate key | `unsubscribe_id` = `member_id` + `list_id` + `unsubscribe_timestamp` | `unsubscribe_id` = `source_relation` + `member_id` + `list_id` + `unsubscribe_timestamp` | Updated to include `source_relation` |
+
+## Feature Update
+- **Union Data Functionality**: This release supports running the package on multiple Mailchimp source connections. See the [README](https://github.com/fivetran/dbt_mailchimp/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 for primary keys. 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.
+  - Note that surrogate keys are unaffected and retain their uniqueness tests.  
+- Primary key tests will be reintroduced once a version-agnostic solution is available.
+
 # dbt_mailchimp v1.0.0
 
 [PR #55](https://github.com/fivetran/dbt_mailchimp/pull/55) includes the following updates:

README.md

@@ -54,7 +54,7 @@ Include the following mailchimp package version in your `packages.yml` file:
 ```yaml
 packages:
   - package: fivetran/mailchimp
-    version: [">=1.0.0", "<1.1.0"] # we recommend using ranges to capture non-breaking changes automatically
+    version: [">=1.1.0", "<1.2.0"] # we recommend using ranges to capture non-breaking changes automatically
 ```
 > All required sources and staging models are now bundled into this transformation package. Do not include `fivetran/mailchimp_source` in your `packages.yml` since this package has been deprecated.
 
@@ -65,13 +65,73 @@ dispatch:
   - macro_namespace: dbt_utils
     search_order: ['spark_utils', 'dbt_utils']
 
-## Step 3: Define database and schema variables
-By default, this package runs using your destination and the `mailchimp` schema. If this is not where your Mailchimp data is (for example, if your Mailchimp schema is named `mailchimp_fivetran`), add the following configuration to your root `dbt_project.yml` file:
+### Step 3: Define database and schema variables
+
+#### 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 `mailchimp` schema. If this is not where your Mailchimp data is (for example, if your Mailchimp schema is named `mailchimp_fivetran`), add the following configuration to your root `dbt_project.yml` file:
 
 ```yml
 vars:
-    mailchimp_schema: your_schema_name
+  mailchimp:
     mailchimp_database: your_database_name
+    mailchimp_schema: your_schema_name
+```
+
+#### Option B: Union multiple connections
+If you have multiple Mailchimp 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 `mailchimp_sources` variable in your root `dbt_project.yml` file:
+
+```yml
+# dbt_project.yml
+
+vars:
+  mailchimp:
+    mailchimp_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 Mailchimp 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 Mailchimp source rather than one set of unioned models.*
+
+By default, this package defines one single-connection source, called `mailchimp`, which will be disabled if you are unioning multiple connections. This means that your DAG will not include your Mailchimp sources, though the package will run successfully.
+
+To properly incorporate all of your Mailchimp connections into your project's DAG:
+1. Define each of your sources in a `.yml` file in 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_mailchimp.yml` [file](https://github.com/fivetran/dbt_mailchimp/blob/main/models/staging/src_mailchimp.yml).
+
+```yml
+# a .yml file in your root project
+
+version: 2
+
+sources:
+  - name: <name> # ex: Should match name in mailchimp_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 mailchimp/models/staging/src_mailchimp.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](#step-4-disable-models-for-non-existent-sources)), 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 `mailchimp` package) to `True`, like such:
+```yml
+# dbt_project.yml
+vars:
+  mailchimp:
+    has_defined_sources: true
 ```
 
 ## Step 4: Disable models for non-existent sources

dbt_project.yml

@@ -1,6 +1,6 @@
 config-version: 2
 name: 'mailchimp'
-version: '1.0.0'
+version: '1.1.0'
 require-dbt-version: [">=1.3.0", "<2.0.0"]
 
 models:

integration_tests/dbt_project.yml

@@ -1,6 +1,6 @@
 config-version: 2
 name: 'mailchimp_integration_tests'
-version: '1.0.0'
+version: '1.1.0'
 profile: 'integration_tests'
 
 models:

macros/union/apply_source_relation.sql

@@ -0,0 +1,15 @@
+{% macro apply_source_relation() -%}
+
+{{ adapter.dispatch('apply_source_relation', 'mailchimp') () }}
+
+{%- endmacro %}
+
+{% macro default__apply_source_relation() -%}
+
+{% if var('mailchimp_sources', []) != [] %}
+, _dbt_source_relation as source_relation
+{% else %}
+, '{{ var("mailchimp_database", target.database) }}' || '.'|| '{{ var("mailchimp_schema", "mailchimp") }}' as source_relation
+{% endif %}
+
+{%- endmacro %}

macros/union/partition_by_source_relation.sql

@@ -0,0 +1,15 @@
+{%- macro partition_by_source_relation(has_other_partitions='yes', alias=None) -%}
+
+{{ adapter.dispatch('partition_by_source_relation', 'mailchimp') (has_other_partitions, alias) }}
+
+{%- endmacro %}
+
+{% macro default__partition_by_source_relation(has_other_partitions='yes', alias=None) -%}
+    {% set prefix = '' if alias is none else alias ~ '.' %}
+
+    {%- if has_other_partitions == 'no' -%}
+        {{ 'partition by ' ~ prefix ~ 'source_relation' if var('mailchimp_sources', [])|length > 1 }}
+    {%- else -%}
+        {{ ', ' ~ prefix ~ 'source_relation' if var('mailchimp_sources', [])|length > 1 }}
+    {%- endif -%}
+{%- endmacro -%}

macros/union/union_connections.sql

@@ -0,0 +1,93 @@
+{% macro mailchimp_union_connections(connection_dictionary, single_source_name, single_table_name, default_identifier=single_table_name) %}
+
+{{ adapter.dispatch('mailchimp_union_connections', 'mailchimp') (connection_dictionary, single_source_name, single_table_name, default_identifier) }}
+
+{%- endmacro %}
+
+{% macro default__mailchimp_union_connections(connection_dictionary, single_source_name, single_table_name, default_identifier=single_table_name) %}
+
+{%- set exception_warning = "\n\nPlease be aware: The " ~ single_source_name|upper ~ "." ~ single_table_name|upper ~ " table was not found in your schema(s). The Fivetran Data Model will create a completely empty staging model as to not break downstream transformations. To turn off these warnings, set the `fivetran__remove_empty_table_warnings` variable to TRUE (see https://github.com/fivetran/dbt_fivetran_utils/tree/releases/v0.4.latest#union_data-source for details).\n"%}
+{%- set using_empty_table_warnings = (execute and not var('fivetran__remove_empty_table_warnings', false)) %}
+{%- set connections = var(connection_dictionary, []) %}
+{%- set using_unioning = connections | length > 0 %}
+
+{%- if using_unioning %}
+{# For unioning #}
+    {%- set relations = [] -%}
+    {%- for connection in connections -%}
+
+    {% if var('has_defined_sources', false) %}
+        {%- set database = source(connection.name, single_table_name).database %}
+        {%- set schema = source(connection.name, single_table_name).schema %}
+        {%- set identifier = source(connection.name, single_table_name).identifier %}
+    {%- else %}
+        {%- set database = connection.database if connection.database else target.database %}
+        {%- set schema = connection.schema if connection.schema else single_source_name %}
+        {%- set identifier = default_identifier %}
+    {%- endif %}
+
+    {%- set relation=adapter.get_relation(
+        database=database,
+        schema=schema,
+        identifier=identifier
+        )
+    -%}
+
+    {%- if relation is not none -%}
+        {%- do relations.append(relation) -%}
+    {%- endif -%}
+
+    -- ** Values passed to adapter.get_relation:
+    {{ '-- database: ' ~ database }}
+    {{ '-- schema: ' ~ schema }}
+    {{ '-- identifier: ' ~ identifier ~ '\n' }}
+
+    {%- endfor -%}
+
+    {%- if relations != [] -%}
+        {{ mailchimp.mailchimp_union_relations(relations) }}
+
+    {%- else -%}
+        {{ exceptions.warn(exception_warning) if using_empty_table_warnings }}
+
+        select
+            cast(null as {{ dbt.type_string() }}) as _dbt_source_relation
+        limit {{ '0' if target.type != 'redshift' else '1' }}
+    {%- endif -%}
+
+{% else %}
+{# Not unioning #}
+
+    {% set identifier_var = single_source_name + "_" + single_table_name + "_identifier"%}
+    {%- set database = source(single_source_name, single_table_name).database %}
+    {%- set schema =   source(single_source_name, single_table_name).schema %}
+    {%- set identifier = source(single_source_name, single_table_name).identifier %}
+
+    {%- set relation=adapter.get_relation(
+        database=database,
+        schema=schema,
+        identifier=identifier
+        )
+    -%}
+
+    -- ** Values passed to adapter.get_relation:
+    {{ '-- full-identifier_var: ' ~ identifier_var }}
+    {{ '-- database: ' ~ database }}                            
+    {{ '-- schema: ' ~ schema }}
+    {{ '-- identifier: ' ~ identifier ~ '\n' }}
+
+    {% if relation is not none -%}
+        select
+            {{ dbt_utils.star(from=source(single_source_name, single_table_name)) }}
+        from {{ source(single_source_name, single_table_name) }} as source_table
+
+    {% else %}
+        {{ exceptions.warn(exception_warning) if using_empty_table_warnings }}
+
+        select
+            cast(null as {{ dbt.type_string() }}) as _dbt_source_relation
+        limit {{ '0' if target.type != 'redshift' else '1' }}
+    {%- endif -%}
+{% endif -%}
+
+{%- endmacro %}

macros/union/union_relations.sql

@@ -0,0 +1,131 @@
+{# Adapted from dbt_utils.union_relations() #}
+
+{%- macro mailchimp_union_relations(relations, aliases=none, column_override=none, include=[], exclude=[], source_column_name='_dbt_source_relation', where=none) -%}
+    {{ return(adapter.dispatch('mailchimp_union_relations', 'mailchimp')(relations, aliases, column_override, include, exclude, source_column_name, where)) }}
+{% endmacro %}
+
+{%- macro default__mailchimp_union_relations(relations, aliases=none, column_override=none, include=[], exclude=[], source_column_name='_dbt_source_relation', where=none) -%}
+
+    {%- if exclude and include -%}
+        {{ exceptions.raise_compiler_error("Both an exclude and include list were provided to the `union` macro. Only one is allowed") }}
+    {%- endif -%}
+
+    {#-- Prevent querying of db in parsing mode. This works because this macro does not create any new refs. -#}
+    {%- if not execute %}
+        {{ return('') }}
+    {% endif -%}
+
+    {%- set column_override = column_override if column_override is not none else {} -%}
+
+    {%- set relation_columns = {} -%}
+    {%- set column_superset = {} -%}
+    {%- set all_excludes = [] -%}
+    {%- set all_includes = [] -%}
+
+    {%- if exclude -%}
+        {%- for exc in exclude -%}
+            {%- do all_excludes.append(exc | lower) -%}
+        {%- endfor -%}
+    {%- endif -%}
+
+    {%- if include -%}
+        {%- for inc in include -%}
+            {%- do all_includes.append(inc | lower) -%}
+        {%- endfor -%}
+    {%- endif -%}
+
+    {%- for relation in relations -%}
+
+        {%- do relation_columns.update({relation: []}) -%}
+
+        {%- do dbt_utils._is_relation(relation, 'mailchimp_union_relations') -%}
+        {%- do dbt_utils._is_ephemeral(relation, 'mailchimp_union_relations') -%}
+        {%- set cols = adapter.get_columns_in_relation(relation) -%}
+        {%- for col in cols -%}
+
+        {#- If an exclude list was provided and the column is in the list, do nothing -#}
+        {%- if exclude and col.column | lower in all_excludes -%}
+
+        {#- If an include list was provided and the column is not in the list, do nothing -#}
+        {%- elif include and col.column | lower not in all_includes -%}
+
+        {#- Otherwise add the column to the column superset -#}
+        {%- else -%}
+
+            {#- update the list of columns in this relation -#}
+            {%- do relation_columns[relation].append(col.column) -%}
+
+            {%- if col.column in column_superset -%}
+
+                {%- set stored = column_superset[col.column] -%}
+                {%- if col.is_string() and stored.is_string() and col.string_size() > stored.string_size() -%}
+
+                    {%- do column_superset.update({col.column: col}) -%}
+
+                {%- endif %}
+
+            {%- else -%}
+
+                {%- do column_superset.update({col.column: col}) -%}
+
+            {%- endif -%}
+
+        {%- endif -%}
+
+        {%- endfor -%}
+    {%- endfor -%}
+
+    {%- set ordered_column_names = column_superset.keys() -%}
+    {%- set dbt_command = flags.WHICH -%}
+
+
+    {% if dbt_command in ['run', 'build'] %}
+    {% if (include | length > 0 or exclude | length > 0) and not column_superset.keys() %}
+        {%- set relations_string -%}
+            {%- for relation in relations -%}
+                {{ relation.name }}
+            {%- if not loop.last %}, {% endif -%}
+            {%- endfor -%}
+        {%- endset -%}
+
+        {%- set error_message -%}
+            There were no columns found to union for relations {{ relations_string }}
+        {%- endset -%}
+
+        {{ exceptions.raise_compiler_error(error_message) }}
+    {%- endif -%}
+    {%- endif -%}
+
+    {%- for relation in relations %}
+
+        (
+            select
+
+                {%- if source_column_name is not none %}
+                cast({{ dbt.string_literal(relation.database ~ '.' ~ relation.schema) }} as {{ dbt.type_string() }}) as {{ source_column_name }},
+                {%- endif %}
+
+                {% for col_name in ordered_column_names -%}
+
+                    {%- set col = column_superset[col_name] %}
+                    {%- set col_type = column_override.get(col.column, col.data_type) %}
+                    {%- set col_name = adapter.quote(col_name) if col_name in relation_columns[relation] else 'null' %}
+                    cast({{ col_name }} as {{ col_type }}) as {{ col.quoted }} {% if not loop.last %},{% endif -%}
+
+                {%- endfor %}
+
+            {# This alias is the only addition made to the dbt_utils.union_relations() code. Avoids errors if the table is named a reserved keyword #}
+            from {{ aliases[loop.index0] if aliases else relation }} as unioned_relation_{{ loop.index }}
+
+            {% if where -%}
+            where {{ where }}
+            {%- endif %}
+        )
+
+        {% if not loop.last -%}
+            union all
+        {% endif -%}
+
+    {%- endfor -%}
+
+{%- endmacro -%}