如何基于数据源合并Snowflake中的重复员工数据？

我在Snowflake中编写SQL，通过UNION ALL合并Workday、Bamboo、ADP、Salesforce、Mihi五个数据源的员工信息，原查询如下：

WITH personal_info_workday AS (
    SELECT
        'Workday' AS source,
        CAST(w.worker_code AS string) AS employee_id,
        w.business_site_summary_name AS office_location
    FROM
        prod_raw.workday_prod.worker w
),
personal_info_bamboo AS (
    SELECT
        'Bamboo' AS source,
        NULL AS employee_id,
        NULL AS office_location
    FROM    
        prod_raw.bamboohr.employee e
),
personal_info_adp AS (
    SELECT
        'ADP' AS source,
        CAST(w.id AS string) AS employee_id,
        NULL AS office_location
    FROM
        prod_raw.adp_workforce_now.worker w
),
personal_info_salesforce AS (
    SELECT
        'Salesforce' AS source,
        CAST(c.workday_id__c AS string) AS employee_id,
        c.office_location__c AS office_location
    FROM 
        prod_raw.salesforce_corp_hub.contact c
    WHERE
        c.full_time_part_time__c = 'Contractor'
),
personal_info_mihi AS (
    SELECT
        'Mihi' AS source,
        CAST(e.employee_number AS string) AS employee_id,
        NULL AS office_location
    FROM 
        prod_raw.mihi.employees e
)
SELECT * FROM personal_info_workday
UNION ALL
SELECT *
FROM personal_info_salesforce s
WHERE NOT EXISTS (
    SELECT 1
    FROM personal_info_workday w
    WHERE s.employee_id = w.employee_id
)
UNION ALL
SELECT * FROM personal_info_bamboo
UNION ALL
SELECT * FROM personal_info_adp
UNION ALL
SELECT * FROM personal_info_mihi
ORDER BY employee_id asc

当前查询生成的结果：

需求：当Workday和Salesforce存在相同EMPLOYEE_ID时，仅保留Workday的行，过滤对应Salesforce重复行，其他数据源的相同ID无需处理，期望结果：

我曾尝试用NOT EXISTS过滤Salesforce重复行，但未生效：

SELECT *
FROM personal_info_salesforce s
WHERE NOT EXISTS (
    SELECT 1
    FROM personal_info_workday w
    WHERE s.employee_id = w.employee_id
)

测试用表创建及插入语句：

CREATE TABLE employee_office_location (
    source VARCHAR(50),
    employee_id VARCHAR(50),
    office_location INT,
    PRIMARY KEY (source, employee_id) -- 确保每个(source, employee_id)唯一
);
INSERT INTO employee_office_location (source, employee_id, office_location) VALUES
('ADP', '1', 1),
('Salesforce', '57', 2),
('Workday', '57', 2),
('Bamboo', '58', 3),
('Salesforce', '59', 4),
('Mihi', '1010013461', 5);

可以通过窗口函数+优先级排序的方式实现需求，先将所有数据源合并，再为每个employee_id按数据源优先级分配排序，保留排序第一的行：

WITH personal_info_workday AS (
    SELECT
        'Workday' AS source,
        CAST(w.worker_code AS string) AS employee_id,
        w.business_site_summary_name AS office_location
    FROM
        prod_raw.workday_prod.worker w
),
personal_info_bamboo AS (
    SELECT
        'Bamboo' AS source,
        NULL AS employee_id,
        NULL AS office_location
    FROM    
        prod_raw.bamboohr.employee e
),
personal_info_adp AS (
    SELECT
        'ADP' AS source,
        CAST(w.id AS string) AS employee_id,
        NULL AS office_location
    FROM
        prod_raw.adp_workforce_now.worker w
),
personal_info_salesforce AS (
    SELECT
        'Salesforce' AS source,
        CAST(c.workday_id__c AS string) AS employee_id,
        c.office_location__c AS office_location
    FROM 
        prod_raw.salesforce_corp_hub.contact c
    WHERE
        c.full_time_part_time__c = 'Contractor'
),
personal_info_mihi AS (
    SELECT
        'Mihi' AS source,
        CAST(e.employee_number AS string) AS employee_id,
        NULL AS office_location
    FROM 
        prod_raw.mihi.employees e
),
-- 合并所有数据源并添加优先级排序
combined_data AS (
    SELECT 
        *,
        -- 给Workday最高优先级，Salesforce次之，其他数据源保持原有顺序
        ROW_NUMBER() OVER (
            PARTITION BY employee_id 
            ORDER BY CASE source WHEN 'Workday' THEN 1 WHEN 'Salesforce' THEN 2 ELSE 3 END
        ) AS rn
    FROM (
        SELECT * FROM personal_info_workday
        UNION ALL
        SELECT * FROM personal_info_salesforce
        UNION ALL
        SELECT * FROM personal_info_bamboo
        UNION ALL
        SELECT * FROM personal_info_adp
        UNION ALL
        SELECT * FROM personal_info_mihi
    ) all_sources
)
-- 只保留每个employee_id的第一行（Workday优先）
SELECT source, employee_id, office_location
FROM combined_data
WHERE rn = 1
ORDER BY employee_id asc;

说明： #

1. 原NOT EXISTS方法未生效的原因：可能是Workday的employee_id与Salesforce的employee_id存在隐式类型转换差异，或者Workday数据源中该ID的记录未被正确查询到。窗口函数的方式可彻底避免这类问题，直接按优先级保留所需行。
2. 优先级逻辑：通过CASE语句给Workday分配排序值1（最高），Salesforce为2，其他数据源为3。同一employee_id下，排序值最小的行会被保留，因此Workday的记录会覆盖同ID的Salesforce记录，其他数据源的同ID记录不受影响。
3. 测试验证：将上述逻辑应用到测试表时，可直接替换CTE部分为测试表查询：

WITH combined_data AS (
    SELECT 
        *,
        ROW_NUMBER() OVER (
            PARTITION BY employee_id 
            ORDER BY CASE source WHEN 'Workday' THEN 1 WHEN 'Salesforce' THEN 2 ELSE 3 END
        ) AS rn
    FROM employee_office_location
)
SELECT source, employee_id, office_location
FROM combined_data
WHERE rn = 1
ORDER BY employee_id asc;

执行后即可得到期望的结果。

内容的提问来源于stack exchange，提问作者NidenK