MDM Migration Guide: Informatica MDM to Purview + Azure SQL / Cosmos DB¶

-- models/mdm/stg_mdm__customer_match_candidates.sql
-- Prepare standardized match keys for candidate identification

SELECT
    customer_id,
    source_system,
    -- Standardize for matching
    UPPER(TRIM(first_name)) AS first_name_std,
    UPPER(TRIM(last_name)) AS last_name_std,
    LOWER(TRIM(email)) AS email_std,
    REPLACE(REPLACE(REPLACE(phone, '-', ''), '(', ''), ')', '') AS phone_std,
    UPPER(TRIM(city)) AS city_std,
    UPPER(TRIM(state)) AS state_std,
    REPLACE(REPLACE(postal_code, '-', ''), ' ', '') AS postal_code_std,

    -- Create match keys (blocking keys for candidate selection)
    CONCAT(UPPER(LEFT(last_name, 4)), UPPER(LEFT(first_name, 2)), LEFT(postal_code, 5)) AS match_key_1,
    email AS match_key_2,
    REPLACE(REPLACE(REPLACE(phone, '-', ''), '(', ''), ')', '') AS match_key_3

FROM {{ ref('stg_crm__customers') }}

-- models/mdm/int_mdm__customer_match_pairs.sql
-- Find candidate pairs using blocking keys

WITH exact_email_match AS (
    SELECT
        a.customer_id AS customer_id_a,
        b.customer_id AS customer_id_b,
        'email_exact' AS match_type,
        1.0 AS match_score
    FROM {{ ref('stg_mdm__customer_match_candidates') }} a
    JOIN {{ ref('stg_mdm__customer_match_candidates') }} b
        ON a.email_std = b.email_std
        AND a.customer_id < b.customer_id  -- avoid self-match and duplicates
    WHERE a.email_std IS NOT NULL
      AND a.email_std != ''
),

exact_phone_match AS (
    SELECT
        a.customer_id AS customer_id_a,
        b.customer_id AS customer_id_b,
        'phone_exact' AS match_type,
        0.9 AS match_score
    FROM {{ ref('stg_mdm__customer_match_candidates') }} a
    JOIN {{ ref('stg_mdm__customer_match_candidates') }} b
        ON a.phone_std = b.phone_std
        AND a.customer_id < b.customer_id
    WHERE a.phone_std IS NOT NULL
      AND LEN(a.phone_std) >= 10
),

name_address_match AS (
    SELECT
        a.customer_id AS customer_id_a,
        b.customer_id AS customer_id_b,
        'name_address' AS match_type,
        0.85 AS match_score
    FROM {{ ref('stg_mdm__customer_match_candidates') }} a
    JOIN {{ ref('stg_mdm__customer_match_candidates') }} b
        ON a.last_name_std = b.last_name_std
        AND a.first_name_std = b.first_name_std
        AND a.postal_code_std = b.postal_code_std
        AND a.customer_id < b.customer_id
)

SELECT * FROM exact_email_match
UNION ALL
SELECT * FROM exact_phone_match
UNION ALL
SELECT * FROM name_address_match

-- models/mdm/int_mdm__customer_match_groups.sql
-- Assign match group IDs using transitive closure

WITH RECURSIVE match_closure AS (
    -- Base: each customer is in its own group
    SELECT
        customer_id_a AS customer_id,
        customer_id_a AS group_root
    FROM {{ ref('int_mdm__customer_match_pairs') }}
    WHERE match_score >= 0.85  -- threshold

    UNION ALL

    -- Recursive: extend groups through match pairs
    SELECT
        p.customer_id_b AS customer_id,
        mc.group_root
    FROM {{ ref('int_mdm__customer_match_pairs') }} p
    JOIN match_closure mc
        ON p.customer_id_a = mc.customer_id
    WHERE p.match_score >= 0.85
)

SELECT
    customer_id,
    MIN(group_root) AS match_group_id
FROM match_closure
GROUP BY customer_id

-- models/mdm/mart_mdm__customer_golden.sql
-- Create golden record using survivorship (trust) rules

WITH ranked AS (
    SELECT
        c.*,
        g.match_group_id,
        -- Source priority: CRM > ERP > Legacy
        CASE c.source_system
            WHEN 'CRM' THEN 1
            WHEN 'ERP' THEN 2
            WHEN 'LEGACY' THEN 3
            ELSE 4
        END AS source_priority,
        -- Recency: prefer most recently updated
        ROW_NUMBER() OVER (
            PARTITION BY g.match_group_id
            ORDER BY
                CASE c.source_system WHEN 'CRM' THEN 1 WHEN 'ERP' THEN 2 ELSE 3 END,
                c.updated_at DESC
        ) AS rn
    FROM {{ ref('stg_crm__customers') }} c
    JOIN {{ ref('int_mdm__customer_match_groups') }} g
        ON c.customer_id = g.customer_id
)

