Resolución de Identidad Determinística en Data Warehouse mediante dbt y SQL

{{ config(
    materialized = 'incremental',
    unique_key = 'edge_id',
    incremental_strategy = 'merge'
) }}
WITH raw_events AS (
    SELECT 
        anonymous_id,
        LOWER(TRIM(user_email)) AS user_email,
        crm_contact_id,
        received_at
    FROM {{ ref('raw_web_events') }}
    WHERE anonymous_id IS NOT NULL 
      AND (user_email IS NOT NULL OR crm_contact_id IS NOT NULL)
      {% if is_incremental() %}
        AND received_at > (SELECT MAX(received_at) FROM {{ this }})
      {% endif %}
),
deduped_edges AS (
    SELECT
        MD5(CONCAT(anonymous_id, COALESCE(user_email, ''), COALESCE(crm_contact_id, ''))) AS edge_id,
        anonymous_id,
        user_email,
        crm_contact_id,
        MIN(received_at) AS first_seen_at,
        MAX(received_at) AS received_at
    FROM raw_events
    GROUP BY 1, 2, 3, 4
)
SELECT * FROM deduped_edges;

-- Asignación del canonical_user_id a partir de la tabla de aristas
WITH ranked_identities AS (
    SELECT
        anonymous_id,
        user_email,
        crm_contact_id,
        FIRST_VALUE(COALESCE(crm_contact_id, user_email, anonymous_id)) OVER (
            PARTITION BY anonymous_id 
            ORDER BY first_seen_at ASC
            ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
        ) AS canonical_user_id
    FROM {{ ref('stg_identity_edges') }}
)
SELECT DISTINCT
    canonical_user_id,
    anonymous_id,
    user_email,
    crm_contact_id
FROM ranked_identities;