-- Databricks notebook source
-- MAGIC %md
-- MAGIC # ATC/NFC DWS hierarchy validation
-- MAGIC Run this notebook after both DWS hierarchy scripts finish.
-- MAGIC The DWS outputs and legacy DWD baselines must represent the same source snapshot.

-- COMMAND ----------

DESCRIBE TABLE dws.dws_ext_td_ims_atc_hierarchy;

-- COMMAND ----------

DESCRIBE TABLE dws.dws_ext_td_ims_nfc_hierarchy;

-- COMMAND ----------

-- Fix the comparison column set and order at the view boundary.
CREATE OR REPLACE TEMP VIEW validation_atc_legacy AS
SELECT
    ATC1_ID,
    ATC1_CODE,
    ATC1_DES,
    ATC2_ID,
    ATC2_CODE,
    ATC2_DES,
    ATC3_ID,
    ATC3_CODE,
    ATC3_DES,
    ATC4_ID,
    ATC4_CODE,
    ATC4_DES
FROM dwd.dwd_ims_atc_hierarchy;

-- COMMAND ----------

CREATE OR REPLACE TEMP VIEW validation_atc_dws AS
SELECT
    ATC1_ID,
    ATC1_CODE,
    ATC1_DES,
    ATC2_ID,
    ATC2_CODE,
    ATC2_DES,
    ATC3_ID,
    ATC3_CODE,
    ATC3_DES,
    ATC4_ID,
    ATC4_CODE,
    ATC4_DES
FROM dws.dws_ext_td_ims_atc_hierarchy;

-- COMMAND ----------

CREATE OR REPLACE TEMP VIEW validation_nfc_legacy AS
SELECT
    NFC1_ID,
    NFC1_CODE,
    NFC1_DES,
    NFC2_ID,
    NFC2_CODE,
    NFC2_DES,
    NFC3_ID,
    NFC3_CODE,
    NFC3_DES
FROM dwd.dwd_ims_nfc_hierarchy;

-- COMMAND ----------

CREATE OR REPLACE TEMP VIEW validation_nfc_dws AS
SELECT
    NFC1_ID,
    NFC1_CODE,
    NFC1_DES,
    NFC2_ID,
    NFC2_CODE,
    NFC2_DES,
    NFC3_ID,
    NFC3_CODE,
    NFC3_DES
FROM dws.dws_ext_td_ims_nfc_hierarchy;

-- COMMAND ----------

CREATE OR REPLACE TEMP VIEW validation_hierarchy_compatibility_checks AS
WITH compatibility_checks AS (
    SELECT
        'atc_row_count' AS check_name,
        (SELECT COUNT(*) FROM validation_atc_dws) AS actual_value,
        (SELECT COUNT(*) FROM validation_atc_legacy) AS expected_value
    UNION ALL
    SELECT
        'atc_legacy_minus_dws' AS check_name,
        (
            SELECT COUNT(*)
            FROM (
                SELECT * FROM validation_atc_legacy
                EXCEPT ALL
                SELECT * FROM validation_atc_dws
            ) AS differences
        ) AS actual_value,
        0 AS expected_value
    UNION ALL
    SELECT
        'atc_dws_minus_legacy' AS check_name,
        (
            SELECT COUNT(*)
            FROM (
                SELECT * FROM validation_atc_dws
                EXCEPT ALL
                SELECT * FROM validation_atc_legacy
            ) AS differences
        ) AS actual_value,
        0 AS expected_value
    UNION ALL
    SELECT
        'nfc_row_count' AS check_name,
        (SELECT COUNT(*) FROM validation_nfc_dws) AS actual_value,
        (SELECT COUNT(*) FROM validation_nfc_legacy) AS expected_value
    UNION ALL
    SELECT
        'nfc_legacy_minus_dws' AS check_name,
        (
            SELECT COUNT(*)
            FROM (
                SELECT * FROM validation_nfc_legacy
                EXCEPT ALL
                SELECT * FROM validation_nfc_dws
            ) AS differences
        ) AS actual_value,
        0 AS expected_value
    UNION ALL
    SELECT
        'nfc_dws_minus_legacy' AS check_name,
        (
            SELECT COUNT(*)
            FROM (
                SELECT * FROM validation_nfc_dws
                EXCEPT ALL
                SELECT * FROM validation_nfc_legacy
            ) AS differences
        ) AS actual_value,
        0 AS expected_value
    UNION ALL
    SELECT
        'except_all_null_semantics' AS check_name,
        (
            SELECT COUNT(*)
            FROM (
                SELECT CAST(NULL AS STRING) AS nullable_value
                EXCEPT ALL
                SELECT CAST(NULL AS STRING) AS nullable_value
            ) AS differences
        ) AS actual_value,
        0 AS expected_value
)
SELECT
    check_name,
    actual_value,
    expected_value,
    actual_value = expected_value AS passed
