refactor CHPA 01 and 02 warehouse jobs

document CHPA warehouse refactor rules
2026-08-20 19:39:11 +08:00 · 2026-08-20 19:17:19 +08:00
26 changed files with 4630 additions and 42 deletions
@@ -0,0 +1,148 @@
 # AGENTS.md
 ## Repository purpose
 This repository contains the maintainable Databricks SQL replacement for the legacy CHPA warehouse jobs.
 - Treat the workspace-level `CHPA/` directory as read-only legacy input.
 - Put all rewritten code, documentation, and validation under `RE/`.
 - Preserve business behavior first. Separate logic refactoring, physical renaming, and business-rule changes.
 - Keep each persisted target in one clearly owned script whenever practical.
 ## Warehouse layers
 The warehouse has exactly three persistent layers and data flows in one direction:
 ```text
 DWD -> DWS -> DM
 ```
 ### DWD
 - Cleaned atomic details and foundational master data.
 - Retains source traceability and stable source grain.
 - Does not contain report-oriented hierarchy flattening or subject metrics.
 ### DWS
 - Reusable dimensions, hierarchy-wide tables, and reusable facts built from DWD.
 - Standardizes shared codes, joins, and business definitions.
 - A job that reads DWD and produces a reusable derived table writes to DWS.
 ### DM
 - Subject-area datasets, metrics, and report-facing outputs built from DWS.
 - Must not become an upstream dependency of DWD or DWS.
 DWD must not read DWS/DM. DWS must not read DM. DWS/DM derived results must not be written back to DWD, and DM derived results must not be written back to DWS.
 ## Table naming
 Use the schema that matches the target layer and one of these fixed table prefixes:
 | Layer | Object type | Required form |
 | --- | --- | --- |
 | DWS | Dimension/master data | `dws.dws_ext_td_<business_entity>` |
 | DWS | Fact data | `dws.dws_ext_tf_<business_entity>` |
 | DM | Dimension/master data | `dm.dm_ext_td_<business_entity>` |
 | DM | Fact data | `dm.dm_ext_tf_<business_entity>` |
 Naming rules:
 - Use lowercase snake_case table names.
 - `td` means dimension/master data; `tf` means fact data.
 - Include the source/domain at the start of the business entity when it avoids ambiguity, for example `ims_atc_hierarchy`.
 - Do not write a newly derived DWS table with a `dwd_` name or into the `dwd` schema.
 - Keep legacy output column names during a logic-only migration. Rename columns only through an explicit coordinated contract change.
 Current hierarchy targets:
 - `dws.dws_ext_td_ims_atc_hierarchy`
 - `dws.dws_ext_td_ims_nfc_hierarchy`
 ## File layout and order
 Use:
 ```text
 sql/<domain>/<stage>/<sequence>_<target_table>.sql
 validation/<domain>/<stage>/<validation_name>.sql
 ```
 Rules:
 - DWD jobs belong in `01_dwd`, DWS jobs in `02_dws`, and DM jobs in `03_dm`.
 - Use two-digit sequence numbers to make execution order explicit.
 - Use lowercase snake_case paths with no spaces.
 - Keep Databricks notebook markers and split executable statements with `-- COMMAND ----------`.
 - Split legacy multi-target scripts by target or responsibility when doing so does not alter transactional behavior.
 ## SQL contracts
 Every persisted build script must include a concise header with:
 - purpose;
 - source tables;
 - target table;
 - output grain;
 - write mode;
 - legacy table replaced;
 - known consumers or migration dependency.
 SQL rules:
 - Use uppercase SQL keywords and descriptive lowercase snake_case CTE aliases.
 - Use four-space indentation and one selected column per line.
 - Use explicit target columns and explicit final projection columns.
 - Never use `SELECT *` in a persisted write.
 - Qualify columns whenever multiple relations are in scope.
 - Replace post-write null cleanup with projection expressions such as `COALESCE` when semantics are unchanged.
 - Replace update-driven staging with ordered CTEs only after proving the result is equivalent.
 - Keep one-time target DDL in a commented, separate Databricks cell above the write.
 - Prefer `CREATE TABLE IF NOT EXISTS <new_table> LIKE <verified_legacy_table>` when the legacy schema is the authoritative type contract.
 - For a stateful rolling/incremental target, seed retained state with an approved clone or backfill before the first run; an empty `LIKE` table is not sufficient.
 - When a legacy `SELECT *` contract is unknown, define and document a deliberate narrowed contract with zero-row CTAS type inference instead of creating unknown columns that remain NULL.
 - Do not guess source field types when no DDL or `DESCRIBE` output is available.
 ## Validation
 Before migrating downstream consumers:
 - compare old and new row counts on the same source snapshot;
 - run `EXCEPT ALL` in both directions with a fixed column order;
 - profile duplicate business keys or paths;
 - count null required keys;
 - count unmatched hierarchy/join levels;
 - make compatibility checks fail the notebook with `raise_error()` when any required check fails.
 Keep old physical tables as validation baselines until downstream migration is complete. Do not add new downstream dependencies to legacy derived tables.
 ## Git workflow
 Repository:
 ```text
 https://git.chenwuzhu.cn/chenwu/REFACTOR-MA.git
 ```
 Commit identity for this checkout:
 ```text
 chenwu <zhuchenwu@chenwuzhu.cn>
 ```
 Workflow rules:
 1. Work on the local `main` branch unless the user requests a review branch.
 2. Check `git status` before editing and never discard unrelated user changes.
 3. Run whitespace/static checks and inspect staged diffs before committing.
 4. Use focused imperative commit messages that describe the warehouse change.
 5. Push to `origin/main` only after the requested batch is complete and locally verified.
 6. Never force-push or rewrite shared history without explicit approval.
 7. Report the pushed commit hash and any Databricks validation that could not be executed locally.
 Recent baseline commits:
 - `3f3dc44`: initial hierarchy SQL refactor.
 - `33e60d0`: move hierarchy outputs to DWS naming.
 - `b0d5f81`: add commented one-time DDL templates.
@@ -2,69 +2,67 @@
 Databricks 数仓 SQL 重构仓库。
-## 数仓分层
+## 设计入口
-仓库采用三层模型，数据依赖保持单向流动：
+- Agent 和工程约束：`AGENTS.md`
 - SQL 重构规范：`docs/sql_refactoring_standard.md`
 - CHPA 01/02 迁移矩阵：`docs/chpa_01_02_migration.md`
 工作区上级 `CHPA/` 目录是只读 legacy 输入；所有新代码、文档和验证都放在本仓库。
 ## 数仓分层
 ```text
 DWD -> DWS -> DM
 ```
 - `DWD`：清洗后的原子明细和基础主数据。
- `DWS`：引用 DWD 构建的公共维度、宽表和可复用事实表。
+- `DWS`：引用 DWD 构建的公共维度、层级宽表和可复用事实表。
- `DM`：面向具体分析主题、指标和报表的数据集市。
+- `DM`：引用 DWS 构建的主题、指标和报表数据集。
-## 首批 review 范围
+固定表名前缀：
-本批次重构原 `CHPA/01` 中两个层级维表脚本：
+- DWS 维度：`dws.dws_ext_td_<业务实体>`
 - DWS 事实：`dws.dws_ext_tf_<业务实体>`
 - DM 维度：`dm.dm_ext_td_<业务实体>`
 - DM 事实：`dm.dm_ext_tf_<业务实体>`
- `sql/chpa/02_dws/01_dws_ext_td_ims_atc_hierarchy.sql`
+## 当前重构范围
 - `sql/chpa/02_dws/02_dws_ext_td_ims_nfc_hierarchy.sql`
-两个脚本读取 DWD，并写入新的 DWS 维度表：
+本批覆盖 `CHPA/` 下全部 6 个 `01` 开头脚本和 11 个 `02` 开头脚本。
- `dws.dws_ext_td_ims_atc_hierarchy`
+主要变化：
 - `dws.dws_ext_td_ims_nfc_hierarchy`