SELECT
    match_group_id AS master_customer_id,
    -- Survivorship: best value per attribute
    MAX(CASE WHEN source_priority = 1 THEN first_name END) AS first_name,
    MAX(CASE WHEN source_priority = 1 THEN last_name END) AS last_name,
    -- Email: prefer CRM, then most recent non-null
    COALESCE(
        MAX(CASE WHEN source_system = 'CRM' AND email IS NOT NULL THEN email END),
        MAX(CASE WHEN email IS NOT NULL THEN email END)
    ) AS email,
    -- Phone: prefer most recent non-null
    MAX(CASE WHEN rn = 1 THEN phone END) AS phone,
    -- Address: prefer CRM source
    MAX(CASE WHEN source_priority = 1 THEN street_line_1 END) AS street_line_1,
    MAX(CASE WHEN source_priority = 1 THEN city END) AS city,
    MAX(CASE WHEN source_priority = 1 THEN state END) AS state,
    MAX(CASE WHEN source_priority = 1 THEN postal_code END) AS postal_code,
    -- Metadata
    COUNT(*) AS source_record_count,
    STRING_AGG(DISTINCT source_system, ', ') AS contributing_sources,
    MAX(updated_at) AS last_updated
FROM ranked
GROUP BY match_group_id

-- Create hierarchy table using HierarchyID
CREATE TABLE dbo.organization_hierarchy (
    org_id INT PRIMARY KEY,
    org_name NVARCHAR(200),
    org_type NVARCHAR(50),
    hierarchy_node HIERARCHYID NOT NULL,
    hierarchy_level AS hierarchy_node.GetLevel() PERSISTED,
    parent_node AS hierarchy_node.GetAncestor(1) PERSISTED,
    CONSTRAINT UQ_hierarchy_node UNIQUE (hierarchy_node)
);

-- Index for efficient descendant queries
CREATE INDEX IX_hierarchy_depth_first
ON dbo.organization_hierarchy (hierarchy_node);

-- Index for efficient breadth-first (sibling) queries
CREATE INDEX IX_hierarchy_breadth_first
ON dbo.organization_hierarchy (hierarchy_level, hierarchy_node);

-- models/mdm/mart_mdm__org_hierarchy.sql
-- Portable hierarchy using recursive CTE (works on any SQL engine)

WITH RECURSIVE org_tree AS (
    -- Root nodes
    SELECT
        org_id,
        org_name,
        org_type,
        parent_org_id,
        0 AS hierarchy_level,
        CAST(org_name AS VARCHAR(4000)) AS hierarchy_path
    FROM {{ ref('stg_org__organizations') }}
    WHERE parent_org_id IS NULL

    UNION ALL

    -- Child nodes
    SELECT
        c.org_id,
        c.org_name,
        c.org_type,
        c.parent_org_id,
        p.hierarchy_level + 1,
        CAST(CONCAT(p.hierarchy_path, ' > ', c.org_name) AS VARCHAR(4000))
    FROM {{ ref('stg_org__organizations') }} c
    JOIN org_tree p ON c.parent_org_id = p.org_id
)

SELECT * FROM org_tree

-- models/mdm/int_mdm__match_features.sql
-- Prepare features for ML-based entity resolution

SELECT
    a.customer_id AS id_a,
    b.customer_id AS id_b,

    -- Name similarity (Jaro-Winkler via SQL CLR or Python UDF)
    dbo.fn_jaro_winkler(a.last_name_std, b.last_name_std) AS name_jaro_winkler_score,

    -- Email exact match
    CASE WHEN a.email_std = b.email_std AND a.email_std IS NOT NULL THEN 1 ELSE 0 END AS email_exact_match,

    -- Phone match
    CASE WHEN a.phone_std = b.phone_std AND a.phone_std IS NOT NULL THEN 1 ELSE 0 END AS phone_match,

    -- Postal code match
    CASE WHEN a.postal_code_std = b.postal_code_std THEN 1 ELSE 0 END AS postal_code_match,

    -- City match
    CASE WHEN a.city_std = b.city_std THEN 1 ELSE 0 END AS city_match

FROM {{ ref('stg_mdm__customer_match_candidates') }} a
JOIN {{ ref('stg_mdm__customer_match_candidates') }} b
    ON a.blocking_key = b.blocking_key  -- blocking key to reduce pairs
    AND a.customer_id < b.customer_id

-- Stewardship audit table
CREATE TABLE mdm.stewardship_audit (
    audit_id INT IDENTITY(1,1) PRIMARY KEY,
    action_type VARCHAR(20),  -- 'merge', 'reject', 'edit', 'approve'
    entity_type VARCHAR(50),  -- 'customer', 'product', 'vendor'
    entity_id_a INT,
    entity_id_b INT,
    steward_email VARCHAR(200),
    decision_reason VARCHAR(500),
    before_state NVARCHAR(MAX),  -- JSON snapshot before change
    after_state NVARCHAR(MAX),   -- JSON snapshot after change
    decided_at DATETIME2 DEFAULT GETDATE()
);