FROM compatibility_checks;

-- COMMAND ----------

SELECT
    check_name,
    actual_value,
    expected_value,
    passed
FROM validation_hierarchy_compatibility_checks
ORDER BY check_name;

-- COMMAND ----------

-- These quality metrics require business review; they do not fail automatically.
WITH atc_duplicate_paths AS (
    SELECT COUNT(*) AS path_count
    FROM validation_atc_dws
    GROUP BY ATC1_CODE, ATC2_CODE, ATC3_CODE, ATC4_CODE
    HAVING COUNT(*) > 1
),
nfc_duplicate_paths AS (
    SELECT COUNT(*) AS path_count
    FROM validation_nfc_dws
    GROUP BY NFC1_CODE, NFC2_CODE, NFC3_CODE
    HAVING COUNT(*) > 1
)
SELECT
    'atc_duplicate_path_rows' AS metric_name,
    COALESCE(SUM(path_count - 1), 0) AS metric_value
FROM atc_duplicate_paths
UNION ALL
SELECT
    'atc_source_level_1_required_key_null' AS metric_name,
    COUNT_IF(therapeutic_id IS NULL OR therapeutic_code IS NULL) AS metric_value
FROM dwd.dwd_ims_td_therapeutic_class
WHERE therapeutic_level = '1'
UNION ALL
SELECT
    'atc_unmatched_level_2' AS metric_name,
    COUNT_IF(ATC2_CODE IS NULL) AS metric_value
FROM validation_atc_dws
UNION ALL
SELECT
    'atc_unmatched_level_3' AS metric_name,
    COUNT_IF(ATC3_CODE IS NULL) AS metric_value
FROM validation_atc_dws
UNION ALL
SELECT
    'atc_unmatched_level_4' AS metric_name,
    COUNT_IF(ATC4_CODE IS NULL) AS metric_value
FROM validation_atc_dws
UNION ALL
SELECT
    'nfc_duplicate_path_rows' AS metric_name,
    COALESCE(SUM(path_count - 1), 0) AS metric_value
FROM nfc_duplicate_paths
UNION ALL
SELECT
    'nfc_source_level_1_required_key_null' AS metric_name,
    COUNT_IF(newformclass_id IS NULL OR newformclass_code IS NULL) AS metric_value
FROM dwd.dwd_ims_td_new_form_class
WHERE newformclass_level = '1'
UNION ALL
SELECT
    'nfc_unmatched_level_2' AS metric_name,
    COUNT_IF(NFC2_CODE IS NULL) AS metric_value
FROM validation_nfc_dws
UNION ALL
SELECT
    'nfc_unmatched_level_3' AS metric_name,
    COUNT_IF(NFC3_CODE IS NULL) AS metric_value
FROM validation_nfc_dws
ORDER BY metric_name;

-- COMMAND ----------

-- Block downstream migration when any DWS-to-DWD compatibility check fails.
SELECT IF(
    COUNT_IF(NOT passed) = 0,
    'All DWS hierarchy compatibility checks passed.',
    raise_error(
        CONCAT(
            'DWS hierarchy compatibility validation failed: ',
            CAST(COUNT_IF(NOT passed) AS STRING),
            ' check(s) did not pass.'
        )
    )
) AS validation_result
FROM validation_hierarchy_compatibility_checks;