Warehouse-Native Data Deduplication: Identifying and Merging Duplicate Records Across Ingestion Sources Using SQL and dbt Macros
Introduction
Imagine you're a data engineer at a mid-sized e-commerce company. Your marketing team uses HubSpot to track leads. Your sales team works out of Salesforce. Your customer support team lives in Zendesk. And once a quarter, someone uploads a CSV of contacts from a trade show. By the time all of that lands in your data warehouse, you've got the same person — Sarah Chen, sarah.chen@acme.com — appearing four times across four different tables, each with slightly different formatting, slightly different field completeness, and a slightly different perspective on who she is.
This is not a hypothetical. It's the everyday reality of modern data pipelines. When data flows in from multiple sources, duplicates are not a bug in the process — they're an expected consequence of it. The question isn't whether you'll have duplicates; it's whether you have a systematic, repeatable strategy for finding them and resolving them before they corrupt your analytics, inflate your customer counts, and erode trust in your data.
By the end of this lesson, you will be able to identify duplicate records across ingestion sources using SQL window functions, score candidate matches using deterministic rules, resolve conflicts between duplicate fields using a coalesce-based priority strategy, and wrap the entire process into a reusable dbt macro. You'll leave with working code you can adapt to a real project.
What you'll learn:
- Why duplicates occur in multi-source ingestion pipelines and how to classify them
- How to use
ROW_NUMBER() and blocking keys to find exact and near-exact duplicates in SQL
- How to implement a field-level "best value wins" merge strategy using
COALESCE
- How to build a parameterized dbt macro that encapsulates the deduplication logic
- How to structure a dbt project so deduplication is a first-class transformation step
Prerequisites
- Basic SQL fluency: you should be comfortable writing
SELECT, JOIN, GROUP BY, and WHERE clauses
- A passing familiarity with what dbt is — you don't need to be an expert, but you should know that dbt is a transformation framework that runs SQL models inside a data warehouse
- Access to a data warehouse (Snowflake, BigQuery, DuckDB, or Postgres will all work for this lesson's examples)
- dbt Core or dbt Cloud installed, with a project connected to your warehouse
Understanding the Anatomy of a Duplicate
Before you write a single line of SQL, you need to understand what kind of duplicate you're dealing with. Treating every duplicate the same way is like using a sledgehammer for every carpentry task — it works sometimes, but it causes a lot of unnecessary damage.
Exact duplicates are rows that are byte-for-byte identical. These typically come from pipeline misconfiguration — your ELT tool ran twice, or someone loaded the same file twice. They're the easiest to fix.
Key duplicates occur when rows share a primary identifier — like customer_id or email — but differ in other fields. Maybe Salesforce has the phone number and HubSpot has the mailing address, but both agree on the email address. These require a merge, not just a drop.
Fuzzy duplicates are the most complex. The names look similar, the addresses are close, but no single field matches exactly. Sarah Chen in one system versus S. Chen in another, both at the same company email domain. Fuzzy matching is a deep topic (Levenshtein distance, phonetic algorithms, ML-based entity resolution) that goes beyond this lesson, but we'll acknowledge where it fits in.
For this lesson, we'll focus on key duplicates across sources, which is the most common problem in multi-source warehouse pipelines and the one most solvable with pure SQL and dbt.
Setting Up a Realistic Scenario
Let's build a concrete example. You have a contacts staging area in your warehouse with three source tables, all loaded by your ELT tool (Fivetran, Airbyte, or similar):
-- stg_hubspot__contacts
SELECT
id AS source_id,
'hubspot' AS source_system,
email,
first_name,
last_name,
phone,
company,
created_at,
updated_at
FROM raw.hubspot.contacts
-- stg_salesforce__contacts
SELECT
id AS source_id,
'salesforce' AS source_system,
email,
first_name,
last_name,
phone,
account_name AS company,
created_date AS created_at,
last_modified_date AS updated_at
FROM raw.salesforce.contact
-- stg_zendesk__users
SELECT
id AS source_id,
'zendesk' AS source_system,
email,
name AS first_name, -- Zendesk stores full name in one field
NULL AS last_name,
phone,
organization_name AS company,
created_at,
updated_at
FROM raw.zendesk.users
Notice a few things already. Zendesk stores names differently. Salesforce uses different column names for created and modified dates. These are the tiny inconsistencies that become landmines when you try to deduplicate naively.
The first step in any deduplication project is building a unified staging layer — a single model that combines all sources into a consistent schema. In dbt, this is typically a model called int_contacts_unioned or similar.
-- models/intermediate/int_contacts_unioned.sql
WITH hubspot AS (
SELECT * FROM {{ ref('stg_hubspot__contacts') }}
),
salesforce AS (
SELECT * FROM {{ ref('stg_salesforce__contacts') }}
),
zendesk AS (
SELECT * FROM {{ ref('stg_zendesk__users') }}
),
unioned AS (
SELECT * FROM hubspot
UNION ALL
SELECT * FROM salesforce
UNION ALL
SELECT * FROM zendesk
)
SELECT
source_system || '_' || source_id AS surrogate_key,
source_system,
source_id,
LOWER(TRIM(email)) AS email,
TRIM(first_name) AS first_name,
TRIM(last_name) AS last_name,
phone,
company,
created_at,
updated_at
FROM unioned
WHERE email IS NOT NULL
Two things worth highlighting here. First, we normalize email immediately — lowercase and trimmed — because Sarah.Chen@Acme.com and sarah.chen@acme.com are the same person, and case differences will silently break your deduplication logic if you don't handle them. Second, we filter out records with no email, because email is our blocking key — the field we'll use to group candidate duplicates.
Tip: The choice of blocking key matters enormously. Email is a good blocking key for contact data because it has high coverage and high precision. Phone numbers are less reliable (people change phones, share numbers). Names alone are almost useless as a sole blocking key.
Identifying Duplicates with Window Functions
Now that you have a unified table, it's time to find the duplicates. The SQL tool you'll reach for is ROW_NUMBER() — a window function that assigns a sequential number to each row within a partition.
Here's the core pattern:
-- models/intermediate/int_contacts_deduped.sql
WITH unioned AS (
SELECT * FROM {{ ref('int_contacts_unioned') }}
),
ranked AS (
SELECT
*,
ROW_NUMBER() OVER (
PARTITION BY email
ORDER BY
CASE source_system
WHEN 'salesforce' THEN 1
WHEN 'hubspot' THEN 2
WHEN 'zendesk' THEN 3
ELSE 4
END,
updated_at DESC
) AS row_rank
FROM unioned
)
SELECT * FROM ranked
WHERE row_rank = 1
This pattern is called pick-one deduplication. For each email address, you rank all the records and keep only the top-ranked one. The ORDER BY clause encodes your business logic: Salesforce is your system of record, so it wins. Within the same source, the most recently updated record wins.
Pick-one deduplication is simple and fast. It's also lossy. When you pick the Salesforce record, you throw away the phone number that only existed in HubSpot. In many real-world cases, you actually want to merge the records — taking the best available value for each field from across all sources. That requires a different approach.
Building a Merge Strategy: Best Value Wins
A merge strategy defines, for each field, how you resolve conflicts when multiple source records provide different values. The most common strategies are:
- Source priority: Salesforce wins, always, for every field
- Most recent wins: The record with the latest
updated_at provides every field
- Best value wins (COALESCE): For each field individually, take the first non-null value from a priority-ordered list
The COALESCE strategy is the most nuanced and usually the most correct. Here's how it works in SQL:
-- models/intermediate/int_contacts_merged.sql
WITH unioned AS (
SELECT * FROM {{ ref('int_contacts_unioned') }}
),
-- Pivot: one row per email, one column per (field, source) combination
sf AS (
SELECT email, first_name, last_name, phone, company, updated_at
FROM unioned WHERE source_system = 'salesforce'
),
hs AS (
SELECT email, first_name, last_name, phone, company, updated_at
FROM unioned WHERE source_system = 'hubspot'
),
zd AS (
SELECT email, first_name, last_name, phone, company, updated_at
FROM unioned WHERE source_system = 'zendesk'
),
all_emails AS (
SELECT DISTINCT email FROM unioned
),
merged AS (
SELECT
ae.email,
-- first_name: prefer Salesforce, fall back to HubSpot, then Zendesk
COALESCE(sf.first_name, hs.first_name, zd.first_name) AS first_name,
COALESCE(sf.last_name, hs.last_name, zd.last_name) AS last_name,
-- phone: prefer HubSpot (marketing keeps it cleaner), then Salesforce
COALESCE(hs.phone, sf.phone, zd.phone) AS phone,
-- company: prefer Salesforce (sales owns account data)
COALESCE(sf.company, hs.company, zd.company) AS company,
-- track which systems have a record for this contact
(sf.email IS NOT NULL) AS in_salesforce,
(hs.email IS NOT NULL) AS in_hubspot,
(zd.email IS NOT NULL) AS in_zendesk,
-- use the most recent updated_at across all sources
GREATEST(
COALESCE(sf.updated_at, '1900-01-01'),
COALESCE(hs.updated_at, '1900-01-01'),
COALESCE(zd.updated_at, '1900-01-01')
) AS last_updated_at
FROM all_emails ae
LEFT JOIN sf ON ae.email = sf.email
LEFT JOIN hs ON ae.email = hs.email
LEFT JOIN zd ON ae.email = zd.email
)
SELECT * FROM merged
This model produces one row per email address — a single golden record — with each field sourced from the system most trusted to have it. The in_salesforce, in_hubspot, in_zendesk flags are extremely useful for downstream debugging and for answering questions like "which contacts exist in Salesforce but not in Zendesk?"
Warning: This approach assumes no email address appears more than once within a single source system. If a source has its own internal duplicates, you need to deduplicate within that source first, before the union step. Add a ROW_NUMBER() within each staging model to handle this.
Wrapping the Logic in a dbt Macro
If you only have one entity to deduplicate, a one-off SQL model is fine. But if you need to deduplicate contacts, companies, products, and orders, you'll find yourself copy-pasting the same structural logic with slight variations. That's the perfect use case for a dbt macro.
A dbt macro is a Jinja-templated SQL function defined in your macros/ directory. It takes parameters and renders into SQL at compile time.
Let's build a macro called merge_sources that handles the source-priority merge pattern:
{# macros/merge_sources.sql #}
{% macro merge_sources(
sources,
join_key,
fields,
updated_at_col='updated_at'
) %}
{#
sources: list of dicts with keys:
- ref: the dbt ref string, e.g. 'stg_salesforce__contacts'
- alias: short name used in CTEs, e.g. 'sf'
- priority: integer, 1 = highest priority
join_key: the column used to group duplicates, e.g. 'email'
fields: list of dicts with keys:
- name: column name
- source_priority_override: optional list of aliases in preferred order
#}
{% set sources_sorted = sources | sort(attribute='priority') %}
WITH
{% for source in sources_sorted %}
{{ source.alias }} AS (
SELECT
{{ join_key }},
{% for field in fields %}
{{ field.name }}{{ "," if not loop.last }}
{% endfor %},
{{ updated_at_col }}
FROM {{ ref(source.ref) }}
WHERE {{ join_key }} IS NOT NULL
),
{% endfor %}
all_keys AS (
{% for source in sources_sorted %}
SELECT DISTINCT {{ join_key }} FROM {{ source.alias }}
{% if not loop.last %}UNION{% endif %}
{% endfor %}
),
merged AS (
SELECT
ak.{{ join_key }},
{% for field in fields %}
{% set override = field.get('source_priority_override', sources_sorted | map(attribute='alias') | list) %}
COALESCE(
{% for alias in override %}
{{ alias }}.{{ field.name }}{{ "," if not loop.last }}
{% endfor %}
) AS {{ field.name }}{{ "," if not loop.last }}
{% endfor %},
GREATEST(
{% for source in sources_sorted %}
COALESCE({{ source.alias }}.{{ updated_at_col }}, '1900-01-01'::TIMESTAMP){{ "," if not loop.last }}
{% endfor %}
) AS last_updated_at
FROM all_keys ak
{% for source in sources_sorted %}
LEFT JOIN {{ source.alias }}
ON ak.{{ join_key }} = {{ source.alias }}.{{ join_key }}
{% endfor %}
)
SELECT * FROM merged
{% endmacro %}
Now your contact merge model becomes dramatically simpler:
{# models/intermediate/int_contacts_merged.sql #}
{{ merge_sources(
sources=[
{'ref': 'stg_salesforce__contacts', 'alias': 'sf', 'priority': 1},
{'ref': 'stg_hubspot__contacts', 'alias': 'hs', 'priority': 2},
{'ref': 'stg_zendesk__users', 'alias': 'zd', 'priority': 3}
],
join_key='email',
fields=[
{'name': 'first_name'},
{'name': 'last_name'},
{'name': 'phone', 'source_priority_override': ['hs', 'sf', 'zd']},
{'name': 'company'}
]
) }}
The source_priority_override for the phone field means: for this specific field, prefer HubSpot over Salesforce, overriding the global source priority. This is the kind of nuance that macros let you encode cleanly without duplicating the structural SQL.
Tip: Run dbt compile to see the rendered SQL that the macro generates. This is invaluable for debugging Jinja logic — the compiled output file lives in target/compiled/.
Testing Your Deduplication Output
Writing deduplication logic without testing it is a bit like mopping a floor with the tap still running. dbt has a built-in testing framework that makes this straightforward.
In your schema.yml file for the intermediate layer, add:
models:
- name: int_contacts_merged
description: >
One golden record per email address, merged across Salesforce,
HubSpot, and Zendesk.
columns:
- name: email
description: The unique identifier and join key for merged contacts
tests:
- unique
- not_null
- name: first_name
tests:
- not_null:
where: "in_salesforce = true"
# If Salesforce has this record, first_name should never be null
The unique test on email is your primary correctness check. If this test fails, your deduplication logic has a bug — you're producing multiple rows for the same email, which defeats the entire purpose.
Run your tests with dbt test --select int_contacts_merged and treat a passing unique test as your definition of done.
Hands-On Exercise
Work through this exercise in sequence. Each step builds on the previous one.
Setup: Create three small seed files in your dbt project's seeds/ directory. Seed files are CSVs that dbt loads directly into your warehouse — they're perfect for testing with controlled data.
Create seeds/seed_hubspot_contacts.csv:
id,email,first_name,last_name,phone,company,updated_at
hs_001,sarah.chen@acme.com,Sarah,Chen,415-555-0101,Acme Corp,2024-01-15
hs_002,marcus.johnson@globex.io,Marcus,Johnson,,Globex,2024-01-10
hs_003,priya.patel@initech.com,Priya,Patel,512-555-0199,Initech,2024-01-20
Create seeds/seed_salesforce_contacts.csv:
id,email,first_name,last_name,phone,company,updated_at
sf_001,sarah.chen@acme.com,Sarah,Chen,,ACME Corporation,2024-01-18
sf_002,marcus.johnson@globex.io,Marcus,Johnson,617-555-0202,Globex Inc,2024-01-12
sf_004,wei.zhang@initech.com,Wei,Zhang,202-555-0303,Initech,2024-01-05
Create seeds/seed_zendesk_users.csv:
id,email,name,phone,company,updated_at
zd_001,sarah.chen@acme.com,Sarah Chen,415-555-0101,Acme,2024-01-08
zd_002,priya.patel@initech.com,Priya Patel,,Initech,2024-01-19
zd_003,ada.lovelace@history.org,Ada Lovelace,800-555-0000,History Dept,2024-01-01
Step 1: Run dbt seed to load these files into your warehouse.
Step 2: Write three staging models that select from each seed, normalizing email with LOWER(TRIM(email)) and splitting Zendesk's name field using SPLIT_PART(name, ' ', 1) for first name and SPLIT_PART(name, ' ', 2) for last name.
Step 3: Write the int_contacts_unioned model that combines all three with UNION ALL.
Step 4: Write a int_contacts_merged model using either the manual COALESCE approach or the macro, that produces one row per email with merged fields.
Step 5: Count the expected output. You should have exactly 5 unique emails:
Step 6: Add the unique and not_null tests to schema.yml and run dbt test. Verify that sarah.chen@acme.com has phone 415-555-0101 (from HubSpot if you prioritize it for phone) and company ACME Corporation (from Salesforce if you prioritize it for company).
Common Mistakes & Troubleshooting
Mistake: Forgetting to normalize the join key before deduplication
If you block on email without lowercasing, Sarah.Chen@Acme.com and sarah.chen@acme.com will not be recognized as duplicates. Always normalize your blocking key in the staging model, not in the dedup model — normalize it once, at the source.
Mistake: Assuming one-to-one mapping within source systems
Your Salesforce source might itself have duplicate emails if the data quality is poor. Always run a GROUP BY email HAVING COUNT(*) > 1 query against each staging model before writing your merge logic. If there are within-source duplicates, add a ROW_NUMBER() to each staging model with WHERE row_rank = 1 before feeding into the union.
Mistake: Using UNION instead of UNION ALL
UNION deduplicates rows at the SQL level, but it does so by comparing entire rows. Since rows from different sources will have different source_system values, UNION won't remove cross-source duplicates — it'll just silently drop exact row duplicates you weren't expecting, and run significantly slower. Always use UNION ALL and handle deduplication explicitly.
Mistake: The macro generates invalid SQL for edge cases
If all your sources have NULL for a given field, COALESCE returns NULL, which is correct. But if you're using GREATEST() for timestamp merging and all values are NULL, the result is NULL — which is fine, but make sure your downstream models handle NULL last_updated_at gracefully. Use COALESCE(last_updated_at, created_at) in downstream models if needed.
Troubleshooting: The unique test is failing
Run this diagnostic query directly in your warehouse:
SELECT email, COUNT(*) AS record_count
FROM {{ ref('int_contacts_merged') }}
GROUP BY email
HAVING COUNT(*) > 1
ORDER BY record_count DESC
This shows you exactly which emails are duplicated in your output and how many times. Trace one of those emails back through the intermediate models to find where the extra row is being generated.
Summary & Next Steps
You've covered the full deduplication lifecycle: understanding why duplicates occur in multi-source pipelines, normalizing data into a unified staging layer, using window functions to identify key duplicates, building a field-level COALESCE merge strategy, wrapping it all in a reusable dbt macro, and testing your output with dbt's built-in test framework.
The pattern you've built here — stage, union, deduplicate, test — is a repeatable architecture that works for contacts, companies, products, transactions, or any entity that flows in from multiple sources. The macro makes it scale.
Where to go from here:
- Fuzzy matching: When email isn't available as a blocking key, explore Levenshtein distance functions (available natively in Snowflake and BigQuery) or the
dbt-utils package's generate_surrogate_key macro for composite blocking
- Incremental deduplication: Right now your dedup models run full refreshes. For large datasets, look into dbt incremental models with
unique_key and merge strategy to deduplicate only new records
- Audit tables: Add a
contact_source_history model that preserves all source records with their surrogate keys, so you can always trace a golden record back to its constituent source rows
- Data observability: Integrate
dbt-expectations or elementary-data to monitor the ratio of records per source over time — a sudden spike in duplicates from one source is a canary for an upstream pipeline problem
