Prod Traceability

---

• sub_versions_list is a list of sub_versions, e.g. ['2024_11', '2024_10', '2024_09', '2024_08']

-- 🪟💲🌟
-- run_id: {{ run_id }}
-- task_name: {{ task_name }}
-- Table: {{ schema_name }}.prod_traceability_to_raw_data


{% set sub_versions = sub_versions_list | sort(reverse=True) %}

CREATE TABLE IF NOT EXISTS {{ schema_name }}.prod_traceability_to_raw_data 

WITH (
    PARTITIONING = ARRAY['payer_id']
)
AS
WITH df AS (
    -- Build payer_rates UNIONs
    {% for sub_version in sub_versions %}
    SELECT DISTINCT
        roid,
        payer_id,
        core_rates_id as raw_id,
        'payer' as source,
        '{{ sub_version }}' as sub_version,
        file_hash as payer_file_hash,
        location_data_source_name as payer_location_data_source_name,
        last_updated_on as payer_last_updated_on,
        NULL as hospital_file_id,
        NULL as hospital_ingested_on,
        NULL as hospital_loaded_on,
        NULL as hospital_filename,
        NULL as hospital_version
    FROM {{ schema_name }}.tmp_raw_mrf_payer_rates_{{ sub_version }}
    {% if not loop.last %}UNION ALL{% endif %}
    {% endfor %}

    UNION ALL

    -- Build hospital_rates UNIONs
    {% for sub_version in sub_versions %}
    SELECT
        roid, 
        payer_id,
        hospital_rates_id as raw_id,
        'hospital' as source,
        '{{ sub_version }}' as sub_version,
        NULL as payer_file_hash,
        NULL as payer_location_data_source_name,
        NULL as payer_last_updated_on,
        file_id as hospital_file_id,
        ingested_on as hospital_ingested_on,
        loaded_on as hospital_loaded_on,
        filename as hospital_filename,
        version as hospital_version
    FROM {{ schema_name }}.tmp_raw_mrf_hospital_rates_{{ sub_version }}
    {% if not loop.last %}UNION ALL{% endif %}
    {% endfor %}
)

SELECT 
    CAST(uuid() AS VARCHAR) as traceability_id,
    roid,
    sub_version,
    source,
    raw_id,
    payer_file_hash,
    payer_location_data_source_name,
    payer_last_updated_on,
    hospital_file_id,
    hospital_ingested_on,
    hospital_loaded_on,
    hospital_filename,
    hospital_version,
    ROW_NUMBER() OVER (PARTITION BY payer_id, raw_id, sub_version, source, roid) as rn,
    payer_id
FROM df