-工作区 `CHPA/` 下的原脚本保持不变。旧表 `dwd.dwd_ims_atc_hierarchy` 和 `dwd.dwd_ims_nfc_hierarchy` 仅作为本批新旧结果校验基线。
+- DWD 配置标准化和外部事实入湖保留在 `sql/chpa/01_dwd/`。
-
+- ATC/NFC hierarchy、pack property、中文维度和 market 等派生宽表迁移到 DWS。
-## 表命名
+- `tmp.tmp_ims_tf_fact_sales` 提升为 `dws.dws_ext_tf_ims_chpa_sales`。
-
+- 产品排序由共享临时表和 UPDATE 改为声明式 DWS 标记。
- DWS 维度表：`dws_ext_td_<业务实体>`
+- 每个持久化脚本包含注释形式的一次性建表语句、显式写入字段和脚本契约头。
 - DWS 事实表：`dws_ext_tf_<业务实体>`
 - DM 维度表：`dm_ext_td_<业务实体>`
 - DM 事实表：`dm_ext_tf_<业务实体>`
 完整规范及首批表名映射见 `docs/sql_refactoring_standard.md`。
 ## 目录结构
 ```text
 RE/
 |-- AGENTS.md
 |-- README.md
 |-- docs/
-|   `-- sql_refactoring_standard.md
+|   |-- sql_refactoring_standard.md
 |   `-- chpa_01_02_migration.md
 |-- sql/
 |   `-- chpa/
 |       |-- 01_dwd/
 |       `-- 02_dws/
 |           |-- 01_dws_ext_td_ims_atc_hierarchy.sql
 |           `-- 02_dws_ext_td_ims_nfc_hierarchy.sql
 `-- validation/
    `-- chpa/
        `-- 02_dws/
            `-- validate_hierarchy_refactor.sql
 ```
-## 执行顺序
+## 执行和迁移
-1. 确认两个 DWS 目标表已按脚本中的显式字段契约创建。
+1. 按 `docs/chpa_01_02_migration.md` 中的顺序执行 DWD 脚本。
-2. 按文件名前缀顺序执行 `sql/chpa/02_dws/` 中的脚本。
+2. 确认 DWS 目标表已通过各脚本中的注释 DDL 创建。
-3. 运行 `validation/chpa/02_dws/validate_hierarchy_refactor.sql`，直接比较新 DWS 输出与旧 DWD 基线。
+3. 按 `sql/chpa/02_dws/` 文件序号执行 DWS 脚本。
-4. 兼容性检查必须全部返回 `passed = true`；同时 review 重复路径和未匹配层级指标。
+4. 在相同源数据快照上运行 `validation/chpa/02_dws/` 下的验证 notebook。
-5. 下游迁移完成前保留旧 DWD 表；下游不得继续新增对旧表的依赖。
+5. 新旧行数、双向差集和必填键检查通过后，再迁移下游消费者。
-## 本次 review 重点
+本批不修改 `03` 开头脚本。`03` 中仍存在对旧 DWD/DWS/tmp 表的引用；在下一批完成消费者迁移之前，旧表必须保留为兼容和验证基线。
 1. `DWD -> DWS -> DM` 的职责边界是否符合现有调度设计。
 2. `dws_ext_td_ims_*` 中是否需要保留来源系统 `ims`。
 3. DWS 目标表的 catalog、字段类型、表属性和权限是否需要统一 DDL 模板。
@@ -0,0 +1,104 @@
 # CHPA 01/02 重构迁移清单
 ## 1. 范围
 本批覆盖工作区 `CHPA/` 下全部 6 个 `01` 开头脚本和 11 个 `02` 开头脚本。原文件保持只读，新代码写入 `RE/sql/chpa/01_dwd` 和 `RE/sql/chpa/02_dws`。
 重构遵循：
 ```text
 DWD -> DWS -> DM
 ```
 本批不修改 `03` 开头脚本。所有 `03` 消费者的旧表引用必须在下一批统一迁移，旧表在此之前保留为验证基线。
 ## 2. 旧文件到新文件映射
 | 旧文件 | 新文件 | 新目标或处理方式 |
 | --- | --- | --- |
 | `01 dwd_update.sql` | `01_dwd/01_standardize_gnd_codes.sql` | DWD 配置编码标准化 |
 | `01 dwd_update.sql` | `01_dwd/02_refresh_ims_market_config.sql` | 保留 `dwd.dwd_gnd_ims_tblmarket` |
 | `01 dwd_update.sql` | `01_dwd/03_refresh_retail_market_config.sql` | 保留 `dwd.dwd_gnd_ext_retail_tblmarket` |
 | `01 dwd_update.sql` | `01_dwd/04_refresh_dtp_market_config.sql` | 保留 `dwd.dwd_gnd_dtp_tblmarket` |
 | `01 dwd_update.sql` | `01_dwd/05_normalize_gnd_time_windows.sql` | DWD 时间窗口标准化，保留旧默认值差异 |
 | `01 dwd_ims_td_manufacturer_corp.sql` | `01_dwd/06_fix_ims_manufacturer_type.sql` | DWD `ManufacturerType_ID=0 -> 2` 修复 |
 | `01_FB_BLOB_TO_DWD.sql` | `01_dwd/07_ingest_pharbers_province_fact.sql` | 保留 `dwd.dwd_gnd_pharbers_prov_fact` |
 | `01 dwd_ims_atc_hierarchy.sql` | `02_dws/01_dws_ext_td_ims_atc_hierarchy.sql` | `dws.dws_ext_td_ims_atc_hierarchy` |
 | `01 dwd_ims_nfc_hierarchy.sql` | `02_dws/02_dws_ext_td_ims_nfc_hierarchy.sql` | `dws.dws_ext_td_ims_nfc_hierarchy` |
 | `01 dwd_ims_td_manufacturer_corp.sql` | `02_dws/03_dws_ext_td_ims_manufacturer_corporation.sql` | `dws.dws_ext_td_ims_manufacturer_corporation` |
 | `01 dwd_ims_td_pack_property.sql` | `02_dws/04_dws_ext_td_ims_pack_property.sql` | `dws.dws_ext_td_ims_pack_property` |
 | `02 DWS_IMS_TD_GEO.sql` | `02_dws/05_dws_ext_td_ims_geo.sql` | `dws.dws_ext_td_ims_geo` |
 | `02 dws_ims_td_corp_cn.sql` | `02_dws/06_dws_ext_td_ims_corporation_cn.sql` | `dws.dws_ext_td_ims_corporation_cn` |
 | `02 dws_ims_td_manu_cn.sql` | `02_dws/07_dws_ext_td_ims_manufacturer_cn.sql` | `dws.dws_ext_td_ims_manufacturer_cn` |
 | `02 tmp_ims_td_prod_tmp.sql` | `02_dws/08_dws_ext_td_ims_product_multi_manufacturer.sql` | 将临时名单升级为声明式 DWS 标记维表 |
 | `02 dws_ims_td_prod_cn.sql` | `02_dws/09_dws_ext_td_ims_product_cn.sql` | `dws.dws_ext_td_ims_product_cn`，声明式生成 `RANK_TYPE` |
 | `02 dws_ims_td_atc_cn.sql` | `02_dws/10_dws_ext_td_ims_atc_cn.sql` | `dws.dws_ext_td_ims_atc_cn` |
 | `02 dws_ims_td_nfc_cn.sql` | `02_dws/11_dws_ext_td_ims_nfc_cn.sql` | `dws.dws_ext_td_ims_nfc_cn` |
 | `02 dws_ims_td_market_ta.sql` | `02_dws/12_dws_ext_td_ims_market_ta.sql` | `dws.dws_ext_td_ims_market_ta` |
 | `02 tmp_ims_tf_fact_sales.sql` | `02_dws/13_dws_ext_td_ims_pack_ym.sql` | `dws.dws_ext_td_ims_pack_ym` |
 | `02 tmp_ims_tf_fact_sales.sql` | `02_dws/14_dws_ext_tf_ims_chpa_sales.sql` | `dws.dws_ext_tf_ims_chpa_sales` |
 | `02 dws_ims_td_date.sql` | `02_dws/15_dws_ext_td_ims_date.sql` | `dws.dws_ext_td_ims_date` |
 | `02 dws_ims_td_market.sql` | `02_dws/16_dws_ext_td_ims_market.sql` | `dws.dws_ext_td_ims_market` |
 ## 3. 执行顺序
 ### DWD
 1. 标准化 GND 配置编码。
 2. 刷新 IMS、Retail、DTP 市场配置快照。
 3. 按旧业务口径补齐各配置表时间窗口。
 4. 修复 manufacturer 类型。
 5. Pharbers 省级事实入湖任务独立运行；无匹配文件时必须失败，不能复用旧 staging 数据。
 ### DWS
 1. ATC/NFC hierarchy。
 2. Manufacturer-corporation mapping。
 3. Pack property。
 4. Geo、corporation/manufacturer 中文维度。
 5. 多厂家产品标记、产品中文维度。
 6. ATC/NFC 中文维度、market-TA mapping。
 7. Pack 月快照、CHPA sales 事实、日期维度。
 8. Market 维度。
 ## 4. 明确保留的兼容行为
 - ATC/NFC 父级编码截取长度保持不变。
 - Pack、product、molecule 编码补零规则保持不变。
 - A5Z/A5ZD corporation/manufacturer 覆盖规则保持不变。
 - GND 时间默认值中 `190001/209901` 与 `200001/299912` 的历史差异暂不统一。
 - CHPA sales 全国数据从 `202201` 开始，省级数据不增加该下限。
 - Market 扩展市场 ratio 的旧脚本净结果保持为 `1`；缺失 `ELSE` 的疑点只记录，不在本批修复。
 - 日期维度保留 YYYYMM 值 `max(YM)-900`（约九年）的旧算法，尽管旧注释称“最近五年”。
 ## 5. 本批修复的工程问题
 - DWD 派生宽表迁移到 DWS，并使用 `dws_ext_td_` / `dws_ext_tf_` 命名。
 - 持久化写入使用显式目标字段和最终投影。
 - 一次性建表语句以注释形式放在独立 Databricks cell。
 - 产品排序从“临时表 + UPDATE”改为声明式构建。
 - CHPA sales 从共享 `tmp` 表提升为正式 DWS 事实表。
 - DWS sales 不再读取 DM geography，改为读取本批 DWS geo。
 - Market 的共享临时表、MERGE DELETE 和 UPDATE 改为 notebook 内 CTE、anti join、rank 和投影。
 - Pharbers 入湖在无文件时失败，避免旧 staging 被再次覆盖到 DWD。
 ## 6. 上线前契约确认
 - Legacy manufacturer build 使用 `T1.*` 且仓库无完整 DDL。新表明确收窄为当前消费者需要的 9 列，注释 CTAS 从源表达式推导字段类型；新增消费者字段前必须先 `DESCRIBE` 源表并协调扩展契约。
 - Market-TA 明确收窄为 `MARKET, TA` 两列，并用零行 CTAS 推导类型；上线前用 `DESCRIBE` 确认旧 source/target 没有仍需保留的业务列。
 - `dws_ext_td_ims_pack_ym` 首次部署必须 `DEEP CLONE` 旧表，不能只复制 schema，否则五年窗口外的历史映射会丢失。
 - CHPA sales 的省份映射从 DM geography 改为 DWS geo 后，必须验证 fact `PROVINCE_C` 到 geo `PROVINCE_C` 的匹配率和 geo 省份唯一性。
 - Legacy DM 消费者读取 `CORP_DES_CN` / `MANU_DES_CN`，而 02 构建契约保留 `CORP_DES_C` / `MANU_DES_C`；03 迁移时必须统一。
 - Market 的扩展 ratio、扩展行 `BU`/`BrandType` 位置互换、KC NULL 去重和无总排序 tie-breaker 均为兼容保留项，业务修复需单独变更。
 ## 7. 下一批必须迁移的 03 消费者
 本批新表上线后，以下类型的旧引用仍存在于 `03` 脚本中：
 - `dws.dws_ims_td_geo`、`dws.dws_ims_td_market`；
 - `dws.dws_ims_td_*_cn`；
 - `dwd.dwd_ims_td_pack_property`；
 - `tmp.tmp_ims_tf_fact_sales`；
 - `dws.dws_ims_td_pack_ym`。
 在 `03` 消费者全部迁移并完成新旧双向差集验证之前，不删除旧表。
@@ -62,15 +62,19 @@ DWD -> DWS -> DM
 - 需要区分来源系统时，将来源放在业务实体开头，例如 `ims_atc_hierarchy`。
 - `atc`、`nfc`、`ims` 等已形成业务共识的缩写可以保留。
-首批及后续建议映射：
+本批核心映射：
 | 旧表 | 新表 | 类型 |
 | --- | --- | --- |
 | `dwd.dwd_ims_atc_hierarchy` | `dws.dws_ext_td_ims_atc_hierarchy` | DWS 维度 |
 | `dwd.dwd_ims_nfc_hierarchy` | `dws.dws_ext_td_ims_nfc_hierarchy` | DWS 维度 |
-| `dwd.dwd_ims_td_manufacturer_corp` | `dws.dws_ext_td_ims_manufacturer_corporation` | DWS 维度，后续处理 |
+| `dwd.dwd_ims_td_manufacturer_corp` | `dws.dws_ext_td_ims_manufacturer_corporation` | DWS 维度 |
-| `dwd.dwd_ims_td_pack_property` | `dws.dws_ext_td_ims_pack_property` | DWS 维度，后续处理 |
+| `dwd.dwd_ims_td_pack_property` | `dws.dws_ext_td_ims_pack_property` | DWS 维度 |
-| `dwd.dwd_gnd_pharbers_prov_fact` | `dws.dws_ext_tf_pharbers_province_sales` | DWS 事实，需先确认粒度 |
+| `tmp.tmp_ims_td_prod_tmp` | `dws.dws_ext_td_ims_product_multi_manufacturer` | DWS 维度 |
 | `tmp.tmp_ims_tf_fact_sales` | `dws.dws_ext_tf_ims_chpa_sales` | DWS 事实 |
 | `dws.dws_ims_td_market` | `dws.dws_ext_td_ims_market` | DWS 维度 |
 `dwd.dwd_gnd_pharbers_prov_fact` 是省级原子事实，仍保留在 DWD；它与 IMS 全国事实组合后的可复用结果写入 `dws.dws_ext_tf_ims_chpa_sales`。全部 01/02 文件级映射见 `docs/chpa_01_02_migration.md`。
 ## 4. 文件和目录命名
@@ -127,7 +131,7 @@ sql/chpa/02_dws/01_dws_ext_td_ims_atc_hierarchy.sql
 - 每个脚本必须声明预期目标粒度。
 - 最少验证新旧行数、双向差集、业务键重复、必填键空值和层级未匹配数。
-## 8. 首批兼容说明
+## 8. 兼容和验证说明
 ATC 和 NFC 脚本保留旧逻辑中的父级编码规则：
@@ -137,4 +141,6 @@ ATC 和 NFC 脚本保留旧逻辑中的父级编码规则：
 - NFC 2 级关联 1 级：取前 1 位。
 - NFC 3 级关联 2 级：取前 2 位。
-两个 DWS 脚本执行后运行 `validation/chpa/02_dws/validate_hierarchy_refactor.sql`。新 DWS 输出与旧 DWD 基线的行数必须一致，双向差集必须为 0；重复路径和未匹配层级作为业务 review 指标记录。
+两个 hierarchy 脚本执行后运行 `validation/chpa/02_dws/validate_hierarchy_refactor.sql`。新 DWS 输出与旧 DWD 基线的行数必须一致，双向差集必须为 0；重复路径和未匹配层级作为业务 review 指标记录。
 完整 01/02 批次执行后运行 `validation/chpa/02_dws/validate_01_02_refactor.sql`。比较时排除重新生成的 ETL 时间戳，并区分两类检查：逻辑等价迁移必须通过行数和双向 `EXCEPT ALL` 硬门禁；主动修复层级违规或已知故障模式的脚本记录质量指标和人工确认项，不伪造等价结论。
@@ -0,0 +1,60 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Standardize GND configuration codes (zero-padding) so join keys
 --              line up across the market configuration tables. Kept on the DWD
 --              config tables themselves (same targets as legacy).
 -- Source     : dwd.dwd_gnd_ims_tblbrandratio,
 --              dwd.dwd_gnd_tblmarket_bymonth,
 --              dwd.dwd_gnd_ims_tblkeycompetitor,
 --              dwd.dwd_gnd_ims_tblbrandtype
 -- Target     : dwd.dwd_gnd_ims_tblbrandratio (UPDATE),
 --              dwd.dwd_gnd_tblmarket_bymonth (UPDATE),
 --              dwd.dwd_gnd_ims_tblkeycompetitor (UPDATE),
 --              dwd.dwd_gnd_ims_tblbrandtype (UPDATE)
 -- Grain      : N/A -- in-place column updates, row grain unchanged.
 -- Write mode : In-place UPDATE (idempotent padding).
 -- Replaces   : Code-padding block of legacy CHPA/01 dwd_update.sql
 --              (time-window defaults moved to
 --              05_normalize_gnd_time_windows.sql).
 -- Consumers  : IMS market config refresh, DWS pack-property/market and other
 --              jobs joining the four updated IMS configuration tables.
 -- Notes      : Padding rules and literals preserved exactly:
 --              - Pack_Code/PACK_COD: left-padded with zeros to 12 digits when
 --                the code starts with a digit, otherwise left unchanged.
 --              - Product_Code: unconditional 9-digit left zero-pad.
 --              - Molecule_Code/CMPS_COD: unconditional 6-digit left zero-pad.
 --              Must run before the IMS market snapshot and DWS consumers.
 -- =============================================================================
 -- COMMAND ----------
 UPDATE dwd.dwd_gnd_ims_tblbrandratio
 SET
    PACK_COD = if(CAST(PACK_COD AS string) REGEXP '^[0-9]', right(concat('000000000000', CAST(PACK_COD AS string)), 12), CAST(PACK_COD AS string)),
    CMPS_COD = RIGHT(concat('000000', CAST(CMPS_COD AS string)), 6)
 ;
 -- COMMAND ----------
 UPDATE dwd.dwd_gnd_tblmarket_bymonth
 SET
    Pack_Code = if(Pack_Code REGEXP '^[0-9]', right(concat('000000000000', Pack_Code), 12), Pack_Code),
    Product_Code = RIGHT(concat('000000000', Product_Code), 9),
    Molecule_Code = RIGHT(concat('000000', Molecule_Code), 6)
 ;
 -- COMMAND ----------
 UPDATE dwd.dwd_gnd_ims_tblkeycompetitor
 SET
    Pack_Code = if(Pack_Code REGEXP '^[0-9]', right(concat('000000000000', Pack_Code), 12), Pack_Code),
    Product_Code = RIGHT(concat('000000000', Product_Code), 9),
    Molecule_Code = RIGHT(concat('000000', Molecule_Code), 6)
 ;
 -- COMMAND ----------
 UPDATE dwd.dwd_gnd_ims_tblbrandtype
 SET
    PACK_COD = if(PACK_COD REGEXP '^[0-9]', right(concat('000000000000', PACK_COD), 12), PACK_COD)
 ;
@@ -0,0 +1,79 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Refresh the IMS market configuration master from the by-month
 --              staging table (full snapshot, deduplicated).
 -- Source     : dwd.dwd_gnd_tblmarket_bymonth (codes pre-padded by
 --              01_standardize_gnd_codes.sql)
 -- Target     : dwd.dwd_gnd_ims_tblmarket
 -- Grain      : One row per distinct market definition (market_no, market, bu,
 --              ATC/NFC, pack/product/molecule, corporation/manufacturer).
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : IMS market snapshot block of legacy CHPA/01 dwd_update.sql.
 -- Consumers  : 02_dws pack-property/market jobs, DM market datasets.
 -- Notes      : Target/source column contract is the legacy 24-column list,
 --              preserved verbatim. The target is the existing legacy config
 --              master; codes must be padded before this job runs.
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: the target is the existing legacy config master. If it must be
 -- recreated, restore the verified 24-column legacy contract listed below; do
 -- not LIKE the by-month staging table (it additionally carries starttime/
 -- endtime).
 -- COMMAND ----------
 INSERT OVERWRITE TABLE dwd.dwd_gnd_ims_tblmarket (
    market_no,
    market,
    bu,
    atc1_code,
    atc2_code,
    atc3_code,
    atc4_code,
    nfc1_code,
    nfc2_code,
    nfc3_code,
    pack_code,
    pack_desc,
    strength,
    product_code,
    product_desc,
    molecule_code,
    molecule_desc,
    not_in_flag,
    extend_market,
    extend_market_ratio,
    corporation_code,
    corporation_desc,
    manufacturer_code,
    manufacturer_desc
 )
 SELECT DISTINCT
    market_no,
    market,
    bu,
    atc1_code,
    atc2_code,
    atc3_code,
    atc4_code,
    nfc1_code,
    nfc2_code,
    nfc3_code,
    pack_code,
    pack_desc,
    strength,
    product_code,
    product_desc,
    molecule_code,
    molecule_desc,
    not_in_flag,
    extend_market,
    extend_market_ratio,
    corporation_code,
    corporation_desc,
    manufacturer_code,
    manufacturer_desc
 FROM dwd.dwd_gnd_tblmarket_bymonth
 ;
@@ -0,0 +1,78 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Refresh the retail (ext) market configuration master from the
 --              by-month staging table (full snapshot, deduplicated).
 -- Source     : dwd.dwd_gnd_retail_tblmarket_bymonth
 -- Target     : dwd.DWD_gnd_ext_retail_tblmarket (legacy table name preserved,
 --              including the historical uppercase DWD_ prefix)
 -- Grain      : One row per distinct market definition (market_no, market, bu,
 --              ATC/NFC, pack/product/molecule, corporation/manufacturer).
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : Retail market snapshot block of legacy CHPA/01 dwd_update.sql.
 -- Consumers  : retail-side market/pack jobs.
 -- Notes      : Target/source column contract is the legacy 24-column list,
 --              preserved verbatim. Codes must be padded before this job runs.
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: the target is the existing legacy config master. If it must be
 -- recreated, restore the verified 24-column legacy contract listed below; do
 -- not LIKE the by-month staging table (it additionally carries starttime/
 -- endtime).
 -- COMMAND ----------
 INSERT OVERWRITE TABLE dwd.DWD_gnd_ext_retail_tblmarket (
    market_no,
    market,
    bu,
    atc1_code,
    atc2_code,
    atc3_code,
    atc4_code,
    nfc1_code,
    nfc2_code,
    nfc3_code,
    pack_code,
    pack_desc,
    strength,
    product_code,
    product_desc,
    molecule_code,
    molecule_desc,
    not_in_flag,
    extend_market,
    extend_market_ratio,
    corporation_code,
    corporation_desc,
    manufacturer_code,
    manufacturer_desc
 )
 SELECT DISTINCT
    market_no,
    market,
    bu,
    atc1_code,
    atc2_code,
    atc3_code,
    atc4_code,
    nfc1_code,
    nfc2_code,
    nfc3_code,
    pack_code,
    pack_desc,
    strength,
    product_code,
    product_desc,
    molecule_code,
    molecule_desc,
    not_in_flag,
    extend_market,
    extend_market_ratio,
    corporation_code,
    corporation_desc,
    manufacturer_code,
    manufacturer_desc
 FROM dwd.dwd_gnd_retail_tblmarket_bymonth
 ;
@@ -0,0 +1,78 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Refresh the DTP market configuration master from the by-month
 --              staging table (full snapshot, deduplicated).
 -- Source     : dwd.dwd_gnd_dtp_tblmarket_bymonth
 -- Target     : DWD.dwd_gnd_dtp_tblmarket (legacy table name preserved,
 --              including the historical uppercase DWD schema prefix)
 -- Grain      : One row per distinct market definition (market_no, market, bu,
 --              ATC/NFC, pack/product/molecule, corporation/manufacturer).
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : DTP market snapshot block of legacy CHPA/01 dwd_update.sql.
 -- Consumers  : DTP-side market/pack jobs.
 -- Notes      : Target/source column contract is the legacy 24-column list,
 --              preserved verbatim. Codes must be padded before this job runs.
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: the target is the existing legacy config master. If it must be
 -- recreated, restore the verified 24-column legacy contract listed below; do
 -- not LIKE the by-month staging table (it additionally carries starttime/
 -- endtime).
 -- COMMAND ----------
 INSERT OVERWRITE TABLE DWD.dwd_gnd_dtp_tblmarket (
    market_no,
    market,
    bu,
    atc1_code,
    atc2_code,
    atc3_code,
    atc4_code,
    nfc1_code,
    nfc2_code,
    nfc3_code,
    pack_code,
    pack_desc,
    strength,
    product_code,
    product_desc,
    molecule_code,
    molecule_desc,
    not_in_flag,
    extend_market,
    extend_market_ratio,
    corporation_code,
    corporation_desc,
    manufacturer_code,
    manufacturer_desc
 )
 SELECT DISTINCT
    market_no,
    market,
    bu,
    atc1_code,
    atc2_code,
    atc3_code,
    atc4_code,
    nfc1_code,
    nfc2_code,
    nfc3_code,
    pack_code,
    pack_desc,
    strength,
    product_code,
    product_desc,
    molecule_code,
    molecule_desc,
    not_in_flag,
    extend_market,
    extend_market_ratio,
    corporation_code,
    corporation_desc,
    manufacturer_code,
    manufacturer_desc
 FROM dwd.dwd_gnd_dtp_tblmarket_bymonth
 ;
@@ -0,0 +1,95 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Normalize the GND market/config time windows. The legacy job
 --              applied two different default conventions; both are preserved
 --              verbatim, including the historical difference between
 --              190001/209901 and 200001/299912 (do not unify without business
 --              sign-off).
 -- Source     : dwd.dwd_gnd_ims_tblbrandratio,
 --              dwd.dwd_gnd_tblmarket_bymonth,
 --              dwd.dwd_gnd_retail_tblmarket_bymonth,
 --              dwd.dwd_gnd_ec_tblmarket_bymonth,
 --              dwd.dwd_gnd_dtp_tblmarket_bymonth
 -- Target     : the same five DWD tables (in-place UPDATE).
 -- Grain      : N/A -- in-place column updates, row grain unchanged.
 -- Write mode : In-place UPDATE.
 -- Replaces   : Time-window blocks of legacy CHPA/01 dwd_update.sql.
 -- Consumers  : downstream market/pack jobs that filter on starttime/endtime.
 -- Notes      : Statement order is significant and matches legacy execution:
 --              1) brandratio defaults via CASE (UPPER(...)='ALL' ->
 --                 190001/209901);
 --              2) blanket defaults 200001/299912 for all five tables where the
 --                 value is NULL/''/'All'.
 --              Because step 1 runs first, brandratio rows originally
 --              NULL/'ALL'/'All'/'all' end at 190001/209901, while '' rows end
 --              at 200001/299912 -- exactly as legacy. The literal checks
 --              ('All' vs UPPER(...)='ALL') are preserved case-sensitively.
 -- =============================================================================
 -- COMMAND ----------
 UPDATE dwd.dwd_gnd_ims_tblbrandratio
 SET
    StartTime = CASE WHEN StartTime IS NULL OR UPPER(StartTime) = 'ALL' THEN '190001' ELSE StartTime END,
    EndTime = CASE WHEN EndTime IS NULL OR UPPER(EndTime) = 'ALL' THEN '209901' ELSE EndTime END
 ;
 -- COMMAND ----------
 UPDATE DWD.dwd_gnd_tblmarket_bymonth
 SET starttime = '200001'
 WHERE starttime IS NULL OR starttime = '' OR starttime = 'All';
 -- COMMAND ----------
 UPDATE DWD.dwd_gnd_tblmarket_bymonth
 SET endtime = '299912'
 WHERE endtime IS NULL OR endtime = '' OR endtime = 'All';
 -- COMMAND ----------
 UPDATE DWD.dwd_gnd_ims_tblbrandratio
 SET starttime = '200001'
 WHERE starttime IS NULL OR starttime = '' OR starttime = 'All';
 -- COMMAND ----------
 UPDATE DWD.dwd_gnd_ims_tblbrandratio
 SET endtime = '299912'
 WHERE endtime IS NULL OR endtime = '' OR endtime = 'All';
 -- COMMAND ----------
 UPDATE DWD.dwd_gnd_retail_tblmarket_bymonth
 SET starttime = '200001'
 WHERE starttime IS NULL OR starttime = '' OR starttime = 'All';
 -- COMMAND ----------
 UPDATE DWD.dwd_gnd_retail_tblmarket_bymonth
 SET endtime = '299912'
 WHERE endtime IS NULL OR endtime = '' OR endtime = 'All';
 -- COMMAND ----------
 UPDATE DWD.dwd_gnd_ec_tblmarket_bymonth
 SET starttime = '200001'
 WHERE starttime IS NULL OR starttime = '' OR starttime = 'All';
 -- COMMAND ----------
 UPDATE DWD.dwd_gnd_ec_tblmarket_bymonth
 SET endtime = '299912'
 WHERE endtime IS NULL OR endtime = '' OR endtime = 'All';
 -- COMMAND ----------
 UPDATE DWD.dwd_gnd_dtp_tblmarket_bymonth
 SET starttime = '200001'
 WHERE starttime IS NULL OR starttime = '' OR starttime = 'All';
 -- COMMAND ----------
 UPDATE DWD.dwd_gnd_dtp_tblmarket_bymonth
 SET endtime = '299912'
 WHERE endtime IS NULL OR endtime = '' OR endtime = 'All';
@@ -0,0 +1,36 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : DWD data-quality fix: manufacturers whose ManufacturerType_ID is
 --              0 cannot be matched to a ManufacturerType. The verified subset
 --              of such rows (or the rows themselves) are MNC, so 0 is remapped
 --              to 2 (MNC) so reports classify them correctly.
 -- Source     : dwd.dwd_ims_td_manufacturer (in place)
 -- Target     : dwd.dwd_ims_td_manufacturer
 -- Grain      : N/A -- in-place column update; only rows with
 --              ManufacturerType_ID = 0.
 -- Write mode : In-place UPDATE.
 -- Replaces   : UPDATE block of legacy CHPA/01 dwd_ims_td_manufacturer_corp.sql
 --              (the mapping build moved to
 --              02_dws/03_dws_ext_td_ims_manufacturer_corporation.sql).
 -- Consumers  : 02_dws/03_dws_ext_td_ims_manufacturer_corporation (and through
 --              it the corporation/manufacturer CN dims and pack-property).
 -- Notes      : Historical manual fixes for manufacturer_id = '93' (SH/ escape
 --              issue) are intentionally NOT reapplied -- the 20240904 review
 --              concluded the upstream value no longer requires them. Kept below
 --              as commented history for traceability.
 -- =============================================================================
 -- COMMAND ----------
 UPDATE dwd.dwd_ims_td_manufacturer
 SET ManufacturerType_ID = 2
 WHERE ManufacturerType_ID = 0;
 -- COMMAND ----------
 -- Historical manual corrections (20240904 review: no longer needed; kept as
 -- commented history only):
 -- update dwd.dwd_ims_td_manufacturer set manufacturer_abbr = 'SH/' where manufacturer_id = '93';
 -- update dwd.dwd_ims_td_manufacturer set corporation_id = '93' where manufacturer_id = '93';
 -- update dwd.dwd_ims_td_manufacturer set corporation_code = '00221' where manufacturer_id = '93';
 -- update dwd.dwd_ims_td_manufacturer set manufacturertype_id = '1' where manufacturer_id = '93';
@@ -0,0 +1,260 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Daily ingestion of the Pharbers provincial sales fact file
 --              (Pharbers_PROV_Fact*.csv) from the ADLS user-upload blob into
 --              the DWD province-level sales fact.
 -- Source     : ADLS blob CSV files (Pharbers_PROV_Fact*.csv under
 --              <...>/ODS/GND/UserUpload/<YYYY/MM/DD>/, environment-specific),
 --              tmp.tmp_chpa_raw_data (staging)
 -- Target     : dwd.dwd_gnd_pharbers_prov_fact
 -- Grain      : One row per province x month x drug as provided by the source
 --              file; grain to be confirmed against the CSV contract.
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : Legacy CHPA/01_FB_BLOB_TO_DWD.sql (same DWD target).
 -- Consumers  : 02 tmp_ims_tf_fact_sales -> DM sales.
 -- Notes      : Two engineering fixes vs legacy (per migration doc):
 --              1) Fail fast when the day's upload path or matching files are
 --                 missing, instead of silently republishing stale staging.
 --              2) withColumnRenamed results are now assigned back to the frame
 --                 (legacy discarded them); renames are applied only to columns
 --                 that exist, preserving legacy tolerance of missing columns.
 --              Everything else is preserved: paths, file filter regex,
 --              CSV options, staging flow, NULL lineage columns, and the UTC+8
 --              insert timestamp.
 -- =============================================================================
 -- COMMAND ----------
 -- MAGIC %run ../../../Common/config
 -- COMMAND ----------
 -- MAGIC %md
 -- MAGIC ### 从 blob 读取 csv 文件作为 CHPA 法伯省级事实表
 -- COMMAND ----------
 -- MAGIC %python
 -- MAGIC from datetime import datetime, timedelta
 -- MAGIC import pandas as pd
 -- COMMAND ----------
 -- MAGIC %python
 -- MAGIC if ENVIRONMENT == PRD_ENVIRONMENT_VALUE:
 -- MAGIC     factsales_file_path_template = "abfss://master@azcdatalakeprd.dfs.core.chinacloudapi.cn/ODS/GND/UserUpload/"
 -- MAGIC elif ENVIRONMENT == TEST_ENVIRONMENT_VALUE:
 -- MAGIC     factsales_file_path_template = "abfss://master@retaildlstoragetest.dfs.core.chinacloudapi.cn/ODS/GND/UserUpload/"
 -- COMMAND ----------
 -- MAGIC %python
 -- MAGIC # 路径是否存在
 -- MAGIC def path_exists(path):
 -- MAGIC     try:
 -- MAGIC         dbutils.fs.ls(path)
 -- MAGIC         return True
 -- MAGIC     except Exception as e:
 -- MAGIC         if "java.io.FileNotFoundException" in str(e):
 -- MAGIC             return False
 -- MAGIC         else:
 -- MAGIC             print(f"检查路径 {path} 时出错: {e}")
 -- MAGIC             raise
 -- COMMAND ----------
 -- MAGIC %python
 -- MAGIC # 列出 blob 上的文件列表
 -- MAGIC def list_file_name(path):
 -- MAGIC     first_path_list = [i.path for i in dbutils.fs.ls(path)]
 -- MAGIC     second_path_list = [dbutils.fs.ls(i)[0] for i in first_path_list]
 -- MAGIC     return second_path_list
 -- COMMAND ----------
 -- MAGIC %python
 -- MAGIC # 从 blob 下载文件到 local
 -- MAGIC def download_file(file_path, local_path):
 -- MAGIC     dbutils.fs.cp(file_path, local_path)
 -- MAGIC     print(f"已下载 {file_path} 到 {local_path}")
 -- MAGIC     return local_path
 -- COMMAND ----------
 -- MAGIC %python
 -- MAGIC # 计算时间得到当天的路径
 -- MAGIC current_date = datetime.utcnow() + timedelta(hours=8)
 -- MAGIC date_path = current_date.strftime("%Y/%m/%d/")
 -- MAGIC base_path = factsales_file_path_template + date_path
 -- COMMAND ----------
 -- MAGIC %md
 -- MAGIC ### 获取路径下的文件名称，并挑出符合条件的文件路径
 -- MAGIC - 无文件时直接失败，避免旧 staging 数据被再次覆盖到 DWD
 -- COMMAND ----------
 -- MAGIC %python
 -- MAGIC if not path_exists(base_path):
 -- MAGIC     raise FileNotFoundError(f"上传路径不存在，任务失败，拒绝沿用旧 staging 数据: {base_path}")
 -- MAGIC
 -- MAGIC all_file_list = list_file_name(base_path)
 -- MAGIC
 -- MAGIC # 生成 df 来筛选内容
 -- MAGIC files_df = pd.DataFrame([{
 -- MAGIC     'path': f.path,
 -- MAGIC     'modificationtime': f.modificationTime,
 -- MAGIC     'name': f.name
 -- MAGIC } for f in all_file_list])
 -- MAGIC
 -- MAGIC # 同名文件保留修改时间最新的一份
 -- MAGIC files_df = files_df.sort_values('modificationtime', ascending=False).drop_duplicates('name').sort_index()
 -- MAGIC files_df = files_df[files_df['name'].str.match(r'^Pharbers_PROV_Fact.*\.csv$')]
 -- MAGIC files_df = files_df.reset_index(drop=True)
 -- MAGIC
 -- MAGIC if files_df.empty:
 -- MAGIC     raise RuntimeError("未找到符合条件的数据文件 (Pharbers_PROV_Fact*.csv)，任务失败，拒绝沿用旧 staging 数据")
 -- MAGIC
 -- MAGIC print(f"找到 {len(files_df)} 个符合条件的数据文件")
 -- COMMAND ----------
 -- MAGIC %python
 -- MAGIC import os
 -- MAGIC
 -- MAGIC # 下载数据到 local，读取并清洗，逐文件收集
 -- MAGIC df_all = []
 -- MAGIC for file in files_df['path'].tolist():
 -- MAGIC     local_path = download_file(file, f"/Volumes/{NGBI_CATALOG}/tmp/volume_tmp/tmp/{os.path.basename(file)}")
 -- MAGIC     file_df = (spark.read
 -- MAGIC                .option("header", "true")
 -- MAGIC                .option("quote", '"')
 -- MAGIC                .option("escape", '"')
 -- MAGIC                .option("multiLine", "true")
 -- MAGIC                .option("mode", "PERMISSIVE")
 -- MAGIC                .csv(local_path))
 -- MAGIC     # 与旧脚本一致：丢弃 TA / Market 两列
 -- MAGIC     file_df = file_df.drop("TA", "Market")
 -- MAGIC
 -- MAGIC     # 修复历史 bug：旧脚本调用 withColumnRenamed 后未把结果赋回原变量，
 -- MAGIC     # 重命名实际从未生效。这里正确赋值；且仅对确实存在的列重命名，
 -- MAGIC     # 保持旧任务对缺失列名的容忍，不因列不存在而失败。
 -- MAGIC     rename_map = {
 -- MAGIC         'IMS.药品ID': 'IMS_DRUG_ID',
 -- MAGIC         '是否法伯编码': 'IS_HOSP_CODE',
 -- MAGIC         '规格': 'SPEC',
 -- MAGIC         '转换比': 'CONVERSION_RATIO',
 -- MAGIC         '剂型': 'DOSAGE_FORM',
 -- MAGIC         '价格': 'PRICE',
 -- MAGIC     }
 -- MAGIC     for old_name, new_name in rename_map.items():
 -- MAGIC         if old_name in file_df.columns:
 -- MAGIC             file_df = file_df.withColumnRenamed(old_name, new_name)
 -- MAGIC
 -- MAGIC     print(f"已读取 {local_path}")
 -- MAGIC     df_all.append(file_df)
 -- MAGIC
 -- MAGIC if not df_all:
 -- MAGIC     raise RuntimeError("没有读取到任何数据文件，任务失败")
 -- COMMAND ----------
 -- MAGIC %python
 -- MAGIC # 先清空 staging，避免旧数据残留，随后逐文件追加
 -- MAGIC spark.sql("TRUNCATE TABLE tmp.tmp_chpa_raw_data")
 -- MAGIC for num, file_df in enumerate(df_all, start=1):
 -- MAGIC     file_df.createOrReplaceTempView("fact_sales")
 -- MAGIC     spark.sql("INSERT INTO tmp.tmp_chpa_raw_data SELECT * FROM fact_sales")
 -- MAGIC     print(f"第{num}个")
 -- COMMAND ----------
 -- 全量覆盖
 INSERT OVERWRITE TABLE dwd.dwd_gnd_pharbers_prov_fact (
    year,
    ym,
    province_c,
    ims_drug_id,
    is_hosp_code,
    prod_corp,
    prod_cod,
    pack_cod,
    phcd,
    prod_des,
    cmps_des,
    corp_des,
    mnfl_cod,
    prod_des_c,
    cmps_c,
    corp_des_c,
    pack_des,
    spec,
    conversion_ratio,
    dosage_form,
    atc4_cod,
    app1_cod,
    app1_des,
    app1_des_c,
    app2_cod,
    app2_des,
    app2_des_c,
    app3_cod,
    app3_des,
    app3_des_c,
    vbp_batch,
    vbp,
    value,
    totalunit,
    countingunit,
    price,
    manu_des,
    manu_des_c,
    source_file_path,
    source_file_name,
    etl_insert_dt
 )
 SELECT
    year,
    ym,
    province_c,
    ims_drug_id,
    is_hosp_code,
    prod_corp,
    prod_cod,
    pack_cod,
    phcd,
    prod_des,
    cmps_des,
    corp_des,
    mnfl_cod,
    prod_des_c,
    cmps_c,
    corp_des_c,
    pack_des,
    spec,
    conversion_ratio,
    dosage_form,
    atc4_cod,
    app1_cod,
    app1_des,
    app1_des_c,
    app2_cod,
    app2_des,
    app2_des_c,
    app3_cod,
    app3_des,
    app3_des_c,
    vbp_batch,
    vbp,
    value,
    totalunit,
    countingunit,
    price,
    manu_des,
    manu_des_c,
    NULL AS source_file_path,
    NULL AS source_file_name,
    FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(), 'UTC+8') AS etl_insert_dt
 FROM tmp.tmp_chpa_raw_data
 ;
@@ -0,0 +1,81 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Build the manufacturer-to-corporation mapping dimension:
 --              one row per manufacturer with its own attributes plus the
 --              matched corporation master code, abbreviation and name.
 -- Source     : dwd.dwd_ims_td_manufacturer (self-join on
 --              Corporation_Code = Manufacturer_CODE)
 -- Target     : dws.dws_ext_td_ims_manufacturer_corporation
 -- Grain      : One row per manufacturer; CORP_MASTER_CODE/CORP_ABBR/CORP_DES
 --              are NULL when no corporation master matches.
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : dwd.dwd_ims_td_manufacturer_corp (legacy output).
 -- Consumers  : 02_dws/04_dws_ext_td_ims_pack_property,
 --              02_dws/06_dws_ext_td_ims_corporation_cn,
 --              02_dws/07_dws_ext_td_ims_manufacturer_cn -- all must read the
 --              DWS target.
 -- Notes      : The legacy job wrote T1.* but no source DDL/DESCRIBE is available
 --              in this repository. This migration therefore defines a deliberate
 --              9-column DWS contract containing every field used by repository
 --              consumers, instead of creating unknown columns that would be
 --              silently populated with NULL. The commented CTAS infers physical
 --              types from source expressions without guessing them. DESCRIBE the
 --              source before adding any future consumer field.
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: infer the explicit 9-column contract without copying data.
 -- CREATE TABLE IF NOT EXISTS dws.dws_ext_td_ims_manufacturer_corporation AS
 -- SELECT
 --     manufacturer.Manufacturer_ID,
 --     manufacturer.Manufacturer_CODE,
 --     manufacturer.Manufacturer_Abbr,
 --     manufacturer.Manufacturer_Name,
 --     manufacturer.ManufacturerType_ID,
 --     manufacturer.Corporation_Code,
 --     corporation_master.Corporation_Code AS CORP_MASTER_CODE,
 --     corporation_master.Manufacturer_Abbr AS CORP_ABBR,
 --     corporation_master.Manufacturer_Name AS CORP_DES
 -- FROM dwd.dwd_ims_td_manufacturer AS manufacturer
 -- LEFT JOIN dwd.dwd_ims_td_manufacturer AS corporation_master
 --     ON manufacturer.Corporation_Code = corporation_master.Manufacturer_CODE
 -- WHERE 1 = 0;
 -- COMMAND ----------
 INSERT OVERWRITE TABLE dws.dws_ext_td_ims_manufacturer_corporation (
    Manufacturer_ID,
    Manufacturer_CODE,
    Manufacturer_Abbr,
    Manufacturer_Name,
    ManufacturerType_ID,
    Corporation_Code,
    CORP_MASTER_CODE,
    CORP_ABBR,
    CORP_DES
 )
 WITH manufacturer AS (
    SELECT
        manufacturer_id,
        manufacturer_code,
        manufacturer_abbr,
        manufacturer_name,
        manufacturertype_id,
        corporation_code
    FROM dwd.dwd_ims_td_manufacturer
 )
 SELECT
    manufacturer.manufacturer_id AS manufacturer_id,
    manufacturer.manufacturer_code AS manufacturer_code,
    manufacturer.manufacturer_abbr AS manufacturer_abbr,
    manufacturer.manufacturer_name AS manufacturer_name,
    manufacturer.manufacturertype_id AS manufacturertype_id,
    manufacturer.corporation_code AS corporation_code,
    corporation_master.corporation_code AS corp_master_code,
    corporation_master.manufacturer_abbr AS corp_abbr,
    corporation_master.manufacturer_name AS corp_des
 FROM manufacturer
 LEFT JOIN dwd.dwd_ims_td_manufacturer AS corporation_master
    ON manufacturer.corporation_code = corporation_master.manufacturer_code
 ;
@@ -0,0 +1,179 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Build the master pack property dimension: one row per pack with
 --              pack/product/molecule codes and descriptions, ATC/NFC hierarchy
 --              codes, molecule attributes, manufacturer/corporation codes and
 --              descriptions, manufacturer type and brand type. Also applies the
 --              AZ (A5Z/A5ZD) business-scope overrides exactly as legacy.
 -- Source     : dwd.dwd_ims_td_pack,
 --              dwd.dwd_ims_td_product,
 --              dwd.dwd_ims_td_new_form_class,
 --              dws.dws_ext_td_ims_nfc_hierarchy (repointed from legacy DWD),
 --              dwd.dwd_ims_td_therapeutic_class,
 --              dws.dws_ext_td_ims_atc_hierarchy (repointed from legacy DWD),
 --              dwd.dwd_ims_td_pack_additional_attribute,
 --              dws.dws_ext_td_ims_manufacturer_corporation (repointed from
 --              legacy DWD),
 --              dwd.dwd_ims_td_manufacturertype,
 --              dwd.dwd_gnd_ims_tblbrandtype,
 --              dwd.dwd_gnd_tbl_corp_change (AZ override key)
 -- Target     : dws.dws_ext_td_ims_pack_property
 -- Grain      : One row per PACK_COD (SELECT DISTINCT; hierarchy/attribute fan-out
 --              is masked by DISTINCT, matching legacy).
 -- Write mode : Full refresh (INSERT OVERWRITE) + 2 business-override UPDATEs
 --              (A5Z/A5ZD), executed in this order.
 -- Replaces   : dwd.dwd_ims_td_pack_property (legacy output).
 -- Consumers  : 02_dws/08_dws_ext_td_ims_product_multi_manufacturer,
 --              02_dws/16_dws_ext_td_ims_market, 03 dm_ims_td_pack_property --
 --              all must migrate to the DWS target.
 -- Notes      : Legacy expressions, padding rules, join keys and DISTINCT are
 --              preserved verbatim. The two post-write NULL cleanups
 --              (CORP_COD='' / STGH_DES='') are folded into COALESCE in the
 --              projection (identical final values; the A5Z/A5ZD overrides key
 --              on PROD_COD/PACK_COD, so they are unaffected). The BrandType
 --              join compares the raw source Pack_Code with the pre-padded
 --              BRANDTYPE.PACK_COD -- a known legacy hazard, intentionally not
 --              changed. Run order: after 01_dwd/01_standardize_gnd_codes.sql
 --              (brandtype padding) and 02_dws/01, 02, 03 (hierarchy and
 --              manufacturer-corporation).
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: copy the verified legacy schema without copying data.
 -- CREATE TABLE IF NOT EXISTS dws.dws_ext_td_ims_pack_property
 -- LIKE dwd.dwd_ims_td_pack_property;
 -- COMMAND ----------
 INSERT OVERWRITE TABLE dws.dws_ext_td_ims_pack_property (
    PACK_COD,
    PACK_DES,
    STGH_DES,
    PACK_LCH,
    PROD_COD,
    CMPS_COD,
    CMPS_DES,
    ATC1_COD,
    ATC2_COD,
    ATC3_COD,
    ATC4_COD,
    APP1_COD,
    APP2_COD,
    APP3_COD,
    BIO_DESC,
    GENE_ORIG_DESC,
    ETH_OTC_DESC,
    NRDL_DESC,
    NRDL_Entry_Date,
    EDL_DESC,
    TCM_DESC,
    PAED_DESC,
    GQCE_DESC,
    VBP_DESC,
    MANU_COD,
    MANU_DES,
    MNFL_COD,
    MNFL_DES,
    CORP_COD,
    CORP_DES,
    BrandType
 )
 SELECT DISTINCT
    if(PACK.Pack_Code REGEXP '^[0-9]', right(concat('000000000000', PACK.Pack_Code), 12), PACK.Pack_Code) AS PACK_COD,
    PACK.Pack_Description AS PACK_DES,
    COALESCE(PACK.STRENGTH, '') AS STGH_DES,
    concat('Y', LEFT(PACK.LAUNCHTIME, 4), 'M', RIGHT(PACK.LAUNCHTIME, 2)) AS PACK_LCH,
    RIGHT(concat('000000000', PROD.Product_Code), 9) AS PROD_COD,
    RIGHT(concat('000000', MOLE.MoleCompCode), 6) AS CMPS_COD,
    MOLE.MoleCompDesc AS CMPS_DES,
    ATCH.ATC1_CODE AS ATC1_COD,
    ATCH.ATC2_CODE AS ATC2_COD,
    ATCH.ATC3_CODE AS ATC3_COD,
    ATCH.ATC4_CODE AS ATC4_COD,
    NFCH.NFC1_CODE AS APP1_COD,
    NFCH.NFC2_CODE AS APP2_COD,
    NFCH.NFC3_CODE AS APP3_COD,
    MOLE.BIO AS BIO_DESC,
    MOLE.Gene_Orig AS GENE_ORIG_DESC,
    MOLE.Rx_Flag AS ETH_OTC_DESC,
    MOLE.NRDL AS NRDL_DESC,
    MOLE.NRDL_Entry_Date,
    MOLE.EDL AS EDL_DESC,
    MOLE.TCMEX AS TCM_DESC,
    MOLE.PAED AS PAED_DESC,
    MOLE.GQCE AS GQCE_DESC,
    MOLE.VBP AS VBP_DESC,
    MANU.Manufacturer_Abbr AS MANU_COD,
    MANU.Manufacturer_Name AS MANU_DES,
    MANUT.ManufacturerType_CODE AS MNFL_COD,
    MANUT.ManufacturerType_Name AS MNFL_DES,
    COALESCE(MANU.CORP_ABBR, '') AS CORP_COD,
    MANU.CORP_DES AS CORP_DES,
    BRANDTYPE.Brand_Type AS BrandType
 FROM dwd.dwd_ims_td_pack AS PACK
 LEFT JOIN dwd.dwd_ims_td_product AS PROD
    ON PACK.Product_ID = PROD.Product_ID
 LEFT JOIN dwd.dwd_ims_td_new_form_class AS NFC
    ON PACK.NewFormClass_ID = NFC.NewFormClass_ID
 LEFT JOIN dws.dws_ext_td_ims_nfc_hierarchy AS NFCH
    ON NFC.NewFormClass_Code = NFCH.NFC3_CODE
 LEFT JOIN dwd.dwd_ims_td_therapeutic_class AS ATC
    ON PACK.Therapeutic_ID = ATC.Therapeutic_ID
 LEFT JOIN dws.dws_ext_td_ims_atc_hierarchy AS ATCH
    ON ATC.Therapeutic_Code = ATCH.ATC4_CODE
 LEFT JOIN dwd.dwd_ims_td_pack_additional_attribute AS MOLE
    ON PACK.Pack_ID = MOLE.Pack_ID
 LEFT JOIN dws.dws_ext_td_ims_manufacturer_corporation AS MANU
    ON PROD.Manufacturer_ID = MANU.Manufacturer_ID
 LEFT JOIN dwd.dwd_ims_td_manufacturertype AS MANUT
    ON MANU.ManufacturerType_ID = MANUT.ManufacturerType_ID
 LEFT JOIN dwd.dwd_gnd_ims_tblbrandtype AS BRANDTYPE
    ON PACK.Pack_Code = BRANDTYPE.PACK_COD
 ;
 -- COMMAND ----------
 -- AZ brand scope override A5Z (legacy business rule, preserved verbatim):
 -- products/packs flagged with corp_cod = 'A5Z' in the corporate-change table
 -- are displayed as ASTRAZENECA GROUP. Matches on product level or pack level.
 UPDATE dws.dws_ext_td_ims_pack_property
 SET
    CORP_COD = 'A5Z',
    CORP_DES = 'ASTRAZENECA GROUP'
 WHERE PROD_COD IN (
        SELECT RIGHT(concat('0000000000', PROD_COD), 9) AS PROD_COD
        FROM dwd.dwd_gnd_tbl_corp_change
        WHERE corp_cod = 'A5Z'
    )
   OR PACK_COD IN (
        SELECT if(PACK_COD REGEXP '^[0-9]', RIGHT(concat('000000000000', PACK_COD), 12), PACK_COD) AS PACK_COD
        FROM dwd.dwd_gnd_tbl_corp_change
        WHERE corp_cod = 'A5Z'
    )
 ;
 -- COMMAND ----------
 -- AZ brand scope override A5ZD (legacy business rule, preserved verbatim):
 -- products/packs flagged with corp_cod = 'A5ZD' are re-labeled as AZDealed on
 -- both corporation and manufacturer fields.
 UPDATE dws.dws_ext_td_ims_pack_property
 SET
    CORP_COD = 'A5ZD',
    CORP_DES = 'AZDealed',
    MANU_COD = 'A5ZD',
    MANU_DES = 'AZDealed'
 WHERE PROD_COD IN (
        SELECT RIGHT(concat('0000000000', PROD_COD), 9) AS PROD_COD
        FROM dwd.dwd_gnd_tbl_corp_change
        WHERE corp_cod = 'A5ZD'
    )
   OR PACK_COD IN (
        SELECT if(PACK_COD REGEXP '^[0-9]', RIGHT(concat('000000000000', PACK_COD), 12), PACK_COD) AS PACK_COD
        FROM dwd.dwd_gnd_tbl_corp_change
        WHERE corp_cod = 'A5ZD'
    )
 ;
@@ -0,0 +1,56 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Build the IMS audit/geo (province) dimension with English and
 --              Chinese names, audit type and tier attributes. Province data is
 --              hard-coded in the staging source; city data is no longer
 --              provided (20260122 note).
 -- Source     : tmp.tmp_province_rawdata (hard-coded province data)
 -- Target     : dws.dws_ext_td_ims_geo
 -- Grain      : One row per AUDIT_COD as provided by the source. The grain is
 --              not enforced by this job (no DISTINCT, matching legacy).
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : dws.dws_ims_td_geo (legacy output).
 -- Consumers  : dm_ims_td_geo (DM build), dm_ims_td_org, dm_ims_td_org_hvh --
 --              all must migrate to the DWS target.
 -- Notes      : Source stays on the tmp staging table (legacy contract; no DWD
 --              equivalent exists yet). ETL_INSERT_DT/ETL_UPDATE_DT are copied
 --              from the source unchanged, matching legacy. 03 dm_ims_td_geo.sql
 --              currently ALSO writes dws.dws_ims_td_geo; that duplicate write
 --              must be removed when consumers migrate.
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: copy the verified legacy schema without copying data.
 -- CREATE TABLE IF NOT EXISTS dws.dws_ext_td_ims_geo
 -- LIKE dws.dws_ims_td_geo;
 -- COMMAND ----------
 INSERT OVERWRITE TABLE dws.dws_ext_td_ims_geo (
    AUDIT_COD,
    AUDIT_DES,
    AUDIT_DES_C,
    AUDIT_TYPE,
    CITY_TIER,
    AZ_CITY_TIER,
    PROVINCE,
    PROVINCE_C,
    REGIONCENTER,
    ETL_INSERT_DT,
    ETL_UPDATE_DT
 )
 SELECT
    audit_cod,
    audit_des,
    audit_des_c,
    audit_type,
    city_tier,
    az_city_tier,
    province,
    province_c,
    regioncenter,
    etl_insert_dt,
    etl_update_dt
 FROM tmp.tmp_province_rawdata
 ;
@@ -0,0 +1,72 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Attach Chinese (CN) corporation names; also forces the
 --              hard-coded AZDealed row (CORP_COD = 'A5ZD') exactly as the
 --              legacy job did.
 -- Source     : dws.dws_ext_td_ims_manufacturer_corporation,
 --              dwd.dwd_gnd_ims_tblmanucn (CN names on CORP_ABBR = abbrev)
 -- Target     : dws.dws_ext_td_ims_corporation_cn
 -- Grain      : One row per distinct CORP_COD (plus the forced A5ZD row).
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : dws.dws_ims_td_corp_cn (legacy output).
 -- Consumers  : tmp_ims_td_prod_tmp (join on CORP_COD); dm_ims_td_pack_property
 --              (DIM_CORP) -- both must migrate to the DWS target.
 -- Notes      : Corporation source repointed from dwd.dwd_ims_td_manufacturer_corp
 --              to dws.dws_ext_td_ims_manufacturer_corporation (same schema,
 --              incl. CORP_ABBR/CORP_DES); that DWS table must be built before
 --              this job runs. The legacy unconditional second INSERT of the
 --              A5ZD row is folded into a single deterministic overwrite via
 --              UNION ALL; the resulting row set is identical to legacy.
 --              Legacy writes CORP_DES_C (no English fallback): a corporation
 --              without a Chinese name keeps a NULL CORP_DES_C.
 --              DIM_CORP currently selects CORP_DES_CN (name mismatch vs the
 --              CORP_DES_C contract); align that consumer during migration.
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: copy the verified legacy schema without copying data.
 -- CREATE TABLE IF NOT EXISTS dws.dws_ext_td_ims_corporation_cn
 -- LIKE dws.dws_ims_td_corp_cn;
 -- COMMAND ----------
 INSERT OVERWRITE TABLE dws.dws_ext_td_ims_corporation_cn (
    CORP_COD,
    CORP_DES,
    CORP_DES_C,
    ETL_INSERT_DT,
    ETL_UPDATE_DT
 )
 WITH corporation AS (
    SELECT
        corp_abbr,
        corp_des
    FROM dws.dws_ext_td_ims_manufacturer_corporation
    WHERE corp_abbr IS NOT NULL
 ),
 corporation_cn AS (
    SELECT DISTINCT
        abbrev,
        namec
    FROM dwd.dwd_gnd_ims_tblmanucn
 )
 SELECT DISTINCT
    corporation.corp_abbr AS corp_cod,
    corporation.corp_des AS corp_des,
    corporation_cn.namec AS corp_des_c,
    FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(), 'UTC+8') AS etl_insert_dt,
    FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(), 'UTC+8') AS etl_update_dt
 FROM corporation
 LEFT JOIN corporation_cn
    ON corporation.corp_abbr = corporation_cn.abbrev
 UNION ALL
 SELECT
    'A5ZD' AS corp_cod,
    'AZDealed' AS corp_des,
    'AZDealed' AS corp_des_c,
    FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(), 'UTC+8') AS etl_insert_dt,
    FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(), 'UTC+8') AS etl_update_dt
 ;
@@ -0,0 +1,72 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Attach Chinese (CN) manufacturer names; also forces the
 --              hard-coded AZDealed row (MANU_COD = 'A5ZD') exactly as the
 --              legacy job did.
 -- Source     : dws.dws_ext_td_ims_manufacturer_corporation,
 --              dwd.dwd_gnd_ims_tblmanucn (CN names on Manufacturer_Abbr = abbrev)
 -- Target     : dws.dws_ext_td_ims_manufacturer_cn
 -- Grain      : One row per distinct MANU_COD (plus the forced A5ZD row).
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : dws.dws_ims_td_manu_cn (legacy output).
 -- Consumers  : dm_ims_td_pack_property (DIM_MANU) -- must migrate to the DWS
 --              target.
 -- Notes      : Manufacturer source repointed from dwd.dwd_ims_td_manufacturer_corp
 --              to dws.dws_ext_td_ims_manufacturer_corporation (same schema,
 --              incl. Manufacturer_Abbr/Manufacturer_Name); that DWS table must
 --              be built before this job runs. The legacy unconditional second
 --              INSERT of the A5ZD row is folded into a single deterministic
 --              overwrite via UNION ALL; the resulting row set is identical to
 --              legacy. Legacy writes MANU_DES_C (no English fallback): a
 --              manufacturer without a Chinese name keeps a NULL MANU_DES_C.
 --              DIM_MANU currently selects MANU_DES_CN (name mismatch vs the
 --              MANU_DES_C contract); align that consumer during migration.
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: copy the verified legacy schema without copying data.
 -- CREATE TABLE IF NOT EXISTS dws.dws_ext_td_ims_manufacturer_cn
 -- LIKE dws.dws_ims_td_manu_cn;
 -- COMMAND ----------
 INSERT OVERWRITE TABLE dws.dws_ext_td_ims_manufacturer_cn (
    MANU_COD,
    MANU_DES,
    MANU_DES_C,
    ETL_INSERT_DT,
    ETL_UPDATE_DT
 )
 WITH manufacturer AS (
    SELECT
        manufacturer_abbr,
        manufacturer_name
    FROM dws.dws_ext_td_ims_manufacturer_corporation
    WHERE manufacturer_abbr IS NOT NULL
 ),
 manufacturer_cn AS (
    SELECT DISTINCT
        abbrev,
        namec
    FROM dwd.dwd_gnd_ims_tblmanucn
 )
 SELECT DISTINCT
    manufacturer.manufacturer_abbr AS manu_cod,
    manufacturer.manufacturer_name AS manu_des,
    manufacturer_cn.namec AS manu_des_c,
    FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(), 'UTC+8') AS etl_insert_dt,
    FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(), 'UTC+8') AS etl_update_dt
 FROM manufacturer
 LEFT JOIN manufacturer_cn
    ON manufacturer.manufacturer_abbr = manufacturer_cn.abbrev
 UNION ALL
 SELECT
    'A5ZD' AS manu_cod,
    'AZDealed' AS manu_des,
    'AZDealed' AS manu_des_c,
    FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(), 'UTC+8') AS etl_insert_dt,
    FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(), 'UTC+8') AS etl_update_dt
 ;
@@ -0,0 +1,62 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Flag products whose Chinese name (PROD_DES_C) is shared by more
 --              than 5 distinct corporations (CORP_DES) or more than 5 distinct
 --              manufacturers (MANU_DES). These products are demoted in report
 --              ordering via RANK_TYPE in dws.dws_ext_td_ims_product_cn.
 -- Source     : dws.dws_ext_td_ims_pack_property (pack-level CORP_DES, MANU_DES)
 --              dwd.dwd_gnd_ims_tblprodcn (prodcode -> PROD_COD, namec -> PROD_DES_C)
 -- Target     : dws.dws_ext_td_ims_product_multi_manufacturer
 -- Grain      : One row per PROD_COD (only flagged products are materialized).
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : tmp.tmp_ims_td_prod_tmp (legacy staging list used by the
 --              legacy UPDATE on dws.dws_ims_td_prod_cn).
 -- Consumers  : dws.dws_ext_td_ims_product_cn (RANK_TYPE 0/1 via join).
 -- Notes      : The legacy >5 distinct CORP_DES OR >5 distinct MANU_DES rule is
 --              preserved, counted at PROD_DES_C level across all packs sharing
 --              the Chinese name, then applied to every PROD_COD carrying that
 --              name. Legacy dead joins (pack_property, corp_cn) are dropped.
 --              PROD_COD is zero-padded to 9 digits exactly as in the legacy
 --              product build (RIGHT(CONCAT('0000000000', prodcode), 9)).
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: copy the verified legacy schema without copying data.
 -- CREATE TABLE IF NOT EXISTS dws.dws_ext_td_ims_product_multi_manufacturer
 -- LIKE tmp.tmp_ims_td_prod_tmp;
 -- COMMAND ----------
 INSERT OVERWRITE TABLE dws.dws_ext_td_ims_product_multi_manufacturer (
    PROD_COD
 )
 WITH product_cn AS (
    -- Chinese-name master derived from the raw source (same zero-padding as
    -- the legacy product build) so this job does not depend on the product_cn
    -- table it feeds.
    SELECT DISTINCT
        RIGHT(CONCAT('0000000000', prodcode), 9) AS prod_cod,
        namec AS prod_des_c
    FROM dwd.dwd_gnd_ims_tblprodcn
 ),
 flagged_names AS (
    -- Chinese names shared by more than 5 distinct CORPs or MANUs.
    SELECT
        product_cn.prod_des_c
    FROM dws.dws_ext_td_ims_pack_property pack_property
    LEFT JOIN product_cn
        ON pack_property.prod_cod = product_cn.prod_cod
    GROUP BY
        product_cn.prod_des_c
    HAVING
        (   COUNT(DISTINCT pack_property.corp_des) > 5
         OR COUNT(DISTINCT pack_property.manu_des) > 5
        )
        AND product_cn.prod_des_c IS NOT NULL
 )
 SELECT DISTINCT
    product_cn.prod_cod
 FROM product_cn
 WHERE product_cn.prod_des_c IN (SELECT prod_des_c FROM flagged_names)
 ;
@@ -0,0 +1,61 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Build the product master with Chinese names and a declarative
 --              RANK_TYPE. RANK_TYPE = 0 marks products demoted in report
 --              ordering (Chinese name shared by >5 CORPs or MANUs, see
 --              dws.dws_ext_td_ims_product_multi_manufacturer); otherwise 1.
 -- Source     : dwd.dwd_gnd_ims_tblprodcn
 --              dws.dws_ext_td_ims_product_multi_manufacturer (flag join)
 -- Target     : dws.dws_ext_td_ims_product_cn
 -- Grain      : One row per PROD_COD (9-digit zero-padded product code).
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : dws.dws_ims_td_prod_cn (legacy 7-column contract preserved) and
 --              the legacy two-step RANK_TYPE demotion (tmp list + UPDATE) which
 --              is folded into a declarative CASE here.
 -- Consumers  : Report/BI layer via RANK_TYPE (sorting priority). This job must
 --              run after dws_ext_td_ims_product_multi_manufacturer (file 08).
 -- Notes      : Legacy DISTINCT semantics preserved. Column names follow the
 --              legacy dws_ims_td_prod_cn contract; ETL timestamps named per
 --              the DWS convention (ETL_INSERT_DT/ETL_UPDATE_DT).
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: copy the verified legacy schema without copying data.
 -- CREATE TABLE IF NOT EXISTS dws.dws_ext_td_ims_product_cn
 -- LIKE dws.dws_ims_td_prod_cn;
 -- COMMAND ----------
 INSERT OVERWRITE TABLE dws.dws_ext_td_ims_product_cn (
    PROD_COD,
    PROD_DES,
    PROD_DES_C,
    CMPS_DES_C,
    RANK_TYPE,
    ETL_INSERT_DT,
    ETL_UPDATE_DT
 )
 WITH product_cn AS (
    SELECT DISTINCT
        RIGHT(CONCAT('0000000000', prodcode), 9) AS prod_cod,
        ename AS prod_des,
        namec AS prod_des_c,
        gene_name AS cmps_des_c
    FROM dwd.dwd_gnd_ims_tblprodcn
 )
 SELECT DISTINCT
    product_cn.prod_cod AS PROD_COD,
    product_cn.prod_des AS PROD_DES,
    product_cn.prod_des_c AS PROD_DES_C,
    product_cn.cmps_des_c AS CMPS_DES_C,
    CASE
        WHEN multi_mfr.prod_cod IS NOT NULL THEN 0
        ELSE 1
    END AS RANK_TYPE,
    from_utc_timestamp(current_timestamp(), 'UTC+8') AS ETL_INSERT_DT,
    from_utc_timestamp(current_timestamp(), 'UTC+8') AS ETL_UPDATE_DT
 FROM product_cn
 LEFT JOIN dws.dws_ext_td_ims_product_multi_manufacturer multi_mfr
    ON product_cn.prod_cod = multi_mfr.prod_cod
 ;
@@ -0,0 +1,108 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Attach Chinese (CN) descriptions to the flattened IMS ATC
 --              level 1-4 hierarchy; the English description is the fallback
 --              when no CN name exists for a code.
 -- Source     : dws.dws_ext_td_ims_atc_hierarchy,
 --              dwd.dwd_gnd_ims_tblATC (per-level CN names)
 -- Target     : dws.dws_ext_td_ims_atc_cn
 -- Grain      : One row per distinct ATC1..ATC4 hierarchy path; lower levels
 --              may be null when a parent code has no matching child.
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : dws.dws_ims_td_atc_cn (legacy output).
 -- Consumers  : dm_ims_td_pack_property (DIM_ATC) -- must migrate to the DWS
 --              target.
 -- Notes      : Hierarchy source repointed from dwd.dwd_ims_atc_hierarchy to
 --              dws.dws_ext_td_ims_atc_hierarchy; the DWS table carries the same
 --              code/name columns, so the joins are unchanged. CN-name joins are
 --              case-sensitive string equality, preserved from legacy. Legacy
 --              typo'd column names ATC2_CODe/ATC3_CODe/ATC4_CODe are normalized
 --              to ATC2_CODE/ATC3_CODE/ATC4_CODE (case-insensitive resolution
 --              made them equivalent).
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: copy the verified legacy schema without copying data.
 -- CREATE TABLE IF NOT EXISTS dws.dws_ext_td_ims_atc_cn
 -- LIKE dws.dws_ims_td_atc_cn;
 -- COMMAND ----------
 INSERT OVERWRITE TABLE dws.dws_ext_td_ims_atc_cn (
    ATC1_COD,
    ATC1_DES,
    ATC1_DES_C,
    ATC2_COD,
    ATC2_DES,
    ATC2_DES_C,
    ATC3_COD,
    ATC3_DES,
    ATC3_DES_C,
    ATC4_COD,
    ATC4_DES,
    ATC4_DES_C,
    ETL_INSERT_DT,
    ETL_UPDATE_DT
 )
 WITH hierarchy AS (
    SELECT
        atc1_code,
        atc1_des,
        atc2_code,
        atc2_des,
        atc3_code,
        atc3_des,
        atc4_code,
        atc4_des
    FROM dws.dws_ext_td_ims_atc_hierarchy
 ),
 atc1_cn AS (
    SELECT DISTINCT
        atc1_cod,
        atc1_des_c
    FROM dwd.dwd_gnd_ims_tblATC
 ),
 atc2_cn AS (
    SELECT DISTINCT
        atc2_cod,
        atc2_des_c
    FROM dwd.dwd_gnd_ims_tblATC
 ),
 atc3_cn AS (
    SELECT DISTINCT
        atc3_cod,
        atc3_des_c
    FROM dwd.dwd_gnd_ims_tblATC
 ),
 atc4_cn AS (
    SELECT DISTINCT
        atc4_cod,
        atc4_des_c
    FROM dwd.dwd_gnd_ims_tblATC
 )
 SELECT DISTINCT
    hierarchy.atc1_code AS atc1_cod,
    hierarchy.atc1_des AS atc1_des,
    COALESCE(atc1_cn.atc1_des_c, hierarchy.atc1_des) AS atc1_des_c,
    hierarchy.atc2_code AS atc2_cod,
    hierarchy.atc2_des AS atc2_des,
    COALESCE(atc2_cn.atc2_des_c, hierarchy.atc2_des) AS atc2_des_c,
    hierarchy.atc3_code AS atc3_cod,
    hierarchy.atc3_des AS atc3_des,
    COALESCE(atc3_cn.atc3_des_c, hierarchy.atc3_des) AS atc3_des_c,
    hierarchy.atc4_code AS atc4_cod,
    hierarchy.atc4_des AS atc4_des,
    COALESCE(atc4_cn.atc4_des_c, hierarchy.atc4_des) AS atc4_des_c,
    FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(), 'UTC+8') AS etl_insert_dt,
    FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(), 'UTC+8') AS etl_update_dt
 FROM hierarchy
 LEFT JOIN atc1_cn
    ON hierarchy.atc1_code = atc1_cn.atc1_cod
 LEFT JOIN atc2_cn
    ON hierarchy.atc2_code = atc2_cn.atc2_cod
 LEFT JOIN atc3_cn
    ON hierarchy.atc3_code = atc3_cn.atc3_cod
 LEFT JOIN atc4_cn
    ON hierarchy.atc4_code = atc4_cn.atc4_cod
 ;
@@ -0,0 +1,89 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Attach Chinese (CN) descriptions to the flattened IMS NFC
 --              level 1-3 hierarchy; the English description is the fallback
 --              when no CN name exists for a code.
 -- Source     : dws.dws_ext_td_ims_nfc_hierarchy,
 --              dwd.dwd_gnd_ims_tblAPP (per-level CN names)
 -- Target     : dws.dws_ext_td_ims_nfc_cn
 -- Grain      : One row per distinct NFC1..NFC3 hierarchy path; lower levels
 --              may be null when a parent code has no matching child.
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : dws.dws_ims_td_nfc_cn (legacy output).
 -- Consumers  : dm_ims_td_pack_property (DIM_NFC) -- must migrate to the DWS
 --              target.
 -- Notes      : Hierarchy source repointed from dwd.dwd_ims_nfc_hierarchy to
 --              dws.dws_ext_td_ims_nfc_hierarchy; the DWS table carries the same
 --              code/name columns, so the joins are unchanged. CN-name joins are
 --              case-sensitive string equality, preserved from legacy.
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: copy the verified legacy schema without copying data.
 -- CREATE TABLE IF NOT EXISTS dws.dws_ext_td_ims_nfc_cn
 -- LIKE dws.dws_ims_td_nfc_cn;
 -- COMMAND ----------
 INSERT OVERWRITE TABLE dws.dws_ext_td_ims_nfc_cn (
    APP1_COD,
    APP1_DES,
    APP1_DES_C,
    APP2_COD,
    APP2_DES,
    APP2_DES_C,
    APP3_COD,
    APP3_DES,
    APP3_DES_C,
    ETL_INSERT_DT,
    ETL_UPDATE_DT
 )
 WITH hierarchy AS (
    SELECT
        nfc1_code,
        nfc1_des,
        nfc2_code,
        nfc2_des,
        nfc3_code,
        nfc3_des
    FROM dws.dws_ext_td_ims_nfc_hierarchy
 ),
 nfc1_cn AS (
    SELECT DISTINCT
        app1_cod,
        app1_des_c
    FROM dwd.dwd_gnd_ims_tblAPP
 ),
 nfc2_cn AS (
    SELECT DISTINCT
        app2_cod,
        app2_des_c
    FROM dwd.dwd_gnd_ims_tblAPP
 ),
 nfc3_cn AS (
    SELECT DISTINCT
        app3_cod,
        app3_des_c
    FROM dwd.dwd_gnd_ims_tblAPP
 )
 SELECT DISTINCT
    hierarchy.nfc1_code AS app1_cod,
    hierarchy.nfc1_des AS app1_des,
    COALESCE(nfc1_cn.app1_des_c, hierarchy.nfc1_des) AS app1_des_c,
    hierarchy.nfc2_code AS app2_cod,
    hierarchy.nfc2_des AS app2_des,
    COALESCE(nfc2_cn.app2_des_c, hierarchy.nfc2_des) AS app2_des_c,
    hierarchy.nfc3_code AS app3_cod,
    hierarchy.nfc3_des AS app3_des,
    COALESCE(nfc3_cn.app3_des_c, hierarchy.nfc3_des) AS app3_des_c,
    FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(), 'UTC+8') AS etl_insert_dt,
    FROM_UTC_TIMESTAMP(CURRENT_TIMESTAMP(), 'UTC+8') AS etl_update_dt
 FROM hierarchy
 LEFT JOIN nfc1_cn
    ON hierarchy.nfc1_code = nfc1_cn.app1_cod
 LEFT JOIN nfc2_cn
    ON hierarchy.nfc2_code = nfc2_cn.app2_cod
 LEFT JOIN nfc3_cn
    ON hierarchy.nfc3_code = nfc3_cn.app3_cod
 ;
@@ -0,0 +1,38 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Copy the market-to-TA mapping as a DWS dimension.
 -- Source     : dwd.dwd_gnd_ims_tblmarket_ta_map
 -- Target     : dws.dws_ext_td_ims_market_ta
 -- Grain      : One row per source mapping row (grain defined by the source;
 --              no DISTINCT, matching legacy).
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : dws.dws_ims_td_market_ta (legacy output).
 -- Consumers  : None found in the repo today (03 reads the DWD map directly).
 --              Table is kept for lineage parity; wire consumers during
 --              migration.
 -- Notes      : Legacy SELECT * is replaced with an explicit MARKET, TA
 --              projection; if the source contains additional columns they are
 --              intentionally not carried -- confirm against the source DDL.
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: infer the explicit two-column contract without copying data.
 -- CREATE TABLE IF NOT EXISTS dws.dws_ext_td_ims_market_ta AS
 -- SELECT
 --     MARKET,
 --     TA
 -- FROM dwd.dwd_gnd_ims_tblmarket_ta_map
 -- WHERE 1 = 0;
 -- COMMAND ----------
 INSERT OVERWRITE TABLE dws.dws_ext_td_ims_market_ta (
    MARKET,
    TA
 )
 SELECT
    market,
    ta
 FROM dwd.dwd_gnd_ims_tblmarket_ta_map
 ;
@@ -0,0 +1,68 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Maintain the rolling 5-year (ym, pack_id) -> pack_code snapshot.
 --              The recent 5-year window is refreshed with the current pack
 --              mapping; rows older than the window are retained unchanged from
 --              previous runs (legacy delete+insert behavior).
 -- Source     : dwd.dwd_ims_tf_fact_sales (ym range)
 --              dwd.dwd_ims_td_pack (distinct pack_id, pack_code mapping)
 -- Target     : dws.dws_ext_td_ims_pack_ym
 -- Grain      : One row per (ym, pack_id).
 -- Write mode : Delete + insert (legacy rolling maintenance preserved; NOT a
 --              full refresh -- converting to INSERT OVERWRITE would drop the
 --              retained older-than-5-year rows).
 -- Replaces   : dws.dws_ims_td_pack_ym (legacy table, same maintenance logic).
 -- Consumers  : dws.dws_ext_tf_ims_chpa_sales (Part 1 national pack_code join).
 -- Notes      : The legacy temp view "dwd_ims_td_pack" (distinct pack_id,
 --              pack_code) is inlined as CTE pack_dim. The inner join between
 --              distinct ym and distinct packs is a cross join, exactly as in
 --              the legacy script. The first deployment must DEEP CLONE the
 --              legacy table so rows outside the rolling refresh window are
 --              retained with their original schema and timestamps.
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: this stateful rolling table must seed retained history as well
 -- as schema. An empty LIKE table would permanently lose rows outside the
 -- five-year refresh window on the first run.
 -- CREATE TABLE IF NOT EXISTS dws.dws_ext_td_ims_pack_ym
 -- DEEP CLONE dws.dws_ims_td_pack_ym;
 -- COMMAND ----------
 DELETE FROM dws.dws_ext_td_ims_pack_ym
 WHERE ym + 500 > (SELECT MAX(year * 100 + month) FROM dwd.dwd_ims_tf_fact_sales);
 -- COMMAND ----------
 INSERT INTO dws.dws_ext_td_ims_pack_ym (
    ym,
    pack_id,
    pack_code,
    ETL_INSERT_DT,
    ETL_UPDATE_DT
 )
 WITH fact_ym AS (
    -- Distinct year-months within the recent 5-year window.
    SELECT DISTINCT
        year * 100 + month AS ym
    FROM dwd.dwd_ims_tf_fact_sales
    WHERE year * 100 + month + 500 > (SELECT MAX(year * 100 + month) FROM dwd.dwd_ims_tf_fact_sales)
 ),
 pack_dim AS (
    -- Current pack_id -> pack_code mapping (legacy temp view inlined).
    SELECT DISTINCT
        pack_id,
        pack_code
    FROM dwd.dwd_ims_td_pack
 )
 SELECT DISTINCT
    fact_ym.ym,
    pack_dim.pack_id,
    pack_dim.pack_code,
    from_utc_timestamp(current_timestamp(), 'UTC+8') AS ETL_INSERT_DT,
    from_utc_timestamp(current_timestamp(), 'UTC+8') AS ETL_UPDATE_DT
 FROM fact_ym
 CROSS JOIN pack_dim
 ;
@@ -0,0 +1,237 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Build the unified CHPA fact sales table: national (CHT) rows
 --              from IMS plus province (CHPA) rows from Pharbers, each row
 --              carrying current and prior-year (LY) measures in MTH00* columns.
 -- Source     : dwd.dwd_gnd_pharbers_prov_fact (province fact)
 --              dws.dws_ext_td_ims_geo (province dimension, replaces the legacy
 --              dm.dm_td_geography normalization -- see Notes)
 --              dwd.dwd_gnd_dept_pack_property (IQVIA pack code / counting-unit
 --              ratio)
 --              dwd.dwd_ims_tf_fact_sales (national fact)
 --              dws.dws_ext_td_ims_pack_ym (pack_id -> pack_code by ym)
 --              dwd.dwd_ims_td_audit (Audit_Code = 'CHT' filter)
 -- Target     : dws.dws_ext_tf_ims_chpa_sales
 -- Grain      : One row per (YM, AUDIT_COD, PACK_COD).
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : tmp.tmp_ims_tf_fact_sales (legacy staging fact).
 -- Consumers  : dws.dws_ext_td_ims_date, dm_ims_tf_sales (L2Y view), DM layer.
 -- Notes      :
 --   * Province mapping now joins source PROVINCE_C directly to
 --     dws.dws_ext_td_ims_geo.PROVINCE_C and outputs geo.AUDIT_COD. This
 --     REPLACES the legacy DIM_PROVINCE normalization over dm.dm_td_geography
 --     (CONCAT of 市/自治区/省 suffixes) and removes the DWS -> DM layer
 --     violation. Contract: dws_ext_td_ims_geo must be province-grain (one row
 --     per province, no city rows) and its PROVINCE_C values must exactly match
 --     dwd_gnd_pharbers_prov_fact.PROVINCE_C; unmatched provinces produce NULL
 --     AUDIT_COD and collapse into one NULL group, exactly as legacy.
 --   * Mixed pack-code domains preserved: Part 1 PACK_COD = IMS pack_code (via
 --     pack_ym), Part 2 PACK_COD = IQVIA_PACK_CODE. Do NOT join the two parts
 --     on PACK_COD.
 --   * Asymmetric time windows preserved: Part 1 filters YM >= 202201
 --     (20260320 chenwu CHPA-only rule); Part 2 has no lower bound. Both parts
 --     are capped at yearmont_range = MIN(max province ym, max national ym).
 --   * LY construction preserved: current rows carry MTH00* and LY-shifted rows
 --     (YM+1 year, values in *LY columns) are unioned and summed into the same
 --     (YM, AUDIT_COD, PACK_COD) key.
 --   * COUNTINGUNIT_RATIO fallback to conversion_ratio preserved.
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: copy the verified legacy schema without copying data.
 -- CREATE TABLE IF NOT EXISTS dws.dws_ext_tf_ims_chpa_sales
 -- LIKE tmp.tmp_ims_tf_fact_sales;
 -- COMMAND ----------
 INSERT OVERWRITE TABLE dws.dws_ext_tf_ims_chpa_sales (
    YM,
    AUDIT_COD,
    PACK_COD,
    MTH00LC,
    MTH00LCLY,
    MTH00CN,
    MTH00CNLY,
    MTH00UN,
    MTH00UNLY
 )
 WITH yearmont_range AS (
    -- Cap both parts at the smaller of the two sources' max month to avoid a
    -- leading-corner mismatch between province and national data.
    SELECT
        MIN(ym) AS ym
    FROM (
        SELECT MAX(ym) AS ym FROM dwd.dwd_gnd_pharbers_prov_fact
        UNION ALL
        SELECT MAX(year * 100 + month) AS ym FROM dwd.dwd_ims_tf_fact_sales
    )
 ),
 prov_fact AS (
    -- Pharbers province fact with normalized types (mirrors the legacy
    -- FACT_CHPA_SALES_TEMP_WITH_PREVIOUS view).
    SELECT
        CAST(ym AS INT) AS ym,
        CAST(year AS INT) AS year,
        CAST(REPLACE(ym, year, '') AS INT) AS month,
        CAST(value AS DECIMAL(38, 10)) AS value,
        CAST(countingunit AS DECIMAL(38, 10)) AS countingunit,
        CAST(totalunit AS DECIMAL(38, 10)) AS totalunit,
        province_c,
        phcd,
        conversion_ratio
    FROM dwd.dwd_gnd_pharbers_prov_fact
 ),
 geo_province AS (
    -- Province dimension from DWS. Replaces the legacy DIM_PROVINCE view that
    -- normalized dm.dm_td_geography province names with CONCAT suffixes; the
    -- fact's PROVINCE_C now matches geo.PROVINCE_C directly.
    SELECT
        province_c,
        audit_cod
    FROM dws.dws_ext_td_ims_geo
 ),
 chpa_pack_info AS (
    -- IQVIA pack code and counting-unit ratio (mirrors the legacy
    -- DIM_CHPA_PACK_INFO view; MAX() dedup preserved).
    SELECT
        pack_cod,
        MAX(iqvia_pack_code) AS iqvia_pack_code,
        MAX(countingunit) AS countingunit_ratio
    FROM dwd.dwd_gnd_dept_pack_property
    GROUP BY
        pack_cod
 ),
 fact_chpa_sales AS (
    -- Province rows: current month plus the same month one year earlier
    -- (LY), unioned so the final GROUP BY recombines them per (YM, audit,
    -- pack). PACK_COD is the IQVIA pack code domain.
    SELECT
        prov_fact.ym,
        prov_fact.year,
        REPLACE(prov_fact.ym, prov_fact.year, '') AS month,
        chpa_pack_info.iqvia_pack_code AS pack_code,
        geo_province.audit_cod,
        prov_fact.value AS mth00lc,
        0 AS mth00lcly,
        CASE
            WHEN chpa_pack_info.countingunit_ratio IS NULL
                THEN prov_fact.totalunit * prov_fact.conversion_ratio
            ELSE prov_fact.totalunit * chpa_pack_info.countingunit_ratio
        END AS mth00cn,
        0 AS mth00cnly,
        prov_fact.totalunit AS mth00un,
        0 AS mth00unly
    FROM prov_fact
    LEFT JOIN geo_province
        ON prov_fact.province_c = geo_province.province_c
    LEFT JOIN chpa_pack_info
        ON prov_fact.phcd = chpa_pack_info.pack_cod
    UNION ALL
    SELECT
        CAST((prov_fact.year + 1) * 100 + REPLACE(prov_fact.ym, prov_fact.year, '') AS INT) AS ym,
        prov_fact.year + 1 AS year,
        CAST(REPLACE(prov_fact.ym, prov_fact.year, '') AS INT) AS month,
        chpa_pack_info.iqvia_pack_code AS pack_code,
        geo_province.audit_cod,
        0 AS mth00lc,
        prov_fact.value AS mth00lcly,
        0 AS mth00cn,
        CASE
            WHEN chpa_pack_info.countingunit_ratio IS NULL
                THEN prov_fact.totalunit * prov_fact.conversion_ratio
            ELSE prov_fact.totalunit * chpa_pack_info.countingunit_ratio
        END AS mth00cnly,
        0 AS mth00un,
        prov_fact.totalunit AS mth00unly
    FROM prov_fact
    LEFT JOIN geo_province
        ON prov_fact.province_c = geo_province.province_c
    LEFT JOIN chpa_pack_info
        ON prov_fact.phcd = chpa_pack_info.pack_cod
 ),
 national_sales AS (
    -- National rows: current month plus LY-shifted rows; pack_code comes from
    -- the rolling pack_ym snapshot (IMS pack_code domain -- mixed with the
    -- IQVIA domain in Part 2 by design).
    SELECT
        year * 100 + month AS ym,
        year,
        month,
        pack_ym.pack_code,
        audit_id,
        sales_value_lc AS mth00lc,
        0 AS mth00lcly,
        counting_unit AS mth00cn,
        0 AS mth00cnly,
        sales_unit AS mth00un,
        0 AS mth00unly
    FROM dwd.dwd_ims_tf_fact_sales fact_sales
    LEFT JOIN dws.dws_ext_td_ims_pack_ym pack_ym
        ON fact_sales.pack_id = pack_ym.pack_id
       AND fact_sales.year * 100 + fact_sales.month = pack_ym.ym
    UNION ALL
    SELECT
        (year + 1) * 100 + month AS ym,
        year + 1 AS year,
        month,
        pack_ym.pack_code,
        audit_id,
        0 AS mth00lc,
        sales_value_lc AS mth00lcly,
        0 AS mth00cn,
        counting_unit AS mth00cnly,
        0 AS mth00un,
        sales_unit AS mth00unly
    FROM dwd.dwd_ims_tf_fact_sales fact_sales
    LEFT JOIN dws.dws_ext_td_ims_pack_ym pack_ym
        ON fact_sales.pack_id = pack_ym.pack_id
       AND fact_sales.year * 100 + fact_sales.month = pack_ym.ym
 )
 -- Part 1: national CHT rows. Asymmetric window preserved: YM >= 202201 and
 -- YM <= yearmont_range (the legacy 20260320 CHPA-only lower bound).
 SELECT
    national_sales.ym AS YM,
    audit.audit_code AS AUDIT_COD,
    national_sales.pack_code AS PACK_COD,
    SUM(national_sales.mth00lc) AS MTH00LC,
    SUM(national_sales.mth00lcly) AS MTH00LCLY,
    SUM(national_sales.mth00cn) AS MTH00CN,
    SUM(national_sales.mth00cnly) AS MTH00CNLY,
    SUM(national_sales.mth00un) AS MTH00UN,
    SUM(national_sales.mth00unly) AS MTH00UNLY
 FROM national_sales
 LEFT JOIN dwd.dwd_ims_td_audit audit
    ON national_sales.audit_id = audit.audit_id
 WHERE national_sales.ym <= (SELECT ym FROM yearmont_range)
  AND national_sales.ym >= 202201
  AND audit.audit_code = 'CHT'
 GROUP BY
    national_sales.ym,
    audit.audit_code,
    national_sales.pack_code
 UNION ALL
 -- Part 2: province CHPA rows. No lower ym bound (asymmetric with Part 1).
 SELECT
    fact_chpa_sales.ym AS YM,
    fact_chpa_sales.audit_cod AS AUDIT_COD,
    fact_chpa_sales.pack_code AS PACK_COD,
    SUM(fact_chpa_sales.mth00lc) AS MTH00LC,
    SUM(fact_chpa_sales.mth00lcly) AS MTH00LCLY,
    SUM(fact_chpa_sales.mth00cn) AS MTH00CN,
    SUM(fact_chpa_sales.mth00cnly) AS MTH00CNLY,
    SUM(fact_chpa_sales.mth00un) AS MTH00UN,
    SUM(fact_chpa_sales.mth00unly) AS MTH00UNLY
 FROM fact_chpa_sales
 WHERE fact_chpa_sales.ym <= (SELECT ym FROM yearmont_range)
 GROUP BY
    fact_chpa_sales.ym,
    fact_chpa_sales.audit_cod,
    fact_chpa_sales.pack_code
 ;
@@ -0,0 +1,84 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Build the dynamic month dimension from the CHPA fact sales
 --              range, with the report month flagged 'R' (max YM present).
 -- Source     : dws.dws_ext_tf_ims_chpa_sales (YM range; reads the promoted
 --              DWS fact instead of the legacy tmp.tmp_ims_tf_fact_sales)
 -- Target     : dws.dws_ext_td_ims_date
 -- Grain      : One row per YM present in the fact (approximately nine years by
 --              the legacy YYYYMM-minus-900 arithmetic).
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : dws.dws_ims_td_date (legacy output, same column contract and
 --              the same 900 arithmetic: YM > max(YM) - 900).
 -- Consumers  : DM calendar / pack-property jobs must migrate to this target.
 -- Notes      : Legacy header comment claims "最近五年" (5 years), but subtracting
 --              900 from a YYYYMM value gives approximately nine years; that
 --              arithmetic is preserved exactly. QUARTER/YQ/HALF_YEAR/DATE_FLAG
 --              preserved verbatim.
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: copy the verified legacy schema without copying data.
 -- CREATE TABLE IF NOT EXISTS dws.dws_ext_td_ims_date
 -- LIKE dws.dws_ims_td_date;
 -- COMMAND ----------
 INSERT OVERWRITE TABLE dws.dws_ext_td_ims_date (
    YM,
    YEAR,
    MONTH,
    QUARTER,
    YQ,
    DATE_FLAG,
    HALF_YEAR,
    ETL_INSERT_DT,
    ETL_UPDATE_DT
 )
 SELECT DISTINCT
    fact_sales.ym AS YM,
    LEFT(CAST(fact_sales.ym AS STRING), 4) AS YEAR,
    RIGHT(CAST(fact_sales.ym AS STRING), 2) AS MONTH,
    CONCAT(
        'Q',
        QUARTER(
            DATE(
                CONCAT(
                    LEFT(CAST(fact_sales.ym AS STRING), 4),
                    '-',
                    RIGHT(CAST(fact_sales.ym AS STRING), 2),
                    '-01'
                )
            )
        )
    ) AS QUARTER,
    CONCAT(
        LEFT(CAST(fact_sales.ym AS STRING), 4),
        'Q',
        QUARTER(
            DATE(
                CONCAT(
                    LEFT(CAST(fact_sales.ym AS STRING), 4),
                    '-',
                    RIGHT(CAST(fact_sales.ym AS STRING), 2),
                    '-01'
                )
            )
        )
    ) AS YQ,
    CASE
        WHEN fact_sales.ym = (SELECT MAX(ym) FROM dws.dws_ext_tf_ims_chpa_sales) THEN 'R'
        ELSE RIGHT(CAST(fact_sales.ym AS STRING), 2)
    END AS DATE_FLAG,
    CASE
        WHEN fact_sales.ym % 100 > 6
            THEN CONCAT(LEFT(CAST(fact_sales.ym AS STRING), 4), 'H2')
        ELSE CONCAT(LEFT(CAST(fact_sales.ym AS STRING), 4), 'H1')
    END AS HALF_YEAR,
    from_utc_timestamp(current_timestamp(), 'UTC+8') AS ETL_INSERT_DT,
    from_utc_timestamp(current_timestamp(), 'UTC+8') AS ETL_UPDATE_DT
 FROM dws.dws_ext_tf_ims_chpa_sales fact_sales
 WHERE fact_sales.ym > (SELECT MAX(ym) - 900 FROM dws.dws_ext_tf_ims_chpa_sales)
 ORDER BY fact_sales.ym DESC
 ;
@@ -0,0 +1,813 @@
 -- Databricks notebook source
 -- =============================================================================
 -- Purpose    : Build the IMS market dictionary: one row per market x pack with
 --              pack attributes, business unit, market ratio and key competitor.
 -- Source     : dws.dws_ext_td_ims_pack_property   (31-column pack hub)
 --              dwd.dwd_gnd_ims_tblmarket          (include / exclude / extend rules)
 --              dwd.dwd_gnd_ims_tblkeycompetitor   (key-competitor rules)
 -- Target     : dws.dws_ext_td_ims_market
 -- Grain      : Distinct market x pack (attributes, bu, market_ratio); at most one
 --              Key_Competitor per (market, PACK_COD, PROD_COD). The same
 --              (market, PACK_COD) may fan out across bu / market_ratio rows.
 -- Write mode : Full refresh (INSERT OVERWRITE).
 -- Replaces   : dws.dws_ims_td_market (legacy output, kept as validation baseline).
 -- Consumers  : dm_ims_td_org / dm_ims_td_org_hvh (LEFT JOIN on pack_cod -> market).
 -- Depends on : 01_dwd pad-zero updates of dwd.dwd_gnd_ims_tblmarket and
 --              dwd.dwd_gnd_ims_tblkeycompetitor, and the pack-property job that
 --              lands dws.dws_ext_td_ims_pack_property.
 -- =============================================================================
 -- Compatibility notes (legacy behavior preserved, NOT fixed):
 -- 1. Extend-market ratio: the legacy Market_Ratio CASE for extended rows had no
 --    ELSE branch, so a configured Extend_Market_Ratio yielded NULL, which the
 --    legacy cell-5 UPDATE then forced to '1'. NET LEGACY BEHAVIOR: every
 --    extended-market row ends with Market_Ratio = '1' and the configured
 --    Extend_Market_Ratio is silently ignored. Reproduced verbatim below (buggy
 --    CASE in market_extended + NULL -> '1' in market_all). Do not "fix" without
 --    business sign-off.
 -- 2. MERGE multi-match failure modes removed: the legacy MERGE ... DELETE cells
 --    could raise DELTA_MULTIPLE_SOURCE_ROW_MATCHING when several source rows
 --    matched one target row (config rows whose 14-key tuple is identical except
 --    bu / market_ratio). The anti-join equivalents below apply the delete
 --    deterministically and match the MERGE result in every non-error case.
 -- 3. KC dedup NULL semantics: the anti-join reproduces the MERGE's SQL equality
 --    on (14 keys + Key_Competitor + no). NULL never equals NULL, so -- exactly
 --    like legacy -- duplicate rows inside one (market, PACK_COD, PROD_COD)
 --    partition that carry NO key-competitor match (Key_Competitor / no NULL) are
 --    NOT deduped; all of them survive and later become 'Others'. A naive
 --    rank = 1 filter would have dropped them; this script does not.
 -- 4. no1 window preserved verbatim, including the duplicated NFC2_CODE sort key
 --    (legacy lines 152-155). ROW_NUMBER() OVER(ORDER BY ...) has no total order;
 --    among exact ties the winner is engine-defined (same as legacy).
 -- 5. Pack hub swap: t1 now reads dws.dws_ext_td_ims_pack_property. Equivalence
 --    holds only if that table is a logic-identical copy of
 --    dwd.dwd_ims_td_pack_property (same 31-column contract incl. zero-padded
 --    codes and non-null CORP_COD / STGH_DES).
 -- 6. Timestamps: the legacy final INSERT had no column list and two anonymous
 --    from_utc_timestamp expressions. They are named ETL_INSERT_DT /
 --    ETL_UPDATE_DT here (repo convention); confirm against the LIKE-created DDL
 --    via DESCRIBE before the first run.
 -- 7. t1.* is replaced by an explicit 31-column projection; if the pack hub ever
 --    gains columns this script pins the contract where legacy positional
 --    inserts would have silently shifted.
 -- 8. DISTINCT is kept at exactly the legacy stages (include, exclude, extend,
 --    KC join, final); no cross-stage dedup is added and the base/extended
 --    UNION ALL does not dedup between its two branches.
 -- 9. Extended rows preserve a legacy positional-write defect: cell 4 selected
 --    BU before BrandType into a target whose final fields are BrandType, bu,
 --    Market_Ratio. Therefore extended rows expose configured BU as BrandType
 --    and the source pack BrandType as bu. This is explicit below so refactoring
 --    does not silently change persisted output; correct only with business
 --    sign-off in a separate change.
 -- =============================================================================
 -- COMMAND ----------
 -- One-time DDL: copy the verified legacy schema without copying data.
 -- CREATE TABLE IF NOT EXISTS dws.dws_ext_td_ims_market
 -- LIKE dws.dws_ims_td_market;
 -- COMMAND ----------
 INSERT OVERWRITE TABLE dws.dws_ext_td_ims_market (
    market,
    PACK_COD,
    PACK_DES,
    STGH_DES,
    PACK_LCH,
    PROD_COD,
    CMPS_COD,
    CMPS_DES,
    ATC1_COD,
    ATC2_COD,
    ATC3_COD,
    ATC4_COD,
    APP1_COD,
    APP2_COD,
    APP3_COD,
    BIO_DESC,
    GENE_ORIG_DESC,
    ETH_OTC_DESC,
    NRDL_DESC,
    NRDL_Entry_Date,
    EDL_DESC,
    TCM_DESC,
    PAED_DESC,
    GQCE_DESC,
    VBP_DESC,
    MANU_COD,
    MANU_DES,
    MNFL_COD,
    MNFL_DES,
    CORP_COD,
    CORP_DES,
    BrandType,
    bu,
    Market_Ratio,
    Key_Competitor,
    ETL_INSERT_DT,
    ETL_UPDATE_DT
 )
 WITH pack_hub AS (
    -- 31-column pack property contract (legacy dwd.dwd_ims_td_pack_property).
    SELECT
        PACK_COD,
        PACK_DES,
        STGH_DES,
        PACK_LCH,
        PROD_COD,
        CMPS_COD,
        CMPS_DES,
        ATC1_COD,
        ATC2_COD,
        ATC3_COD,
        ATC4_COD,
        APP1_COD,
        APP2_COD,
        APP3_COD,
        BIO_DESC,
        GENE_ORIG_DESC,
        ETH_OTC_DESC,
        NRDL_DESC,
        NRDL_Entry_Date,
        EDL_DESC,
        TCM_DESC,
        PAED_DESC,
        GQCE_DESC,
        VBP_DESC,
        MANU_COD,
        MANU_DES,
        MNFL_COD,
        MNFL_DES,
        CORP_COD,
        CORP_DES,
        BrandType
    FROM dws.dws_ext_td_ims_pack_property
 ),
 market_config_include AS (
    -- Explicitly defined markets (legacy cell 1 filter): Extend_Market IS NULL
    -- and NOT_IN_FLAG IS NULL or '1' (despite the name, NOT_IN_FLAG = '1' is an
    -- include rule; '0' is the exclude rule handled below).
    SELECT
        market,
        bu,
        ATC1_Code AS atc1_code,
        ATC2_Code AS atc2_code,
        ATC3_Code AS atc3_code,
        ATC4_Code AS atc4_code,
        NFC1_Code AS nfc1_code,
        NFC2_Code AS nfc2_code,
        NFC3_Code AS nfc3_code,
        corporation_code,
        Manufacturer_Code AS manufacturer_code,
        Product_Code AS product_code,
        Pack_Code AS pack_code,
        Strength AS strength,
        Molecule_Code AS molecule_code,
        extend_market_ratio
    FROM dwd.dwd_gnd_ims_tblmarket
    WHERE Extend_Market IS NULL
      AND (NOT_IN_FLAG IS NULL OR NOT_IN_FLAG = '1')
 ),
 market_config_exclude AS (
    -- Anti rules (legacy cell 2 filter): Extend_Market IS NULL and NOT_IN_FLAG = '0'.
    SELECT
        market,
        bu,
        ATC1_Code AS atc1_code,
        ATC2_Code AS atc2_code,
        ATC3_Code AS atc3_code,
        ATC4_Code AS atc4_code,
        NFC1_Code AS nfc1_code,
        NFC2_Code AS nfc2_code,
        NFC3_Code AS nfc3_code,
        corporation_code,
        Manufacturer_Code AS manufacturer_code,
        Product_Code AS product_code,
        Pack_Code AS pack_code,
        Strength AS strength,
        Molecule_Code AS molecule_code,
        extend_market_ratio
    FROM dwd.dwd_gnd_ims_tblmarket
    WHERE Extend_Market IS NULL
      AND NOT_IN_FLAG = '0'
 ),
 market_config_extend AS (
    -- Extend rules (legacy cell 4 filter): Extend_Market IS NOT NULL.
    SELECT
        Market AS market,
        BU AS bu,
        Extend_Market AS extend_market,
        Extend_Market_Ratio AS extend_market_ratio
    FROM dwd.dwd_gnd_ims_tblmarket
    WHERE Extend_Market IS NOT NULL
 ),
 market_include_rows AS (
    -- Legacy cell 1: wildcard equality -- a config code constrains the match
    -- only when non-null (NULL config code = wildcard). Kept as LEFT JOIN with
    -- WHERE t2.market IS NOT NULL, exactly like legacy (effectively inner).
    SELECT DISTINCT
        t2.market AS market,
        t1.PACK_COD,
        t1.PACK_DES,
        t1.STGH_DES,
        t1.PACK_LCH,
        t1.PROD_COD,
        t1.CMPS_COD,
        t1.CMPS_DES,
        t1.ATC1_COD,
        t1.ATC2_COD,
        t1.ATC3_COD,
        t1.ATC4_COD,
        t1.APP1_COD,
        t1.APP2_COD,
        t1.APP3_COD,
        t1.BIO_DESC,
        t1.GENE_ORIG_DESC,
        t1.ETH_OTC_DESC,
        t1.NRDL_DESC,
        t1.NRDL_Entry_Date,
        t1.EDL_DESC,
        t1.TCM_DESC,
        t1.PAED_DESC,
        t1.GQCE_DESC,
        t1.VBP_DESC,
        t1.MANU_COD,
        t1.MANU_DES,
        t1.MNFL_COD,
        t1.MNFL_DES,
        t1.CORP_COD,
        t1.CORP_DES,
        t1.BrandType,
        t2.bu AS bu,
        CASE
            WHEN t2.extend_market_ratio IS NULL THEN '1'
            ELSE t2.extend_market_ratio
        END AS Market_Ratio
    FROM pack_hub t1
    LEFT JOIN market_config_include t2
        ON t1.ATC1_COD = CASE WHEN t2.atc1_code IS NULL THEN t1.ATC1_COD ELSE t2.atc1_code END
        AND t1.ATC2_COD = CASE WHEN t2.atc2_code IS NULL THEN t1.ATC2_COD ELSE t2.atc2_code END
        AND t1.ATC3_COD = CASE WHEN t2.atc3_code IS NULL THEN t1.ATC3_COD ELSE t2.atc3_code END
        AND t1.ATC4_COD = CASE WHEN t2.atc4_code IS NULL THEN t1.ATC4_COD ELSE t2.atc4_code END
        AND t1.APP1_COD = CASE WHEN t2.nfc1_code IS NULL THEN t1.APP1_COD ELSE t2.nfc1_code END
        AND t1.APP2_COD = CASE WHEN t2.nfc2_code IS NULL THEN t1.APP2_COD ELSE t2.nfc2_code END
        AND t1.APP3_COD = CASE WHEN t2.nfc3_code IS NULL THEN t1.APP3_COD ELSE t2.nfc3_code END
        AND t1.CORP_COD = CASE WHEN t2.corporation_code IS NULL THEN t1.CORP_COD ELSE t2.corporation_code END
        AND t1.MANU_COD = CASE WHEN t2.manufacturer_code IS NULL THEN t1.MANU_COD ELSE t2.manufacturer_code END
        AND t1.PROD_COD = CASE WHEN t2.product_code IS NULL THEN t1.PROD_COD ELSE t2.product_code END
        AND t1.PACK_COD = CASE WHEN t2.pack_code IS NULL THEN t1.PACK_COD ELSE t2.pack_code END
        AND t1.STGH_DES = CASE WHEN t2.strength IS NULL THEN t1.STGH_DES ELSE t2.strength END
        AND t1.CMPS_COD = CASE WHEN t2.molecule_code IS NULL THEN t1.CMPS_COD ELSE t2.molecule_code END
    WHERE t2.market IS NOT NULL
 ),
 market_exclude_rows AS (
    -- Legacy cell 2: same wildcard join against the exclude rules.
    SELECT DISTINCT
        t2.market AS market,
        t1.PACK_COD,
        t1.PACK_DES,
        t1.STGH_DES,
        t1.PACK_LCH,
        t1.PROD_COD,
        t1.CMPS_COD,
        t1.CMPS_DES,
        t1.ATC1_COD,
        t1.ATC2_COD,
        t1.ATC3_COD,
        t1.ATC4_COD,
        t1.APP1_COD,
        t1.APP2_COD,
        t1.APP3_COD,
        t1.BIO_DESC,
        t1.GENE_ORIG_DESC,
        t1.ETH_OTC_DESC,
        t1.NRDL_DESC,
        t1.NRDL_Entry_Date,
        t1.EDL_DESC,
        t1.TCM_DESC,
        t1.PAED_DESC,
        t1.GQCE_DESC,
        t1.VBP_DESC,
        t1.MANU_COD,
        t1.MANU_DES,
        t1.MNFL_COD,
        t1.MNFL_DES,
        t1.CORP_COD,
        t1.CORP_DES,
        t1.BrandType,
        t2.bu AS bu,
        CASE
            WHEN t2.extend_market_ratio IS NULL THEN '1'
            ELSE t2.extend_market_ratio
        END AS Market_Ratio
    FROM pack_hub t1
    LEFT JOIN market_config_exclude t2
        ON t1.ATC1_COD = CASE WHEN t2.atc1_code IS NULL THEN t1.ATC1_COD ELSE t2.atc1_code END
        AND t1.ATC2_COD = CASE WHEN t2.atc2_code IS NULL THEN t1.ATC2_COD ELSE t2.atc2_code END
        AND t1.ATC3_COD = CASE WHEN t2.atc3_code IS NULL THEN t1.ATC3_COD ELSE t2.atc3_code END
        AND t1.ATC4_COD = CASE WHEN t2.atc4_code IS NULL THEN t1.ATC4_COD ELSE t2.atc4_code END
        AND t1.APP1_COD = CASE WHEN t2.nfc1_code IS NULL THEN t1.APP1_COD ELSE t2.nfc1_code END
        AND t1.APP2_COD = CASE WHEN t2.nfc2_code IS NULL THEN t1.APP2_COD ELSE t2.nfc2_code END
        AND t1.APP3_COD = CASE WHEN t2.nfc3_code IS NULL THEN t1.APP3_COD ELSE t2.nfc3_code END
        AND t1.CORP_COD = CASE WHEN t2.corporation_code IS NULL THEN t1.CORP_COD ELSE t2.corporation_code END
        AND t1.MANU_COD = CASE WHEN t2.manufacturer_code IS NULL THEN t1.MANU_COD ELSE t2.manufacturer_code END
        AND t1.PROD_COD = CASE WHEN t2.product_code IS NULL THEN t1.PROD_COD ELSE t2.product_code END
        AND t1.PACK_COD = CASE WHEN t2.pack_code IS NULL THEN t1.PACK_COD ELSE t2.pack_code END
        AND t1.STGH_DES = CASE WHEN t2.strength IS NULL THEN t1.STGH_DES ELSE t2.strength END
        AND t1.CMPS_COD = CASE WHEN t2.molecule_code IS NULL THEN t1.CMPS_COD ELSE t2.molecule_code END
    WHERE t2.market IS NOT NULL
 ),
 market_after_exclude AS (
    -- Legacy cell 3 MERGE ... WHEN MATCHED THEN DELETE, expressed as an
    -- anti-join on the 14 materialized keys (13 codes + market). NULL never
    -- equals NULL, exactly like the MERGE join.
    SELECT
        t1.market,
        t1.PACK_COD,
        t1.PACK_DES,
        t1.STGH_DES,
        t1.PACK_LCH,
        t1.PROD_COD,
        t1.CMPS_COD,
        t1.CMPS_DES,
        t1.ATC1_COD,
        t1.ATC2_COD,
        t1.ATC3_COD,
        t1.ATC4_COD,
        t1.APP1_COD,
        t1.APP2_COD,
        t1.APP3_COD,
        t1.BIO_DESC,
        t1.GENE_ORIG_DESC,
        t1.ETH_OTC_DESC,
        t1.NRDL_DESC,
        t1.NRDL_Entry_Date,
        t1.EDL_DESC,
        t1.TCM_DESC,
        t1.PAED_DESC,
        t1.GQCE_DESC,
        t1.VBP_DESC,
        t1.MANU_COD,
        t1.MANU_DES,
        t1.MNFL_COD,
        t1.MNFL_DES,
        t1.CORP_COD,
        t1.CORP_DES,
        t1.BrandType,
        t1.bu,
        t1.Market_Ratio
    FROM market_include_rows t1
    LEFT ANTI JOIN market_exclude_rows t2
        ON t1.ATC1_COD = t2.ATC1_COD
        AND t1.ATC2_COD = t2.ATC2_COD
        AND t1.ATC3_COD = t2.ATC3_COD
        AND t1.ATC4_COD = t2.ATC4_COD
        AND t1.APP1_COD = t2.APP1_COD
        AND t1.APP2_COD = t2.APP2_COD
        AND t1.APP3_COD = t2.APP3_COD
        AND t1.CORP_COD = t2.CORP_COD
        AND t1.MANU_COD = t2.MANU_COD
        AND t1.PROD_COD = t2.PROD_COD
        AND t1.PACK_COD = t2.PACK_COD
        AND t1.STGH_DES = t2.STGH_DES
        AND t1.CMPS_COD = t2.CMPS_COD
        AND t1.market = t2.market
 ),
 market_extended AS (
    -- Legacy cell 4: clone every pack of an existing market into the extend
    -- market (append). Market_Ratio keeps the legacy missing-ELSE CASE exactly
    -- (a configured extend_market_ratio yields NULL here; see compatibility
    -- note 1 -- it is intentionally NOT fixed).
    SELECT DISTINCT
        t2.market AS market,
        t1.PACK_COD,
        t1.PACK_DES,
        t1.STGH_DES,
        t1.PACK_LCH,
        t1.PROD_COD,
        t1.CMPS_COD,
        t1.CMPS_DES,
        t1.ATC1_COD,
        t1.ATC2_COD,
        t1.ATC3_COD,
        t1.ATC4_COD,
        t1.APP1_COD,
        t1.APP2_COD,
        t1.APP3_COD,
        t1.BIO_DESC,
        t1.GENE_ORIG_DESC,
        t1.ETH_OTC_DESC,
        t1.NRDL_DESC,
        t1.NRDL_Entry_Date,
        t1.EDL_DESC,
        t1.TCM_DESC,
        t1.PAED_DESC,
        t1.GQCE_DESC,
        t1.VBP_DESC,
        t1.MANU_COD,
        t1.MANU_DES,
        t1.MNFL_COD,
        t1.MNFL_DES,
        t1.CORP_COD,
        t1.CORP_DES,
        t2.bu AS BrandType,
        t1.BrandType AS bu,
        CASE
            WHEN t2.extend_market_ratio IS NULL THEN '1'
        END AS Market_Ratio
    FROM market_after_exclude t1
    LEFT JOIN market_config_extend t2
        ON t1.market = t2.extend_market
    WHERE t2.market IS NOT NULL
 ),
 market_all AS (
    -- Legacy cells 4 (INSERT INTO append) + 5 (UPDATE Market_Ratio = 1 WHERE
    -- Market_Ratio IS NULL): union of base and extend rows, then NULL ratios
    -- default to '1'. Together with the missing-ELSE CASE above this makes every
    -- extended row end with Market_Ratio = '1' (compatibility note 1).
    SELECT
        market,
        PACK_COD,
        PACK_DES,
        STGH_DES,
        PACK_LCH,
        PROD_COD,
        CMPS_COD,
        CMPS_DES,
        ATC1_COD,
        ATC2_COD,
        ATC3_COD,
        ATC4_COD,
        APP1_COD,
        APP2_COD,
        APP3_COD,
        BIO_DESC,
        GENE_ORIG_DESC,
        ETH_OTC_DESC,
        NRDL_DESC,
        NRDL_Entry_Date,
        EDL_DESC,
        TCM_DESC,
        PAED_DESC,
        GQCE_DESC,
        VBP_DESC,
        MANU_COD,
        MANU_DES,
        MNFL_COD,
        MNFL_DES,
        CORP_COD,
        CORP_DES,
        BrandType,
        bu,
        CASE
            WHEN Market_Ratio IS NULL THEN '1'
            ELSE Market_Ratio
        END AS Market_Ratio
    FROM (
        SELECT
            market,
            PACK_COD,
            PACK_DES,
            STGH_DES,
            PACK_LCH,
            PROD_COD,
            CMPS_COD,
            CMPS_DES,
            ATC1_COD,
            ATC2_COD,
            ATC3_COD,
            ATC4_COD,
            APP1_COD,
            APP2_COD,
            APP3_COD,
            BIO_DESC,
            GENE_ORIG_DESC,
            ETH_OTC_DESC,
            NRDL_DESC,
            NRDL_Entry_Date,
            EDL_DESC,
            TCM_DESC,
            PAED_DESC,
            GQCE_DESC,
            VBP_DESC,
            MANU_COD,
            MANU_DES,
            MNFL_COD,
            MNFL_DES,
            CORP_COD,
            CORP_DES,
            BrandType,
            bu,
            Market_Ratio
        FROM market_after_exclude
        UNION ALL
        SELECT
            market,
            PACK_COD,
            PACK_DES,
            STGH_DES,
            PACK_LCH,
            PROD_COD,
            CMPS_COD,
            CMPS_DES,
            ATC1_COD,
            ATC2_COD,
            ATC3_COD,
            ATC4_COD,
            APP1_COD,
            APP2_COD,
            APP3_COD,
            BIO_DESC,
            GENE_ORIG_DESC,
            ETH_OTC_DESC,
            NRDL_DESC,
            NRDL_Entry_Date,
            EDL_DESC,
            TCM_DESC,
            PAED_DESC,
            GQCE_DESC,
            VBP_DESC,
            MANU_COD,
            MANU_DES,
            MNFL_COD,
            MNFL_DES,
            CORP_COD,
            CORP_DES,
            BrandType,
            bu,
            Market_Ratio
        FROM market_extended
    ) market_union
 ),
 key_competitor_rules AS (
    -- Legacy cell 6 source: no1 = specificity rank over all KC rules, computed
    -- verbatim -- including the duplicated NFC2_CODE sort key (compatibility
    -- note 4; ROW_NUMBER() has no total order, ties are engine-defined).
    SELECT
        ROW_NUMBER() OVER (
            ORDER BY
                CASE
                    WHEN ATC1_Code IS NOT NULL THEN 1
                    WHEN ATC2_Code IS NOT NULL THEN 2
                    WHEN ATC3_Code IS NOT NULL THEN 3
                    WHEN ATC4_Code IS NOT NULL THEN 4
                    WHEN Molecule_Code IS NOT NULL THEN 5
                    WHEN Product_Code IS NOT NULL THEN 6
                    WHEN Pack_Code IS NOT NULL THEN 7
                    ELSE 999
                END,
                CASE WHEN NFC1_CODE IS NULL THEN 0 ELSE 1 END,
                CASE WHEN NFC2_CODE IS NULL THEN 0 ELSE 1 END,
                CASE WHEN NFC2_CODE IS NULL THEN 0 ELSE 1 END,
                CASE WHEN NFC3_CODE IS NULL THEN 0 ELSE 1 END
        ) AS no1,
        keycompetitor,
        no,
        market,
        ATC1_Code AS atc1_code,
        ATC2_Code AS atc2_code,
        ATC3_Code AS atc3_code,
        ATC4_Code AS atc4_code,
        NFC1_CODE AS nfc1_code,
        NFC2_CODE AS nfc2_code,
        NFC3_CODE AS nfc3_code,
        corporation_code,
        Manufacturer_Code AS manufacturer_code,
        Product_Code AS product_code,
        Pack_Code AS pack_code,
        Strength AS strength,
        Molecule_Code AS molecule_code
    FROM dwd.dwd_gnd_ims_tblkeycompetitor
 ),
 market_kc AS (
    -- Legacy cell 6: wildcard LEFT JOIN to key-competitor rules on the 13 codes
    -- plus market; unmatched packs keep NULL Key_Competitor / no / no1.
    SELECT DISTINCT
        t2.keycompetitor AS Key_Competitor,
        t2.no AS no,
        t2.no1 AS no1,
        t1.market,
        t1.PACK_COD,
        t1.PACK_DES,
        t1.STGH_DES,
        t1.PACK_LCH,
        t1.PROD_COD,
        t1.CMPS_COD,
        t1.CMPS_DES,
        t1.ATC1_COD,
        t1.ATC2_COD,
        t1.ATC3_COD,
        t1.ATC4_COD,
        t1.APP1_COD,
        t1.APP2_COD,
        t1.APP3_COD,
        t1.BIO_DESC,
        t1.GENE_ORIG_DESC,
        t1.ETH_OTC_DESC,
        t1.NRDL_DESC,
        t1.NRDL_Entry_Date,
        t1.EDL_DESC,
        t1.TCM_DESC,
        t1.PAED_DESC,
        t1.GQCE_DESC,
        t1.VBP_DESC,
        t1.MANU_COD,
        t1.MANU_DES,
        t1.MNFL_COD,
        t1.MNFL_DES,
        t1.CORP_COD,
        t1.CORP_DES,
        t1.BrandType,
        t1.bu,
        t1.Market_Ratio
    FROM market_all t1
    LEFT JOIN key_competitor_rules t2
        ON t1.ATC1_COD = CASE WHEN t2.atc1_code IS NULL THEN t1.ATC1_COD ELSE t2.atc1_code END
        AND t1.ATC2_COD = CASE WHEN t2.atc2_code IS NULL THEN t1.ATC2_COD ELSE t2.atc2_code END
        AND t1.ATC3_COD = CASE WHEN t2.atc3_code IS NULL THEN t1.ATC3_COD ELSE t2.atc3_code END
        AND t1.ATC4_COD = CASE WHEN t2.atc4_code IS NULL THEN t1.ATC4_COD ELSE t2.atc4_code END
        AND t1.APP1_COD = CASE WHEN t2.nfc1_code IS NULL THEN t1.APP1_COD ELSE t2.nfc1_code END
        AND t1.APP2_COD = CASE WHEN t2.nfc2_code IS NULL THEN t1.APP2_COD ELSE t2.nfc2_code END
        AND t1.APP3_COD = CASE WHEN t2.nfc3_code IS NULL THEN t1.APP3_COD ELSE t2.nfc3_code END
        AND t1.CORP_COD = CASE WHEN t2.corporation_code IS NULL THEN t1.CORP_COD ELSE t2.corporation_code END
        AND t1.MANU_COD = CASE WHEN t2.manufacturer_code IS NULL THEN t1.MANU_COD ELSE t2.manufacturer_code END
        AND t1.PROD_COD = CASE WHEN t2.product_code IS NULL THEN t1.PROD_COD ELSE t2.product_code END
        AND t1.PACK_COD = CASE WHEN t2.pack_code IS NULL THEN t1.PACK_COD ELSE t2.pack_code END
        AND t1.STGH_DES = CASE WHEN t2.strength IS NULL THEN t1.STGH_DES ELSE t2.strength END
        AND t1.CMPS_COD = CASE WHEN t2.molecule_code IS NULL THEN t1.CMPS_COD ELSE t2.molecule_code END
        AND t1.market = CASE WHEN t2.market IS NULL THEN t1.market ELSE t2.market END
 ),
 market_kc_ranked AS (
    -- Legacy cell 7: rank KC rules per (market, PACK_COD, PROD_COD); the winner
    -- is the highest no1, ties broken by the highest no (latest rule). Ordering
    -- is preserved verbatim (no1 DESC, no DESC).
    SELECT
        row_number() OVER (
            PARTITION BY market, PACK_COD, PROD_COD
            ORDER BY no1 DESC, no DESC
        ) AS rank_id,
        market,
        PACK_COD,
        PACK_DES,
        STGH_DES,
        PACK_LCH,
        PROD_COD,
        CMPS_COD,
        CMPS_DES,
        ATC1_COD,
        ATC2_COD,
        ATC3_COD,
        ATC4_COD,
        APP1_COD,
        APP2_COD,
        APP3_COD,
        BIO_DESC,
        GENE_ORIG_DESC,
        ETH_OTC_DESC,
        NRDL_DESC,
        NRDL_Entry_Date,
        EDL_DESC,
        TCM_DESC,
        PAED_DESC,
        GQCE_DESC,
        VBP_DESC,
        MANU_COD,
        MANU_DES,
        MNFL_COD,
        MNFL_DES,
        CORP_COD,
        CORP_DES,
        BrandType,
        bu,
        Market_Ratio,
        Key_Competitor,
        no,
        no1
    FROM market_kc
 ),
 market_kc_losers AS (
    -- Legacy cell 7 output (id > 1): every non-winning row, keyed by the 16
    -- columns the legacy MERGE used to delete it.
    SELECT
        ATC1_COD,
        ATC2_COD,
        ATC3_COD,
        ATC4_COD,
        APP1_COD,
        APP2_COD,
        APP3_COD,
        CORP_COD,
        MANU_COD,
        PROD_COD,
        PACK_COD,
        STGH_DES,
        CMPS_COD,
        market,
        Key_Competitor,
        no
    FROM market_kc_ranked
    WHERE rank_id > 1
 ),
 market_kc_kept AS (
    -- Legacy cell 8 MERGE ... WHEN MATCHED THEN DELETE as an anti-join on
    -- (14 keys + Key_Competitor + no). SQL equality semantics are preserved:
    -- a row is deleted only when every join column is non-null and equal, so
    -- KC-less duplicates inside one partition survive exactly as in legacy
    -- (compatibility note 3).
    SELECT
        t1.market,
        t1.PACK_COD,
        t1.PACK_DES,
        t1.STGH_DES,
        t1.PACK_LCH,
        t1.PROD_COD,
        t1.CMPS_COD,
        t1.CMPS_DES,
        t1.ATC1_COD,
        t1.ATC2_COD,
        t1.ATC3_COD,
        t1.ATC4_COD,
        t1.APP1_COD,
        t1.APP2_COD,
        t1.APP3_COD,
        t1.BIO_DESC,
        t1.GENE_ORIG_DESC,
        t1.ETH_OTC_DESC,
        t1.NRDL_DESC,
        t1.NRDL_Entry_Date,
        t1.EDL_DESC,
        t1.TCM_DESC,
        t1.PAED_DESC,
        t1.GQCE_DESC,
        t1.VBP_DESC,
        t1.MANU_COD,
        t1.MANU_DES,
        t1.MNFL_COD,
        t1.MNFL_DES,
        t1.CORP_COD,
        t1.CORP_DES,
        t1.BrandType,
        t1.bu,
        t1.Market_Ratio,
        t1.Key_Competitor
    FROM market_kc t1
    LEFT ANTI JOIN market_kc_losers t2
        ON t1.ATC1_COD = t2.ATC1_COD
        AND t1.ATC2_COD = t2.ATC2_COD
        AND t1.ATC3_COD = t2.ATC3_COD
        AND t1.ATC4_COD = t2.ATC4_COD
        AND t1.APP1_COD = t2.APP1_COD
        AND t1.APP2_COD = t2.APP2_COD
        AND t1.APP3_COD = t2.APP3_COD
        AND t1.CORP_COD = t2.CORP_COD
        AND t1.MANU_COD = t2.MANU_COD
        AND t1.PROD_COD = t2.PROD_COD
        AND t1.PACK_COD = t2.PACK_COD
        AND t1.STGH_DES = t2.STGH_DES
        AND t1.CMPS_COD = t2.CMPS_COD
        AND t1.market = t2.market
        AND t1.Key_Competitor = t2.Key_Competitor
        AND t1.no = t2.no
 )
 -- Legacy cell 9 (Key_Competitor = 'Others' WHERE NULL) is the COALESCE below and
 -- legacy cell 10 (drop no / no1) is handled by the explicit projection.
 SELECT DISTINCT
    market,
    PACK_COD,
    PACK_DES,
    STGH_DES,
    PACK_LCH,
    PROD_COD,
    CMPS_COD,
    CMPS_DES,
    ATC1_COD,
    ATC2_COD,
    ATC3_COD,
    ATC4_COD,
    APP1_COD,
    APP2_COD,
    APP3_COD,
    BIO_DESC,
    GENE_ORIG_DESC,
    ETH_OTC_DESC,
    NRDL_DESC,
    NRDL_Entry_Date,
    EDL_DESC,
    TCM_DESC,
    PAED_DESC,
    GQCE_DESC,
    VBP_DESC,
    MANU_COD,
    MANU_DES,
    MNFL_COD,
    MNFL_DES,
    CORP_COD,
    CORP_DES,
    BrandType,
    bu,
    Market_Ratio,
    COALESCE(Key_Competitor, 'Others') AS Key_Competitor,
    from_utc_timestamp(current_timestamp(), 'UTC+8') AS ETL_INSERT_DT,
    from_utc_timestamp(current_timestamp(), 'UTC+8') AS ETL_UPDATE_DT
 FROM market_kc_kept
 ;
Author	SHA1	Message	Date
chenwu	fd1355a4fb	refactor CHPA 01 and 02 warehouse jobs	2026-08-20 19:39:11 +08:00
chenwu	8c3f45f178	document CHPA warehouse refactor rules	2026-08-20 19:17:19 +08:00